mirror of
https://github.com/PerpetualSoftware/pad.git
synced 2026-10-26 14:30:13 +00:00
feat(store): enqueue push notifications at four store sites (TASK-3937 part B) (#2268)
* feat(store): enqueue push notifications at four store sites (TASK-3937 part B)
notifications.go writes queue rows on the caller's transaction, only for a
user with a registered device: item assignment (create and update),
comment and reply, fenced (app) comment, and reminder fire. The account
claim also deletes the old holder's devices, queue and settings.
Claude-Session: https://claude.ai/code/session_01JRXHP9ZVfEN3YEuzmucxP3
* fix(store): comment recipients collected without a closure so the appstore boundary walk resolves them (TASK-3937 part B)
Claude-Session: https://claude.ai/code/session_01JRXHP9ZVfEN3YEuzmucxP3
* docs(store): with the enqueue built, the Go comments say it exists (TASK-3937 part B)
A (07dc06285) future-tensed every sentence about the enqueue, because
part B had not merged. In part B's tree they read as false, so the
current description lives in Go comments:
- notifications.go's header: rows only for a user with a registered
device, so nothing is written until TASK-3938; idempotent per
(user, dedupe_key).
- migrated.go's push note.
- the settings model.
- nulExcluded's contract header.
Migration 155 is NOT edited: merged migrations are immutable, comments
included, so its owed-tense prose stays as history, and the
notifications.go header says which description is current.
Claude-Session: https://claude.ai/code/session_01JRXHP9ZVfEN3YEuzmucxP3
* docs(store): the enqueue says recipients are gated at send, and mentions are TASK-3946's (TASK-3937 part B r1)
Ruled on part B r1. notifications.go's header now says recipients are
not access-checked at enqueue: a removed member stays created_by and
possibly assignee, and the dispatcher (TASK-3939) is to re-check
membership and visibility at send. It also says the 'mention' kind is
enqueued by TASK-3946, from comment_mentions, not here.
Claude-Session: https://claude.ai/code/session_01JRXHP9ZVfEN3YEuzmucxP3
This commit is contained in:
@@ -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",
|
||||
|
||||
@@ -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"`
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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,
|
||||
}})
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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)",
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user