diff --git a/server/CLAUDE.md b/server/CLAUDE.md index 22929f62..fd7eb371 100644 --- a/server/CLAUDE.md +++ b/server/CLAUDE.md @@ -85,7 +85,7 @@ Uploaded images live in MinIO under `organizations/{org}/documents/{doc}/files/{ - **`internal/document/file/manager`** ticks every 5 minutes: it trims expired changelog snapshots, then reclaims files. A file is referenced if its id appears in any branch content of its document, any retained `document_branch_changelogs` snapshot, or any comment/reply — checked in SQL (`CheckDocumentFileReferenced`), so a spurious match only ever retains. Unreferenced files are stamped `unreferenced_at` and deleted a day later; rows with NULL FKs skip the wait; files younger than a day are never touched, which covers the gap between an upload and the first content persist that mentions it. - **`internal/document/hook/manager`** does the same for hooks: a row whose branch, document or organization went NULL gets `Hook.Delete` (external teardown) before the row is dropped. - **Organization deletion is announced, not discovered.** Better Auth owns the organizations table, so core cannot notice a deletion on its own. `beforeDeleteOrganization` (`auth-realtime/src/auth.ts`) calls `POST /api/x/organizations/{id}/teardown` while every row still exists: core tears down the org's hooks, queues the search removal, drops the `slack_apps` and `github_installations` rows, and deletes the logo object. The hook throws if core fails, so the organization survives instead of leaving orphans nothing can address. Document files still need no action — the cascade nulls their keys and the file manager reclaims them. -- **Ordering is inverted between create and delete.** Creating writes the row before the object (upload, duplication); deleting removes the object before the row. Both leave at most a row without an object, which the sweep resolves — an object without a row is invisible to every cleanup path. +- **Ordering is inverted between create and delete.** Creating writes the row before the object (upload, duplication); deleting removes the object before the row. Hooks follow the same rule from outside the transaction: `copyHooksToBranch` runs after the fork or merge commits, because `hook.NewHook` creates the watcher as a side effect and a rollback cannot take that back; a failed insert tears the just-created watcher down again. Both leave at most a row without an object, which the sweep resolves — an object without a row is invisible to every cleanup path. - **Duplication copies, never shares.** `Document.Duplicate` regenerates every block uid, rewrites image `src` paths to the new document (swapping only the path, never the host), and returns the old→new id map the handler uses to copy each object server-side into the duplicate's own folder. Files are therefore owned by exactly one document. - **Search index entries are removed by the delete itself.** Only default-branch blocks are indexed, keyed by block uid with a `documentId` field. `agent.DeleteDocument` collects the subtree ids before the row goes away and queues a search job carrying them, so the manager clears the index by `documentId` filter — the descendants the cascade destroys have no other trace to clean up by. Deletion paths therefore never queue the removal themselves. - Snapshots pin the files they reference, so `DB_MAX_DOCUMENT_CHANGELOGS` (count, trimmed on insert) and `DB_DOCUMENT_CHANGELOG_RETENTION` (age, trimmed in the sweep) also decide how long a removed image survives. diff --git a/server/core/internal/document/hook/manager/manager.go b/server/core/internal/document/hook/manager/manager.go index 12620e66..724ac067 100644 --- a/server/core/internal/document/hook/manager/manager.go +++ b/server/core/internal/document/hook/manager/manager.go @@ -157,28 +157,18 @@ func (m *Manager) processHooks(ctx context.Context) error { continue } - m.updateHook(ctx, h) + // a notification for a score that was not persisted would be + // published again on every cycle, since the next one recomputes + // the same transition from the same stored score. + if !m.updateHook(ctx, h) { + continue + } - if previousScore.Equal(_fullScore) && h.Score.Equal(decimal.Zero) { - maintainers, err := m.db.FetchDocumentMaintainers(ctx, h.DocumentID.V, h.OrganizationID.String) - if err != nil { - m.log.With("hook_id", h.ID). - With("error", err). - Error("fetching document maintainers for hook notification") - - continue - } - - m.notifPub.PublishNotifications( - h.OrganizationID.String, - notification.NewDocumentHookTriggeredNotification( - h.DocumentID.V, - h.Type, - h.BlockID, - h.BranchID.V, - ), - maintainers..., - ) + // the score decays gradually — a scheduled reminder walks down + // through 99…1 — so the transition to watch for is the arrival + // at zero, not an exact full-to-zero jump within one cycle. + if !previousScore.IsZero() && h.Score.IsZero() { + m.notifyMaintainers(ctx, h) } } @@ -192,6 +182,30 @@ func (m *Manager) processHooks(ctx context.Context) error { return nil } +// notifyMaintainers tells the document's maintainers that the hook ran out of +// freshness. +func (m *Manager) notifyMaintainers(ctx context.Context, h hook.Hook) { + maintainers, err := m.db.FetchDocumentMaintainers(ctx, h.DocumentID.V, h.OrganizationID.String) + if err != nil { + m.log.With("hook_id", h.ID). + With("error", err). + Error("fetching document maintainers for hook notification") + + return + } + + m.notifPub.PublishNotifications( + h.OrganizationID.String, + notification.NewDocumentHookTriggeredNotification( + h.DocumentID.V, + h.Type, + h.BlockID, + h.BranchID.V, + ), + maintainers..., + ) +} + // ensureHook ensures the hook is valid and handles deletions if necessary. func (m *Manager) ensureHook( ctx context.Context, @@ -265,14 +279,18 @@ func (m *Manager) deleteHook(ctx context.Context, h *hook.Hook) { } } -// updateHook persists the hook's current state. -func (m *Manager) updateHook(ctx context.Context, h hook.Hook) { +// updateHook persists the hook's current state and reports whether it stuck. +func (m *Manager) updateHook(ctx context.Context, h hook.Hook) bool { err := m.db.UpdateDocumentHook(ctx, h) if err != nil { m.log.With("hook_id", h.ID). With("error", err). Error("updating document hook") + + return false } + + return true } // DB defines the database operations required by the Manager. diff --git a/server/core/internal/document/hook/manager/manager_test.go b/server/core/internal/document/hook/manager/manager_test.go index 40c00975..4e46c634 100644 --- a/server/core/internal/document/hook/manager/manager_test.go +++ b/server/core/internal/document/hook/manager/manager_test.go @@ -141,12 +141,13 @@ func Test_Manager_processHooks(t *testing.T) { } tests := map[string]struct { - Hooks func(t *testing.T) []hook.Hook - FetchErr error - DocErr error - Doc *document.Document - MaintErr error - Checks []check + Hooks func(t *testing.T) []hook.Hook + FetchErr error + DocErr error + Doc *document.Document + MaintErr error + UpdateErr error + Checks []check }{ "Hook fetch failure is propagated": { Hooks: func(*testing.T) []hook.Hook { return nil }, @@ -281,6 +282,51 @@ func Test_Manager_processHooks(t *testing.T) { wasPublished(1), ), }, + // the score decays gradually, so by the time it reaches zero the + // previous one is somewhere below full: the notification has to + // trigger on the arrival at zero, not on a full-to-zero jump. + "Partly decayed score reaching zero notifies the maintainers": { + Hooks: func(t *testing.T) []hook.Hook { + h := stubHook(t, branchID, time.Now().Add(-time.Hour), time.Now().Add(-2*time.Hour)) + h.Score = decimal.NewFromInt(40) + + return []hook.Hook{h} + }, + Doc: stubDocument(), + Checks: checks( + hasError(false), + hasUpdatedScore(decimal.Zero), + wasPublished(1), + ), + }, + "Already zero score is not re-notified": { + Hooks: func(t *testing.T) []hook.Hook { + h := stubHook(t, branchID, time.Now().Add(-time.Hour), time.Now().Add(-2*time.Hour)) + h.Score = decimal.Zero + + return []hook.Hook{h} + }, + Doc: stubDocument(), + Checks: checks( + hasError(false), + wasUpdateCalled(1), + wasPublished(0), + ), + }, + // the unpersisted score would be recomputed into the same + // transition on the next cycle, notifying again every five minutes. + "Failed score persist suppresses the notification": { + Hooks: func(t *testing.T) []hook.Hook { + return []hook.Hook{stubHook(t, branchID, time.Now().Add(-time.Hour), time.Now().Add(-2*time.Hour))} + }, + Doc: stubDocument(), + UpdateErr: assert.AnError, + Checks: checks( + hasError(false), + wasUpdateCalled(1), + wasPublished(0), + ), + }, "Maintainer fetch failure suppresses the notification": { Hooks: func(t *testing.T) []hook.Hook { return []hook.Hook{stubHook(t, branchID, time.Now().Add(-time.Hour), time.Now().Add(-2*time.Hour))} @@ -381,6 +427,9 @@ func Test_Manager_processHooks(t *testing.T) { FetchDocumentMaintainersFunc: func(context.Context, xid.ID, string) ([]string, error) { return []string{"user-1"}, tc.MaintErr }, + UpdateDocumentHookFunc: func(context.Context, hook.Hook) error { + return tc.UpdateErr + }, } pub := &fakePublisher{} diff --git a/server/core/internal/server/internal/document/branch_http_test.go b/server/core/internal/server/internal/document/branch_http_test.go index 42fa76e7..de9056db 100644 --- a/server/core/internal/server/internal/document/branch_http_test.go +++ b/server/core/internal/server/internal/document/branch_http_test.go @@ -401,29 +401,39 @@ func Test_Handler_MergeBranches(t *testing.T) { Body: validBody, RespCode: http.StatusInternalServerError, }, + // the hooks are copied after the commit, so a failure there leaves + // the merge itself standing. "Hook copy fetch error": { - DB: &DBMock{FetchDocumentByBranchIDFunc: fetchByBranch}, - Tx: &TxMock{ + DB: &DBMock{ + FetchDocumentByBranchIDFunc: fetchByBranch, FetchDocumentHooksByBranchIDFunc: func(context.Context, xid.ID, string) ([]hookCore.Hook, error) { return nil, errors.New("boom") }, }, - Body: validBody, - RespCode: http.StatusInternalServerError, + Tx: &TxMock{}, + Body: validBody, + RespCode: http.StatusOK, + Committed: 1, + Metadata: 1, + Reviewers: 1, }, "Hook re-creation error": { - DB: &DBMock{FetchDocumentByBranchIDFunc: fetchByBranch}, - Tx: &TxMock{ + DB: &DBMock{ + FetchDocumentByBranchIDFunc: fetchByBranch, FetchDocumentHooksByBranchIDFunc: func(context.Context, xid.ID, string) ([]hookCore.Hook, error) { return []hookCore.Hook{storedHook("bogus")}, nil }, }, - Body: validBody, - RespCode: http.StatusInternalServerError, + Tx: &TxMock{}, + Body: validBody, + RespCode: http.StatusOK, + Committed: 1, + Metadata: 1, + Reviewers: 1, }, "Hook insertion error": { - DB: &DBMock{FetchDocumentByBranchIDFunc: fetchByBranch}, - Tx: &TxMock{ + DB: &DBMock{ + FetchDocumentByBranchIDFunc: fetchByBranch, FetchDocumentHooksByBranchIDFunc: func(context.Context, xid.ID, string) ([]hookCore.Hook, error) { return []hookCore.Hook{storedHook(hookCore.TypeScheduledReminder)}, nil }, @@ -431,8 +441,12 @@ func Test_Handler_MergeBranches(t *testing.T) { return errors.New("boom") }, }, - Body: validBody, - RespCode: http.StatusInternalServerError, + Tx: &TxMock{}, + Body: validBody, + RespCode: http.StatusOK, + Committed: 1, + Metadata: 1, + Reviewers: 1, }, "Document update error": { DB: &DBMock{FetchDocumentByBranchIDFunc: fetchByBranch}, @@ -466,12 +480,13 @@ func Test_Handler_MergeBranches(t *testing.T) { Committed: 1, }, "Successful merge": { - DB: &DBMock{FetchDocumentByBranchIDFunc: fetchByBranch}, - Tx: &TxMock{ + DB: &DBMock{ + FetchDocumentByBranchIDFunc: fetchByBranch, FetchDocumentHooksByBranchIDFunc: func(context.Context, xid.ID, string) ([]hookCore.Hook, error) { return []hookCore.Hook{storedHook(hookCore.TypeScheduledReminder)}, nil }, }, + Tx: &TxMock{}, Body: validBody, RespCode: http.StatusOK, Committed: 1, @@ -501,15 +516,22 @@ func Test_Handler_MergeBranches(t *testing.T) { assert.Equal(t, c.Reviewers, cnt.reviewers) if c.RespCode == http.StatusOK { - // hooks are soft-deleted on the target and re-created - // from the source branch. + // hooks are soft-deleted on the target inside the + // transaction and re-created from the source branch once it + // commits, since creating one creates its watcher. require.Len(t, c.Tx.SoftDeleteDocumentHooksByBranchIDCalls(), 1) assert.Equal(t, _branchID, c.Tx.SoftDeleteDocumentHooksByBranchIDCalls()[0].BranchID) - require.Len(t, c.Tx.FetchDocumentHooksByBranchIDCalls(), 1) - assert.Equal(t, _branchID2, c.Tx.FetchDocumentHooksByBranchIDCalls()[0].BranchID) - require.Len(t, c.Tx.InsertDocumentHookCalls(), 1) - assert.Equal(t, null.ValueFrom(_branchID), c.Tx.InsertDocumentHookCalls()[0].Hk.BranchID) + assert.Empty(t, c.Tx.InsertDocumentHookCalls()) require.Len(t, c.Tx.PromoteBranchApprovalsCalls(), 1) + + ff := c.DB.FetchDocumentHooksByBranchIDCalls() + require.Len(t, ff, 1) + assert.Equal(t, _branchID2, ff[0].BranchID) + } + + if cn == "Successful merge" { + require.Len(t, c.DB.InsertDocumentHookCalls(), 1) + assert.Equal(t, null.ValueFrom(_branchID), c.DB.InsertDocumentHookCalls()[0].Hk.BranchID) } }) } @@ -977,13 +999,15 @@ func Test_Handler_CreateDocumentBranch(t *testing.T) { validBody := `{"branch":"feature","sourceBranchId":"` + _branchID.String() + `"}` cc := map[string]struct { - DB *DBMock - Tx *TxMock - BeginErr error - NoSession bool - Body string - RespCode int - Committed int + DB *DBMock + Tx *TxMock + BeginErr error + NoSession bool + Body string + CopiedHooks []hookCore.Hook + HookFetchErr error + RespCode int + Committed int }{ "No session in context": { DB: &DBMock{}, @@ -1063,12 +1087,13 @@ func Test_Handler_CreateDocumentBranch(t *testing.T) { FetchDocumentFunc: func(context.Context, xid.ID, string, string) (*documentCore.Document, error) { return branchDoc(_branchID2), nil }, - FetchDocumentHooksByBranchIDFunc: func(context.Context, xid.ID, string) ([]hookCore.Hook, error) { - return nil, errors.New("boom") - }, }, - Body: validBody, - RespCode: http.StatusInternalServerError, + HookFetchErr: errors.New("boom"), + Body: validBody, + // the hooks are copied after the commit, so a failure there + // leaves the branch itself standing. + RespCode: http.StatusCreated, + Committed: 1, }, "Commit error": { DB: &DBMock{ @@ -1098,13 +1123,11 @@ func Test_Handler_CreateDocumentBranch(t *testing.T) { FetchDocumentFunc: func(context.Context, xid.ID, string, string) (*documentCore.Document, error) { return branchDoc(_branchID2), nil }, - FetchDocumentHooksByBranchIDFunc: func(context.Context, xid.ID, string) ([]hookCore.Hook, error) { - return []hookCore.Hook{storedHook(hookCore.TypeScheduledReminder)}, nil - }, }, - Body: validBody, - RespCode: http.StatusCreated, - Committed: 1, + CopiedHooks: []hookCore.Hook{storedHook(hookCore.TypeScheduledReminder)}, + Body: validBody, + RespCode: http.StatusCreated, + Committed: 1, }, } @@ -1112,6 +1135,10 @@ func Test_Handler_CreateDocumentBranch(t *testing.T) { t.Run(cn, func(t *testing.T) { t.Parallel() + c.DB.FetchDocumentHooksByBranchIDFunc = func(context.Context, xid.ID, string) ([]hookCore.Hook, error) { + return c.CopiedHooks, c.HookFetchErr + } + hdl, _ := newTestHandler(withTx(c.DB, c.Tx, c.BeginErr), &fakePublisher{}) rec := httptest.NewRecorder() @@ -1120,14 +1147,21 @@ func Test_Handler_CreateDocumentBranch(t *testing.T) { assert.Equal(t, c.RespCode, rec.Code) assert.Len(t, c.Tx.CommitCalls(), c.Committed) + assert.Empty(t, c.Tx.InsertDocumentHookCalls()) if c.RespCode == http.StatusCreated { require.Len(t, c.Tx.ForkDocumentBranchCalls(), 1) assert.Equal(t, "feature", c.Tx.ForkDocumentBranchCalls()[0].TargetBranch) - // hooks are copied from the source to the new branch. - require.Len(t, c.Tx.InsertDocumentHookCalls(), 1) - assert.Equal(t, null.ValueFrom(_branchID2), c.Tx.InsertDocumentHookCalls()[0].Hk.BranchID) } + + if len(c.CopiedHooks) == 0 { + return + } + + // hooks are copied from the source to the new branch once the + // fork commits, since creating one creates its watcher. + require.Len(t, c.DB.InsertDocumentHookCalls(), 1) + assert.Equal(t, null.ValueFrom(_branchID2), c.DB.InsertDocumentHookCalls()[0].Hk.BranchID) }) } } @@ -1240,3 +1274,30 @@ func Test_Handler_DeleteDocumentBranch(t *testing.T) { }) } } + +func Test_Handler_copyHooksToBranch(t *testing.T) { + t.Parallel() + + // creating a hook creates its external resource, so a failed insert has + // to hand it back rather than leave it running with no row pointing at + // it. A url-watcher would call changedetection.io here, which the test + // cannot reach; the scheduled reminder exercises the same path with a + // teardown that needs nothing external. + db := &DBMock{ + FetchDocumentHooksByBranchIDFunc: func(context.Context, xid.ID, string) ([]hookCore.Hook, error) { + return []hookCore.Hook{storedHook(hookCore.TypeScheduledReminder)}, nil + }, + InsertDocumentHookFunc: func(context.Context, hookCore.Hook) error { + return errors.New("boom") + }, + } + + hdl, _ := newTestHandler(db, &fakePublisher{}) + + err := hdl.copyHooksToBranch(context.Background(), _branchID2, _branchID, _documentID, "org1") + require.Error(t, err) + + // the failure stops the copy rather than working through the rest of + // the branch's hooks and creating a watcher for each. + require.Len(t, db.InsertDocumentHookCalls(), 1) +} diff --git a/server/core/internal/server/internal/document/http.go b/server/core/internal/server/internal/document/http.go index ee0df4ea..a791cd3a 100644 --- a/server/core/internal/server/internal/document/http.go +++ b/server/core/internal/server/internal/document/http.go @@ -882,11 +882,6 @@ func (h *Handler) MergeBranches(w http.ResponseWriter, r *http.Request) { return } - if err := h.copyHooksToBranch(r.Context(), tx, fromDoc.BranchID, toDoc.BranchID, toDoc.ID, session.ActiveOrganizationID); err != nil { - httpserver.RespondError(h.log, w, err) - return - } - if err := tx.UpdateDocument(r.Context(), ndoc); err != nil { httpserver.RespondError(h.log, w, err) return @@ -907,6 +902,21 @@ func (h *Handler) MergeBranches(w http.ResponseWriter, r *http.Request) { return } + // the hooks are copied after the commit: creating one creates its + // external watcher too, and a rollback cannot take that back. + if err := h.copyHooksToBranch( + r.Context(), + fromDoc.BranchID, + toDoc.BranchID, + toDoc.ID, + session.ActiveOrganizationID, + ); err != nil { + logutil.Critical(h.log, err).Error( + "cannot copy hooks to the merged branch", + slog.String("branch_id", toDoc.BranchID.String()), + ) + } + if h.metadata.changeCallback != nil { h.metadata.changeCallback(session.ActiveOrganizationID, ndoc) } @@ -1229,14 +1239,24 @@ func (h *Handler) CreateDocumentBranch(w http.ResponseWriter, r *http.Request) { return } - if err := h.copyHooksToBranch(r.Context(), tx, sourceDoc.BranchID, newDoc.BranchID, sourceDoc.ID, session.ActiveOrganizationID); err != nil { + if err := tx.Commit(); err != nil { httpserver.RespondError(h.log, w, err) return } - if err := tx.Commit(); err != nil { - httpserver.RespondError(h.log, w, err) - return + // the hooks are copied after the commit: creating one creates its + // external watcher too, and a rollback cannot take that back. + if err := h.copyHooksToBranch( + r.Context(), + sourceDoc.BranchID, + newDoc.BranchID, + sourceDoc.ID, + session.ActiveOrganizationID, + ); err != nil { + logutil.Critical(h.log, err).Error( + "cannot copy hooks to the forked branch", + slog.String("branch_id", newDoc.BranchID.String()), + ) } httpserver.Respond( @@ -1300,8 +1320,13 @@ func (h *Handler) DeleteDocumentBranch(w http.ResponseWriter, r *http.Request) { // copyHooksToBranch fetches all hooks from fromBranchID and re-creates them on // toBranchID with fresh state. This is handler-level business logic; the DB // layer is not involved in the re-creation decision. -func (h *Handler) copyHooksToBranch(ctx context.Context, db HooksDBAgent, fromBranchID, toBranchID, documentID xid.ID, organizationID string) error { - hooks, err := db.FetchDocumentHooksByBranchID(ctx, fromBranchID, organizationID) +// +// Creating a hook creates its external resource as a side effect, so this runs +// outside the caller's transaction: a rollback cannot take a changedetection.io +// watcher back, and the row that would have pointed at it is gone. A failed +// insert tears the watcher down again for the same reason. +func (h *Handler) copyHooksToBranch(ctx context.Context, fromBranchID, toBranchID, documentID xid.ID, organizationID string) error { + hooks, err := h.db.FetchDocumentHooksByBranchID(ctx, fromBranchID, organizationID) if err != nil { return err } @@ -1319,7 +1344,14 @@ func (h *Handler) copyHooksToBranch(ctx context.Context, db HooksDBAgent, fromBr return err } - if err := db.InsertDocumentHook(ctx, *newHk); err != nil { + if err := h.db.InsertDocumentHook(ctx, *newHk); err != nil { + if derr := newHk.Delete(ctx, inp); derr != nil { + logutil.Critical(h.log, derr).Error( + "cannot delete the hook external resource after a failed insert", + slog.String("hook_id", newHk.ID.String()), + ) + } + return err } }