posts/store.go

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()
}