diff --git a/internal/notify/gateway.go b/internal/notify/gateway.go index e13c8a931..b9ffa408d 100644 --- a/internal/notify/gateway.go +++ b/internal/notify/gateway.go @@ -63,6 +63,11 @@ type Gateway struct { HTTP *http.Client } +// SendTimeout is the HTTP client's whole-request timeout, above the gateway's +// own 20s deadline. The dispatcher's claim lease must exceed it by a margin +// (internal/server/push_dispatch.go). +const SendTimeout = 25 * time.Second + // NewGateway returns a Gateway with a client that does not follow redirects: // the secret rides in a header, and a redirect would carry it to wherever the // answer points. @@ -70,7 +75,7 @@ func NewGateway(url, secret string) *Gateway { return &Gateway{ URL: url, Secret: secret, HTTP: &http.Client{ - Timeout: 25 * time.Second, // above the gateway's own 20s deadline + Timeout: SendTimeout, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }, diff --git a/internal/server/handlers_watch_access.go b/internal/server/handlers_watch_access.go index 689108218..b820fbadf 100644 --- a/internal/server/handlers_watch_access.go +++ b/internal/server/handlers_watch_access.go @@ -200,15 +200,25 @@ func (v watchAccessVisibility) allows(collectionID, itemID string) bool { // resolver computeSSEVisibility (handlers_events.go) has the same shape // and was NOT changed, because no counting caller reads it. func (s *Server) computeWatchAccessVisibility(bearerAuth bool, user *models.User, workspaceID string) (watchAccessVisibility, error) { + vis, msg, err := s.resolveWatchAccessVisibility(bearerAuth, user, workspaceID) + if err != nil { + slog.Warn(msg, "workspace_id", workspaceID, "user_id", user.ID, "error", err) + } + return vis, err +} + +// resolveWatchAccessVisibility is computeWatchAccessVisibility without the +// logging: on error it returns the log message for the failing lookup instead +// of writing it, so the push dispatcher (TASK-3939 B1), which must never log a +// workspace id, a user id or a store error, can call it. +func (s *Server) resolveWatchAccessVisibility(bearerAuth bool, user *models.User, workspaceID string) (watchAccessVisibility, string, error) { if user.Role == "admin" && !bearerAuth { - return watchAccessVisibility{fullAccess: true}, nil + return watchAccessVisibility{fullAccess: true}, "", nil } member, err := s.store.GetWorkspaceMember(workspaceID, user.ID) if err != nil { - slog.Warn("watch-access: failed to resolve workspace membership, denying all in workspace", - "workspace_id", workspaceID, "user_id", user.ID, "error", err) - return watchAccessVisibility{}, err + return watchAccessVisibility{}, "watch-access: failed to resolve workspace membership, denying all in workspace", err } if member != nil && member.CollectionAccess != "specific" { // "all" access (or the legacy "" default) — unrestricted. Note @@ -217,7 +227,7 @@ func (s *Server) computeWatchAccessVisibility(bearerAuth bool, user *models.User // member of this specific workspace legitimately sees it in // full; the gate above only denies the platform-wide, not-a- // member-here bypass. - return watchAccessVisibility{fullAccess: true}, nil + return watchAccessVisibility{fullAccess: true}, "", nil } // From here: either a "specific"-access member, a non-member guest, @@ -227,9 +237,7 @@ func (s *Server) computeWatchAccessVisibility(bearerAuth bool, user *models.User // VisibleCollectionIDs. fullCollIDs, grantedItemIDs, gerr := s.store.GuestVisibleResources(workspaceID, user.ID) if gerr != nil { - slog.Warn("watch-access: failed to resolve grants, denying all in workspace", - "workspace_id", workspaceID, "user_id", user.ID, "error", gerr) - return watchAccessVisibility{}, gerr + return watchAccessVisibility{}, "watch-access: failed to resolve grants, denying all in workspace", gerr } fullCollSet := make(map[string]bool, len(fullCollIDs)) @@ -244,9 +252,7 @@ func (s *Server) computeWatchAccessVisibility(bearerAuth bool, user *models.User // collection_grants from GuestVisibleResources above. memberColls, mcErr := s.store.GetMemberCollectionAccess(workspaceID, user.ID) if mcErr != nil { - slog.Warn("watch-access: failed to resolve member collection access, denying all in workspace", - "workspace_id", workspaceID, "user_id", user.ID, "error", mcErr) - return watchAccessVisibility{}, mcErr + return watchAccessVisibility{}, "watch-access: failed to resolve member collection access, denying all in workspace", mcErr } for _, id := range memberColls { fullCollSet[id] = true @@ -261,5 +267,5 @@ func (s *Server) computeWatchAccessVisibility(bearerAuth bool, user *models.User return watchAccessVisibility{ visibleCollIDs: fullCollSet, grantedItemIDs: grantedSet, - }, nil + }, "", nil } diff --git a/internal/server/push_dispatch.go b/internal/server/push_dispatch.go new file mode 100644 index 000000000..132487988 --- /dev/null +++ b/internal/server/push_dispatch.go @@ -0,0 +1,453 @@ +package server + +import ( + "context" + "errors" + "log/slog" + "sort" + "sync" + "time" + + "github.com/google/uuid" + + "github.com/PerpetualSoftware/pad/internal/notify" + "github.com/PerpetualSoftware/pad/internal/store" +) + +// The push dispatcher (TASK-3939 part B1): claims settled user_notifications +// rows, runs every privacy gate (push_gates.go) on each, groups the survivors +// per (user, item), sends one content-free push per live device, and settles +// the rows. B1 is the decide-and-send core; StartPushDispatch is NOT called +// from cmd/pad yet. +// +// LEASE INVARIANT (r6-Q1): lease >= sendTimeout + leaseMargin. A claim is +// stamped with second precision (up to 1s behind the real clock), and a Send +// can take sendTimeout (notify.SendTimeout), so a lease shorter than that lets +// another instance reclaim a row mid-send and push it twice. It is enforced +// three ways: the setter refuses a violating config, all gateway I/O runs +// under a deadline of start+lease-leaseMargin, and a chunk only starts when +// sendTimeout is still left before that deadline (otherwise its rows are +// unclaimed, no attempt burned). +// +// Logging: fixed strings and counts only. A store error can carry a row id or +// a value, so it is never logged here. +const ( + defaultPushInterval = 5 * time.Second + defaultPushSettle = 10 * time.Second + defaultPushLease = 60 * time.Second + defaultPushLeaseMargin = 15 * time.Second +) + +var errPushLeaseTooShort = errors.New("push dispatch: lease must be at least sendTimeout + leaseMargin") + +// pushSender is the gateway as the dispatcher sees it; *notify.Gateway +// satisfies it, and tests substitute a fake. +type pushSender interface { + Send(ctx context.Context, msgs []notify.Message) ([]notify.Result, error) +} + +type pushDispatchConfig struct { + mu sync.Mutex + interval time.Duration + settle time.Duration + lease time.Duration + leaseMargin time.Duration + sendTimeout time.Duration + sender pushSender // test seam; nil = the server's gateway + now func() time.Time // test seam; nil = time.Now + stop chan struct{} + running bool + instanceID string + // tick, when non-nil, replaces the interval ticker so a test drives passes. + tick <-chan time.Time +} + +type pushDispatchSettings struct { + interval, settle, lease, leaseMargin, sendTimeout time.Duration + instanceID string +} + +// SetPushDispatchConfig overrides the timings; zero keeps the current value. +// A combination violating the lease invariant is refused whole and nothing +// changes. +func (s *Server) SetPushDispatchConfig(interval, settle, lease, leaseMargin, sendTimeout time.Duration) error { + cur := s.resolvePushDispatchSettings() + pick := func(v, d time.Duration) time.Duration { + if v > 0 { + return v + } + return d + } + next := pushDispatchSettings{ + interval: pick(interval, cur.interval), settle: pick(settle, cur.settle), + lease: pick(lease, cur.lease), leaseMargin: pick(leaseMargin, cur.leaseMargin), + sendTimeout: pick(sendTimeout, cur.sendTimeout), + } + if next.lease < next.sendTimeout+next.leaseMargin { + slog.Warn("push dispatch: config refused", "reason", "lease shorter than send timeout plus margin") + return errPushLeaseTooShort + } + d := &s.pushDispatch + d.mu.Lock() + defer d.mu.Unlock() + d.interval, d.settle, d.lease, d.leaseMargin, d.sendTimeout = next.interval, next.settle, next.lease, next.leaseMargin, next.sendTimeout + return nil +} + +// SetPushSender substitutes the gateway client (a test seam). +func (s *Server) SetPushSender(p pushSender) { + s.pushDispatch.mu.Lock() + defer s.pushDispatch.mu.Unlock() + s.pushDispatch.sender = p +} + +// resolvePushDispatchSettings fills every unset knob and stores the result, +// so a tick reached directly never sees a zero (the outbox drain's lesson). +func (s *Server) resolvePushDispatchSettings() pushDispatchSettings { + d := &s.pushDispatch + d.mu.Lock() + defer d.mu.Unlock() + if d.interval <= 0 { + d.interval = defaultPushInterval + } + if d.settle <= 0 { + d.settle = defaultPushSettle + } + if d.sendTimeout <= 0 { + d.sendTimeout = notify.SendTimeout + } + if d.leaseMargin <= 0 { + d.leaseMargin = defaultPushLeaseMargin + } + if d.lease <= 0 { + d.lease = defaultPushLease + } + if d.instanceID == "" { + d.instanceID = uuid.NewString() + } + return pushDispatchSettings{d.interval, d.settle, d.lease, d.leaseMargin, d.sendTimeout, d.instanceID} +} + +func (s *Server) pushNow() time.Time { + s.pushDispatch.mu.Lock() + now := s.pushDispatch.now + s.pushDispatch.mu.Unlock() + if now == nil { + return time.Now() + } + return now() +} + +func (s *Server) pushSenderOrNil() pushSender { + s.pushDispatch.mu.Lock() + defer s.pushDispatch.mu.Unlock() + if s.pushDispatch.sender != nil { + return s.pushDispatch.sender + } + if s.pushGateway != nil { + return s.pushGateway + } + return nil +} + +// StartPushDispatch starts the dispatch loop. Idempotent; tracked on s.bg. +func (s *Server) StartPushDispatch() { + cfg := s.resolvePushDispatchSettings() + d := &s.pushDispatch + d.mu.Lock() + if d.running { + d.mu.Unlock() + return + } + d.stop = make(chan struct{}) + d.running = true + stop, tick := d.stop, d.tick + d.mu.Unlock() + + s.bg.Add(1) + go func() { + defer s.bg.Done() + defer s.recoverSweeper("push-dispatch") + // A stop cancels the tick in flight, which unclaims what it holds. + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go func() { + select { + case <-stop: + cancel() + case <-ctx.Done(): + } + }() + var c <-chan time.Time + if tick != nil { + c = tick + } else { + t := time.NewTicker(cfg.interval) + defer t.Stop() + c = t.C + } + for { + select { + case <-stop: + return + case <-c: + s.runPushDispatchTick(ctx) + } + } + }() +} + +// stopPushDispatch signals the loop to exit. Safe when it never started. +func (s *Server) stopPushDispatch() { + d := &s.pushDispatch + d.mu.Lock() + defer d.mu.Unlock() + if !d.running { + return + } + close(d.stop) + d.running = false +} + +// pushUnit is the rows of one (user, item) group after gating, which become +// one push per live device. nid and kind are the newest row's. +type pushUnit struct { + rowIDs []string + nid string + kind string + devices []store.PushDevice +} + +// kindPriority breaks a created_at tie between rows of one group, so the kind +// of a mixed group is deterministic (ids are random). +var kindPriority = map[string]int{ + notify.KindReminder: 0, notify.KindComment: 1, notify.KindAssigned: 2, notify.KindReply: 3, notify.KindMention: 4, +} + +// runPushDispatchTick is one pass. ctx is the shutdown signal. +func (s *Server) runPushDispatchTick(ctx context.Context) { + cfg := s.resolvePushDispatchSettings() + sender := s.pushSenderOrNil() + if sender == nil { + return + } + start := s.pushNow() + // Fresh per tick and never kept: the store appends its own nonce, and + // every settle below uses Claim.Token. + token := "pd-" + cfg.instanceID + "-" + uuid.NewString() + claim, err := s.store.ClaimPendingNotifications(token, notify.MaxBatch, start.Add(-cfg.settle), start.Add(-cfg.lease)) + if err != nil { + slog.Error("push dispatch: claim failed", "op", "claim") + return + } + if len(claim.Rows) == 0 { + return + } + deadline := start.Add(cfg.lease - cfg.leaseMargin) + + cache := newPushGateCache() + skips := map[string][]string{} + var unclaim []string + groups := map[string][]store.Notification{} + var order []string + for _, n := range claim.Rows { + _, reason, gerr := s.notificationAccess(cache, n) + if gerr == nil && reason == "" { + reason, gerr = s.preferenceGates(cache, n, start) + } + switch { + case gerr != nil: + // Not a denial: never send, never record a skip, burn no attempt. + unclaim = append(unclaim, n.ID) + case reason != "": + skips[reason] = append(skips[reason], n.ID) + default: + key := n.UserID + "/" + n.ItemID + if _, seen := groups[key]; !seen { + order = append(order, key) + } + groups[key] = append(groups[key], n) + } + } + s.settleSkips(claim.Token, skips) + s.unclaim(claim.Token, unclaim) + + var units []pushUnit + for _, key := range order { + rows := groups[key] + sort.SliceStable(rows, func(i, j int) bool { + if rows[i].CreatedAt != rows[j].CreatedAt { + return rows[i].CreatedAt < rows[j].CreatedAt + } + if pi, pj := kindPriority[rows[i].Kind], kindPriority[rows[j].Kind]; pi != pj { + return pi < pj + } + return rows[i].ID < rows[j].ID + }) + newest := rows[len(rows)-1] + u := pushUnit{nid: newest.ID, kind: newest.Kind, devices: cache.devices[newest.UserID]} + for _, r := range rows { + u.rowIDs = append(u.rowIDs, r.ID) + } + units = append(units, u) + } + + // Chunks hold whole units, so a unit's outcome is decided in one place. + for len(units) > 0 { + var chunk []pushUnit + msgs := 0 + for len(units) > 0 { + n := min(len(units[0].devices), notify.MaxBatch) + if len(chunk) > 0 && msgs+n > notify.MaxBatch { + break + } + chunk = append(chunk, units[0]) + units = units[1:] + msgs += n + } + if ctx.Err() != nil || deadline.Sub(s.pushNow()) < cfg.sendTimeout { + s.unclaim(claim.Token, unitRowIDs(append(chunk, units...))) + return + } + s.sendChunk(ctx, sender, claim.Token, chunk, deadline) + } +} + +func unitRowIDs(units []pushUnit) []string { + var ids []string + for _, u := range units { + ids = append(ids, u.rowIDs...) + } + return ids +} + +// preferenceGates is G5/G6 and G7/G8 with their cached lookups. +func (s *Server) preferenceGates(c *pushGateCache, n store.Notification, at time.Time) (string, error) { + set, ok := c.settings[n.UserID] + if !ok { + var err error + if set, err = s.store.GetNotificationSettings(n.UserID); err != nil { + return "", err + } + c.settings[n.UserID] = set + } + if reason := kindEnabled(set, n); reason != "" { + return reason, nil + } + devs, ok := c.devices[n.UserID] + if !ok { + var err error + if devs, err = s.store.LivePushDevices(n.UserID, at); err != nil { + return "", err + } + c.devices[n.UserID] = devs + } + return deviceGate(n, devs), nil +} + +func (s *Server) settleSkips(token string, skips map[string][]string) { + for reason, ids := range skips { + if _, err := s.store.MarkNotificationsSkipped(token, ids, reason); err != nil { + slog.Warn("push dispatch: settle failed", "op", "skip") + } + } +} + +func (s *Server) unclaim(token string, ids []string) { + if len(ids) == 0 { + return + } + if _, err := s.store.UnclaimNotifications(token, ids); err != nil { + slog.Warn("push dispatch: settle failed", "op", "unclaim") + } +} + +// sendChunk sends one chunk of whole units and settles it immediately. B1: a +// unit any of whose devices was sent is sent; every other outcome releases +// its rows (attempts + 1). +func (s *Server) sendChunk(ctx context.Context, sender pushSender, token string, chunk []pushUnit, deadline time.Time) { + // Drop rows no longer held (reclaimed, settled or deleted meanwhile). + all := unitRowIDs(chunk) + held, err := s.store.HeldNotifications(token, all) + if err != nil { + slog.Warn("push dispatch: settle failed", "op", "held") + s.unclaim(token, all) + return + } + heldSet := make(map[string]bool, len(held)) + for _, id := range held { + heldSet[id] = true + } + var msgs []notify.Message + owner := map[string]int{} // message id -> unit index + var live []pushUnit + var noMsg []string // units none of whose devices could be given a message + for _, u := range chunk { + var ids []string + for _, id := range u.rowIDs { + if heldSet[id] { + ids = append(ids, id) + } + } + if len(ids) == 0 { + continue + } + u.rowIDs = ids + first := len(msgs) + for _, d := range u.devices[:min(len(u.devices), notify.MaxBatch)] { + id := notify.BatchID(len(msgs)) + m, ok := notify.NewMessage(id, d.Platform, d.Token, u.kind, u.nid, d.ConnectionID, 1) + if !ok { + continue + } + owner[id] = len(live) + msgs = append(msgs, m) + } + if len(msgs) == first { + noMsg = append(noMsg, ids...) + continue + } + live = append(live, u) + } + if len(noMsg) > 0 { + s.settleSkips(token, map[string][]string{skipNoDevice: noMsg}) + } + if len(msgs) == 0 { + s.unclaim(token, unitRowIDs(live)) + return + } + + sctx, cancel := context.WithDeadline(ctx, deadline) + results, serr := sender.Send(sctx, msgs) + cancel() + if serr != nil && ctx.Err() != nil { + s.unclaim(token, unitRowIDs(live)) // shutdown: no verdict, no attempt + return + } + sent := make([]bool, len(live)) + if serr == nil { + for _, r := range results { + if i, ok := owner[r.ID]; ok && r.Status == notify.StatusSent { + sent[i] = true + } + } + } + var sentIDs, releaseIDs []string + for i, u := range live { + if sent[i] { + sentIDs = append(sentIDs, u.rowIDs...) + } else { + releaseIDs = append(releaseIDs, u.rowIDs...) + } + } + if len(sentIDs) > 0 { + if _, err := s.store.RecordNotificationsSent(sentIDs); err != nil { + slog.Warn("push dispatch: settle failed", "op", "sent") + } + } + if len(releaseIDs) > 0 { + if _, err := s.store.ReleaseNotifications(token, releaseIDs); err != nil { + slog.Warn("push dispatch: settle failed", "op", "release") + } + } +} diff --git a/internal/server/push_gates.go b/internal/server/push_gates.go new file mode 100644 index 000000000..92c7112ea --- /dev/null +++ b/internal/server/push_gates.go @@ -0,0 +1,195 @@ +package server + +import ( + "errors" + + "github.com/PerpetualSoftware/pad/internal/models" + "github.com/PerpetualSoftware/pad/internal/notify" + "github.com/PerpetualSoftware/pad/internal/store" +) + +// The push dispatcher's privacy gates (TASK-3939 part B1). The enqueue writes +// a row for anyone with a device and checks nothing, so THIS is where a row +// stops being about something its recipient may no longer see. A row that +// fails a gate is skipped with a recorded reason and never reaches the +// gateway. A gate that cannot be evaluated is not a denial: the caller +// unclaims the row, never sends it, and never records a skip. + +// Skip reasons the dispatcher records. +const ( + skipUserDisabled = "user_disabled" + skipNotMember = "not_member" + skipItemGone = "item_gone" + skipWorkspaceGone = "workspace_gone" + skipNotVisible = "not_visible" + skipNoLongerAssign = "no_longer_assigned" + skipCommentGone = "comment_gone" + skipSettingOff = "setting_off" + skipAgentOff = "agent_activity_off" + skipUnknownKind = "unknown_kind" + skipNoDevice = "no_device" + skipPredatesDevice = "predates_device" +) + +// pushGateCache memoises one tick's lookups. Errors are never cached. +type pushGateCache struct { + users map[string]*models.User + members map[string]*models.WorkspaceMember + vis map[string]watchAccessVisibility + live map[string]bool + subjects map[string]*models.Item + comments map[string]*models.Comment + settings map[string]models.NotificationSettings + devices map[string][]store.PushDevice +} + +func newPushGateCache() *pushGateCache { + return &pushGateCache{ + users: map[string]*models.User{}, + members: map[string]*models.WorkspaceMember{}, + vis: map[string]watchAccessVisibility{}, + live: map[string]bool{}, + subjects: map[string]*models.Item{}, + comments: map[string]*models.Comment{}, + settings: map[string]models.NotificationSettings{}, + devices: map[string][]store.PushDevice{}, + } +} + +// notificationAccess is the SHARED gate: may n's recipient still be told +// about n's subject. Order is privacy first, so when several fail the +// recorded reason is the privacy one: user (G4), membership (G1, separate from +// visibility), subject liveness (G3), visibility (G2), then whether the +// reason for the row still holds (G9). reason != "" is a denial; err != nil +// means a gate could not be evaluated. A nil cache is a fresh one. +func (s *Server) notificationAccess(c *pushGateCache, n store.Notification) (item *models.Item, reason string, err error) { + if c == nil { + c = newPushGateCache() + } + user, ok := c.users[n.UserID] + if !ok { + if user, err = s.store.GetUser(n.UserID); err != nil { + return nil, "", err + } + c.users[n.UserID] = user + } + // user == nil guards a user deleted between the claim and this gate + // (user_notifications.user_id cascades on delete, migration 155). + if user == nil || user.IsDisabled() || user.IsApp() { + return nil, skipUserDisabled, nil + } + + mkey := n.WorkspaceID + "/" + n.UserID + member, ok := c.members[mkey] + if !ok { + if member, err = s.store.GetWorkspaceMember(n.WorkspaceID, n.UserID); err != nil { + return nil, "", err + } + c.members[mkey] = member + } + if member == nil { + return nil, skipNotMember, nil + } + + alive, ok := c.live[n.WorkspaceID] + if !ok { + ws, werr := s.store.GetWorkspaceByID(n.WorkspaceID) + if werr != nil { + return nil, "", werr + } + alive = ws != nil + c.live[n.WorkspaceID] = alive + } + if !alive { + return nil, skipWorkspaceGone, nil + } + + item, ok = c.subjects[n.ItemID] + if !ok { + item, err = s.store.NotificationSubject(n.ItemID) + if errors.Is(err, store.ErrNotificationSubjectGone) { + item, err = nil, nil + } + if err != nil { + return nil, "", err + } + c.subjects[n.ItemID] = item + } + if item == nil || item.WorkspaceID != n.WorkspaceID { + return nil, skipItemGone, nil + } + + vis, ok := c.vis[mkey] + if !ok { + // bearerAuth=true: a platform admin's cookie bypass is a property of a + // live browser connection, not of a push to a phone. + if vis, _, err = s.resolveWatchAccessVisibility(true, user, n.WorkspaceID); err != nil { + return nil, "", err + } + c.vis[mkey] = vis + } + if !vis.allows(item.CollectionID, item.ID) { + return nil, skipNotVisible, nil + } + + switch n.Kind { + case notify.KindAssigned: + if item.AssignedUserID == nil || *item.AssignedUserID != n.UserID { + return nil, skipNoLongerAssign, nil + } + case notify.KindComment, notify.KindReply, notify.KindMention: + cm, ok := c.comments[n.CommentID] + if !ok { + if n.CommentID != "" { + if cm, err = s.store.GetComment(n.CommentID); err != nil { + return nil, "", err + } + } + c.comments[n.CommentID] = cm + } + if cm == nil || cm.Deleted || cm.ItemID != n.ItemID || cm.WorkspaceID != n.WorkspaceID { + return nil, skipCommentGone, nil + } + } + return item, "", nil +} + +// kindEnabled applies the recipient's settings (G5) and the agent-activity +// switch (G6) to n. A kind the dispatcher has no text for is skipped, not sent. +func kindEnabled(set models.NotificationSettings, n store.Notification) (reason string) { + var on bool + switch n.Kind { + case notify.KindAssigned: + on = set.NotifyAssigned + case notify.KindComment: + on = set.NotifyComments + case notify.KindReply: + on = set.NotifyReplies + case notify.KindReminder: + on = set.NotifyReminders + case notify.KindMention: + on = set.NotifyMentions + default: + return skipUnknownKind + } + if !set.PushEnabled || !on { + return skipSettingOff + } + if n.ActorIsAgent && !set.IncludeAgentActivity { + return skipAgentOff + } + return "" +} + +// deviceGate is G7/G8: the user's live devices (the newest live registrant +// owns a shared token, an expired or revoked session is no device), and the +// row must not predate the earliest of them. No live device is a skip too. +func deviceGate(n store.Notification, devs []store.PushDevice) string { + if len(devs) == 0 { + return skipNoDevice + } + if n.CreatedAt < devs[0].CreatedAt { + return skipPredatesDevice + } + return "" +} diff --git a/internal/server/server.go b/internal/server/server.go index 898b3245b..0aad4103b 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -427,6 +427,10 @@ type Server struct { // loop via stopOutboxDrain. outboxDrain outboxDrainConfig + // pushDispatch holds the push dispatcher's config and lifecycle + // (TASK-3939). Stop() signals the loop via stopPushDispatch. + pushDispatch pushDispatchConfig + // reminderTick holds the periodic config + lifecycle for the item-reminder // scheduler (IDEA-2641). Mirrors outboxDrain. Configured via // SetReminderTickConfig + started via StartReminderTick; Stop() signals @@ -759,6 +763,7 @@ func (s *Server) Stop() { // SPEC-3 event outbox drain (TASK-2714). Same lifecycle pattern; an // in-flight delivery is tracked on s.bg and awaited below. s.stopOutboxDrain() + s.stopPushDispatch() // Item reminder tick (IDEA-2641). Same lifecycle pattern; an in-flight // pass is tracked on s.bg and awaited below. s.stopReminderTick() diff --git a/internal/server/task3939b_push_dispatch_test.go b/internal/server/task3939b_push_dispatch_test.go new file mode 100644 index 000000000..a09b69565 --- /dev/null +++ b/internal/server/task3939b_push_dispatch_test.go @@ -0,0 +1,445 @@ +package server + +import ( + "bytes" + "context" + "errors" + "fmt" + "log/slog" + "strings" + "sync" + "testing" + "time" + + "github.com/PerpetualSoftware/pad/internal/notify" +) + +// TASK-3939 part B1: the dispatch core. Gate cases live in +// task3939b_push_gates_test.go. + +func TestPushDispatch_SendsOnePerUnitAndSettlesSent(t *testing.T) { + needsBackends(t, func(t *testing.T, srv *Server) { + f := newPdx(t, srv) + f.device(f.bob, "tok-bob-2", 4*time.Hour) + other := f.newItem("Other", &f.bob.ID) + a1, a2 := f.row("comment"), f.row("comment", func(o *rowOpt) { o.age = 3 * time.Minute }) + b := f.row("assigned", func(o *rowOpt) { o.item = other }) + f.tick() + for _, id := range []string{a1, a2, b} { + if st := f.state(id); st.State != "sent" || st.SentAt == "" || st.Claim != "" { + t.Fatalf("row %s = %+v, want sent, stamped, unclaimed", id, st) + } + } + // two units (one per item) x two devices, in one request + if n := f.fake.callCount(); n != 1 { + t.Fatalf("%d Send calls, want 1", n) + } + msgs := f.fake.all() + if len(msgs) != 4 { + t.Fatalf("%d messages, want 2 units x 2 devices", len(msgs)) + } + seen := map[string]bool{} + for _, m := range msgs { + if seen[m.ID] { + t.Fatalf("duplicate message id %s", m.ID) + } + seen[m.ID] = true + } + nids := map[string]int{} + for _, m := range msgs { + nids[m.Data.NID]++ + } + if nids[a1] != 2 || nids[b] != 2 || nids[a2] != 0 { + t.Fatalf("nids = %v: the item's two rows are one push carrying the newest row's id", nids) + } + }) +} + +func TestPushDispatch_NonSentResultsReleaseWithAnAttempt(t *testing.T) { + needsBackends(t, func(t *testing.T, srv *Server) { + f := newPdx(t, srv) + f.fake.fn = func(_ context.Context, msgs []notify.Message) ([]notify.Result, error) { + return allStatus(msgs, notify.StatusRetry), nil + } + id := f.row("comment") + f.tick() + if st := f.state(id); st.State != "pending" || st.Attempts != 1 || st.Claim != "" { + t.Fatalf("after retry: %+v, want pending, 1 attempt, unclaimed", st) + } + f.fake.fn = func(context.Context, []notify.Message) ([]notify.Result, error) { return nil, notify.ErrUnavailable } + f.tick() + if st := f.state(id); st.State != "pending" || st.Attempts != 2 { + t.Fatalf("after a failed request: %+v, want 2 attempts", st) + } + }) +} + +// A unit with one device sent and another retried is sent: re-pushing would +// duplicate to the device that already has it. +func TestPushDispatch_OneDeviceSentIsSent(t *testing.T) { + needsBackends(t, func(t *testing.T, srv *Server) { + f := newPdx(t, srv) + f.device(f.bob, "tok-bob-2", 4*time.Hour) + f.fake.fn = func(_ context.Context, msgs []notify.Message) ([]notify.Result, error) { + out := allStatus(msgs, notify.StatusRetry) + out[0].Status = notify.StatusSent + return out, nil + } + id := f.row("comment") + f.tick() + if st := f.state(id); st.State != "sent" || st.Attempts != 0 { + t.Fatalf("row = %+v, want sent", st) + } + }) +} + +// --- lease invariant (r6-Q1) --- + +func TestPushDispatch_LeaseExceedsSendTimeout(t *testing.T) { + srv := testServer(t) + cfg := srv.resolvePushDispatchSettings() + if cfg.sendTimeout != notify.SendTimeout { + t.Fatalf("send timeout = %v, want notify.SendTimeout", cfg.sendTimeout) + } + if cfg.lease < notify.SendTimeout+cfg.leaseMargin { + t.Fatalf("default lease %v < send timeout %v + margin %v", cfg.lease, notify.SendTimeout, cfg.leaseMargin) + } + if got := notify.NewGateway("http://x.test", "s").HTTP.Timeout; got != notify.SendTimeout { + t.Fatalf("gateway client timeout = %v, want notify.SendTimeout", got) + } + for _, lease := range []time.Duration{20 * time.Second, notify.SendTimeout, cfg.sendTimeout + cfg.leaseMargin - time.Second} { + if err := srv.SetPushDispatchConfig(0, 0, lease, 0, 0); !errors.Is(err, errPushLeaseTooShort) { + t.Fatalf("lease %v: err = %v, want refused", lease, err) + } + } + if got := srv.resolvePushDispatchSettings().lease; got != cfg.lease { + t.Fatalf("a refused config changed the lease to %v", got) + } + if err := srv.SetPushDispatchConfig(0, 0, cfg.sendTimeout+cfg.leaseMargin, 0, 0); err != nil { + t.Fatalf("a lease exactly at the invariant was refused: %v", err) + } + // Raising the send timeout past the lease is the same violation. + if err := srv.SetPushDispatchConfig(0, 0, 0, 0, time.Hour); !errors.Is(err, errPushLeaseTooShort) { + t.Fatalf("send timeout raised past the lease: err = %v", err) + } +} + +// The Send runs under the tick's deadline (start + lease - margin), so it ends +// before the claim lapses; while it runs, and while the tick still holds the +// row after the deadline fired, no other claimer can take the row. The clock +// is the dispatcher's seam: the tick "started" an hour ago, so its deadline is +// already past and the Send's context must already be done. +func TestPushDispatch_SlowSendNotReclaimedMidFlight(t *testing.T) { + needsBackends(t, func(t *testing.T, srv *Server) { + f := newPdx(t, srv) + must(t, srv.SetPushDispatchConfig(0, 0, 3*time.Second, time.Second, time.Second)) + clk := f.clock(time.Now().Add(-time.Hour)) + id := f.row("comment", func(o *rowOpt) { o.age = 3 * time.Hour }) + compete := func(when string) { + c, err := srv.store.ClaimPendingNotifications("rival", 10, time.Now().Add(-10*time.Second), time.Now().Add(-3*time.Second)) + if err != nil || len(c.Rows) != 0 { + t.Errorf("%s: a rival claimed %d rows (err %v)", when, len(c.Rows), err) + } + } + f.fake.fn = func(ctx context.Context, _ []notify.Message) ([]notify.Result, error) { + compete("mid-flight") + dl, ok := ctx.Deadline() + if !ok || !dl.Equal(clk.start.Add(2*time.Second)) { + t.Errorf("Send deadline = %v (set %v), want tick start + lease - margin = %v", dl, ok, clk.start.Add(2*time.Second)) + } + if ctx.Err() == nil { + t.Error("the deadline had passed but the Send context is live: no deadline on the gateway call") + } + compete("after the deadline fired, tick still holding") + return nil, notify.ErrUnavailable + } + f.tick() + if n := f.fake.callCount(); n != 1 { + t.Errorf("%d Send calls while the row was held", n) + } + if st := f.state(id); st.State != "pending" || st.Attempts != 1 || st.Claim != "" { + t.Fatalf("released row = %+v", st) + } + c, err := srv.store.ClaimPendingNotifications("rival", 10, time.Now().Add(-10*time.Second), time.Now().Add(-3*time.Second)) + if err != nil || len(c.Rows) != 1 { + t.Fatalf("after release the rival claimed %d rows (err %v)", len(c.Rows), err) + } + }) +} + +// manyUnits gives bob 51 devices and two items with a row each: 102 messages, +// so two chunks of one unit each. +func manyUnits(f *pdx) (first, second string) { + for i := 0; i < 50; i++ { + f.device(f.bob, fmt.Sprintf("tok-many-%d-%d", f.n, i), 4*time.Hour) + } + other := f.newItem("Second", &f.bob.ID) + first = f.row("assigned", func(o *rowOpt) { o.age = 5 * time.Minute }) + second = f.row("assigned", func(o *rowOpt) { o.item = other }) + return +} + +// A chunk starts only if a whole send timeout is left before the deadline; +// otherwise its rows are unclaimed with no attempt burned. +func TestPushDispatch_ChunkNotStartedWithoutTimeLeft(t *testing.T) { + needsBackends(t, func(t *testing.T, srv *Server) { + f := newPdx(t, srv) + must(t, srv.SetPushDispatchConfig(0, 0, 3*time.Second, time.Second, time.Second)) + first, second := manyUnits(f) + clk := f.clock(time.Now()) + f.fake.fn = func(_ context.Context, msgs []notify.Message) ([]notify.Result, error) { + clk.advance(1200 * time.Millisecond) // deadline (2s) minus this is under the 1s send timeout + return allStatus(msgs, notify.StatusSent), nil + } + f.tick() + if n := f.fake.callCount(); n != 1 { + t.Fatalf("%d Send calls, want only the first chunk", n) + } + if st := f.state(first); st.State != "sent" { + t.Fatalf("first = %+v", st) + } + if st := f.state(second); st.State != "pending" || st.Attempts != 0 || st.Claim != "" { + t.Fatalf("second = %+v, want pending, unclaimed, no attempt", st) + } + }) +} + +// --- fresh token, settle under the claim token --- + +func TestPushDispatch_FreshClaimTokenPerTick(t *testing.T) { + needsBackends(t, func(t *testing.T, srv *Server) { + f := newPdx(t, srv) + var prefixes []string + f.fake.fn = func(_ context.Context, msgs []notify.Message) ([]notify.Result, error) { + var tok string + row := srv.store.DB().QueryRow(`SELECT claim_token FROM user_notifications WHERE claim_token <> '' LIMIT 1`) + if err := row.Scan(&tok); err != nil { + t.Errorf("no claimed row mid-send: %v", err) + } + prefixes = append(prefixes, tok[:strings.LastIndex(tok, ".")]) + return allStatus(msgs, notify.StatusRetry), nil + } + id := f.row("comment") + f.tick() + f.tick() + if len(prefixes) != 2 || prefixes[0] == prefixes[1] || !strings.HasPrefix(prefixes[0], "pd-") { + t.Fatalf("caller tokens = %v, want two different pd-... tokens", prefixes) + } + if st := f.state(id); st.Attempts != 2 { + t.Fatalf("row = %+v, want both ticks to have settled their release", st) + } + }) +} + +// --- r4-L4: a send that outlives its lease --- + +func parkedSend(f *pdx) (entered, release chan struct{}) { + entered, release = make(chan struct{}), make(chan struct{}) + var once sync.Once + f.fake.fn = func(_ context.Context, msgs []notify.Message) ([]notify.Result, error) { + once.Do(func() { close(entered) }) + <-release + return allStatus(msgs, notify.StatusSent), nil + } + return +} + +func (f *pdx) lapse(ids ...string) { + f.t.Helper() + for _, id := range ids { + f.exec(`UPDATE user_notifications SET claimed_at = ? WHERE id = ?`, f.ts(2*time.Minute), id) + } +} + +func TestPushDispatch_LateSentRecordedWhenReclaimed(t *testing.T) { + needsBackends(t, func(t *testing.T, srv *Server) { + f := newPdx(t, srv) + id := f.row("comment") + entered, release := parkedSend(f) + done := make(chan struct{}) + go func() { f.tick(); close(done) }() + <-entered + f.lapse(id) + b, err := srv.store.ClaimPendingNotifications("B", 10, time.Now().Add(-10*time.Second), time.Now().Add(-time.Minute)) + if err != nil || len(b.Rows) != 1 { + t.Fatalf("B reclaimed %d rows (%v)", len(b.Rows), err) + } + close(release) + <-done + if st := f.state(id); st.State != "sent" || st.SentAt == "" { + t.Fatalf("row = %+v: A's delivery must be recorded although B holds the row", st) + } + if n, _ := srv.store.MarkNotificationsSkipped(b.Token, []string{id}, "late"); n != 0 { + t.Fatal("B's settle changed a row recorded sent") + } + }) +} + +func TestPushDispatch_LateSentRecordedWhenExpired(t *testing.T) { + needsBackends(t, func(t *testing.T, srv *Server) { + f := newPdx(t, srv) + id := f.row("comment", func(o *rowOpt) { o.age = 2 * time.Minute }) + entered, release := parkedSend(f) + done := make(chan struct{}) + go func() { f.tick(); close(done) }() + <-entered + f.exec(`UPDATE user_notifications SET created_at = ? WHERE id = ?`, f.ts(2*time.Hour), id) + f.lapse(id) + if n, err := srv.store.ExpireStaleNotifications(time.Now().Add(-time.Hour), time.Now().Add(-time.Minute)); err != nil || n != 1 { + t.Fatalf("expire = %d, %v", n, err) + } + close(release) + <-done + if st := f.state(id); st.State != "sent" { + t.Fatalf("row = %+v: a delivered row must not read expired", st) + } + }) +} + +// The pre-send re-check: rows taken by another instance while an earlier +// chunk was in flight are dropped from the next one. +func TestPushDispatch_HeldRecheckDropsRowsTakenMidTick(t *testing.T) { + needsBackends(t, func(t *testing.T, srv *Server) { + f := newPdx(t, srv) + first, second := manyUnits(f) + entered, release := parkedSend(f) + done := make(chan struct{}) + go func() { f.tick(); close(done) }() + <-entered + f.lapse(second) + b, err := srv.store.ClaimPendingNotifications("B", 10, time.Now().Add(-10*time.Second), time.Now().Add(-time.Minute)) + if err != nil || len(b.Rows) != 1 || b.Rows[0].ID != second { + t.Fatalf("B claimed %v (%v), want the second row only", b.Rows, err) + } + close(release) + <-done + if n := f.fake.callCount(); n != 1 { + t.Fatalf("%d Send calls: the row B holds was sent again", n) + } + if st := f.state(first); st.State != "sent" { + t.Fatalf("first = %+v", st) + } + if st := f.state(second); st.State != "pending" || st.Claim != b.Token { + t.Fatalf("second = %+v, want still B's", st) + } + }) +} + +// 3946's supersede can DELETE a pending row the dispatcher has claimed. The +// settle that finds it gone is a no-op, not an error and not a log line. +func TestPushDispatch_RowDeletedMidSendIsNotAnError(t *testing.T) { + needsBackends(t, func(t *testing.T, srv *Server) { + f := newPdx(t, srv) + id := f.row("comment") + logs := pdCaptureLogs(t) + entered, release := parkedSend(f) + done := make(chan struct{}) + go func() { f.tick(); close(done) }() + <-entered + f.exec(`DELETE FROM user_notifications WHERE id = ?`, id) + close(release) + <-done + if out := logs.String(); strings.Contains(out, "level=WARN") || strings.Contains(out, "level=ERROR") { + t.Fatalf("a deleted row logged: %s", out) + } + }) +} + +// --- shutdown --- + +func TestPushDispatch_ShutdownUnclaimsWithoutAnAttempt(t *testing.T) { + needsBackends(t, func(t *testing.T, srv *Server) { + f := newPdx(t, srv) + id := f.row("comment") + ctx, cancel := context.WithCancel(context.Background()) + entered := make(chan struct{}) + f.fake.fn = func(c context.Context, _ []notify.Message) ([]notify.Result, error) { + close(entered) + <-c.Done() + return nil, notify.ErrUnavailable + } + done := make(chan struct{}) + go func() { srv.runPushDispatchTick(ctx); close(done) }() + <-entered + cancel() + <-done + if st := f.state(id); st.State != "pending" || st.Attempts != 0 || st.Claim != "" { + t.Fatalf("after shutdown mid-send: %+v, want pending, unclaimed, no attempt", st) + } + + // and a tick that begins cancelled sends nothing + f.fake.fn = nil + before := f.fake.callCount() + srv.runPushDispatchTick(ctx) + if f.fake.callCount() != before { + t.Fatal("a cancelled tick sent") + } + if st := f.state(id); st.State != "pending" || st.Claim != "" { + t.Fatalf("cancelled tick left %+v", st) + } + }) +} + +func TestPushDispatch_LoopRunsTicksAndStops(t *testing.T) { + srv := testServer(t) + f := newPdx(t, srv) + tick := make(chan time.Time) + srv.pushDispatch.tick = tick + id := f.row("comment") + srv.StartPushDispatch() + srv.StartPushDispatch() // idempotent + tick <- time.Now() + deadline := time.Now().Add(5 * time.Second) + for f.state(id).State != "sent" { + if time.Now().After(deadline) { + t.Fatal("the loop never dispatched the row") + } + time.Sleep(20 * time.Millisecond) + } + srv.stopPushDispatch() + srv.stopPushDispatch() // safe twice +} + +// --- logging --- + +func pdCaptureLogs(t *testing.T) *bytes.Buffer { + t.Helper() + var buf bytes.Buffer + prev := slog.Default() + slog.SetDefault(slog.New(slog.NewTextHandler(&pdLogWriter{w: &buf}, &slog.HandlerOptions{Level: slog.LevelDebug}))) + t.Cleanup(func() { slog.SetDefault(prev) }) + return &buf +} + +type pdLogWriter struct { + mu sync.Mutex + w *bytes.Buffer +} + +func (l *pdLogWriter) Write(p []byte) (int, error) { + l.mu.Lock() + defer l.mu.Unlock() + return l.w.Write(p) +} + +// A failed claim logs a fixed line: never the store's error text, which can +// echo a value, and never an id. +func TestPushDispatch_FailedClaimLogsNoStoreError(t *testing.T) { + needsBackends(t, func(t *testing.T, srv *Server) { + f := newPdx(t, srv) + id := f.row("comment") + logs := pdCaptureLogs(t) + f.exec(`ALTER TABLE user_notifications RENAME TO user_notifications_broken`) + defer f.exec(`ALTER TABLE user_notifications_broken RENAME TO user_notifications`) + f.tick() + out := logs.String() + if !strings.Contains(out, "claim failed") { + t.Fatalf("no claim failure logged: %q", out) + } + for _, leak := range []string{id, f.bob.ID, "user_notifications", "no such table", "does not exist", "claim notifications"} { + if strings.Contains(out, leak) { + t.Fatalf("log leaks %q: %s", leak, out) + } + } + }) +} diff --git a/internal/server/task3939b_push_fixture_test.go b/internal/server/task3939b_push_fixture_test.go new file mode 100644 index 000000000..cd577088b --- /dev/null +++ b/internal/server/task3939b_push_fixture_test.go @@ -0,0 +1,254 @@ +package server + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/PerpetualSoftware/pad/internal/models" + "github.com/PerpetualSoftware/pad/internal/notify" + "github.com/PerpetualSoftware/pad/internal/store" +) + +// Fixtures for the push dispatcher tests (TASK-3939 part B1). Every test runs +// on SQLite and, under `make test-pg`, on Postgres (needsBackends). + +// fakePush is the gateway: it records every Send and answers sent for every +// message unless fn says otherwise. +type fakePush struct { + mu sync.Mutex + calls int + msgs [][]notify.Message + fn func(ctx context.Context, msgs []notify.Message) ([]notify.Result, error) +} + +func (p *fakePush) Send(ctx context.Context, msgs []notify.Message) ([]notify.Result, error) { + p.mu.Lock() + p.calls++ + p.msgs = append(p.msgs, append([]notify.Message(nil), msgs...)) + fn := p.fn + p.mu.Unlock() + if fn != nil { + return fn(ctx, msgs) + } + return allStatus(msgs, notify.StatusSent), nil +} + +func (p *fakePush) callCount() int { + p.mu.Lock() + defer p.mu.Unlock() + return p.calls +} + +// all returns every message sent so far, flattened. +func (p *fakePush) all() []notify.Message { + p.mu.Lock() + defer p.mu.Unlock() + var out []notify.Message + for _, b := range p.msgs { + out = append(out, b...) + } + return out +} + +func allStatus(msgs []notify.Message, status string) []notify.Result { + out := make([]notify.Result, len(msgs)) + for i, m := range msgs { + out[i] = notify.Result{ID: m.ID, Status: status} + } + return out +} + +var pdxSeq atomic.Int64 + +// pdx is a workspace with an owner (the actor), a recipient (bob) who is a +// member, assigned the item and has one live device registered long ago, a +// collection, an item, and a comment on it. +type pdx struct { + t *testing.T + srv *Server + fake *fakePush + n int64 + owner *models.User + bob *models.User + ws *models.Workspace + coll *models.Collection + item *models.Item + comment *models.Comment + devID string + devSess string +} + +func newPdx(t *testing.T, srv *Server) *pdx { + t.Helper() + f := &pdx{t: t, srv: srv, fake: &fakePush{}, n: pdxSeq.Add(1)} + srv.SetPushSender(f.fake) + f.owner = f.user("owner") + f.bob = f.user("bob") + ws, err := srv.store.CreateWorkspace(models.WorkspaceCreate{Name: fmt.Sprintf("Push %d", f.n), OwnerID: f.owner.ID}) + if err != nil { + t.Fatal(err) + } + f.ws = ws + must(t, srv.store.AddWorkspaceMember(ws.ID, f.owner.ID, "owner")) + must(t, srv.store.AddWorkspaceMember(ws.ID, f.bob.ID, "editor")) + f.coll, err = srv.store.CreateCollection(ws.ID, models.CollectionCreate{Name: "Tasks", Schema: nfxSchema}) + if err != nil { + t.Fatal(err) + } + f.item = f.newItem("SECRET-ITEM-TITLE", &f.bob.ID) + f.comment, err = srv.store.CreateComment(ws.ID, f.item.ID, f.owner.ID, models.CommentCreate{Body: "SECRET-COMMENT", Author: "Owner", CreatedBy: "user", Source: "web"}) + if err != nil { + t.Fatal(err) + } + // The device comes last, so nothing above enqueued a row for bob. + f.devID, f.devSess = f.device(f.bob, "tok-bob-"+fmt.Sprint(f.n), 5*time.Hour) + f.exec(`DELETE FROM user_notifications`) + return f +} + +func must(t *testing.T, err error) { + t.Helper() + if err != nil { + t.Fatal(err) + } +} + +func (f *pdx) user(name string) *models.User { + f.t.Helper() + u, err := f.srv.store.CreateUser(models.UserCreate{ + Email: fmt.Sprintf("%s%d@push.test", name, f.n), Name: name, Username: fmt.Sprintf("%s-%d", name, f.n), Password: "pw-test-12345", + }) + must(f.t, err) + return u +} + +func (f *pdx) newItem(title string, assignee *string) *models.Item { + f.t.Helper() + it, err := f.srv.store.CreateItem(f.ws.ID, f.coll.ID, models.ItemCreate{Title: title, Fields: `{"status":"open"}`, AssignedUserID: assignee}) + must(f.t, err) + return it +} + +func (f *pdx) exec(q string, args ...any) { + f.t.Helper() + if _, err := f.srv.store.DB().Exec(f.srv.store.D().Rebind(q), args...); err != nil { + f.t.Fatalf("%s: %v", q, err) + } +} + +func (f *pdx) ts(ago time.Duration) string { return time.Now().UTC().Add(-ago).Format(time.RFC3339) } + +func (f *pdx) pg() bool { return f.srv.store.D().Driver() == store.DriverPostgres } + +// device registers a device for u on a fresh live session, created ago ago. +func (f *pdx) device(u *models.User, token string, ago time.Duration) (deviceID, sessionID string) { + f.t.Helper() + plain, err := f.srv.store.CreateSession(u.ID, "android", "", "", time.Hour) + must(f.t, err) + h := sha256.Sum256([]byte(plain)) + row := f.srv.store.DB().QueryRow(f.srv.store.D().Rebind(`SELECT id FROM sessions WHERE token_hash = ?`), hex.EncodeToString(h[:])) + must(f.t, row.Scan(&sessionID)) + deviceID = fmt.Sprintf("dev-%d-%d", f.n, pdxSeq.Add(1)) + ts := f.ts(ago) + f.exec(`INSERT INTO push_devices (id, user_id, session_id, connection_id, platform, token, app_version, created_at, updated_at) + VALUES (?, ?, ?, ?, 'android', ?, '1', ?, ?)`, deviceID, u.ID, sessionID, "conn-"+deviceID, token, ts, ts) + return +} + +type rowOpt struct { + user *models.User + item *models.Item + comment string // "" = none; "default" = f.comment + age time.Duration + agent bool +} + +// row seeds a pending queue row for bob (or opts.user) about the fixture item. +func (f *pdx) row(kind string, opts ...func(*rowOpt)) string { + f.t.Helper() + o := rowOpt{user: f.bob, item: f.item, age: 2 * time.Minute} + if kind == notify.KindComment || kind == notify.KindReply || kind == notify.KindMention { + o.comment = "default" + } + for _, fn := range opts { + fn(&o) + } + cid := "" + switch o.comment { + case "": + case "default": + cid = f.comment.ID + default: + cid = o.comment + } + id := fmt.Sprintf("n-%d-%d", f.n, pdxSeq.Add(1)) + f.exec(`INSERT INTO user_notifications (id, user_id, workspace_id, item_id, comment_id, kind, actor_user_id, dedupe_key, created_at) + VALUES (?, ?, ?, ?, NULLIF(?, ''), ?, ?, ?, ?)`, id, o.user.ID, f.ws.ID, o.item.ID, cid, kind, f.owner.ID, "k-"+id, f.ts(o.age)) + if o.agent { + lit := "1" + if f.pg() { + lit = "TRUE" + } + f.exec(`UPDATE user_notifications SET actor_is_agent = `+lit+` WHERE id = ?`, id) + } + return id +} + +type rowState struct { + State, Reason, Claim string + Attempts int + SentAt string +} + +func (f *pdx) state(id string) rowState { + f.t.Helper() + var r rowState + row := f.srv.store.DB().QueryRow(f.srv.store.D().Rebind( + `SELECT push_state, skip_reason, claim_token, attempts, COALESCE(sent_at, '') FROM user_notifications WHERE id = ?`), id) + must(f.t, row.Scan(&r.State, &r.Reason, &r.Claim, &r.Attempts, &r.SentAt)) + return r +} + +func (f *pdx) tick() { f.srv.runPushDispatchTick(context.Background()) } + +func (f *pdx) settings(mut func(*models.NotificationSettings)) { + f.t.Helper() + set := models.DefaultNotificationSettings() + mut(&set) + must(f.t, f.srv.store.PutNotificationSettings(f.bob.ID, set)) +} + +// pdClock is the dispatcher's clock (Server.pushDispatch.now) under a test's +// control, so lease and deadline behaviour never depends on wall-clock timing. +type pdClock struct { + mu sync.Mutex + t time.Time + start time.Time +} + +func (c *pdClock) now() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + return c.t +} + +func (c *pdClock) advance(d time.Duration) { + c.mu.Lock() + defer c.mu.Unlock() + c.t = c.t.Add(d) +} + +// clock installs a fake dispatcher clock reading at. +func (f *pdx) clock(at time.Time) *pdClock { + c := &pdClock{t: at, start: at} + f.srv.pushDispatch.mu.Lock() + f.srv.pushDispatch.now = c.now + f.srv.pushDispatch.mu.Unlock() + return c +} diff --git a/internal/server/task3939b_push_gates_test.go b/internal/server/task3939b_push_gates_test.go new file mode 100644 index 000000000..53e9a6bcd --- /dev/null +++ b/internal/server/task3939b_push_gates_test.go @@ -0,0 +1,502 @@ +package server + +import ( + "context" + "fmt" + "strings" + "testing" + "time" + + "github.com/PerpetualSoftware/pad/internal/models" + "github.com/PerpetualSoftware/pad/internal/notify" + "github.com/PerpetualSoftware/pad/internal/store" +) + +// TASK-3939 part B1: the dispatcher IS the privacy gate. Every case seeds a +// pending row for a recipient who has a device, breaks ONE thing, ticks, and +// asserts the row is skipped with that reason and the gateway was never +// called; every skip case has a control that sends, so a skip is not just a +// misconfigured fixture. Each gate has a mutant in the report that turns its +// case red. + +type gateCase struct { + name string + kind string // default comment + setup func(f *pdx, id string) + row []func(*rowOpt) + reason string // "" = the row is pushed + check func(f *pdx, id string) // extra assertions on a pushed row +} + +func gateCases() []gateCase { + removeBob := func(f *pdx, _ string) { must(f.t, f.srv.store.RemoveWorkspaceMember(f.ws.ID, f.bob.ID)) } + specificNone := func(f *pdx, _ string) { + must(f.t, f.srv.store.SetMemberCollectionAccess(f.ws.ID, f.bob.ID, "specific", []string{})) + } + offKind := func(mut func(*models.NotificationSettings)) func(*pdx, string) { + return func(f *pdx, _ string) { f.settings(mut) } + } + otherWorkspace := func(f *pdx) *models.Workspace { + ws2, err := f.srv.store.CreateWorkspace(models.WorkspaceCreate{Name: fmt.Sprintf("Other %d", f.n), OwnerID: f.owner.ID}) + must(f.t, err) + must(f.t, f.srv.store.AddWorkspaceMember(ws2.ID, f.bob.ID, "editor")) + return ws2 + } + return []gateCase{ + // controls: each kind, as the fixture leaves it + {name: "control_comment", kind: "comment"}, + {name: "control_reply", kind: "reply"}, + {name: "control_assigned", kind: "assigned"}, + {name: "control_reminder", kind: "reminder"}, + {name: "control_mention_default_settings", kind: "mention"}, + + // G1 membership, separate from visibility + {name: "removed_member_still_assignee", kind: "assigned", setup: removeBob, reason: skipNotMember}, + {name: "removed_member_still_creator_comment", kind: "comment", setup: removeBob, reason: skipNotMember}, + {name: "platform_admin_not_member", kind: "comment", reason: skipNotMember, setup: func(f *pdx, id string) { + f.exec(`UPDATE users SET role = 'admin' WHERE id = ?`, f.bob.ID) + removeBob(f, id) + }}, + + // G2 visibility + {name: "specific_access_without_the_collection", kind: "comment", setup: specificNone, reason: skipNotVisible}, + {name: "specific_access_with_the_collection", kind: "comment", setup: func(f *pdx, _ string) { + must(f.t, f.srv.store.SetMemberCollectionAccess(f.ws.ID, f.bob.ID, "specific", []string{f.coll.ID})) + }}, + {name: "admin_member_restricted_is_not_bypassed", kind: "comment", reason: skipNotVisible, setup: func(f *pdx, id string) { + f.exec(`UPDATE users SET role = 'admin' WHERE id = ?`, f.bob.ID) + specificNone(f, id) + }}, + + // G3 liveness of the subject + {name: "item_soft_deleted", kind: "assigned", reason: skipItemGone, setup: func(f *pdx, _ string) { + must(f.t, f.srv.store.DeleteItem(f.item.ID)) + }}, + {name: "collection_archived", kind: "assigned", reason: skipItemGone, setup: func(f *pdx, _ string) { + f.exec(`UPDATE collections SET deleted_at = ? WHERE id = ?`, f.ts(time.Minute), f.coll.ID) + }}, + // user_notifications.item_id has no foreign key: the row outlives a purged item. + {name: "item_hard_deleted", kind: "reminder", reason: skipItemGone, setup: func(f *pdx, id string) { + f.exec(`UPDATE user_notifications SET item_id = 'purged-item' WHERE id = ?`, id) + }}, + {name: "workspace_soft_deleted", kind: "assigned", reason: skipWorkspaceGone, setup: func(f *pdx, _ string) { + must(f.t, f.srv.store.DeleteWorkspace(f.ws.Slug)) + }}, + {name: "row_workspace_is_not_the_items", kind: "assigned", reason: skipItemGone, setup: func(f *pdx, id string) { + f.exec(`UPDATE user_notifications SET workspace_id = ? WHERE id = ?`, otherWorkspace(f).ID, id) + }}, + + // G4 the user + {name: "user_disabled", kind: "comment", reason: skipUserDisabled, setup: func(f *pdx, _ string) { + f.exec(`UPDATE users SET disabled_at = ? WHERE id = ?`, f.ts(time.Minute), f.bob.ID) + }}, + {name: "app_principal", kind: "comment", reason: skipUserDisabled, setup: func(f *pdx, _ string) { + f.exec(`UPDATE users SET kind = 'app' WHERE id = ?`, f.bob.ID) + }}, + + // G9 the reason for the row still holds + {name: "assigned_away", kind: "assigned", reason: skipNoLongerAssign, setup: func(f *pdx, _ string) { + f.exec(`UPDATE items SET assigned_user_id = ? WHERE id = ?`, f.owner.ID, f.item.ID) + }}, + {name: "unassigned", kind: "assigned", reason: skipNoLongerAssign, setup: func(f *pdx, _ string) { + f.exec(`UPDATE items SET assigned_user_id = NULL WHERE id = ?`, f.item.ID) + }}, + {name: "assigned_control", kind: "assigned", setup: func(f *pdx, _ string) { + f.exec(`UPDATE items SET assigned_user_id = ? WHERE id = ?`, f.owner.ID, f.item.ID) + f.exec(`UPDATE items SET assigned_user_id = ? WHERE id = ?`, f.bob.ID, f.item.ID) + }}, + {name: "comment_tombstoned", kind: "comment", reason: skipCommentGone, setup: func(f *pdx, _ string) { + f.exec(`UPDATE comments SET deleted_at = ? WHERE id = ?`, f.ts(time.Minute), f.comment.ID) + }}, + {name: "reply_comment_hard_deleted", kind: "reply", reason: skipCommentGone, setup: func(f *pdx, _ string) { + f.exec(`DELETE FROM comments WHERE id = ?`, f.comment.ID) + }}, + {name: "mention_comment_hard_deleted", kind: "mention", reason: skipCommentGone, setup: func(f *pdx, _ string) { + f.exec(`DELETE FROM comments WHERE id = ?`, f.comment.ID) + }}, + {name: "comment_row_without_a_comment_id", kind: "comment", reason: skipCommentGone, row: []func(*rowOpt){func(o *rowOpt) { o.comment = "" }}}, + + // the row's comment must be a comment on the row's item, in its workspace + {name: "comment_belongs_to_another_item", kind: "comment", reason: skipCommentGone, setup: func(f *pdx, id string) { + other := f.newItem("Other", &f.bob.ID) + cm, err := f.srv.store.CreateComment(f.ws.ID, other.ID, f.owner.ID, models.CommentCreate{Body: "x", Author: "Owner", CreatedBy: "user", Source: "web"}) + must(f.t, err) + f.exec(`UPDATE user_notifications SET comment_id = ? WHERE id = ?`, cm.ID, id) + }}, + {name: "comment_belongs_to_another_workspace", kind: "comment", reason: skipCommentGone, setup: func(f *pdx, _ string) { + f.exec(`UPDATE comments SET workspace_id = ? WHERE id = ?`, otherWorkspace(f).ID, f.comment.ID) + }}, + + // G5 settings + {name: "push_disabled", kind: "comment", reason: skipSettingOff, setup: offKind(func(s *models.NotificationSettings) { s.PushEnabled = false })}, + {name: "assigned_off", kind: "assigned", reason: skipSettingOff, setup: offKind(func(s *models.NotificationSettings) { s.NotifyAssigned = false })}, + {name: "comments_off", kind: "comment", reason: skipSettingOff, setup: offKind(func(s *models.NotificationSettings) { s.NotifyComments = false })}, + {name: "replies_off", kind: "reply", reason: skipSettingOff, setup: offKind(func(s *models.NotificationSettings) { s.NotifyReplies = false })}, + {name: "reminders_off", kind: "reminder", reason: skipSettingOff, setup: offKind(func(s *models.NotificationSettings) { s.NotifyReminders = false })}, + {name: "mentions_off", kind: "mention", reason: skipSettingOff, setup: offKind(func(s *models.NotificationSettings) { s.NotifyMentions = false })}, + // comment and reply are separate switches: each off leaves the other on + {name: "comments_off_reply_still_sends", kind: "reply", setup: offKind(func(s *models.NotificationSettings) { s.NotifyComments = false })}, + {name: "replies_off_comment_still_sends", kind: "comment", setup: offKind(func(s *models.NotificationSettings) { s.NotifyReplies = false })}, + {name: "mentions_off_comment_still_sends", kind: "comment", setup: offKind(func(s *models.NotificationSettings) { s.NotifyMentions = false })}, + + // G6 agent activity + {name: "agent_actor_default_off", kind: "comment", reason: skipAgentOff, row: []func(*rowOpt){func(o *rowOpt) { o.agent = true }}}, + {name: "agent_actor_opted_in", kind: "comment", row: []func(*rowOpt){func(o *rowOpt) { o.agent = true }}, + setup: offKind(func(s *models.NotificationSettings) { s.IncludeAgentActivity = true })}, + + // G7 devices + {name: "no_device", kind: "comment", reason: skipNoDevice, setup: func(f *pdx, _ string) { + f.exec(`DELETE FROM push_devices WHERE user_id = ?`, f.bob.ID) + }}, + {name: "row_predates_the_only_device", kind: "comment", reason: skipPredatesDevice, setup: func(f *pdx, _ string) { + f.exec(`UPDATE push_devices SET created_at = ? WHERE id = ?`, f.ts(30*time.Second), f.devID) + }}, + {name: "claim_race_devices_deleted_then_new_device_registers", kind: "assigned", reason: skipPredatesDevice, setup: func(f *pdx, _ string) { + f.exec(`DELETE FROM push_devices WHERE user_id = ?`, f.bob.ID) + f.device(f.bob, "tok-next-holder", 30*time.Second) + }}, + {name: "row_after_the_device_control", kind: "assigned", setup: func(f *pdx, _ string) { + f.exec(`UPDATE push_devices SET created_at = ? WHERE id = ?`, f.ts(time.Hour), f.devID) + }}, + {name: "second_device_after_the_row_older_device_live", kind: "comment", setup: func(f *pdx, _ string) { + f.device(f.bob, "tok-second", 30*time.Second) + }, check: func(f *pdx, _ string) { + devs, err := f.srv.store.LivePushDevices(f.bob.ID, time.Now()) + must(f.t, err) + if len(devs) != 2 || devs[0].ID != f.devID { + f.t.Fatalf("live devices = %+v, want the earliest (%s) first of two", devs, f.devID) + } + got := map[string]int{} + for _, m := range f.fake.all() { + got[m.Token]++ + } + if len(got) != 2 || got["tok-bob-"+fmt.Sprint(f.n)] != 1 || got["tok-second"] != 1 { + f.t.Fatalf("messages per token = %v, want exactly one to each live device", got) + } + }}, + {name: "device_session_expired", kind: "comment", reason: skipNoDevice, setup: func(f *pdx, _ string) { + f.exec(`UPDATE sessions SET expires_at = ? WHERE id = ?`, f.ts(time.Minute), f.devSess) + }}, + + // ordering: when several gates fail the recorded reason is the privacy one + {name: "privacy_reason_wins_over_preference", kind: "comment", reason: skipNotMember, setup: func(f *pdx, id string) { + removeBob(f, id) + f.settings(func(s *models.NotificationSettings) { s.PushEnabled = false }) + }}, + {name: "user_reason_wins_over_membership", kind: "comment", reason: skipUserDisabled, setup: func(f *pdx, id string) { + removeBob(f, id) + f.exec(`UPDATE users SET disabled_at = ? WHERE id = ?`, f.ts(time.Minute), f.bob.ID) + }}, + {name: "membership_reason_wins_over_item_gone", kind: "assigned", reason: skipNotMember, setup: func(f *pdx, id string) { + removeBob(f, id) + must(f.t, f.srv.store.DeleteItem(f.item.ID)) + }}, + {name: "workspace_gone_reason_wins_over_item_gone", kind: "assigned", reason: skipWorkspaceGone, setup: func(f *pdx, _ string) { + must(f.t, f.srv.store.DeleteItem(f.item.ID)) + must(f.t, f.srv.store.DeleteWorkspace(f.ws.Slug)) + }}, + {name: "item_gone_reason_wins_over_visibility", kind: "assigned", reason: skipItemGone, setup: func(f *pdx, id string) { + specificNone(f, id) + must(f.t, f.srv.store.DeleteItem(f.item.ID)) + }}, + {name: "visibility_reason_wins_over_assignment", kind: "assigned", reason: skipNotVisible, setup: func(f *pdx, id string) { + specificNone(f, id) + f.exec(`UPDATE items SET assigned_user_id = ? WHERE id = ?`, f.owner.ID, f.item.ID) + }}, + } +} + +func TestPushDispatch_Gates(t *testing.T) { + needsBackends(t, func(t *testing.T, srv *Server) { + for _, c := range gateCases() { + t.Run(c.name, func(t *testing.T) { + f := newPdx(t, srv) + kind := c.kind + id := f.row(kind, c.row...) + if c.setup != nil { + c.setup(f, id) + } + f.tick() + st := f.state(id) + if c.reason == "" { + if st.State != "sent" || f.fake.callCount() != 1 { + t.Fatalf("control: state=%s/%q calls=%d, want sent and 1 call", st.State, st.Reason, f.fake.callCount()) + } + if c.check != nil { + c.check(f, id) + } + return + } + if st.State != "skipped" || st.Reason != c.reason { + t.Fatalf("state=%s reason=%q, want skipped/%q", st.State, st.Reason, c.reason) + } + if n := f.fake.callCount(); n != 0 { + t.Fatalf("gateway called %d times for a skipped row", n) + } + }) + } + }) +} + +// Shared FCM token: push_devices is unique on (user, token), so two users can +// hold one phone's token. Only the newest LIVE registrant owns it. +func TestPushDispatch_SharedTokenNeverCrossesUsers(t *testing.T) { + needsBackends(t, func(t *testing.T, srv *Server) { + f := newPdx(t, srv) + carol := f.user("carol") + must(t, srv.store.AddWorkspaceMember(f.ws.ID, carol.ID, "editor")) + f.exec(`DELETE FROM push_devices WHERE user_id = ?`, f.bob.ID) + f.device(f.bob, "tok-phone", 3*time.Hour) + _, carolSess := f.device(carol, "tok-phone", time.Hour) // carol signed in later + bobRow := f.row("comment") + carolRow := f.row("comment", func(o *rowOpt) { o.user = carol }) + f.tick() + if st := f.state(bobRow); st.State != "skipped" || st.Reason != skipNoDevice { + t.Fatalf("bob's row = %s/%q, want skipped/no_device: the phone is carol's now", st.State, st.Reason) + } + if st := f.state(carolRow); st.State != "sent" { + t.Fatalf("carol's row = %s/%q", st.State, st.Reason) + } + for _, m := range f.fake.all() { + if m.Data.NID == bobRow { + t.Fatalf("bob's notification went out to the shared token") + } + } + if n := len(f.fake.all()); n != 1 { + t.Fatalf("%d messages, want carol's one", n) + } + + // Carol's session ends: the phone is bob's again. + f.exec(`UPDATE sessions SET expires_at = ? WHERE id = ?`, f.ts(time.Minute), carolSess) + again := f.row("comment") + f.tick() + if st := f.state(again); st.State != "sent" { + t.Fatalf("after carol's session expired bob's row = %s/%q, want sent", st.State, st.Reason) + } + }) +} + +// A comment row and a mention row for one comment are two rows of one group: +// both settle sent, neither is dropped as a duplicate, and they share ONE +// push whose kind is the higher-priority one (mention) on a created_at tie. +func TestPushDispatch_MentionAndCommentRowsShareOnePush(t *testing.T) { + needsBackends(t, func(t *testing.T, srv *Server) { + f := newPdx(t, srv) + same := f.ts(2 * time.Minute) + mk := func(kind string) string { + id := f.row(kind) + f.exec(`UPDATE user_notifications SET created_at = ? WHERE id = ?`, same, id) + return id + } + c, m := mk("comment"), mk("mention") + f.tick() + for _, id := range []string{c, m} { + if st := f.state(id); st.State != "sent" { + t.Fatalf("row %s = %s/%q, want sent: neither is a duplicate of the other", id, st.State, st.Reason) + } + } + msgs := f.fake.all() + if len(msgs) != 1 || msgs[0].Data.Kind != "mention" || msgs[0].Data.NID != m || msgs[0].Data.Count != 1 { + t.Fatalf("messages = %+v, want one mention push carrying the mention row's id", msgs) + } + + // A newer comment row beats an older mention row on created_at. + older := f.row("mention", func(o *rowOpt) { o.age = 3 * time.Minute }) + newer := f.row("comment", func(o *rowOpt) { o.age = 2 * time.Minute }) + f.tick() + msgs = f.fake.all() + last := msgs[len(msgs)-1] + if last.Data.Kind != "comment" || last.Data.NID != newer { + t.Fatalf("last push = %+v, want the newer comment row (%s, not %s)", last.Data, newer, older) + } + }) +} + +// gateLookups is every lookup a tick makes that the fixture can force to fail: +// the table whose rename breaks it, and the access it needs to be reached. +var gateLookups = []struct { + name, table string + specific bool // bob needs "specific" collection access to reach it +}{ + {"settings", "user_notification_settings", false}, + {"comment", "comments", false}, + {"user", "users", false}, + {"membership", "workspace_members", false}, + {"workspace", "workspaces", false}, + {"subject", "items", false}, + {"devices", "push_devices", false}, + {"visibility_collection_grants", "collection_grants", true}, + {"visibility_item_grants", "item_grants", true}, + {"visibility_member_collections", "member_collection_access", true}, +} + +// breakTable makes every query on table fail and returns the undo. +func (f *pdx) breakTable(table string) (restore func()) { + f.exec(`ALTER TABLE ` + table + ` RENAME TO ` + table + `_broken`) + restored := false + restore = func() { + if !restored { + restored = true + f.exec(`ALTER TABLE ` + table + `_broken RENAME TO ` + table) + } + } + f.t.Cleanup(restore) + return restore +} + +func (f *pdx) grantSpecific() { + must(f.t, f.srv.store.SetMemberCollectionAccess(f.ws.ID, f.bob.ID, "specific", []string{f.coll.ID})) +} + +// A gate that cannot be evaluated is not a denial: the row is unclaimed, no +// attempt is burned, no skip is recorded, and nothing is sent. Every lookup +// the dispatcher makes is forced to fail in turn. +func TestPushDispatch_GateErrorUnclaimsNeverSends(t *testing.T) { + needsBackends(t, func(t *testing.T, srv *Server) { + for _, lk := range gateLookups { + t.Run(lk.name, func(t *testing.T) { + f := newPdx(t, srv) + if lk.specific { + f.grantSpecific() + } + id := f.row("comment") + restore := f.breakTable(lk.table) + f.tick() + st := f.state(id) + if st.State != "pending" || st.Claim != "" || st.Attempts != 0 || st.Reason != "" { + t.Fatalf("after a gate error: %+v, want pending, unclaimed, 0 attempts, no skip reason", st) + } + if n := f.fake.callCount(); n != 0 { + t.Fatalf("gateway called %d times with a gate unevaluable", n) + } + restore() + f.tick() + if st := f.state(id); st.State != "sent" { + t.Fatalf("after the gate recovered: %+v, want sent", st) + } + }) + } + }) +} + +// The dispatcher shares computeWatchAccessVisibility's lookups but not its +// logging: that function logs the workspace id, user id and raw store error, +// and a tick must log none of them (the handler keeps its own log). +func TestPushDispatch_VisibilityErrorsLeakNothingToTheLog(t *testing.T) { + needsBackends(t, func(t *testing.T, srv *Server) { + for _, lk := range gateLookups { + if !lk.specific && lk.name != "membership" { + continue + } + t.Run(lk.name, func(t *testing.T) { + f := newPdx(t, srv) + f.grantSpecific() + id := f.row("comment") + logs := pdCaptureLogs(t) + f.breakTable(lk.table) + f.tick() + out := logs.String() + for _, leak := range []string{f.ws.ID, f.bob.ID, id, lk.table, "_broken", "no such table", "does not exist", "failed to resolve"} { + if strings.Contains(out, leak) { + t.Fatalf("log leaks %q: %s", leak, out) + } + } + if st := f.state(id); st.State != "pending" || st.Claim != "" || st.Attempts != 0 || st.Reason != "" { + t.Fatalf("row = %+v, want pending, unclaimed, no attempt, no skip", st) + } + }) + } + }) +} + +// A unit none of whose devices can be given a message (NewMessage refuses an +// unknown kind, which the gates exclude, so the unit is built by hand) is +// settled up front as a skip, whether it is alone in its chunk or another unit +// sends beside it. It is neither unclaimed nor released with an attempt. +func TestPushDispatch_ZeroMessageUnitIsSkipped(t *testing.T) { + needsBackends(t, func(t *testing.T, srv *Server) { + run := func(t *testing.T, withSender bool) { + f := newPdx(t, srv) + other := f.newItem("Other", &f.bob.ID) + f.exec(`DELETE FROM user_notifications`) // the assignment enqueued a row + bad := f.row("comment") + ids := []string{bad} + good := "" + if withSender { + good = f.row("assigned", func(o *rowOpt) { o.item = other }) + ids = append(ids, good) + } + claim, err := srv.store.ClaimPendingNotifications("zm", 10, time.Now(), time.Now().Add(-time.Minute)) + must(t, err) + if len(claim.Rows) != len(ids) { + t.Fatalf("claimed %d rows, want %d", len(claim.Rows), len(ids)) + } + devs, err := srv.store.LivePushDevices(f.bob.ID, time.Now()) + must(t, err) + chunk := []pushUnit{{rowIDs: []string{bad}, nid: bad, kind: "bogus", devices: devs}} + if withSender { + chunk = append(chunk, pushUnit{rowIDs: []string{good}, nid: good, kind: "assigned", devices: devs}) + } + srv.sendChunk(context.Background(), f.fake, claim.Token, chunk, time.Now().Add(time.Minute)) + if st := f.state(bad); st.State != "skipped" || st.Reason != skipNoDevice || st.Attempts != 0 { + t.Fatalf("zero-message unit = %+v, want skipped/%s, no attempt", st, skipNoDevice) + } + wantCalls := 0 + if withSender { + wantCalls = 1 + if st := f.state(good); st.State != "sent" { + t.Fatalf("the sending unit = %+v", st) + } + } + if n := f.fake.callCount(); n != wantCalls { + t.Fatalf("%d Send calls, want %d", n, wantCalls) + } + } + t.Run("alone_in_a_chunk", func(t *testing.T) { run(t, false) }) + t.Run("beside_a_unit_that_sends", func(t *testing.T) { run(t, true) }) + }) +} + +// The message is content-free: no title, comment text or names, and its +// fields come from the right places. +func TestPushDispatch_MessageShape(t *testing.T) { + needsBackends(t, func(t *testing.T, srv *Server) { + f := newPdx(t, srv) + id := f.row("comment") + f.tick() + msgs := f.fake.all() + if len(msgs) != 1 { + t.Fatalf("%d messages", len(msgs)) + } + m := msgs[0] + wantBody, _ := notify.Text("comment", 1) + if m.Title != notify.Title || m.Body != wantBody || m.Platform != "android" || m.Token != "tok-bob-"+fmt.Sprint(f.n) || + m.Data.NID != id || m.Data.CID != "conn-"+f.devID || m.Data.Kind != "comment" || m.Data.Count != 1 { + t.Fatalf("message = %+v", m) + } + // Belt and braces: the exact-field comparison above is the real + // assertion; this scan only catches a secret that lands in a field + // the comparison does not name. + blob := fmt.Sprintf("%+v", msgs) + for _, secret := range []string{"SECRET", "owner", "Push "} { + if strings.Contains(blob, secret) { + t.Fatalf("message carries %q: %s", secret, blob) + } + } + }) +} + +// user_notifications.kind has a CHECK, so an unknown kind cannot be seeded; +// the guard is for a kind added to the schema before the dispatcher learns its +// text. It must skip, not send. +func TestKindEnabled_UnknownKindIsSkipped(t *testing.T) { + set := models.DefaultNotificationSettings() + if got := kindEnabled(set, store.Notification{Kind: "bogus"}); got != skipUnknownKind { + t.Fatalf("unknown kind = %q, want %q", got, skipUnknownKind) + } + for _, k := range []string{"assigned", "comment", "reply", "reminder", "mention"} { + if got := kindEnabled(set, store.Notification{Kind: k}); got != "" { + t.Fatalf("%s with default settings = %q, want enabled", k, got) + } + } +} diff --git a/internal/store/notifications.go b/internal/store/notifications.go index a8ddba408..cf10af0ed 100644 --- a/internal/store/notifications.go +++ b/internal/store/notifications.go @@ -23,9 +23,10 @@ import ( // // 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. +// someone who can no longer see the item. Once the dispatcher is wired +// into the server (TASK-3939 part B2; part B1 only builds it), it re-checks +// current membership and item visibility at send and skips the row: that is +// the privacy gate. // // @mention notifications (TASK-3946) are built from the users // reindexCommentMentionsTx newly indexed, at the comment create and edit diff --git a/internal/store/push_dispatch.go b/internal/store/push_dispatch.go index df82279cc..53bf02201 100644 --- a/internal/store/push_dispatch.go +++ b/internal/store/push_dispatch.go @@ -192,7 +192,7 @@ func (s *Store) claimNotificationIDs(callerToken string, ids []string, settle, l AND push_state = 'pending' AND created_at <= ? AND (claimed_at IS NULL OR claimed_at < ?)`), token, ts, id, settle, lease); err != nil { - return Claim{}, fmt.Errorf("claim notification %s: %w", id, err) + return Claim{}, fmt.Errorf("claim notification: %w", err) } } if err := tx.Commit(); err != nil { @@ -242,7 +242,8 @@ func (s *Store) ExpireStaleNotifications(createdCutoff, leaseCutoff time.Time) ( // settleNotifications runs one UPDATE per chunk of ids, each conditioned on // the claim and on the row still being pending, so a claimer whose lease // lapsed cannot stamp a row another instance now holds. It returns the rows -// changed. +// changed. Every settle is under the claim EXCEPT recording a delivery +// (RecordNotificationsSent): a push that went out is true whoever holds the row. func (s *Store) settleNotifications(token string, ids []string, set string, setArgs ...any) (int64, error) { if token == "" { return 0, errEmptyClaimToken @@ -271,7 +272,9 @@ func (s *Store) settleNotifications(token string, ids []string, set string, setA } // MarkNotificationsSent records ids as pushed. token is Claim.Token. A row not -// (or no longer) pending under it is left alone. +// (or no longer) pending under it is left alone. The dispatcher does not use +// it: SENT is the one deliberate exception to settle-under-the-claim, and is +// recorded by id with RecordNotificationsSent (push_dispatch_settle.go). func (s *Store) MarkNotificationsSent(token string, ids []string) (int64, error) { return s.settleNotifications(token, ids, `push_state = 'sent', sent_at = ?`, now()) } diff --git a/internal/store/push_dispatch_settle.go b/internal/store/push_dispatch_settle.go new file mode 100644 index 000000000..0ba73eafe --- /dev/null +++ b/internal/store/push_dispatch_settle.go @@ -0,0 +1,74 @@ +package store + +import "fmt" + +// The dispatcher's settle primitives beyond push_dispatch.go's claim-guarded +// ones (TASK-3939 part B). + +// RecordNotificationsSent records ids as pushed, BY ID AND NOT BY CLAIM: this +// is the one deliberate exception to "settle only under the claim" (r4-L4). +// A delivery is a fact. A claimer whose Send outlived its lease has still +// delivered, so its late 'sent' must land on a row another instance has since +// reclaimed (that instance's own settle is then a no-op) and on a row +// ExpireStaleNotifications took after the lease lapsed; a token guard would +// lose the delivered row or record it twice. Only pending and expired rows +// move: a skipped row stays skipped. Skip and release stay token-guarded. +func (s *Store) RecordNotificationsSent(ids []string) (int64, error) { + var total int64 + for start := 0; start < len(ids); start += notificationIDChunk { + end := min(start+notificationIDChunk, len(ids)) + ph, args := idPlaceholders(ids[start:end]) + res, err := s.db.Exec(s.q(`UPDATE user_notifications + SET push_state = 'sent', sent_at = ?, claim_token = '', claimed_at = NULL + WHERE push_state IN ('pending', 'expired') AND id IN (`+ph+`)`), append([]any{now()}, args...)...) + if err != nil { + return total, fmt.Errorf("record notifications sent: %w", err) + } + n, err := res.RowsAffected() + if err != nil { + return total, fmt.Errorf("record notifications sent: %w", err) + } + total += n + } + return total, nil +} + +// UnclaimNotifications hands claimed rows back WITHOUT counting an attempt: +// for rows that were never tried (a shutdown, a deadline, a gate that could +// not be evaluated). token is Claim.Token; same guard as the other settles. +func (s *Store) UnclaimNotifications(token string, ids []string) (int64, error) { + return s.settleNotifications(token, ids, `claim_token = '', claimed_at = NULL`) +} + +// HeldNotifications returns the subset of ids still pending under token: the +// dispatcher's check just before a Send that a row was not reclaimed, +// settled or deleted while it worked. +func (s *Store) HeldNotifications(token string, ids []string) ([]string, error) { + if token == "" { + return nil, errEmptyClaimToken + } + var held []string + for start := 0; start < len(ids); start += notificationIDChunk { + end := min(start+notificationIDChunk, len(ids)) + ph, args := idPlaceholders(ids[start:end]) + rows, err := s.db.Query(s.q(`SELECT id FROM user_notifications + WHERE claim_token = ? AND push_state = 'pending' AND id IN (`+ph+`)`), append([]any{token}, args...)...) + if err != nil { + return nil, fmt.Errorf("held notifications: %w", err) + } + for rows.Next() { + var id string + if err := rows.Scan(&id); err != nil { + rows.Close() + return nil, fmt.Errorf("held notifications: %w", err) + } + held = append(held, id) + } + err = rows.Err() + rows.Close() + if err != nil { + return nil, fmt.Errorf("held notifications: %w", err) + } + } + return held, nil +} diff --git a/internal/store/task3939b_settle_test.go b/internal/store/task3939b_settle_test.go new file mode 100644 index 000000000..62e127d32 --- /dev/null +++ b/internal/store/task3939b_settle_test.go @@ -0,0 +1,157 @@ +package store + +import ( + "slices" + "strings" + "testing" + "time" +) + +// TASK-3939 part B1: the dispatcher's settle primitives. SQLite, and Postgres +// under `make test-pg`. + +// Kills: token-guarding RecordNotificationsSent (the reclaimed leg goes red); +// dropping 'expired' from its state set (the expired leg goes red); letting it +// move a skipped row (the skipped leg goes red). +func TestRecordNotificationsSent_IsByIDForPendingAndExpiredOnly(t *testing.T) { + t.Parallel() + f := newNotifyFixture(t) + u := notifyUser(t, f.s, "rec@example.com") + settle := at(10 * time.Second) + + reclaimed := f.seedNotif(t, u.ID, ago(time.Minute)) + xc, _ := f.s.ClaimPendingNotifications("X", 10, settle, at(time.Minute)) + if len(xc.Rows) != 1 { + t.Fatal("X claimed nothing") + } + zc, _ := f.s.ClaimPendingNotifications("Z", 10, settle, at(-time.Hour)) // X's lease lapsed + if len(zc.Rows) != 1 { + t.Fatal("Z did not reclaim") + } + expired := f.seedNotif(t, u.ID, ago(2*time.Hour)) + f.s.mustExec(t, `UPDATE user_notifications SET push_state = 'expired' WHERE id = ?`, expired) + skipped := f.seedNotif(t, u.ID, ago(time.Minute)) + f.s.mustExec(t, `UPDATE user_notifications SET push_state = 'skipped', skip_reason = 'not_member' WHERE id = ?`, skipped) + + n, err := f.s.RecordNotificationsSent([]string{reclaimed, expired, skipped}) + if err != nil || n != 2 { + t.Fatalf("RecordNotificationsSent = %d, %v; want 2 (the reclaimed and the expired row)", n, err) + } + for _, c := range []struct{ id, state string }{{reclaimed, "sent"}, {expired, "sent"}, {skipped, "skipped"}} { + state, _, tok, attempts := f.notifRow(t, c.id) + if state != c.state || (c.state == "sent" && tok != "") { + t.Fatalf("row %s = %s/%q, want %s with the claim cleared", c.id, state, tok, c.state) + } + if attempts != 0 { + t.Fatalf("row %s has %d attempts: a delivery is not a failed attempt", c.id, attempts) + } + var sentAt string + if err := f.s.db.QueryRow(f.s.q(`SELECT COALESCE(sent_at, '') FROM user_notifications WHERE id = ?`), c.id).Scan(&sentAt); err != nil { + t.Fatal(err) + } + if (sentAt != "") != (c.state == "sent") { + t.Fatalf("row %s (%s) has sent_at %q", c.id, c.state, sentAt) + } + } + // The reclaimer's own token-guarded settle is now a no-op. + if n, _ := f.s.MarkNotificationsSkipped(zc.Token, []string{reclaimed}, "late"); n != 0 { + t.Fatal("Z's settle changed a row that was recorded sent") + } + if n, err := f.s.RecordNotificationsSent(nil); err != nil || n != 0 { + t.Fatalf("empty ids: %d %v", n, err) + } +} + +// Kills: UnclaimNotifications burning an attempt; clearing nothing; ignoring +// the token guard. +func TestUnclaimNotifications_NoAttemptAndTokenGuarded(t *testing.T) { + t.Parallel() + f := newNotifyFixture(t) + u := notifyUser(t, f.s, "unc@example.com") + id := f.seedNotif(t, u.ID, ago(time.Minute)) + c, _ := f.s.ClaimPendingNotifications("X", 10, at(10*time.Second), at(time.Minute)) + if len(c.Rows) != 1 { + t.Fatal("no claim") + } + if n, _ := f.s.UnclaimNotifications("X.not-the-nonce", []string{id}); n != 0 { + t.Fatal("a foreign token unclaimed the row") + } + if n, err := f.s.UnclaimNotifications(c.Token, []string{id}); err != nil || n != 1 { + t.Fatalf("unclaim = %d, %v", n, err) + } + state, _, tok, attempts := f.notifRow(t, id) + if state != "pending" || tok != "" || attempts != 0 { + t.Fatalf("row = %s/%q/%d, want pending, unclaimed, 0 attempts", state, tok, attempts) + } + if again, _ := f.s.ClaimPendingNotifications("Y", 10, at(10*time.Second), at(time.Minute)); len(again.Rows) != 1 { + t.Fatal("an unclaimed row is claimable at once") + } + if _, err := f.s.UnclaimNotifications("", []string{id}); err == nil { + t.Fatal("empty token accepted") + } +} + +// Kills: HeldNotifications ignoring the token, or the pending state. +func TestHeldNotifications(t *testing.T) { + t.Parallel() + f := newNotifyFixture(t) + u := notifyUser(t, f.s, "held@example.com") + a, b, c := f.seedNotif(t, u.ID, ago(time.Minute)), f.seedNotif(t, u.ID, ago(time.Minute)), f.seedNotif(t, u.ID, ago(time.Minute)) + cl, _ := f.s.ClaimPendingNotifications("X", 10, at(10*time.Second), at(time.Minute)) + if len(cl.Rows) != 3 { + t.Fatal("claimed", len(cl.Rows)) + } + f.s.mustExec(t, `UPDATE user_notifications SET push_state = 'sent' WHERE id = ?`, b) + f.s.mustExec(t, `DELETE FROM user_notifications WHERE id = ?`, c) + got, err := f.s.HeldNotifications(cl.Token, []string{a, b, c}) + if err != nil || !slices.Equal(got, []string{a}) { + t.Fatalf("held = %v, %v; want only %s", got, err, a) + } + if got, _ := f.s.HeldNotifications("X.other", []string{a}); len(got) != 0 { + t.Fatalf("another token holds %v", got) + } + if _, err := f.s.HeldNotifications("", []string{a}); err == nil { + t.Fatal("empty token accepted") + } +} + +// Kills: putting a row id back into the claim error text (Q8). +func TestClaimNotificationErrorCarriesNoRowID(t *testing.T) { + t.Parallel() + f := newNotifyFixture(t) + u := notifyUser(t, f.s, "errtxt@example.com") + id := f.seedNotif(t, u.ID, ago(time.Minute)) + f.s.mustExec(t, `ALTER TABLE user_notifications RENAME TO user_notifications_x`) + defer f.s.mustExec(t, `ALTER TABLE user_notifications_x RENAME TO user_notifications`) + _, err := f.s.claimNotificationIDs("X", []string{id}, cutoffTS(at(10*time.Second)), cutoffTS(at(time.Minute)), now()) + if err == nil { + t.Fatal("expected an error with the table renamed") + } + if strings.Contains(err.Error(), id) { + t.Fatalf("claim error carries the row id: %v", err) + } +} + +// Kills: RecordNotificationsSent or HeldNotifications handling only the first +// notificationIDChunk ids of a longer list. +func TestNotificationSettles_CoverMoreIDsThanOneChunk(t *testing.T) { + t.Parallel() + f := newNotifyFixture(t) + u := notifyUser(t, f.s, "chunk@example.com") + ids := make([]string, notificationIDChunk*2+7) + for i := range ids { + ids[i] = f.seedNotif(t, u.ID, ago(time.Minute)) + } + f.s.mustExec(t, `UPDATE user_notifications SET claim_token = 'L.n', claimed_at = ? WHERE user_id = ?`, ago(time.Second), u.ID) + held, err := f.s.HeldNotifications("L.n", ids) + if err != nil || len(held) != len(ids) { + t.Fatalf("held %d of %d ids, %v", len(held), len(ids), err) + } + if n, err := f.s.RecordNotificationsSent(ids); err != nil || n != int64(len(ids)) { + t.Fatalf("recorded %d of %d ids, %v", n, len(ids), err) + } + last := ids[len(ids)-1] + if state, _, _, _ := f.notifRow(t, last); state != "sent" { + t.Fatalf("the last id is %s, want sent", state) + } +}