feat(notes): compact the operation log and purge old tombstones

- Background job at start and every 6 h: log entries older than 30 days
  (private notes: 7) are removed, documents.log_floor marks the boundary;
  clients behind it get resync and keep their queued changes
- Deleted blocks are purged 30 days after deleted_at via a "purge" log
  effect, so open editors drop them without reloading; a tombstone stays
  while a deleted child still points to it
- Text for a purged block becomes a new block at the end; inserts under a
  purged parent go to the top level (server and browser model)
- doc_op_results kept for 90 days; expired lease/session maps are swept
- Tests: storage, API, engine resync, simulation with compactions

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

internal/api/docsync.go

@@ -2,6 +2,8 @@ package api
 
 import (
 	"encoding/json"
+	"errors"
+	"log/slog"
 	"net/http"
 	"strconv"
 	"time"
@@ -31,7 +33,7 @@ type syncResponse struct {
 	Ops      []storage.DocLogEntry `json:"ops"`
 	HeadSeq  int64                 `json:"headSeq"`
 	HasMore  bool                  `json:"hasMore"`
-	Resync   bool                  `json:"resync,omitempty"` // client state is ahead of the server: reload the snapshot
+	Resync   bool                  `json:"resync,omitempty"` // reload the snapshot (state ahead of the server or log compacted)
 	Role     string                `json:"role"`
 	Shared   bool                  `json:"shared"`
 	Sessions int                   `json:"sessions"`
@@ -155,11 +157,13 @@ func (s *Server) handleDocSync(w http.ResponseWriter, r *http.Request) {
 		logLimit = 0
 	}
 	entries, head, err := s.store.GetDocOpsSince(docID, req.LastSeq, logLimit)
-	if err != nil {
+	compacted := errors.Is(err, storage.ErrLogCompacted)
+	if err != nil && !compacted {
 		writeError(w, http.StatusInternalServerError, err.Error())
 		return
 	}
-	if req.LastSeq > head && !req.SkipLog {
+	// Client state is ahead of the server, or so far behind that the log entries are compacted away
+	if (req.LastSeq > head || compacted) && !req.SkipLog {
 		resp.Resync = true
 		entries = []storage.DocLogEntry{}
 	}
@@ -253,3 +257,28 @@ func (s *Server) handleDocLeave(w http.ResponseWriter, r *http.Request) {
 	s.notifyDocChanged(docID, head)
 	writeJSON(w, http.StatusOK, map[string]bool{"success": true})
 }
+
+// compactionInterval is how often the operation log and tombstones are compacted (also once at start).
+const compactionInterval = 6 * time.Hour
+
+func (s *Server) startCompaction() {
+	for {
+		s.compactOnce()
+		time.Sleep(compactionInterval)
+	}
+}
+
+func (s *Server) compactOnce() {
+	s.leases.Sweep()
+	st, err := s.store.CompactDocs(time.Now())
+	if err != nil {
+		slog.Error("Verdichtung des Änderungsprotokolls fehlgeschlagen", "error", err)
+	}
+	// Open editors drop the purged tombstones with their next sync
+	for docID, head := range st.Purged {
+		s.notifyDocChanged(docID, head)
+	}
+	if st.Docs > 0 || st.OpResults > 0 {
+		slog.Info("Änderungsprotokoll verdichtet", "notes", st.Docs, "logEntries", st.LogEntries, "tombstones", st.Tombstones, "opResults", st.OpResults)
+	}
+}

internal/api/docsync_test.go

@@ -196,3 +196,36 @@ func itoa(n int64) string {
 	b, _ := json.Marshal(n)
 	return string(b)
 }
+
+func TestDocSyncAfterCompaction(t *testing.T) {
+	srv, store, ts, owner, ownerTok := startRealtimeServer(t)
+	doc, err := store.CreateDocument(owner.ID, "", "Alt", "📄", "", []domain.DocumentBlock{
+		{ID: "b1", Type: "bullet", Content: "Bleibt"}, {ID: "b2", Type: "bullet", Content: "Weg"},
+	}, nil, false)
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, live := syncDoc(t, ts, ownerTok, doc.ID, map[string]interface{}{
+		"clientId": "live", "lastSeq": 0, "ops": []collab.Op{{OpID: "d", Kind: collab.OpDelete, BlockID: "b2"}},
+	})
+
+	// 40 days later: log and tombstone are past their retention
+	st, err := srv.store.CompactDocs(time.Now().Add(40 * 24 * time.Hour))
+	if err != nil || st.Tombstones != 1 {
+		t.Fatalf("compaction: %+v %v", st, err)
+	}
+
+	// A client that was up to date only gets the purge entry
+	_, up := syncDoc(t, ts, ownerTok, doc.ID, map[string]interface{}{"clientId": "live", "lastSeq": live.HeadSeq})
+	if up.Resync || len(up.Ops) != 1 || up.Ops[0].Effect.Kind != collab.OpPurge {
+		t.Fatalf("up-to-date client: resync=%v ops=%+v", up.Resync, up.Ops)
+	}
+	// A client behind the compacted log must reload; its own changes are still applied
+	code, old := syncDoc(t, ts, ownerTok, doc.ID, map[string]interface{}{
+		"clientId": "old", "lastSeq": 1,
+		"ops": []collab.Op{{OpID: "late", Kind: collab.OpSetContent, BlockID: "b1", BaseVersion: 1, Content: strp("Spät")}},
+	})
+	if code != http.StatusOK || !old.Resync || len(old.Ops) != 0 || old.Results[0].Status != collab.StatusApplied {
+		t.Fatalf("stale client: %d resync=%v results=%+v", code, old.Resync, old.Results)
+	}
+}

internal/api/server.go

@@ -46,6 +46,7 @@ func NewServer(cfg *config.Config, store *storage.Store, wsHub *ws.Hub, staticFS
 	}
 	wsHub.SetRoomAuthorizer(srv.canJoinRoom)
 	go srv.startAlarmTicker()
+	go srv.startCompaction()
 	return srv
 }
 

internal/collab/lease.go

@@ -195,3 +195,23 @@ func (s leaseSnapshot) HolderOf(blockID string) (string, bool) {
 	c, ok := s[blockID]
 	return c, ok
 }
+
+// Sweep drops expired leases and sessions of all documents and forgets documents nobody uses any more,
+// so the maps do not grow with every note ever opened.
+func (m *LeaseManager) Sweep() {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	now := m.now()
+	for docID := range m.leases {
+		m.prune(docID, now)
+		if len(m.leases[docID]) == 0 {
+			delete(m.leases, docID)
+		}
+	}
+	for docID := range m.sessions {
+		m.prune(docID, now)
+		if len(m.sessions[docID]) == 0 {
+			delete(m.sessions, docID)
+		}
+	}
+}

internal/collab/ops.go

@@ -14,6 +14,8 @@ const (
 	OpDelete     = "delete"
 	OpRestore    = "restore"
 	OpSetTitle   = "set_title"
+	// OpPurge is only produced by log compaction: deleted blocks older than the retention are removed for good
+	OpPurge = "purge"
 
 	StatusApplied  = "applied"
 	StatusConflict = "conflict"
@@ -131,6 +133,10 @@ func (d *Doc) ApplyEffect(e Effect) {
 		}
 	case OpSetTitle:
 		d.Title, d.TitleVersion = e.Title, e.TitleVersion
+	case OpPurge:
+		for _, id := range e.IDs {
+			delete(d.Blocks, id)
+		}
 	}
 }
 
@@ -189,10 +195,11 @@ func (ap *applier) apply(op Op) Result {
 			parent = *op.ParentID
 		}
 		if parent != "" {
-			if _, ok := d.Blocks[parent]; !ok {
-				return rejected(op, "parent_missing")
+			if _, ok := d.Blocks[parent]; ok {
+				parent = d.liveAncestor(parent)
+			} else {
+				parent = "" // parent was purged after the retention period: keep the block at the top level
 			}
-			parent = d.liveAncestor(parent)
 		}
 		if !ValidOrderKey(op.OrderKey) {
 			return rejected(op, "invalid_order_key")
@@ -221,7 +228,7 @@ func (ap *applier) apply(op Op) Result {
 	case OpSetContent:
 		b, ok := d.Blocks[op.BlockID]
 		if !ok {
-			return rejected(op, "missing")
+			return ap.reviveAsNewBlock(op)
 		}
 		if op.Content == nil {
 			return rejected(op, "no_content")
@@ -372,6 +379,61 @@ func (ap *applier) restore(ids []string) {
 	}
 }
 
+// reviveAsNewBlock handles text for a block that no longer exists (its tombstone was purged after the
+// retention period, e.g. a device was offline for weeks): the text becomes a new block at the end.
+func (ap *applier) reviveAsNewBlock(op Op) Result {
+	if op.Content == nil || *op.Content == "" || op.BlockID == "" || len(op.BlockID) > 64 {
+		return rejected(op, "missing")
+	}
+	if ap.env.IDTaken != nil && ap.env.IDTaken(op.BlockID) {
+		return rejected(op, "missing")
+	}
+	if len(*op.Content) > maxContentLength {
+		return rejected(op, "too_long")
+	}
+	last := ""
+	if roots := ap.d.Children(""); len(roots) > 0 {
+		last = roots[len(roots)-1].OrderKey
+	}
+	key, err := KeyBetween(last, "")
+	if err != nil {
+		return rejected(op, "missing")
+	}
+	ap.emit(Effect{Kind: OpInsert, BlockID: op.BlockID, OrderKey: key, Type: "bullet", Content: *op.Content, Props: &Props{}, Version: 1})
+	return Result{Status: StatusApplied, Version: 1, Reason: "revived"}
+}
+
+// Purge removes tombstones for good. Only blocks whose whole deleted subtree is purged too may go, so no
+// remaining block refers to a missing parent. Returns the effect (nil if nothing to purge).
+func Purge(d *Doc, candidates []string) *Effect {
+	set := make(map[string]bool, len(candidates))
+	for _, id := range candidates {
+		if b, ok := d.Blocks[id]; ok && b.Deleted {
+			set[id] = true
+		}
+	}
+	for changed := true; changed; {
+		changed = false
+		for _, b := range d.Blocks {
+			if !set[b.ID] && set[b.ParentID] {
+				delete(set, b.ParentID) // a remaining block still points to it
+				changed = true
+			}
+		}
+	}
+	if len(set) == 0 {
+		return nil
+	}
+	ids := make([]string, 0, len(set))
+	for id := range set {
+		ids = append(ids, id)
+	}
+	sort.Strings(ids)
+	e := Effect{Kind: OpPurge, IDs: ids}
+	d.ApplyEffect(e)
+	return &e
+}
+
 // conflictCopy keeps the losing text as a new block directly below the contested one.
 func (ap *applier) conflictCopy(op Op, b *Block, content string) Result {
 	id := ConflictBlockID(op.OpID)

internal/collab/ops_test.go

@@ -272,3 +272,33 @@ func TestClearConflictMarking(t *testing.T) {
 		t.Fatalf("conflict marking not cleared: %+v", p)
 	}
 }
+
+func TestPurgeKeepsTreeValid(t *testing.T) {
+	d := docWith(t, [3]string{"p", "", "Eltern"}, [3]string{"c", "p", "Kind"}, [3]string{"x", "", "Anderer"})
+	Apply(d, Op{OpID: "1", Kind: OpDelete, BlockID: "c"}, anna, Env{})
+	Apply(d, Op{OpID: "2", Kind: OpDelete, BlockID: "p"}, anna, Env{})
+	// Only "p" is old enough: it must stay because its deleted child "c" still points to it
+	if e := Purge(d, []string{"p"}); e != nil {
+		t.Fatalf("parent of a remaining tombstone must not be purged: %+v", e)
+	}
+	e := Purge(d, []string{"p", "c"})
+	if e == nil || len(e.IDs) != 2 || len(d.Blocks) != 1 {
+		t.Fatalf("both tombstones must go: %+v, blocks %d", e, len(d.Blocks))
+	}
+	mustValid(t, d)
+}
+
+func TestTextForPurgedBlockBecomesNewBlock(t *testing.T) {
+	d := docWith(t, [3]string{"a", "", "A"}, [3]string{"b", "", "B"})
+	Apply(d, Op{OpID: "1", Kind: OpDelete, BlockID: "b"}, anna, Env{})
+	Purge(d, []string{"b"})
+	_, res := Apply(d, Op{OpID: "2", Kind: OpSetContent, BlockID: "b", BaseVersion: 1, Content: str("Wochen offline getippt")}, ben, Env{})
+	if res.Status != StatusApplied {
+		t.Fatalf("late text must not be lost: %+v", res)
+	}
+	flat := d.Flatten()
+	if len(flat) != 2 || flat[1].Content != "Wochen offline getippt" {
+		t.Fatalf("text must be revived at the end: %+v", flat)
+	}
+	mustValid(t, d)
+}

internal/storage/docsync.go

@@ -9,6 +9,7 @@ import (
 	"sort"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/sandrodach/mindsift/internal/collab"
 	"github.com/sandrodach/mindsift/internal/domain"
@@ -20,6 +21,17 @@ import (
 // ErrDocNotFound is returned when a note does not exist or the user has no access to it.
 var ErrDocNotFound = errors.New("Notiz nicht gefunden oder kein Zugriff")
 
+// ErrLogCompacted means the requested log entries were removed by compaction; the client reloads the snapshot.
+var ErrLogCompacted = errors.New("Änderungsprotokoll verdichtet")
+
+// Retention of the operation log and of tombstones (see CompactDocs).
+const (
+	LogRetentionShared  = 30 * 24 * time.Hour
+	LogRetentionPrivate = 7 * 24 * time.Hour
+	TombstoneRetention  = 30 * 24 * time.Hour
+	OpResultRetention   = 90 * 24 * time.Hour
+)
+
 // Roles a user can have on a note.
 const (
 	DocRoleOwner  = "owner"
@@ -92,12 +104,20 @@ func (s *Store) migrateDocSync() {
 			created_at TEXT NOT NULL
 		);`,
 		"CREATE INDEX IF NOT EXISTS idx_doc_op_results_doc ON doc_op_results(doc_id);",
+		"ALTER TABLE blocks ADD COLUMN deleted_at TEXT NOT NULL DEFAULT '';",
+		"ALTER TABLE documents ADD COLUMN log_floor INTEGER NOT NULL DEFAULT 0;",
+		"CREATE INDEX IF NOT EXISTS idx_doc_ops_created ON doc_ops(created_at);",
+		"CREATE INDEX IF NOT EXISTS idx_doc_op_results_created ON doc_op_results(created_at);",
 	}
 	for _, q := range stmts {
 		if _, err := s.db.Exec(q); err != nil && !strings.Contains(err.Error(), "duplicate column") {
 			slog.Error("Migration der Notiz-Synchronisation fehlgeschlagen", "error", err, "statement", q)
 		}
 	}
+	// Tombstones from before deleted_at existed start their retention now
+	if _, err := s.db.Exec(`UPDATE blocks SET deleted_at = ? WHERE deleted = 1 AND deleted_at = ''`, domain.NowISO()); err != nil {
+		slog.Error("Migration der Notiz-Synchronisation fehlgeschlagen", "error", err)
+	}
 	s.backfillBlockTree()
 }
 
@@ -321,13 +341,14 @@ func (s *Store) applyDocChange(docID string, a collab.Actor, leases collab.Lease
 func persistDocState(tx *sql.Tx, docID string, before, after *collab.Doc, head int64, now string) error {
 	stmt, err := tx.Prepare(`
 		INSERT INTO blocks (id, document_id, parent_id, order_key, type, content, checked, collapsed, lang,
-		                    conflict_of, conflict_author, version, deleted, updated_by, level, position)
-		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, 0)
+		                    conflict_of, conflict_author, version, deleted, deleted_at, updated_by, level, position)
+		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, 0)
 		ON CONFLICT(id) DO UPDATE SET
 			parent_id = excluded.parent_id, order_key = excluded.order_key, type = excluded.type,
 			content = excluded.content, checked = excluded.checked, collapsed = excluded.collapsed,
 			lang = excluded.lang, conflict_of = excluded.conflict_of, conflict_author = excluded.conflict_author,
-			version = excluded.version, deleted = excluded.deleted, updated_by = excluded.updated_by
+			version = excluded.version, deleted = excluded.deleted, updated_by = excluded.updated_by,
+			deleted_at = CASE WHEN excluded.deleted = 0 THEN '' WHEN blocks.deleted = 1 THEN blocks.deleted_at ELSE excluded.deleted_at END
 		WHERE blocks.document_id = excluded.document_id`)
 	if err != nil {
 		return err
@@ -339,12 +360,23 @@ func persistDocState(tx *sql.Tx, docID string, before, after *collab.Doc, head i
 			continue
 		}
 		tagsChanged = true
+		deletedAt := ""
+		if b.Deleted {
+			deletedAt = now
+		}
 		if _, err := stmt.Exec(b.ID, docID, b.ParentID, b.OrderKey, b.Type, b.Content, boolInt(b.Props.Checked),
 			boolInt(b.Props.Collapsed), b.Props.Lang, b.Props.ConflictOf, b.Props.ConflictAuthor, b.Version,
-			boolInt(b.Deleted), b.UpdatedBy); err != nil {
+			boolInt(b.Deleted), deletedAt, b.UpdatedBy); err != nil {
 			return err
 		}
 	}
+	for id := range before.Blocks {
+		if _, ok := after.Blocks[id]; !ok {
+			if _, err := tx.Exec(`DELETE FROM blocks WHERE id = ? AND document_id = ?`, id, docID); err != nil {
+				return err
+			}
+		}
+	}
 
 	flat := flatDocumentBlocks(docID, after)
 	var plain strings.Builder
@@ -454,13 +486,23 @@ func (s *Store) GetThoughtByIDUnchecked(thoughtID string) (*domain.Thought, erro
 		 FROM thoughts WHERE id = ?`, thoughtID))
 }
 
-// GetDocOpsSince returns up to limit log entries after seq, plus the current head.
+// GetDocOpsSince returns up to limit log entries after seq, plus the current head. If entries after seq
+// were already compacted away it returns ErrLogCompacted (with the head): the client must reload the snapshot.
 func (s *Store) GetDocOpsSince(docID string, since int64, limit int) ([]DocLogEntry, int64, error) {
-	var head int64
-	if err := s.db.QueryRow(`SELECT head_seq FROM documents WHERE id = ?`, docID).Scan(&head); err != nil {
+	// One read transaction, so a compaction running in between cannot leave an unnoticed gap
+	tx, err := s.db.Begin()
+	if err != nil {
 		return nil, 0, err
 	}
-	rows, err := s.db.Query(
+	defer tx.Rollback()
+	var head, floor int64
+	if err := tx.QueryRow(`SELECT head_seq, log_floor FROM documents WHERE id = ?`, docID).Scan(&head, &floor); err != nil {
+		return nil, 0, err
+	}
+	if limit > 0 && since < floor {
+		return nil, head, ErrLogCompacted
+	}
+	rows, err := tx.Query(
 		`SELECT seq, op_id, user_id, client_id, payload FROM doc_ops WHERE doc_id = ? AND seq > ? ORDER BY seq ASC LIMIT ?`,
 		docID, since, limit)
 	if err != nil {
@@ -776,3 +818,128 @@ func longestIncreasing(keys []string) []bool {
 	}
 	return keep
 }
+
+// CompactStats counts what one compaction run removed.
+type CompactStats struct {
+	Docs       int
+	LogEntries int64
+	Tombstones int
+	OpResults  int64
+	Purged     map[string]int64 // docID -> new head, for notes whose tombstones were purged
+}
+
+// CompactDocs keeps the collaboration tables small:
+//   - log entries older than the retention (30 days for shared notes, 7 for private ones) are removed and
+//     the note remembers the boundary in log_floor; clients that are further behind reload the snapshot,
+//     their unsent changes stay queued and are sent afterwards,
+//   - deleted blocks older than 30 days are purged; this is itself a log entry ("purge"), so connected
+//     clients drop them without a reload,
+//   - idempotency results older than 90 days are removed.
+func (s *Store) CompactDocs(now time.Time) (CompactStats, error) {
+	st := CompactStats{Purged: map[string]int64{}}
+	ts := func(d time.Duration) string { return now.Add(-d).UTC().Format(time.RFC3339) }
+	tombCutoff := ts(TombstoneRetention)
+	rows, err := s.db.Query(
+		`SELECT doc_id FROM doc_ops WHERE created_at < ?
+		 UNION SELECT document_id FROM blocks WHERE deleted = 1 AND deleted_at != '' AND deleted_at < ?`,
+		ts(LogRetentionPrivate), tombCutoff)
+	if err != nil {
+		return st, err
+	}
+	var docIDs []string
+	for rows.Next() {
+		var id string
+		if rows.Scan(&id) == nil {
+			docIDs = append(docIDs, id)
+		}
+	}
+	rows.Close()
+
+	for _, docID := range docIDs {
+		cutoff := ts(LogRetentionPrivate)
+		if s.IsDocShared(docID) {
+			cutoff = ts(LogRetentionShared)
+		}
+		removed, purged, head, err := s.compactDoc(docID, cutoff, tombCutoff, now.UTC().Format(time.RFC3339))
+		if err != nil {
+			return st, fmt.Errorf("compact %s: %w", docID, err)
+		}
+		if purged > 0 {
+			st.Purged[docID] = head
+		}
+		if removed > 0 || purged > 0 {
+			st.Docs++
+		}
+		st.LogEntries += removed
+		st.Tombstones += purged
+	}
+
+	res, err := s.db.Exec(`DELETE FROM doc_op_results WHERE created_at < ?`, ts(OpResultRetention))
+	if err != nil {
+		return st, err
+	}
+	st.OpResults, _ = res.RowsAffected()
+	return st, nil
+}
+
+func (s *Store) compactDoc(docID, logCutoff, tombCutoff, now string) (removed int64, purged int, head int64, err error) {
+	unlock := lockDoc(docID)
+	defer unlock()
+	tx, err := s.db.Begin()
+	if err != nil {
+		return 0, 0, 0, err
+	}
+	defer tx.Rollback()
+
+	doc, head, err := loadCollabDoc(tx, docID)
+	if err != nil {
+		return 0, 0, 0, err
+	}
+	var candidates []string
+	rows, err := tx.Query(`SELECT id FROM blocks WHERE document_id = ? AND deleted = 1 AND deleted_at != '' AND deleted_at < ?`, docID, tombCutoff)
+	if err != nil {
+		return 0, 0, 0, err
+	}
+	for rows.Next() {
+		var id string
+		if rows.Scan(&id) == nil {
+			candidates = append(candidates, id)
+		}
+	}
+	rows.Close()
+	if e := collab.Purge(doc, candidates); e != nil {
+		head++
+		payload, _ := json.Marshal(e)
+		if _, err := tx.Exec(
+			`INSERT INTO doc_ops (doc_id, seq, op_id, user_id, client_id, kind, block_id, payload, created_at) VALUES (?, ?, '', '', '', ?, '', ?, ?)`,
+			docID, head, e.Kind, string(payload), now); err != nil {
+			return 0, 0, 0, err
+		}
+		for _, id := range e.IDs {
+			if _, err := tx.Exec(`DELETE FROM blocks WHERE id = ? AND document_id = ?`, id, docID); err != nil {
+				return 0, 0, 0, err
+			}
+		}
+		// Only the position changes; tombstones are not part of the note's text, so nothing derived changes
+		if _, err := tx.Exec(`UPDATE documents SET head_seq = ? WHERE id = ?`, head, docID); err != nil {
+			return 0, 0, 0, err
+		}
+		purged = len(e.IDs)
+	}
+
+	var floor int64
+	if err := tx.QueryRow(`SELECT COALESCE(MAX(seq), 0) FROM doc_ops WHERE doc_id = ? AND created_at < ?`, docID, logCutoff).Scan(&floor); err != nil {
+		return 0, 0, 0, err
+	}
+	if floor > 0 {
+		res, err := tx.Exec(`DELETE FROM doc_ops WHERE doc_id = ? AND seq <= ?`, docID, floor)
+		if err != nil {
+			return 0, 0, 0, err
+		}
+		removed, _ = res.RowsAffected()
+		if _, err := tx.Exec(`UPDATE documents SET log_floor = MAX(log_floor, ?) WHERE id = ?`, floor, docID); err != nil {
+			return 0, 0, 0, err
+		}
+	}
+	return removed, purged, head, tx.Commit()
+}

internal/storage/docsync_sim_test.go

@@ -1,6 +1,7 @@
 package storage
 
 import (
+	"errors"
 	"fmt"
 	"math/rand"
 	"sort"
@@ -25,15 +26,18 @@ type simClient struct {
 }
 
 type simWorld struct {
-	t       *testing.T
-	r       *rand.Rand
-	store   *Store
-	docID   string
-	leases  *collab.LeaseManager
-	now     time.Time
-	clients []*simClient
-	sent    map[string]collab.Op     // every op ever sent, by ID
-	results map[string]collab.Result // first result received per op ID
+	t           *testing.T
+	r           *rand.Rand
+	store       *Store
+	docID       string
+	leases      *collab.LeaseManager
+	now         time.Time
+	clients     []*simClient
+	sent        map[string]collab.Op     // every op ever sent, by ID
+	results     map[string]collab.Result // first result received per op ID
+	fullLog     []DocLogEntry            // copy of the log taken before each compaction, for the final checks
+	logSeen     int64
+	compactions int
 }
 
 func (w *simWorld) newOpID(c *simClient) string {
@@ -209,9 +213,17 @@ func (w *simWorld) serverSync(c *simClient, batch []collab.Op, holds []string) [
 func (w *simWorld) pull(c *simClient) {
 	for {
 		entries, head, err := w.store.GetDocOpsSince(w.docID, c.lastSeq, 7)
+		if errors.Is(err, ErrLogCompacted) {
+			// Too far behind: reload the snapshot, unsent operations stay queued
+			w.loadSnapshot(c)
+			continue
+		}
 		if err != nil {
 			w.t.Fatalf("GetDocOpsSince: %v", err)
 		}
+		if len(entries) == 0 && c.lastSeq < head {
+			w.t.Fatalf("%s: stuck at seq %d, head %d, no log entries", c.actor.ClientID, c.lastSeq, head)
+		}
 		for _, e := range entries {
 			if e.Seq != c.lastSeq+1 {
 				w.t.Fatalf("%s: log gap, expected seq %d got %d", c.actor.ClientID, c.lastSeq+1, e.Seq)
@@ -225,6 +237,44 @@ func (w *simWorld) pull(c *simClient) {
 	}
 }
 
+func (w *simWorld) loadSnapshot(c *simClient) {
+	snap, err := w.store.GetDocSnapshot(w.docID)
+	if err != nil {
+		w.t.Fatal(err)
+	}
+	c.confirmed = collab.NewDoc()
+	c.confirmed.Title, c.confirmed.TitleVersion = snap.Title, snap.TitleVersion
+	for _, b := range snap.Blocks {
+		b := b
+		c.confirmed.Blocks[b.ID] = &b
+	}
+	c.lastSeq = snap.HeadSeq
+}
+
+// captureLog copies log entries the simulation has not seen yet, before compaction removes them.
+func (w *simWorld) captureLog() {
+	entries, _, err := w.store.GetDocOpsSince(w.docID, w.logSeen, 1_000_000)
+	if err != nil {
+		w.t.Fatalf("captureLog: %v", err)
+	}
+	for _, e := range entries {
+		w.fullLog = append(w.fullLog, e)
+		w.logSeen = e.Seq
+	}
+}
+
+// compact ages the whole log and every tombstone past their retention and runs the compaction job.
+func (w *simWorld) compact() {
+	w.captureLog()
+	old := time.Now().Add(-40 * 24 * time.Hour).UTC().Format(time.RFC3339)
+	w.store.db.Exec(`UPDATE doc_ops SET created_at = ? WHERE doc_id = ?`, old, w.docID)
+	w.store.db.Exec(`UPDATE blocks SET deleted_at = ? WHERE document_id = ? AND deleted = 1`, old, w.docID)
+	if _, err := w.store.CompactDocs(time.Now()); err != nil {
+		w.t.Fatalf("CompactDocs: %v", err)
+	}
+	w.compactions++
+}
+
 // sync runs one round trip over the unreliable (or, with reliable=true, a perfect) network.
 func (w *simWorld) sync(c *simClient, reliable bool) {
 	if c.inflight == nil {
@@ -299,8 +349,10 @@ func runSimulation(t *testing.T, seed int64, steps int) {
 			}
 		case n < 95:
 			c.offline = !c.offline
-		default:
+		case n < 99:
 			w.now = w.now.Add(time.Duration(w.r.Intn(25)) * time.Second) // leases may expire
+		default:
+			w.compact()
 		}
 	}
 
@@ -339,12 +391,14 @@ func runSimulation(t *testing.T, seed int64, steps int) {
 	if stored != len(w.sent) {
 		t.Fatalf("seed %d: %d operations sent but %d results stored", seed, len(w.sent), stored)
 	}
-	log, _, _ := store.GetDocOpsSince(doc.ID, 0, 1_000_000)
+	w.captureLog()
+	log := w.fullLog
 	bySeq := map[int64]collab.Effect{}
 	inserted := map[string]string{}
 	for _, e := range log {
 		bySeq[e.Seq] = e.Effect
-		if e.Effect.Kind == collab.OpInsert {
+		// First insert only: a purged block that is edited again comes back with the same ID
+		if _, seen := inserted[e.Effect.BlockID]; e.Effect.Kind == collab.OpInsert && !seen {
 			inserted[e.Effect.BlockID] = e.Effect.Content
 		}
 	}
@@ -356,12 +410,18 @@ func runSimulation(t *testing.T, seed int64, steps int) {
 		}
 		switch op.Kind {
 		case collab.OpInsert, collab.OpSetContent, collab.OpSetTitle:
-			if res.Status == collab.StatusRejected {
+			// Clearing the text of a block that was purged meanwhile has nothing left to keep
+			emptied := op.Kind == collab.OpSetContent && *op.Content == "" && res.Reason == "missing"
+			if res.Status == collab.StatusRejected && !emptied {
 				t.Fatalf("seed %d: text-carrying op %s was rejected: %+v", seed, id, res)
 			}
 		}
 		if op.Kind == collab.OpSetContent && res.Status == collab.StatusApplied && res.Seq > 0 {
-			if e := bySeq[res.Seq]; e.Kind != collab.OpSetContent || e.Content != *op.Content {
+			want := collab.OpSetContent
+			if res.Reason == "revived" { // block was purged: the text came back as a new block
+				want = collab.OpInsert
+			}
+			if e := bySeq[res.Seq]; e.Kind != want || e.Content != *op.Content {
 				t.Fatalf("seed %d: applied text of %s not in the log", seed, id)
 			}
 		}
@@ -374,11 +434,11 @@ func runSimulation(t *testing.T, seed int64, steps int) {
 				text = *op.Title
 			}
 			if got, ok := inserted[res.ConflictBlockID]; !ok || got != text {
-				t.Fatalf("seed %d: conflict copy of %s missing or wrong", seed, id)
+				t.Fatalf("seed %d: conflict copy of %s missing or wrong: got %q (present %v), want %q", seed, id, got, ok, text)
 			}
 		}
 	}
-	t.Logf("seed %d: %d ops, %d log entries, %d conflict copies, %d blocks", seed, len(w.sent), len(log), conflicts, len(server.Blocks))
+	t.Logf("seed %d: %d ops, %d log entries, %d conflict copies, %d blocks, %d compactions", seed, len(w.sent), len(log), conflicts, len(server.Blocks), w.compactions)
 }
 
 func docsEqual(a, b *collab.Doc) bool {

internal/storage/docsync_test.go

@@ -1,8 +1,10 @@
 package storage
 
 import (
+	"errors"
 	"path/filepath"
 	"testing"
+	"time"
 
 	"github.com/sandrodach/mindsift/internal/collab"
 	"github.com/sandrodach/mindsift/internal/domain"
@@ -52,7 +54,7 @@ func sameOutline(t *testing.T, got []domain.DocumentBlock, want []domain.Documen
 
 func headSeq(t *testing.T, s *Store, docID string) int64 {
 	t.Helper()
-	_, head, err := s.GetDocOpsSince(docID, 0, 1)
+	_, head, err := s.GetDocOpsSince(docID, 0, 0)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -234,3 +236,129 @@ func TestFixedBlockIDsWorkInSeveralNotes(t *testing.T) {
 		t.Fatal("block IDs must differ between notes")
 	}
 }
+
+func backdate(t *testing.T, s *Store, docID string, age time.Duration) {
+	t.Helper()
+	old := time.Now().Add(-age).UTC().Format(time.RFC3339)
+	for _, q := range []string{
+		`UPDATE doc_ops SET created_at = ? WHERE doc_id = ?`,
+		`UPDATE doc_op_results SET created_at = ? WHERE doc_id = ?`,
+		`UPDATE blocks SET deleted_at = ? WHERE document_id = ? AND deleted = 1`,
+	} {
+		if _, err := s.db.Exec(q, old, docID); err != nil {
+			t.Fatal(err)
+		}
+	}
+}
+
+func TestCompactionKeepsStateAndForcesResync(t *testing.T) {
+	s, u := newSyncTestStore(t)
+	doc, _ := s.CreateDocument(u.ID, "", "Notiz", "📄", "", flatBlocks(
+		[3]interface{}{"a", 0, "Bleibt"},
+		[3]interface{}{"b", 0, "Gelöscht"},
+		[3]interface{}{"c", 1, "Kind von b"},
+	), nil, false)
+	a := collab.Actor{UserID: u.ID, ClientID: "tab1"}
+	if res, _, _ := s.ApplyDocOps(doc.ID, a, []collab.Op{{OpID: "del-b", Kind: collab.OpDelete, BlockID: "b"}}, nil); res[0].Status != collab.StatusApplied {
+		t.Fatalf("delete: %+v", res)
+	}
+	backdate(t, s, doc.ID, 40*24*time.Hour)
+	// Idempotency results stay longer (90 days): only the very old one goes
+	s.db.Exec(`UPDATE doc_op_results SET created_at = ? WHERE op_id = 'del-b'`, time.Now().Add(-100*24*time.Hour).UTC().Format(time.RFC3339))
+
+	// A client that is up to date before compaction must only need the remaining log afterwards
+	snap, _ := s.GetDocSnapshot(doc.ID)
+	client := collab.NewDoc()
+	for i := range snap.Blocks {
+		b := snap.Blocks[i]
+		client.Blocks[b.ID] = &b
+	}
+	text := "Frisch"
+	if _, _, err := s.ApplyDocOps(doc.ID, a, []collab.Op{{OpID: "fresh", Kind: collab.OpSetContent, BlockID: "a", BaseVersion: 1, Content: &text}}, nil); err != nil {
+		t.Fatal(err)
+	}
+
+	st, err := s.CompactDocs(time.Now())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if st.Tombstones != 2 || st.LogEntries == 0 || st.OpResults != 1 || st.Purged[doc.ID] == 0 {
+		t.Fatalf("unexpected stats: %+v", st)
+	}
+	var rows int
+	_ = s.db.QueryRow(`SELECT COUNT(*) FROM blocks WHERE document_id = ? AND id IN ('b', 'c')`, doc.ID).Scan(&rows)
+	if rows != 0 {
+		t.Fatalf("tombstones must be removed from the table, %d left", rows)
+	}
+
+	// Behind the floor: reload needed
+	if _, _, err := s.GetDocOpsSince(doc.ID, 0, 100); !errors.Is(err, ErrLogCompacted) {
+		t.Fatalf("client behind the floor must be told to resync, got %v", err)
+	}
+	entries, head, err := s.GetDocOpsSince(doc.ID, snap.HeadSeq, 100)
+	if err != nil || len(entries) != 2 || entries[1].Effect.Kind != collab.OpPurge {
+		t.Fatalf("fresh edit and purge must remain: %+v %v", entries, err)
+	}
+	for _, e := range entries {
+		client.ApplyEffect(e.Effect)
+	}
+	final, _ := s.GetDocSnapshot(doc.ID)
+	if final.HeadSeq != head || len(final.Blocks) != len(client.Blocks) {
+		t.Fatalf("client after log (%d blocks) differs from snapshot (%d blocks)", len(client.Blocks), len(final.Blocks))
+	}
+	for _, b := range final.Blocks {
+		if cb := client.Blocks[b.ID]; cb == nil || *cb != b {
+			t.Fatalf("block %s differs: client %+v, server %+v", b.ID, cb, b)
+		}
+	}
+	got, _ := s.GetDocumentByID(u.ID, doc.ID)
+	sameOutline(t, got.Blocks, flatBlocks([3]interface{}{"a", 0, "Frisch"}))
+
+	// A second run has nothing left to do
+	if st, _ := s.CompactDocs(time.Now()); st.Docs != 0 || st.OpResults != 0 {
+		t.Fatalf("second run changed something: %+v", st)
+	}
+}
+
+func TestCompactionRetentionDependsOnSharing(t *testing.T) {
+	s, u := newSyncTestStore(t)
+	other, _ := s.CreateUser("ben@example.com", "hash", "Ben", "member")
+	private, _ := s.CreateDocument(u.ID, "", "Privat", "📄", "", flatBlocks([3]interface{}{"p", 0, "x"}), nil, false)
+	shared, _ := s.CreateDocument(u.ID, "", "Geteilt", "📄", "", flatBlocks([3]interface{}{"s", 0, "x"}), nil, false)
+	if _, err := s.CreateDocShare(u.ID, shared.ID, other.Email, "write"); err != nil {
+		t.Fatal(err)
+	}
+	backdate(t, s, private.ID, 10*24*time.Hour)
+	backdate(t, s, shared.ID, 10*24*time.Hour)
+	if _, err := s.CompactDocs(time.Now()); err != nil {
+		t.Fatal(err)
+	}
+	if _, _, err := s.GetDocOpsSince(private.ID, 0, 10); !errors.Is(err, ErrLogCompacted) {
+		t.Fatalf("private note: 10 days old log must be compacted, got %v", err)
+	}
+	if entries, _, err := s.GetDocOpsSince(shared.ID, 0, 10); err != nil || len(entries) == 0 {
+		t.Fatalf("shared note: log must be kept for 30 days: %d entries, %v", len(entries), err)
+	}
+}
+
+func TestDeletedAtFollowsDeleteAndRestore(t *testing.T) {
+	s, u := newSyncTestStore(t)
+	doc, _ := s.CreateDocument(u.ID, "", "Notiz", "📄", "", flatBlocks([3]interface{}{"a", 0, "x"}, [3]interface{}{"b", 0, "y"}), nil, false)
+	a := collab.Actor{UserID: u.ID, ClientID: "tab1"}
+	deletedAt := func() string {
+		var v string
+		_ = s.db.QueryRow(`SELECT deleted_at FROM blocks WHERE id = 'b'`).Scan(&v)
+		return v
+	}
+	s.ApplyDocOps(doc.ID, a, []collab.Op{{OpID: "1", Kind: collab.OpDelete, BlockID: "b"}}, nil)
+	first := deletedAt()
+	if first == "" {
+		t.Fatal("deleting must set deleted_at")
+	}
+	// Other changes to the tombstone (none possible here) must not restart the clock; a restore clears it
+	text := "zurück"
+	s.ApplyDocOps(doc.ID, a, []collab.Op{{OpID: "2", Kind: collab.OpSetContent, BlockID: "b", BaseVersion: 1, Content: &text}}, nil)
+	if deletedAt() != "" {
+		t.Fatal("restoring must clear deleted_at")
+	}
+}