From 58a4e1db8fc83030e6eeb2d676afc72d33bd3925 Mon Sep 17 00:00:00 2001 From: Hermes Date: Tue, 23 Jun 2026 01:31:00 -0700 Subject: [PATCH] feat(store): in-memory state machine, indexes, MemUsage --- store.go | 296 +++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 296 insertions(+) create mode 100644 store.go diff --git a/store.go b/store.go new file mode 100644 index 0000000..40cdc52 --- /dev/null +++ b/store.go @@ -0,0 +1,296 @@ +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() +}