tasks.ts · Source

tasks.tsapps/worker/src/processors/tasks.tsraw ← processors7,492 B← Back to the app
1/**
2 * T-160 — Task due-date reminders (worker core).
3 *
4 * Hourly sweep (maintenance job `task-due-sweep`): for every OPEN task whose
5 * `dueAt` falls in `(now − 1h, now + 24h]`, notify each assignee with a
6 * `TASK_DUE` in-app notification — the 1h look-back covers a sweep that runs
7 * just after the deadline. Idempotency WITHOUT a schema change (additive-only
8 * invariant): a Redis flag per (task, dueAt) — `worker:task-due:{taskId}:
9 * {dueAtMs}` — so a re-run is a no-op and a CHANGED dueAt re-reminds (the
10 * key embeds the timestamp). Same dedupe pattern as the webhook DLQ report.
11 */
12import type { Redis } from "ioredis";
13import type { PrismaClient } from "@prisma/client";
14import { asJson } from "../lib/prisma.js";
15import { getLogger } from "../logger.js";
16 
17const log = getLogger("tasks");
18 
19const REMIND_WINDOW_MS = 24 * 3_600_000; // remind up to 24h before the due date
20const LOOKBACK_MS = 3_600_000; // cover the hour a sweep runs past the due date
21const FLAG_TTL_S = 3 * 86_400; // the flag outlives the window (no re-remind)
22 
23const dueFlag = (taskId: string, dueAt: Date): string =>
24  `worker:task-due:${taskId}:${Math.floor(dueAt.getTime() / 1000)}`;
25 
26const reviewDueFlag = (reviewId: string, deadline: Date): string =>
27  `worker:review-due:${reviewId}:${Math.floor(deadline.getTime() / 1000)}`;
28 
29/**
30 * Claim a once-only reminder (Redis NX+EX). The FIRST caller to claim a key
31 * wins (returns true); later runs / concurrent workers get false → skip. The
32 * key embeds the due/deadline timestamp, so a CHANGED time re-reminds.
33 * Best-effort: if Redis is down we still send (a duplicate beat a missed
34 * reminder here).
35 */
36const claimOnce = async (redis: Redis, key: string): Promise<boolean> =>
37  redis
38    .set(key, "1", "EX", FLAG_TTL_S, "NX")
39    .then((r) => r === "OK")
40    .catch(() => true);
41 
42/**
43 * Run the due-reminder sweep. Returns counts (the report lands in the
44 * maintenance job log). `now` is injectable for deterministic tests.
45 */
46export async function runTaskDueSweep(
47  prisma: PrismaClient,
48  redis: Redis,
49  now: Date = new Date()
50): Promise<{ checked: number; reminded: number; notifications: number }> {
51  const from = new Date(now.getTime() - LOOKBACK_MS);
52  const to = new Date(now.getTime() + REMIND_WINDOW_MS);
53  const rows = await prisma.task.findMany({
54    where: {
55      status: { in: ["todo", "in_progress"] },
56      dueAt: { gt: from, lte: to }
57    },
58    select: { id: true, workspaceId: true, pageId: true, title: true, dueAt: true, assigneeIds: true, status: true }
59  });
60  const pageRows = await prisma.page.findMany({
61    where: { id: { in: rows.map((r) => r.pageId) } },
62    select: { id: true, trashedAt: true }
63  });
64  const pages = new Map(pageRows.map((r) => [r.id, r]));
65 
66  let reminded = 0;
67  let notifications = 0;
68  for (const t of rows) {
69    if (!t.assigneeIds.length) continue; // no assignees → nothing to remind (before the flag claim)
70    const page = pages.get(t.pageId);
71    if (!page || page.trashedAt) continue; // trashed → no reminder (before the flag claim)
72    const dueAt = t.dueAt!;
73    // NX + EX: the FIRST sweep to claim a (task, dueAt) sends the reminders;
74    // every later run (or a concurrent worker) skips.
75    if (!(await claimOnce(redis, dueFlag(t.id, dueAt)))) continue;
76    reminded++;
77    const dueIso = dueAt.toISOString();
78    const title = `Task due: ${t.title || "Untitled"}`;
79    const body = `Due ${dueIso} — status: ${t.status}.`;
80    for (const userId of t.assigneeIds) {
81      try {
82        const row = await prisma.notification.create({
83          data: {
84            workspaceId: t.workspaceId,
85            userId,
86            type: "TASK_DUE",
87            channel: "IN_APP",
88            title,
89            body,
90            link: asJson({ type: "page", id: t.pageId, pageId: t.pageId })
91          },
92          select: { id: true }
93        });
94        // The worker owns the email channel: publish the same
95        // `notification.created` event the API publishes (the email-decision
96        // consumer applies per-user prefs; no actorId → SYSTEM actor).
97        await redis
98          .xadd("factory:events", "*", "event", "notification.created", "payload", JSON.stringify({
99            workspaceId: t.workspaceId,
100            notificationId: row.id,
101            userId,
102            type: "TASK_DUE",
103            title,
104            deepLink: { type: "page", id: t.pageId, pageId: t.pageId }
105          }))
106          .catch(() => undefined);
107        notifications++;
108      } catch (err) {
109        log.warn({ err: (err as Error).message, taskId: t.id, userId }, "task-due notification failed (non-fatal)");
110      }
111    }
112  }
113  log.info({ checked: rows.length, reminded, notifications }, "task due sweep done");
114  return { checked: rows.length, reminded, notifications };
115}
116 
117/**
118 * T-162/T-163 — review DEADLINE reminders (reuses the T-160 pattern). For
119 * every OPEN review whose `deadline` falls in `(now − 1h, now + 24h]`, remind
120 * each reviewer of record (not the requester) with a `REVIEW_REQUESTED`
121 * "deadline" notification. Same NX-flag idempotency (key = reviewId +
122 * deadline ts). Hourly, offset from the task sweep.
123 */
124export async function runReviewDueSweep(
125  prisma: PrismaClient,
126  redis: Redis,
127  now: Date = new Date()
128): Promise<{ checked: number; reminded: number; notifications: number }> {
129  const from = new Date(now.getTime() - LOOKBACK_MS);
130  const to = new Date(now.getTime() + REMIND_WINDOW_MS);
131  const rows = await prisma.review.findMany({
132    where: { closedAt: null, state: "in_review", deadline: { gt: from, lte: to } },
133    select: {
134      id: true,
135      pageId: true,
136      reviewerIds: true,
137      requestedBy: true,
138      deadline: true,
139      page: { select: { workspaceId: true, title: true, trashedAt: true } }
140    }
141  });
142  let reminded = 0;
143  let notifications = 0;
144  for (const r of rows) {
145    if (!r.page || r.page.trashedAt) continue; // trashed page → no reminder
146    if (!r.reviewerIds.length) continue;
147    const deadline = r.deadline!;
148    if (!(await claimOnce(redis, reviewDueFlag(r.id, deadline)))) continue;
149    reminded++;
150    const title = `Review deadline: ${r.page.title || "Untitled page"}`;
151    const body = `Decision due ${deadline.toISOString()}.`;
152    for (const userId of r.reviewerIds) {
153      if (userId === r.requestedBy) continue; // never self-remind the requester
154      try {
155        const row = await prisma.notification.create({
156          data: {
157            workspaceId: r.page.workspaceId,
158            userId,
159            type: "REVIEW_REQUESTED",
160            channel: "IN_APP",
161            title,
162            body,
163            link: asJson({ type: "page", id: r.pageId, pageId: r.pageId })
164          },
165          select: { id: true }
166        });
167        await redis
168          .xadd("factory:events", "*", "event", "notification.created", "payload", JSON.stringify({
169            workspaceId: r.page.workspaceId,
170            notificationId: row.id,
171            userId,
172            type: "REVIEW_REQUESTED",
173            title,
174            deepLink: { type: "page", id: r.pageId, pageId: r.pageId }
175          }))
176          .catch(() => undefined);
177        notifications++;
178      } catch (err) {
179        log.warn({ err: (err as Error).message, reviewId: r.id, userId }, "review-due notification failed (non-fatal)");
180      }
181    }
182  }
183  log.info({ checked: rows.length, reminded, notifications }, "review due sweep done");
184  return { checked: rows.length, reminded, notifications };
185}
186