diff --git a/internal/appstore/boundary_check_test.go b/internal/appstore/boundary_check_test.go index 5fe88a420..1ff7ba1e7 100644 --- a/internal/appstore/boundary_check_test.go +++ b/internal/appstore/boundary_check_test.go @@ -91,6 +91,9 @@ var allowedStoreMethodsInsideDoor = map[string]string{ "emitItemEventAsInstallTx": "INSERT INTO event_outbox (through writeOutboxTx) for the item it is given; sets the block's actor_via_app", "emitItemUpdateEventsAsInstallTx": "INSERT INTO event_outbox for the item it is given (status_changed and/or updated); sets the block's actor_via_app", "emitCommentEventAsInstallTx": "INSERT INTO event_outbox (through writeOutboxTx) for the comment it is given; sets the block's actor_via_app", + // TASK-3937: the push queue, written at the tail of the fenced comment create. + "enqueueCommentNotificationsTx": "reads the comment's item (assignee, creator) and its parent comment's author, then enqueueNotificationsTx for the comment it is given", + "enqueueNotificationsTx": "INSERT INTO user_notifications ... ON CONFLICT (user_id, dedupe_key) DO NOTHING, only for a user with a row in push_devices; on the tx it is handed", // BUG-3653 part B: the storage cap on an app upload. BeginFenced takes the // lock first when the spec sets StorageLimit; CheckStorageLimit reads. "lockStorageLimitTx": "reads workspaces.owner_id, then acquirePlanLimitLock; writes nothing", diff --git a/internal/models/notification_settings.go b/internal/models/notification_settings.go index 972421e93..df66ec837 100644 --- a/internal/models/notification_settings.go +++ b/internal/models/notification_settings.go @@ -1,8 +1,8 @@ package models // NotificationSettings are a person's recipient-side push filters (TASK-3937). -// The dispatcher (TASK-3939) is to apply them and the enqueue (TASK-3937 -// part B) is not to, so changing one will never rewrite the queue. +// The dispatcher (TASK-3939) is to apply them; the enqueue does not, so +// changing one never rewrites the queue. type NotificationSettings struct { PushEnabled bool `json:"push_enabled"` NotifyAssigned bool `json:"notify_assigned"` diff --git a/internal/store/account_claim.go b/internal/store/account_claim.go index f56a92900..7e1ac8ba9 100644 --- a/internal/store/account_claim.go +++ b/internal/store/account_claim.go @@ -380,6 +380,16 @@ func (s *Store) claimAccountTx(tx *sql.Tx, userID string, unusable []byte, ts, c {"delete provider bindings", `DELETE FROM user_oauth_identities WHERE user_id = ?`, []any{userID}}, {"delete sessions", `DELETE FROM sessions WHERE user_id = ?`, []any{userID}}, {"delete api tokens", `DELETE FROM api_tokens WHERE user_id = ?`, []any{userID}}, + // The old owner's registered devices would otherwise receive the new + // person's pushes, and their queued pushes describe items the new + // person was never told about (TASK-3937). The sessions delete above + // already removes devices through push_devices.session_id ON DELETE + // CASCADE; this explicit delete is belt-and-braces over that cascade, + // naming the intent and surviving a future change to the FK. The + // claim test asserts the outcome, not which line did the work. + {"delete push devices", `DELETE FROM push_devices WHERE user_id = ?`, []any{userID}}, + {"delete queued notifications", `DELETE FROM user_notifications WHERE user_id = ?`, []any{userID}}, + {"delete notification settings", `DELETE FROM user_notification_settings WHERE user_id = ?`, []any{userID}}, // Every grant the account held is recorded as revoked, so a refresh // or code exchange that validated before the claim mints nothing // after it (BUG-3678). The connection rows, read before the delete diff --git a/internal/store/comments.go b/internal/store/comments.go index 46b211cf5..41a66e62f 100644 --- a/internal/store/comments.go +++ b/internal/store/comments.go @@ -183,6 +183,10 @@ func (s *Store) createCommentTx(tx *sql.Tx, workspaceID, itemID, userID string, if err := s.enqueueDecisionJobsForItemTx(tx, itemID); err != nil { return "", err } + // TASK-3937: at the tail, after every other in-tx step. + if err := s.enqueueCommentNotificationsTx(tx, created, userID, createdBy == "agent"); err != nil { + return "", err + } return id, nil } diff --git a/internal/store/fenced_comments.go b/internal/store/fenced_comments.go index 1c249598e..6a8204d5e 100644 --- a/internal/store/fenced_comments.go +++ b/internal/store/fenced_comments.go @@ -129,6 +129,10 @@ func (f *FencedTx) CreateComment(in FencedCommentCreate) (*models.Comment, error if err := f.s.enqueueDecisionJobsForItemTx(f.tx, in.ItemID); err != nil { return nil, err } + // TASK-3937: at the tail, after every other in-tx step. + if err := f.s.enqueueCommentNotificationsTx(f.tx, created, in.Actor.UserID, in.Actor.Kind == "agent"); err != nil { + return nil, err + } return created, nil } diff --git a/internal/store/items.go b/internal/store/items.go index 1bf83ecee..bf4908952 100644 --- a/internal/store/items.go +++ b/internal/store/items.go @@ -658,6 +658,13 @@ func (s *Store) createItemTxWithID(tx *sql.Tx, id, workspaceID, collectionID str if err := s.enqueueDecisionJobsTx(tx, item.WorkspaceID, item.ID, item.CollectionID); err != nil { return nil, err } + // TASK-3937: an item created with an assignee tells them, here rather + // than in tryCreateItem so a cross-workspace copy is covered too. + if input.AssignedUserID != nil { + if err := s.enqueueAssignedNotificationTx(tx, item, *input.AssignedUserID, input.ActorUserID, item.CreatedBy == "agent"); err != nil { + return nil, err + } + } // TASK-3585: the caller's last word inside this transaction. if input.AfterWriteTx != nil { if err := input.AfterWriteTx(tx, item); err != nil { @@ -3572,6 +3579,13 @@ func (s *Store) updateItemWithParentLinkOnce( if err := s.emitItemUpdateEventsTx(tx, existing, updated, mutSignal.StatusChanged, mutSignal.FromStatus, doneKey, opt.batchID, hierarchyChanged); err != nil { return nil, err } + // TASK-3937: tell a new assignee, on this transaction. Self-assignment + // notifies nobody; an empty actor (a system writer) is not the assignee. + if mutSignal.AssignmentChanged { + if err := s.enqueueAssignedNotificationTx(tx, updated, mutSignal.ToAssignedUserID, input.ActorUserID, updated.LastModifiedBy == "agent"); err != nil { + return nil, err + } + } // Enqueued on EVERY update, including ones that change nothing a // question can see: whether the state changed is the runner's state-hash // check, not a guess made here (TASK-3117). No-op with no provider. diff --git a/internal/store/migrated.go b/internal/store/migrated.go index dc8fadc1a..f15f65ea2 100644 --- a/internal/store/migrated.go +++ b/internal/store/migrated.go @@ -117,7 +117,8 @@ const MigratedMarker = "PAD_MIGRATED_TO_POSTGRES" // a device row is bound to the session that registers it (TASK-3938), // and the migration does not carry sessions, so every device is to // re-register after the operator signs in again; the queue is delivery state, not a -// person's data: a lost row is one push not sent. (Its dispatcher and +// person's data: a lost row is one push not sent. The enqueue +// (notifications.go) writes it, gated on a registered device. (Its dispatcher and // 7-day sweep land with TASK-3939; until then nothing reads or deletes // it.) Both are excluded as "push-delivery" in migratedExcludedTables, // which TestMigratedEveryLiveTableIsClassified holds complete. diff --git a/internal/store/notifications.go b/internal/store/notifications.go new file mode 100644 index 000000000..469479ca0 --- /dev/null +++ b/internal/store/notifications.go @@ -0,0 +1,177 @@ +package store + +import ( + "database/sql" + "errors" + "fmt" + + "github.com/PerpetualSoftware/pad/internal/models" +) + +// The push queue's enqueue side (TASK-3937). Every function here runs on the +// CALLER's transaction, so a notification commits with the write that caused +// it or not at all. The settings a recipient has chosen are the dispatcher's +// to apply, not the enqueue's. +// +// A row is written ONLY for a user with at least one push_devices row (the +// EXISTS in enqueueNotificationsTx), so until device registration lands +// (TASK-3938) the enqueue writes nothing. dedupe_key's ON CONFLICT makes it +// idempotent per (user, key). Migration 155's header predates this file and +// calls the enqueue owed: merged migrations are immutable, so this comment, +// not that one, is the current description. +// +// Recipients are NOT access-checked here. Removing a member leaves them an +// item's created_by and possibly its assignee, so a row can be written for +// someone who can no longer see the item. The dispatcher (TASK-3939) is to +// re-check current membership and item visibility at send and skip the row: +// that is the privacy gate. +// +// Nothing here enqueues the 'mention' kind: @mention notifications are +// TASK-3946's, built from the comment_mentions index. + +const ( + notifyAssigned = "assigned" + notifyComment = "comment" + notifyReply = "reply" + notifyReminder = "reminder" +) + +// notifyFact is one notification owed to one user. +type notifyFact struct { + UserID string + WorkspaceID string + ItemID string + CommentID string + Kind string + ActorUserID string + ActorIsAgent bool + // DedupeKey is unique per user: enqueuing the same fact twice writes one + // row. Built from a kind and server-minted ids, never from request text. + DedupeKey string +} + +// enqueueNotificationsTx writes each fact for its user, and only for a user +// who has at least one registered device: a person who never installed the +// app accumulates nothing. The EXISTS gate is in the statement, not a prior +// read, so a device registered in the same instant cannot be missed or +// double-counted. The SELECT form carries a WHERE because SQLite misparses +// INSERT ... SELECT ... ON CONFLICT without one. +func (s *Store) enqueueNotificationsTx(tx *sql.Tx, facts []notifyFact) error { + agentParam := "CAST(? AS INTEGER)" + if s.dialect.Driver() == DriverPostgres { + agentParam = "CAST(? AS BOOLEAN)" + } + q := s.q(` + INSERT INTO user_notifications + (id, user_id, workspace_id, item_id, comment_id, kind, actor_user_id, actor_is_agent, dedupe_key, created_at) + SELECT ?, ?, ?, ?, NULLIF(?, ''), ?, NULLIF(?, ''), ` + agentParam + `, ?, ? + WHERE EXISTS (SELECT 1 FROM push_devices WHERE user_id = ?) + ON CONFLICT (user_id, dedupe_key) DO NOTHING`) + ts := now() + for _, f := range facts { + if f.UserID == "" { + continue + } + if _, err := tx.Exec(q, + newID(), f.UserID, f.WorkspaceID, f.ItemID, f.CommentID, f.Kind, f.ActorUserID, + s.dialect.BoolToInt(f.ActorIsAgent), f.DedupeKey, ts, f.UserID); err != nil { + return fmt.Errorf("enqueue %s notification: %w", f.Kind, err) + } + } + return nil +} + +// enqueueAssignedNotificationTx tells toUser they were given item. Assigning +// yourself notifies nobody; an empty actor (a system write) is not "you". +func (s *Store) enqueueAssignedNotificationTx(tx *sql.Tx, item *models.Item, toUser, actorUserID string, actorIsAgent bool) error { + if toUser == "" || toUser == actorUserID { + return nil + } + return s.enqueueNotificationsTx(tx, []notifyFact{{ + UserID: toUser, WorkspaceID: item.WorkspaceID, ItemID: item.ID, Kind: notifyAssigned, + ActorUserID: actorUserID, ActorIsAgent: actorIsAgent, + DedupeKey: fmt.Sprintf("assigned:%s:%d", item.ID, item.Seq), + }}) +} + +// commentRecipients collects who a comment concerns, first kind wins. It is a +// type rather than a closure so the appstore boundary walk can resolve it. +type commentRecipients struct { + order []string + kinds map[string]string +} + +// add records user under kind unless they are the actor, empty, or already in. +func (r *commentRecipients) add(user, kind, actorUserID string) { + if user == "" || user == actorUserID { + return + } + if r.kinds == nil { + r.kinds = map[string]string{} + } + if _, seen := r.kinds[user]; !seen { + r.order = append(r.order, user) + r.kinds[user] = kind + } +} + +// enqueueCommentNotificationsTx tells the people a new comment concerns: the +// item's assignee and creator ("comment"), and the author of the comment it +// replies to ("reply"). The author is never told about their own comment. A +// person who is both a reply target and an assignee gets the one "reply". +// A tombstoned parent still names its author; the dispatcher filters later. +func (s *Store) enqueueCommentNotificationsTx(tx *sql.Tx, c *models.Comment, actorUserID string, actorIsAgent bool) error { + var assigned, creator sql.NullString + if err := tx.QueryRow(s.q(`SELECT assigned_user_id, created_by_user_id FROM items WHERE id = ?`), c.ItemID). + Scan(&assigned, &creator); err != nil { + return fmt.Errorf("read comment notification recipients: %w", err) + } + // The parent's author is added first, so a person who is also the assignee + // keeps the "reply" and gets no second row. + var to commentRecipients + if c.ParentID != "" { + var parentAuthor sql.NullString + err := tx.QueryRow(s.q(`SELECT user_id FROM comments WHERE id = ?`), c.ParentID).Scan(&parentAuthor) + if err != nil && !errors.Is(err, sql.ErrNoRows) { + return fmt.Errorf("read parent comment author: %w", err) + } + to.add(parentAuthor.String, notifyReply, actorUserID) + } + to.add(assigned.String, notifyComment, actorUserID) + to.add(creator.String, notifyComment, actorUserID) + facts := make([]notifyFact, 0, len(to.order)) + for _, u := range to.order { + facts = append(facts, notifyFact{ + UserID: u, WorkspaceID: c.WorkspaceID, ItemID: c.ItemID, CommentID: c.ID, Kind: to.kinds[u], + ActorUserID: actorUserID, ActorIsAgent: actorIsAgent, + DedupeKey: "comment:" + c.ID, + }) + } + return s.enqueueNotificationsTx(tx, facts) +} + +// enqueueReminderNotificationTx tells the item's assignee, or its creator when +// nobody holds it, that a reminder fired. A reminder has no owner column, and +// the key carries fired_at so a re-armed reminder that fires again is a new +// notification. There is no actor: the clock fired it. +func (s *Store) enqueueReminderNotificationTx(tx *sql.Tx, r *models.Reminder, item *models.Item) error { + to := "" + if item.AssignedUserID != nil { + to = *item.AssignedUserID + } + if to == "" { + var creator sql.NullString + if err := tx.QueryRow(s.q(`SELECT created_by_user_id FROM items WHERE id = ?`), item.ID).Scan(&creator); err != nil { + return fmt.Errorf("read reminder notification recipient: %w", err) + } + to = creator.String + } + fired := "" + if r.FiredAt != nil { + fired = *r.FiredAt + } + return s.enqueueNotificationsTx(tx, []notifyFact{{ + UserID: to, WorkspaceID: r.WorkspaceID, ItemID: r.ItemID, Kind: notifyReminder, + DedupeKey: "reminder:" + r.ID + ":" + fired, + }}) +} diff --git a/internal/store/notifications_test.go b/internal/store/notifications_test.go new file mode 100644 index 000000000..f8e2d7ea2 --- /dev/null +++ b/internal/store/notifications_test.go @@ -0,0 +1,311 @@ +package store + +import ( + "context" + "testing" + "time" + + "github.com/PerpetualSoftware/pad/internal/models" +) + +func (f notifyFixture) kinds(t *testing.T, userID string) []string { + t.Helper() + rows, err := f.s.db.Query(f.s.q(`SELECT kind FROM user_notifications WHERE user_id = ? ORDER BY kind`), userID) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + var out []string + for rows.Next() { + var k string + if err := rows.Scan(&k); err != nil { + t.Fatal(err) + } + out = append(out, k) + } + return out +} + +func wantKinds(t *testing.T, what string, got []string, want ...string) { + t.Helper() + if len(got) != len(want) { + t.Fatalf("%s: kinds = %v, want %v", what, got, want) + } + for i := range want { + if got[i] != want[i] { + t.Fatalf("%s: kinds = %v, want %v", what, got, want) + } + } +} + +// Assigning to someone else enqueues for them; assigning yourself does not. +// Kills: dropping the UpdateItem hook (no row); dropping the `!= actor` check +// (a self row appears). +func TestNotify_AssignByOtherEnqueues_SelfAssignDoesNot(t *testing.T) { + t.Parallel() + f := newNotifyFixture(t) + a, b := notifyUser(t, f.s, "a@example.com"), notifyUser(t, f.s, "b@example.com") + f.join(t, a, b) + addDevice(t, f.s, a.ID) + addDevice(t, f.s, b.ID) + + if _, err := f.s.UpdateItem(f.item.ID, models.ItemUpdate{AssignedUserID: &b.ID, ActorUserID: a.ID}); err != nil { + t.Fatal(err) + } + wantKinds(t, "A assigns B", f.kinds(t, b.ID), "assigned") + + item2 := createTestItem(t, f.s, f.ws.ID, f.coll.ID, "Second", "") + if _, err := f.s.UpdateItem(item2.ID, models.ItemUpdate{AssignedUserID: &a.ID, ActorUserID: a.ID}); err != nil { + t.Fatal(err) + } + wantKinds(t, "A assigns A", f.kinds(t, a.ID)) + + // An empty actor (a system writer) is not "self". + item3 := createTestItem(t, f.s, f.ws.ID, f.coll.ID, "Third", "") + if _, err := f.s.UpdateItem(item3.ID, models.ItemUpdate{AssignedUserID: &a.ID}); err != nil { + t.Fatal(err) + } + wantKinds(t, "system assigns A", f.kinds(t, a.ID), "assigned") +} + +// Kills: dropping the createItemTxWithID hook. +func TestNotify_CreateWithAssigneeEnqueues(t *testing.T) { + t.Parallel() + f := newNotifyFixture(t) + a, b := notifyUser(t, f.s, "a@example.com"), notifyUser(t, f.s, "b@example.com") + f.join(t, a, b) + addDevice(t, f.s, a.ID) + addDevice(t, f.s, b.ID) + + if _, err := f.s.CreateItem(f.ws.ID, f.coll.ID, models.ItemCreate{ + Title: "Born assigned", Fields: `{"status":"open"}`, AssignedUserID: &b.ID, ActorUserID: a.ID, + }); err != nil { + t.Fatal(err) + } + wantKinds(t, "create assigned to B", f.kinds(t, b.ID), "assigned") + if _, err := f.s.CreateItem(f.ws.ID, f.coll.ID, models.ItemCreate{ + Title: "Born mine", Fields: `{"status":"open"}`, AssignedUserID: &a.ID, ActorUserID: a.ID, + }); err != nil { + t.Fatal(err) + } + wantKinds(t, "create assigned to self", f.kinds(t, a.ID)) +} + +// A user with no device accumulates nothing. Kills: removing the +// WHERE EXISTS (push_devices) gate. +func TestNotify_NoDeviceNoRow(t *testing.T) { + t.Parallel() + f := newNotifyFixture(t) + a, b := notifyUser(t, f.s, "a@example.com"), notifyUser(t, f.s, "b@example.com") + f.join(t, a, b) + if _, err := f.s.UpdateItem(f.item.ID, models.ItemUpdate{AssignedUserID: &b.ID, ActorUserID: a.ID}); err != nil { + t.Fatal(err) + } + if n := f.s.countRows(t, `SELECT COUNT(*) FROM user_notifications`); n != 0 { + t.Fatalf("%d rows for a user with no device", n) + } +} + +// The row commits with the write or not at all. Kills: writing the row on a +// connection outside the caller's transaction. +func TestNotify_RollbackDropsRow(t *testing.T) { + t.Parallel() + f := newNotifyFixture(t) + a, b := notifyUser(t, f.s, "a@example.com"), notifyUser(t, f.s, "b@example.com") + f.join(t, a, b) + addDevice(t, f.s, b.ID) + tx, err := f.s.db.Begin() + if err != nil { + t.Fatal(err) + } + if err := f.s.enqueueAssignedNotificationTx(tx, f.item, b.ID, a.ID, false); err != nil { + t.Fatal(err) + } + if err := tx.Rollback(); err != nil { + t.Fatal(err) + } + if n := f.s.countRows(t, `SELECT COUNT(*) FROM user_notifications`); n != 0 { + t.Fatalf("%d rows survived a rollback", n) + } +} + +func TestNotify_CommentRecipients(t *testing.T) { + t.Parallel() + f := newNotifyFixture(t) + // creator C, assignee D, parent author E, comment author A. + c, d, e, a := notifyUser(t, f.s, "c@example.com"), notifyUser(t, f.s, "d@example.com"), + notifyUser(t, f.s, "e@example.com"), notifyUser(t, f.s, "a@example.com") + f.s.mustExec(t, `UPDATE items SET created_by_user_id = ?, assigned_user_id = ? WHERE id = ?`, c.ID, d.ID, f.item.ID) + for _, u := range []*models.User{c, d, e, a} { + addDevice(t, f.s, u.ID) + } + parent, err := f.s.CreateComment(f.ws.ID, f.item.ID, e.ID, models.CommentCreate{Body: "first"}) + if err != nil { + t.Fatal(err) + } + // E's own top-level comment notified C and D, not E. + wantKinds(t, "E top-level, C", f.kinds(t, c.ID), "comment") + wantKinds(t, "E top-level, E", f.kinds(t, e.ID)) + f.s.mustExec(t, `DELETE FROM user_notifications`) + + if _, err := f.s.CreateComment(f.ws.ID, f.item.ID, a.ID, models.CommentCreate{Body: "re", ParentID: parent.ID}); err != nil { + t.Fatal(err) + } + wantKinds(t, "reply, creator", f.kinds(t, c.ID), "comment") + wantKinds(t, "reply, assignee", f.kinds(t, d.ID), "comment") + wantKinds(t, "reply, parent author", f.kinds(t, e.ID), "reply") + wantKinds(t, "reply, author", f.kinds(t, a.ID)) + + // The assignee commenting on their own item is told nothing. + f.s.mustExec(t, `DELETE FROM user_notifications`) + if _, err := f.s.CreateComment(f.ws.ID, f.item.ID, d.ID, models.CommentCreate{Body: "mine"}); err != nil { + t.Fatal(err) + } + wantKinds(t, "assignee's own comment", f.kinds(t, d.ID)) + + // A person who is both the parent's author and the assignee gets one + // "reply", not a reply and a comment. + f.s.mustExec(t, `DELETE FROM user_notifications`) + f.s.mustExec(t, `UPDATE items SET assigned_user_id = ? WHERE id = ?`, e.ID, f.item.ID) + if _, err := f.s.CreateComment(f.ws.ID, f.item.ID, a.ID, models.CommentCreate{Body: "again", ParentID: parent.ID}); err != nil { + t.Fatal(err) + } + wantKinds(t, "reply, parent author who is assignee", f.kinds(t, e.ID), "reply") +} + +// The fenced (app) comment path enqueues too. Kills: dropping the fenced hook. +func TestNotify_FencedCommentEnqueues(t *testing.T) { + t.Parallel() + ff := newFenceFixture(t) + s := ff.s + owner, actor := notifyUser(t, s, "owner@example.com"), notifyUser(t, s, "actor@example.com") + item := createTestItem(t, s, ff.ws.ID, ff.companion.ID, "Ticket", "") + s.mustExec(t, `UPDATE items SET assigned_user_id = ? WHERE id = ?`, owner.ID, item.ID) + addDevice(t, s, owner.ID) + addDevice(t, s, actor.ID) + + ftx, err := s.BeginFenced(context.Background(), ff.spec()) + if err != nil { + t.Fatal(err) + } + defer ftx.Rollback() + if _, err := ftx.CreateComment(FencedCommentCreate{ItemID: item.ID, Body: "from the app", Actor: FencedActor{Kind: "agent", UserID: actor.ID}}); err != nil { + t.Fatal(err) + } + if err := ftx.Commit(); err != nil { + t.Fatal(err) + } + f := notifyFixture{s: s} + wantKinds(t, "fenced comment, assignee", f.kinds(t, owner.ID), "comment") + wantKinds(t, "fenced comment, author", f.kinds(t, actor.ID)) + if n := s.countRows(t, `SELECT COUNT(*) FROM user_notifications WHERE actor_is_agent`+agentTrue(s)); n != 1 { + t.Fatalf("agent-authored rows = %d, want 1", n) + } +} + +func agentTrue(s *Store) string { + if s.dialect.Driver() == DriverPostgres { + return " = TRUE" + } + return " = 1" +} + +// Kills: firing without the hook; inverting assignee-else-creator; dropping +// fired_at from the key. +func TestNotify_ReminderRecipient(t *testing.T) { + t.Parallel() + f := newNotifyFixture(t) + creator, assignee := notifyUser(t, f.s, "creator@example.com"), notifyUser(t, f.s, "assignee@example.com") + addDevice(t, f.s, creator.ID) + addDevice(t, f.s, assignee.ID) + f.s.mustExec(t, `UPDATE items SET created_by_user_id = ? WHERE id = ?`, creator.ID, f.item.ID) + + id := armReminder(t, f.s, f.ws.ID, f.item.ID, past) + if _, err := f.s.FireDueReminders(nowTS(), 0); err != nil { + t.Fatal(err) + } + wantKinds(t, "unassigned, creator", f.kinds(t, creator.ID), "reminder") + wantKinds(t, "unassigned, assignee", f.kinds(t, assignee.ID)) + + refire := func() { + t.Helper() + if _, err := f.s.RearmReminder(f.ws.ID, id, past); err != nil { + t.Fatal(err) + } + time.Sleep(1100 * time.Millisecond) // fired_at is whole seconds + if _, err := f.s.FireDueReminders(nowTS(), 0); err != nil { + t.Fatal(err) + } + } + // A re-armed reminder that fires again is a new notification. + refire() + wantKinds(t, "re-fired, creator", f.kinds(t, creator.ID), "reminder", "reminder") + + f.s.mustExec(t, `UPDATE items SET assigned_user_id = ? WHERE id = ?`, assignee.ID, f.item.ID) + refire() + wantKinds(t, "assigned, assignee", f.kinds(t, assignee.ID), "reminder") + wantKinds(t, "assigned, creator unchanged", f.kinds(t, creator.ID), "reminder", "reminder") +} + +// Kills: removing ON CONFLICT (an error) or the unique index. +func TestNotify_DedupeKeyIdempotent(t *testing.T) { + t.Parallel() + f := newNotifyFixture(t) + a, b := notifyUser(t, f.s, "a@example.com"), notifyUser(t, f.s, "b@example.com") + f.join(t, a, b) + addDevice(t, f.s, b.ID) + for i := 0; i < 2; i++ { + tx, err := f.s.db.Begin() + if err != nil { + t.Fatal(err) + } + if err := f.s.enqueueAssignedNotificationTx(tx, f.item, b.ID, a.ID, false); err != nil { + t.Fatal(err) + } + if err := tx.Commit(); err != nil { + t.Fatal(err) + } + } + wantKinds(t, "same fact twice", f.kinds(t, b.ID), "assigned") +} + +// A claim keeps the user row for a new person, so the old holder's devices, +// queued pushes and settings must not follow. The devices also die with the +// sessions the claim deletes (FK cascade), so the test asserts the outcome +// (no rows after the claim) whichever line does the work; the explicit +// push_devices delete is a known mutation survivor for that reason. +// Kills: dropping the queue delete, dropping the settings delete. +func TestNotify_ClaimEmptiesDevicesQueueAndSettings(t *testing.T) { + t.Parallel() + s := testStore(t) + ws := createTestWorkspace(t, s, "Claim") + coll := createTestCollection(t, s, ws.ID, "Tasks") + item := createTestItem(t, s, ws.ID, coll.ID, "Subject", "") + squatter := createUnverifiedUser(t, s, "squat@example.com") + other := notifyUser(t, s, "other@example.com") + for _, u := range []*models.User{squatter, other} { + if err := s.AddWorkspaceMember(ws.ID, u.ID, "editor"); err != nil { + t.Fatal(err) + } + } + addDevice(t, s, squatter.ID) + if err := s.PutNotificationSettings(squatter.ID, models.NotificationSettings{NotifyAssigned: true}); err != nil { + t.Fatal(err) + } + if _, err := s.UpdateItem(item.ID, models.ItemUpdate{AssignedUserID: &squatter.ID, ActorUserID: other.ID}); err != nil { + t.Fatal(err) + } + for _, tbl := range []string{"push_devices", "user_notifications", "user_notification_settings"} { + if n := s.countRows(t, `SELECT COUNT(*) FROM `+tbl+` WHERE user_id = ?`, squatter.ID); n != 1 { + t.Fatalf("fixture: %s has %d rows for the user, want 1", tbl, n) + } + } + if _, err := s.ClaimAccountByProvider(squatter.ID, "google", "", ""); err != nil { + t.Fatalf("claim: %v", err) + } + for _, tbl := range []string{"push_devices", "user_notifications", "user_notification_settings"} { + if n := s.countRows(t, `SELECT COUNT(*) FROM `+tbl+` WHERE user_id = ?`, squatter.ID); n != 0 { + t.Errorf("%s has %d rows for the user after a claim, want 0", tbl, n) + } + } +} diff --git a/internal/store/nulcolumns.go b/internal/store/nulcolumns.go index eae8bd7c2..864b22680 100644 --- a/internal/store/nulcolumns.go +++ b/internal/store/nulcolumns.go @@ -436,11 +436,11 @@ func NULProtectedColumns() []nulColumn { // class entirely. // // The push_devices and user_notifications reasons (TASK-3937) state the -// CONTRACT their writers must keep, and none of those writers exists yet: the -// device register endpoint is TASK-3938, the enqueue is TASK-3937 part B, and -// the dispatcher's claim and send columns are TASK-3939. Each names the unit -// that owes it. user_notification_settings is written today, by -// PutNotificationSettings. +// CONTRACT their writers must keep, and each names the unit that owns the +// writer: the enqueue (TASK-3937 part B, notifications.go) writes the queue +// columns tagged part B; the device register endpoint (TASK-3938) and the +// dispatcher's claim and send columns (TASK-3939) do not exist yet. +// user_notification_settings is written by PutNotificationSettings. var nulExcluded = map[string]string{ "push_devices.id": "server-minted device UUID (TASK-3938, migration 155)", "push_devices.user_id": "the authenticated user's server-minted UUID (TASK-3938)", diff --git a/internal/store/reminders.go b/internal/store/reminders.go index 19d406d96..40b070d6a 100644 --- a/internal/store/reminders.go +++ b/internal/store/reminders.go @@ -747,6 +747,11 @@ func (s *Store) fireOneReminder(id, nowTS string) (*models.Reminder, error) { if err := s.emitReminderEventTx(tx, r, item); err != nil { return nil, fmt.Errorf("emit reminder event %s: %w", id, err) } + // TASK-3937: after the arbiter's UPDATE, so a pass that lost the row + // writes nothing. + if err := s.enqueueReminderNotificationTx(tx, r, item); err != nil { + return nil, fmt.Errorf("enqueue reminder notification %s: %w", id, err) + } if err := tx.Commit(); err != nil { return nil, fmt.Errorf("fire reminder %s: %w", id, err)