package main import ( "sort" "sync" "time" ) // Store — the in-memory state machine. Hot, container-style, single-writer per pod. // Persistence is handled by Snapshot/Restore (see snapshot.go) and the cold-flush // pipeline (see flush.go). The store does NOT touch the filesystem on the hot path. type Store struct { mu sync.RWMutex // base lock; per-author/per-post locks layered on top (see locks.go) posts map[string]*Post comments map[string]*Comment authors map[string]*Author channels map[string]*Channel postChannels map[string][]string // post_id -> []channel (sorted) mentions map[string][]*Mention readState map[[2]string]time.Time // Indexes — derived from the maps above, maintained on every write. postsByBumpedAt *SortedIndex[*Post] // DESC by bumped_at commentsByPostPath map[string]*SortedIndex[*Comment] // post_id -> sorted by path // Adaptive lock controller (see locks.go). Nil means single-writer mode. lockCtrl *LockController // Cold-flush plumbing (see flush.go). flushCh chan struct{} flushInterval time.Duration flushPath string // Injected for testability. nil defaults are wired in NewStore. now func() time.Time newID func() string } // NewStore — production constructor. Wires defaults, starts the flush goroutine. func NewStore() *Store { return NewStoreWithOptions(DefaultStoreOptions()) } // StoreOptions — knobs for NewStoreWithOptions. Production uses DefaultStoreOptions. // Tests inject custom clock, ID generator, and a no-op flush goroutine. type StoreOptions struct { Now func() time.Time NewID func() string FlushInterval time.Duration FlushPath string StartFlush bool // false in tests, true in main } func DefaultStoreOptions() StoreOptions { return StoreOptions{ Now: func() time.Time { return time.Now().UTC() }, NewID: defaultNewID, FlushInterval: 5 * time.Minute, FlushPath: "/mnt/nas/posts-cold", StartFlush: true, } } func NewStoreWithOptions(opts StoreOptions) *Store { s := &Store{ posts: make(map[string]*Post, 1024), comments: make(map[string]*Comment, 4096), authors: make(map[string]*Author), channels: make(map[string]*Channel), postChannels: make(map[string][]string), mentions: make(map[string][]*Mention), readState: make(map[[2]string]time.Time), postsByBumpedAt: NewSortedIndex[*Post](func(a, b *Post) bool { return a.BumpedAt.After(b.BumpedAt) }), commentsByPostPath: make(map[string]*SortedIndex[*Comment]), flushCh: make(chan struct{}, 1), flushInterval: opts.FlushInterval, flushPath: opts.FlushPath, now: opts.Now, newID: opts.NewID, lockCtrl: NewLockController(), } if opts.StartFlush { go s.flushLoop() } return s } // Now — central time helper. Injected for deterministic tests. func (s *Store) Now() time.Time { return s.now() } // NewID — central ID generator. Injected for deterministic tests. func (s *Store) NewID() string { return s.newID() } // MemUsage — reports approximate live bytes per type. Feeds the VOPR harness (#14) // and the deferred memory-budget decision (posts#1, Q7). type MemUsage struct { Posts int Comments int Authors int Channels int Mentions int ReadState int TotalApprox int } func (s *Store) MemUsage() MemUsage { s.mu.RLock() defer s.mu.RUnlock() return MemUsage{ Posts: len(s.posts) * approxPostSize, Comments: len(s.comments) * approxCommentSize, Authors: len(s.authors) * approxAuthorSize, Channels: len(s.channels) * approxChannelSize, Mentions: sumMentionSize(s.mentions), ReadState: len(s.readState) * 40, // author_id+post_id+time.Time ~40 bytes TotalApprox: 0, } } // Approximate per-object sizes; the VOPR harness will measure precisely. // These are upper bounds, not accounting for interning or sync.Pool reuse. const ( approxPostSize = 512 approxCommentSize = 384 approxAuthorSize = 256 approxChannelSize = 192 approxMentionSize = 256 ) func sumMentionSize(m map[string][]*Mention) int { total := 0 for _, v := range m { total += len(v) * approxMentionSize } return total } // --- Read methods (RLock) --- func (s *Store) GetPost(id string) (*Post, bool) { s.mu.RLock() defer s.mu.RUnlock() p, ok := s.posts[id] return p, ok } func (s *Store) GetAuthor(id string) (*Author, bool) { s.mu.RLock() defer s.mu.RUnlock() a, ok := s.authors[id] return a, ok } func (s *Store) GetChannel(slug string) (*Channel, bool) { s.mu.RLock() defer s.mu.RUnlock() c, ok := s.channels[slug] return c, ok } func (s *Store) ListChannels() []*Channel { s.mu.RLock() defer s.mu.RUnlock() out := make([]*Channel, 0, len(s.channels)) for _, c := range s.channels { out = append(out, c) } sort.Slice(out, func(i, j int) bool { return out[i].Slug < out[j].Slug }) return out } // ListPosts — newest-bumped first, optional channel filter. Hot path. func (s *Store) ListPosts(channel string, limit int) []*Post { if limit <= 0 || limit > 100 { limit = 20 } s.mu.RLock() defer s.mu.RUnlock() all := s.postsByBumpedAt.Items() out := make([]*Post, 0, limit) for _, p := range all { if p.IsDeleted { continue } if channel != "" { chs := s.postChannels[p.ID] if !contains(chs, channel) { continue } } out = append(out, p) if len(out) >= limit { break } } return out } func (s *Store) ListComments(postID string) []*Comment { s.mu.RLock() defer s.mu.RUnlock() idx, ok := s.commentsByPostPath[postID] if !ok { return nil } return idx.Items() } // ListMentions — for an author, all unseen (or seen) mentions. func (s *Store) ListMentions(authorID string, seen bool) []*Mention { s.mu.RLock() defer s.mu.RUnlock() all := s.mentions[authorID] out := make([]*Mention, 0, len(all)) for _, m := range all { if m.Seen == seen { out = append(out, m) } } return out } // --- Write methods (Lock) — direct, used by tests and bootstrap. Production // writes go through the Mutation interface (see mutation.go). --- func (s *Store) UpsertAuthor(a *Author) { s.mu.Lock() defer s.mu.Unlock() a.UpdatedAt = s.Now() if _, exists := s.authors[a.ID]; !exists { a.CreatedAt = a.UpdatedAt } s.authors[a.ID] = a } func (s *Store) UpsertChannel(c *Channel) { s.mu.Lock() defer s.mu.Unlock() if _, exists := s.channels[c.Slug]; !exists { c.CreatedAt = s.Now() } s.channels[c.Slug] = c } func (s *Store) ChannelsForPost(postID string) []string { s.mu.RLock() defer s.mu.RUnlock() chs := s.postChannels[postID] out := make([]string, len(chs)) copy(out, chs) return out } // --- Index maintenance (called under write lock by mutation Apply methods) --- func (s *Store) indexInsertPost(p *Post) { s.postsByBumpedAt.Insert(p) } func (s *Store) indexUpdatePost(p *Post) { s.postsByBumpedAt.Update(p) } func (s *Store) indexDeletePost(p *Post) { s.postsByBumpedAt.Delete(p) } func (s *Store) indexInsertComment(c *Comment) { idx, ok := s.commentsByPostPath[c.PostID] if !ok { idx = NewSortedIndex[*Comment](func(a, b *Comment) bool { return a.Path < b.Path }) s.commentsByPostPath[c.PostID] = idx } idx.Insert(c) } func (s *Store) commentsForPostIndex(postID string) *SortedIndex[*Comment] { return s.commentsByPostPath[postID] } // --- helpers --- func contains(xs []string, x string) bool { for _, v := range xs { if v == x { return true } } return false } // defaultNewID — production ID generator. 8 bytes crypto/rand -> 16 hex chars. func defaultNewID() string { return NewID() }