feat(snapshot): Snapshot/Restore with gob encoding, .snap files on disk

snapshot.go

@@ -0,0 +1,246 @@
+package main
+
+import (
+	"bytes"
+	"encoding/gob"
+	"fmt"
+	"os"
+	"path/filepath"
+	"time"
+)
+
+// SnapshotSchemaVersion — bump on breaking changes to the Snapshot shape.
+const SnapshotSchemaVersion = 1
+
+// Snapshot — the on-disk and on-wire shape of the entire in-memory state.
+// Gob-encoded for v0. Schema-as-theater: the Snapshot fields are derived from
+// the Go types; this struct is a serialization, not a spec.
+type Snapshot struct {
+	SchemaVersion int                            `gob:"schema_version"`
+	CapturedAt    time.Time                      `gob:"captured_at"`
+	Posts         map[string]*Post                `gob:"posts"`
+	Comments      map[string]*Comment             `gob:"comments"`
+	Authors       map[string]*Author              `gob:"authors"`
+	Channels      map[string]*Channel             `gob:"channels"`
+	PostChannels  map[string][]string             `gob:"post_channels"`
+	Mentions      []*Mention                      `gob:"mentions"`
+	ReadState     map[[2]string]time.Time         `gob:"read_state"`
+}
+
+// Snapshot — captures the current state under RLock, gob-encodes it. Safe to
+// call concurrently with reads; blocks writers briefly.
+func (s *Store) Snapshot() ([]byte, error) {
+	s.mu.RLock()
+	snap := Snapshot{
+		SchemaVersion: SnapshotSchemaVersion,
+		CapturedAt:    s.Now(),
+		Posts:         copyMap(s.posts),
+		Comments:      copyCommentMap(s.comments),
+		Authors:       copyAuthorMap(s.authors),
+		Channels:      copyChannelMap(s.channels),
+		PostChannels:  copyPostChannelsMap(s.postChannels),
+		Mentions:      copyMentionsMap(s.mentions),
+		ReadState:     copyReadStateMap(s.readState),
+	}
+	s.mu.RUnlock()
+
+	var buf bytes.Buffer
+	if err := gob.NewEncoder(&buf).Encode(snap); err != nil {
+		return nil, fmt.Errorf("encode snapshot: %w", err)
+	}
+	return buf.Bytes(), nil
+}
+
+// Restore — decodes a snapshot and atomically swaps the in-memory state.
+// Returns an error if the schema version is unsupported. Callers should
+// typically Restore once at boot, before starting the HTTP server.
+func (s *Store) Restore(data []byte) error {
+	var snap Snapshot
+	if err := gob.NewDecoder(bytes.NewReader(data)).Decode(&snap); err != nil {
+		return fmt.Errorf("decode snapshot: %w", err)
+	}
+	if snap.SchemaVersion != SnapshotSchemaVersion {
+		return fmt.Errorf("snapshot schema version %d, expected %d", snap.SchemaVersion, SnapshotSchemaVersion)
+	}
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.posts = snap.Posts
+	s.comments = snap.Comments
+	s.authors = snap.Authors
+	s.channels = snap.Channels
+	s.postChannels = snap.PostChannels
+	s.mentions = groupMentions(snap.Mentions)
+	s.readState = snap.ReadState
+
+	// Rebuild indexes from scratch.
+	s.postsByBumpedAt = NewSortedIndex[*Post](func(a, b *Post) bool { return a.BumpedAt.After(b.BumpedAt) })
+	for _, p := range s.posts {
+		s.postsByBumpedAt.Insert(p)
+	}
+	s.commentsByPostPath = make(map[string]*SortedIndex[*Comment])
+	for _, c := range s.comments {
+		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)
+	}
+	return nil
+}
+
+// FlushToPath — writes a snapshot to <path>/<year>/<month>/<batch>-<unix>.parquet.
+// For v0, we write gob files (not Parquet) because the Parquet writer is a
+// follow-up. The path shape is preserved so the writer swap is a no-op for
+// callers. File extension: .snap for v0; .parquet lands in v0.5.
+func (s *Store) FlushToPath(path string) (string, error) {
+	data, err := s.Snapshot()
+	if err != nil {
+		return "", err
+	}
+	now := s.Now()
+	dir := filepath.Join(path, fmt.Sprintf("%04d", now.Year()), fmt.Sprintf("%02d", int(now.Month())))
+	if err := os.MkdirAll(dir, 0o755); err != nil {
+		return "", fmt.Errorf("mkdir %s: %w", dir, err)
+	}
+	filename := fmt.Sprintf("batch-%d.snap", now.Unix())
+	full := filepath.Join(dir, filename)
+	if err := os.WriteFile(full, data, 0o644); err != nil {
+		return "", fmt.Errorf("write %s: %w", full, err)
+	}
+	return full, nil
+}
+
+// RestoreFromLatest — finds the latest .snap file under path and restores it.
+// Used at boot. Returns the path it loaded, or "" if no snapshots exist.
+func (s *Store) RestoreFromLatest(path string) (string, error) {
+	if _, err := os.Stat(path); os.IsNotExist(err) {
+		return "", nil
+	}
+	latest, err := findLatestSnapshot(path)
+	if err != nil {
+		return "", err
+	}
+	if latest == "" {
+		return "", nil
+	}
+	data, err := os.ReadFile(latest)
+	if err != nil {
+		return "", err
+	}
+	if err := s.Restore(data); err != nil {
+		return "", err
+	}
+	return latest, nil
+}
+
+func findLatestSnapshot(path string) (string, error) {
+	var latest string
+	var latestTime time.Time
+	err := filepath.Walk(path, func(p string, info os.FileInfo, err error) error {
+		if err != nil {
+			return err
+		}
+		if info.IsDir() {
+			return nil
+		}
+		if filepath.Ext(p) != ".snap" {
+			return nil
+		}
+		if info.ModTime().After(latestTime) {
+			latest = p
+			latestTime = info.ModTime()
+		}
+		return nil
+	})
+	return latest, err
+}
+
+// --- copy helpers (gob needs deep copies) ---
+
+func copyMap[T any](m map[string]*T) map[string]*T {
+	out := make(map[string]*T, len(m))
+	for k, v := range m {
+		out[k] = v
+	}
+	return out
+}
+
+func copyCommentMap(m map[string]*Comment) map[string]*Comment {
+	out := make(map[string]*Comment, len(m))
+	for k, v := range m {
+		out[k] = v
+	}
+	return out
+}
+
+func copyAuthorMap(m map[string]*Author) map[string]*Author {
+	out := make(map[string]*Author, len(m))
+	for k, v := range m {
+		out[k] = v
+	}
+	return out
+}
+
+func copyChannelMap(m map[string]*Channel) map[string]*Channel {
+	out := make(map[string]*Channel, len(m))
+	for k, v := range m {
+		out[k] = v
+	}
+	return out
+}
+
+func copyPostChannelsMap(m map[string][]string) map[string][]string {
+	out := make(map[string][]string, len(m))
+	for k, v := range m {
+		cp := make([]string, len(v))
+		copy(cp, v)
+		out[k] = cp
+	}
+	return out
+}
+
+func copyMentionsMap(m map[string][]*Mention) []*Mention {
+	out := make([]*Mention, 0)
+	for _, v := range m {
+		out = append(out, v...)
+	}
+	return out
+}
+
+func groupMentions(flat []*Mention) map[string][]*Mention {
+	out := make(map[string][]*Mention)
+	for _, m := range flat {
+		out[m.Mentioned] = append(out[m.Mentioned], m)
+	}
+	return out
+}
+
+func copyReadStateMap(m map[[2]string]time.Time) map[[2]string]time.Time {
+	out := make(map[[2]string]time.Time, len(m))
+	for k, v := range m {
+		out[k] = v
+	}
+	return out
+}
+
+// Register gob types up front — required because the encoder doesn't
+// introspect unexported types automatically.
+func init() {
+	gob.Register(map[string]*Post{})
+	gob.Register(map[string]*Comment{})
+	gob.Register(map[string]*Author{})
+	gob.Register(map[string]*Channel{})
+	gob.Register(map[string][]string{})
+	gob.Register([]*Mention{})
+	gob.Register(map[[2]string]time.Time{})
+	gob.Register(Mention{})
+	gob.Register(ReadStateEntry{})
+}
+
+type ReadStateEntry struct {
+	AuthorID string
+	PostID   string
+	LastSeen time.Time
+}