thread-master/apps/backend/internal/module/scout/repository/memory.go

624 lines
16 KiB
Go

package repository
import (
"context"
"sort"
"sync"
"apps/backend/internal/module/scout/domain"
)
type MemoryStore struct {
mu sync.Mutex
brands map[string]*domain.Brand
products map[string]*domain.Product
active map[int64]string
posts map[string]*domain.Post
runs map[string]*domain.Run
seen map[string]struct{}
hw map[string]*domain.Homework // key owner|theme
crawler map[int64]*domain.CrawlerSession
}
func NewMemory() *MemoryStore {
return &MemoryStore{
brands: map[string]*domain.Brand{}, products: map[string]*domain.Product{},
active: map[int64]string{}, posts: map[string]*domain.Post{},
runs: map[string]*domain.Run{}, seen: map[string]struct{}{},
hw: map[string]*domain.Homework{}, crawler: map[int64]*domain.CrawlerSession{},
}
}
func cloneRun(r *domain.Run) *domain.Run {
if r == nil {
return nil
}
cp := *r
cp.ShortfallReasons = append([]string(nil), r.ShortfallReasons...)
return &cp
}
func clonePost(p *domain.Post) *domain.Post {
if p == nil {
return nil
}
cp := *p
return &cp
}
func compareRuns(a, b *domain.Run) bool {
if a.CreatedAt != b.CreatedAt {
return a.CreatedAt > b.CreatedAt
}
return a.ID > b.ID
}
func comparePosts(a, b *domain.Post) bool {
if a.Score != b.Score {
return a.Score > b.Score
}
aKnown := a.PostedAt > 0
bKnown := b.PostedAt > 0
if aKnown != bKnown {
return aKnown
}
if aKnown && a.PostedAt != b.PostedAt {
return a.PostedAt > b.PostedAt
}
if a.CreatedAt != b.CreatedAt {
return a.CreatedAt > b.CreatedAt
}
return a.ID > b.ID
}
func postIdentity(p *domain.Post) string {
if p.ExternalID != "" {
return "external:" + p.ExternalID
}
if p.Permalink != "" {
return "url:" + p.Permalink
}
return "id:" + p.ID
}
func uniquePosts(items []*domain.Post) []*domain.Post {
sort.Slice(items, func(i, j int) bool { return comparePosts(items[i], items[j]) })
seen := make(map[string]struct{}, len(items))
out := make([]*domain.Post, 0, len(items))
for _, p := range items {
identity := postIdentity(p)
if _, exists := seen[identity]; exists {
continue
}
seen[identity] = struct{}{}
out = append(out, p)
}
return out
}
func legacyRunFromPosts(ownerUID int64, themeKey string, posts []*domain.Post) *domain.Run {
var latest int64
var label string
var brandID, mode string
for _, p := range posts {
if p.CreatedAt > latest {
latest = p.CreatedAt
label = p.ThemeLabel
brandID = p.BrandID
mode = p.ScoutMode
}
}
if latest == 0 {
latest = domain.NowNano()
}
return &domain.Run{
ID: domain.LegacyRunID(themeKey), OwnerUID: ownerUID, JobID: "legacy",
ThemeKey: themeKey, ThemeLabel: label, Intent: label, Mode: mode,
BrandID: brandID, TargetCount: len(posts), Status: domain.RunSucceeded,
SearchedCount: len(posts), EligibleCount: len(posts), PendingCount: countPending(posts),
CreatedAt: latest, CompletedAt: latest,
}
}
func countPending(posts []*domain.Post) int {
n := 0
for _, p := range posts {
if p.OutreachStatus == domain.OutreachNew || p.OutreachStatus == domain.OutreachDrafted {
n++
}
}
return n
}
func identityKey(uid int64, identity string) string {
return formatUID(uid) + "|" + identity
}
func key(uid int64, theme string) string {
return formatUID(uid) + "|" + theme
}
func formatUID(uid int64) string {
if uid == 0 {
return "0"
}
neg := uid < 0
if neg {
uid = -uid
}
var b [32]byte
i := len(b)
for uid > 0 {
i--
b[i] = byte('0' + uid%10)
uid /= 10
}
if neg {
i--
b[i] = '-'
}
return string(b[i:])
}
func (s *MemoryStore) SaveBrand(_ context.Context, b *domain.Brand) error {
s.mu.Lock()
defer s.mu.Unlock()
cp := *b
s.brands[b.ID] = &cp
return nil
}
func (s *MemoryStore) GetBrand(_ context.Context, id string) (*domain.Brand, error) {
s.mu.Lock()
defer s.mu.Unlock()
b, ok := s.brands[id]
if !ok {
return nil, domain.ErrNotFound
}
cp := *b
return &cp, nil
}
func (s *MemoryStore) ListBrands(_ context.Context, ownerUID int64) ([]*domain.Brand, error) {
s.mu.Lock()
defer s.mu.Unlock()
var out []*domain.Brand
for _, b := range s.brands {
if b.OwnerUID == ownerUID {
cp := *b
out = append(out, &cp)
}
}
return out, nil
}
func (s *MemoryStore) DeleteBrand(_ context.Context, id string) error {
s.mu.Lock()
defer s.mu.Unlock()
if _, ok := s.brands[id]; !ok {
return domain.ErrNotFound
}
delete(s.brands, id)
return nil
}
func (s *MemoryStore) SaveProduct(_ context.Context, p *domain.Product) error {
s.mu.Lock()
defer s.mu.Unlock()
cp := *p
if p.MatchTags != nil {
cp.MatchTags = append([]string(nil), p.MatchTags...)
}
if p.PainPoints != nil {
cp.PainPoints = append([]string(nil), p.PainPoints...)
}
if p.ProviderCapabilityTerms != nil {
cp.ProviderCapabilityTerms = append([]string(nil), p.ProviderCapabilityTerms...)
}
if p.ProviderExcludeTerms != nil {
cp.ProviderExcludeTerms = append([]string(nil), p.ProviderExcludeTerms...)
}
s.products[p.ID] = &cp
return nil
}
func (s *MemoryStore) GetProduct(_ context.Context, id string) (*domain.Product, error) {
s.mu.Lock()
defer s.mu.Unlock()
p, ok := s.products[id]
if !ok {
return nil, domain.ErrNotFound
}
return copyProduct(p), nil
}
func (s *MemoryStore) ListProducts(_ context.Context, ownerUID int64, brandID string) ([]*domain.Product, error) {
s.mu.Lock()
defer s.mu.Unlock()
var out []*domain.Product
for _, p := range s.products {
if p.OwnerUID != ownerUID {
continue
}
if brandID != "" && p.BrandID != brandID {
continue
}
out = append(out, copyProduct(p))
}
return out, nil
}
func (s *MemoryStore) DeleteProduct(_ context.Context, id string) error {
s.mu.Lock()
defer s.mu.Unlock()
if _, ok := s.products[id]; !ok {
return domain.ErrNotFound
}
delete(s.products, id)
return nil
}
func copyProduct(p *domain.Product) *domain.Product {
cp := *p
if p.MatchTags != nil {
cp.MatchTags = append([]string(nil), p.MatchTags...)
}
if p.PainPoints != nil {
cp.PainPoints = append([]string(nil), p.PainPoints...)
}
if p.ProviderCapabilityTerms != nil {
cp.ProviderCapabilityTerms = append([]string(nil), p.ProviderCapabilityTerms...)
}
if p.ProviderExcludeTerms != nil {
cp.ProviderExcludeTerms = append([]string(nil), p.ProviderExcludeTerms...)
}
return &cp
}
func (s *MemoryStore) GetActiveBrandID(_ context.Context, ownerUID int64) (string, error) {
s.mu.Lock()
defer s.mu.Unlock()
return s.active[ownerUID], nil
}
func (s *MemoryStore) SetActiveBrandID(_ context.Context, ownerUID int64, brandID string) error {
s.mu.Lock()
defer s.mu.Unlock()
s.active[ownerUID] = brandID
return nil
}
func (s *MemoryStore) SavePost(_ context.Context, p *domain.Post) error {
s.mu.Lock()
defer s.mu.Unlock()
cp := *p
s.posts[p.ID] = &cp
return nil
}
func (s *MemoryStore) GetPost(_ context.Context, id string) (*domain.Post, error) {
s.mu.Lock()
defer s.mu.Unlock()
p, ok := s.posts[id]
if !ok {
return nil, domain.ErrNotFound
}
cp := *p
return &cp, nil
}
func (s *MemoryStore) ListPosts(_ context.Context, ownerUID int64, brandID string) ([]*domain.Post, error) {
s.mu.Lock()
defer s.mu.Unlock()
var out []*domain.Post
for _, p := range s.posts {
if p.OwnerUID != ownerUID {
continue
}
if brandID != "" && p.BrandID != brandID {
continue
}
cp := *p
out = append(out, &cp)
}
sort.Slice(out, func(i, j int) bool { return comparePosts(out[i], out[j]) })
return out, nil
}
func (s *MemoryStore) DeletePost(_ context.Context, id string) error {
s.mu.Lock()
defer s.mu.Unlock()
if _, ok := s.posts[id]; !ok {
return domain.ErrNotFound
}
delete(s.posts, id)
return nil
}
func (s *MemoryStore) DeletePostsByTheme(_ context.Context, ownerUID int64, themeKey string) error {
s.mu.Lock()
defer s.mu.Unlock()
for id, p := range s.posts {
if p.OwnerUID == ownerUID && p.ThemeKey == themeKey {
delete(s.posts, id)
}
}
return nil
}
func (s *MemoryStore) SaveHomework(_ context.Context, h *domain.Homework) error {
s.mu.Lock()
defer s.mu.Unlock()
cp := *h
s.hw[key(h.OwnerUID, h.ThemeKey)] = &cp
return nil
}
func (s *MemoryStore) GetHomework(_ context.Context, ownerUID int64, themeKey string) (*domain.Homework, error) {
s.mu.Lock()
defer s.mu.Unlock()
h, ok := s.hw[key(ownerUID, themeKey)]
if !ok {
return nil, domain.ErrNotFound
}
cp := *h
return &cp, nil
}
func (s *MemoryStore) ListHomework(_ context.Context, ownerUID int64) ([]*domain.Homework, error) {
s.mu.Lock()
defer s.mu.Unlock()
var out []*domain.Homework
for _, h := range s.hw {
if h.OwnerUID == ownerUID {
cp := *h
out = append(out, &cp)
}
}
return out, nil
}
func (s *MemoryStore) DeleteHomework(_ context.Context, ownerUID int64, themeKey string) error {
s.mu.Lock()
defer s.mu.Unlock()
k := key(ownerUID, themeKey)
if _, ok := s.hw[k]; !ok {
return domain.ErrNotFound
}
delete(s.hw, k)
return nil
}
func (s *MemoryStore) GetCrawlerSession(_ context.Context, ownerUID int64) (*domain.CrawlerSession, error) {
s.mu.Lock()
defer s.mu.Unlock()
c, ok := s.crawler[ownerUID]
if !ok || c.StorageStateEnc == "" {
return nil, domain.ErrNotFound
}
cp := *c
return &cp, nil
}
func (s *MemoryStore) SetCrawlerSession(_ context.Context, sess *domain.CrawlerSession) error {
s.mu.Lock()
defer s.mu.Unlock()
cp := *sess
s.crawler[sess.OwnerUID] = &cp
return nil
}
func (s *MemoryStore) ClearCrawlerSession(_ context.Context, ownerUID int64) error {
s.mu.Lock()
defer s.mu.Unlock()
delete(s.crawler, ownerUID)
return nil
}
func (s *MemoryStore) CreateRun(_ context.Context, r *domain.Run) error {
if r == nil || r.ID == "" || r.OwnerUID == 0 {
return domain.ErrValidation
}
s.mu.Lock()
defer s.mu.Unlock()
if _, exists := s.runs[r.ID]; exists {
return domain.ErrValidation
}
s.runs[r.ID] = cloneRun(r)
return nil
}
func (s *MemoryStore) GetRun(_ context.Context, ownerUID int64, id string) (*domain.Run, error) {
s.mu.Lock()
defer s.mu.Unlock()
r, ok := s.runs[id]
if ok && r.OwnerUID == ownerUID {
return cloneRun(r), nil
}
if themeKey, legacy := domain.LegacyThemeKey(id); legacy {
items := make([]*domain.Post, 0)
for _, p := range s.posts {
if p.OwnerUID == ownerUID && p.RunID == "" && p.ThemeKey == themeKey {
items = append(items, clonePost(p))
}
}
items = uniquePosts(items)
if len(items) > 0 {
return legacyRunFromPosts(ownerUID, themeKey, items), nil
}
}
return nil, domain.ErrNotFound
}
func (s *MemoryStore) ListRuns(_ context.Context, filter domain.RunFilter) (domain.RunPage, error) {
s.mu.Lock()
defer s.mu.Unlock()
page := domain.NormalizePage(filter.Page, filter.PageSize)
items := make([]*domain.Run, 0)
for _, r := range s.runs {
if r.OwnerUID != filter.OwnerUID || (filter.BrandID != "" && r.BrandID != filter.BrandID) || (filter.Mode != "" && r.Mode != filter.Mode) {
continue
}
items = append(items, cloneRun(r))
}
legacyGroups := map[string][]*domain.Post{}
for _, p := range s.posts {
if p.OwnerUID != filter.OwnerUID || p.RunID != "" {
continue
}
if filter.BrandID != "" && p.BrandID != filter.BrandID {
continue
}
if filter.Mode != "" && p.ScoutMode != filter.Mode {
continue
}
legacyGroups[p.ThemeKey] = append(legacyGroups[p.ThemeKey], clonePost(p))
}
for themeKey, posts := range legacyGroups {
items = append(items, legacyRunFromPosts(filter.OwnerUID, themeKey, uniquePosts(posts)))
}
sort.Slice(items, func(i, j int) bool { return compareRuns(items[i], items[j]) })
total := int64(len(items))
page = page.WithTotal(total)
start := (page.Page - 1) * page.PageSize
if start >= len(items) {
return domain.RunPage{Items: []*domain.Run{}, Pagination: page}, nil
}
end := start + page.PageSize
if end > len(items) {
end = len(items)
}
return domain.RunPage{Items: items[start:end], Pagination: page}, nil
}
func (s *MemoryStore) ListRunPosts(_ context.Context, ownerUID int64, runID string, requestedPage, requestedSize int) (domain.RunPostPage, error) {
s.mu.Lock()
defer s.mu.Unlock()
r, ok := s.runs[runID]
legacyTheme, isLegacy := domain.LegacyThemeKey(runID)
if isLegacy {
items := make([]*domain.Post, 0)
for _, p := range s.posts {
if p.OwnerUID == ownerUID && p.RunID == "" && p.ThemeKey == legacyTheme {
items = append(items, clonePost(p))
}
}
items = uniquePosts(items)
if len(items) == 0 {
return domain.RunPostPage{}, domain.ErrNotFound
}
r = legacyRunFromPosts(ownerUID, legacyTheme, items)
ok = true
}
if !ok || r.OwnerUID != ownerUID {
return domain.RunPostPage{}, domain.ErrNotFound
}
page := domain.NormalizePage(requestedPage, requestedSize)
if r.Status != domain.RunSucceeded {
return domain.RunPostPage{Run: cloneRun(r), Items: []*domain.Post{}, Pagination: page.WithTotal(0)}, nil
}
items := make([]*domain.Post, 0)
for _, p := range s.posts {
if p.OwnerUID != ownerUID {
continue
}
if (isLegacy && p.RunID == "" && p.ThemeKey == legacyTheme) || (!isLegacy && p.RunID == runID) {
items = append(items, clonePost(p))
}
}
items = uniquePosts(items)
page = page.WithTotal(int64(len(items)))
start := (page.Page - 1) * page.PageSize
if start >= len(items) {
return domain.RunPostPage{Run: cloneRun(r), Items: []*domain.Post{}, Pagination: page}, nil
}
end := start + page.PageSize
if end > len(items) {
end = len(items)
}
return domain.RunPostPage{Run: cloneRun(r), Items: items[start:end], Pagination: page}, nil
}
func (s *MemoryStore) ReplaceRunGuarded(_ context.Context, ownerUID int64, id string, expected []string, replacement *domain.Run) error {
s.mu.Lock()
defer s.mu.Unlock()
current, ok := s.runs[id]
if !ok || current.OwnerUID != ownerUID {
return domain.ErrNotFound
}
if len(expected) > 0 {
allowed := false
for _, status := range expected {
if current.Status == status {
allowed = true
break
}
}
if !allowed {
return domain.ErrIllegalRunStatus
}
}
if replacement == nil || replacement.ID != id {
return domain.ErrValidation
}
if replacement.Status != current.Status && !domain.CanTransitionRun(current.Status, replacement.Status) {
return domain.ErrIllegalRunStatus
}
cp := cloneRun(replacement)
cp.OwnerUID = ownerUID
s.runs[id] = cp
return nil
}
func (s *MemoryStore) DeleteRun(_ context.Context, ownerUID int64, id string) error {
s.mu.Lock()
defer s.mu.Unlock()
r, ok := s.runs[id]
if !ok {
if themeKey, legacy := domain.LegacyThemeKey(id); legacy {
found := false
for postID, p := range s.posts {
if p.OwnerUID == ownerUID && p.RunID == "" && p.ThemeKey == themeKey {
delete(s.posts, postID)
found = true
}
}
if found {
return nil
}
}
}
if !ok || r.OwnerUID != ownerUID {
return domain.ErrNotFound
}
delete(s.runs, id)
for postID, p := range s.posts {
if p.OwnerUID == ownerUID && p.RunID == id {
delete(s.posts, postID)
}
}
return nil
}
func (s *MemoryStore) PublishRunPosts(_ context.Context, ownerUID int64, runID string, posts []*domain.Post) error {
s.mu.Lock()
defer s.mu.Unlock()
r, ok := s.runs[runID]
if !ok || r.OwnerUID != ownerUID {
return domain.ErrNotFound
}
for _, p := range posts {
if p == nil || p.ID == "" || p.OwnerUID != ownerUID || p.RunID != runID {
return domain.ErrValidation
}
}
for _, p := range posts {
s.posts[p.ID] = clonePost(p)
}
return nil
}
func (s *MemoryStore) HasSeenIdentity(_ context.Context, ownerUID int64, identity string) (bool, error) {
s.mu.Lock()
defer s.mu.Unlock()
if identity == "" {
return false, domain.ErrValidation
}
_, ok := s.seen[identityKey(ownerUID, identity)]
return ok, nil
}
func (s *MemoryStore) MarkSeenIdentity(_ context.Context, ownerUID int64, identity, _ string, _ int64) error {
s.mu.Lock()
defer s.mu.Unlock()
if identity == "" {
return domain.ErrValidation
}
s.seen[identityKey(ownerUID, identity)] = struct{}{}
return nil
}
var _ domain.Repository = (*MemoryStore)(nil)