From 6e3314a368797b057ac09db467a22e8f78fd9f08 Mon Sep 17 00:00:00 2001 From: Hermes Date: Tue, 23 Jun 2026 01:31:04 -0700 Subject: [PATCH] feat(snapshot): Snapshot/Restore with gob encoding, .snap files on disk --- snapshot.go | 246 ++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 246 insertions(+) create mode 100644 snapshot.go diff --git a/snapshot.go b/snapshot.go new file mode 100644 index 0000000..6bb2998 --- /dev/null +++ b/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 ///-.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 +}