improve hook notification & duplication handling

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.

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.

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{}

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

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