1// @effect-diagnostics-next-line nodeBuiltinImport:off -- Keys need a deterministic digest, not random bytes.
2import * as NodeCrypto from "node:crypto";
3import { ScheduledTask, ScheduledTaskError } from "@t3tools/contracts";
4import * as Effect from "effect/Effect";
5import * as Schema from "effect/Schema";
6import * as SqlClient from "effect/sql/SqlClient";
7import { renderWebhookPrompt, type WebhookRequest } from "./webhookTemplate.ts";
8
9function key(template: string | undefined, request: WebhookRequest): string {
10 if (template === undefined) return "";
11 const rendered = renderWebhookPrompt(template, request, "", true);
12 return rendered.missing.length > 0 || rendered.prompt === ""
13 ? ""
14 : NodeCrypto.createHash("sha256").update(rendered.prompt).digest("hex");
15}
16
17export const releaseWebhookReplay = Effect.fn("ScheduledTaskService.releaseWebhookReplay")(
18 function* (deliveryId: string) {
19 const sql = yield* SqlClient.SqlClient;
20 yield* sql`DELETE FROM scheduled_task_webhook_replay_keys WHERE delivery_id = ${deliveryId}`;
21 },
22);
23
24const hasReplayKey = Effect.fn(function* (taskId: string, replayKey: string, now: number) {
25 const sql = yield* SqlClient.SqlClient;
26 return (
27 replayKey !== "" &&
28 (yield* sql`SELECT 1 FROM scheduled_task_webhook_replay_keys
29 WHERE task_id = ${taskId} AND replay_key = ${replayKey} AND expires_at > ${now}`).length > 0
30 );
31});
32const claimReplayKey = Effect.fn(function* (
33 taskId: string,
34 replayKey: string,
35 deliveryId: string,
36 now: number,
37) {
38 if (replayKey === "") return;
39 const sql = yield* SqlClient.SqlClient;
40 yield* sql`DELETE FROM scheduled_task_webhook_replay_keys WHERE expires_at <= ${now}`;
41 yield* sql`INSERT INTO scheduled_task_webhook_replay_keys (task_id, replay_key, delivery_id, expires_at)
42 VALUES (${taskId}, ${replayKey}, ${deliveryId}, ${now + 7 * 24 * 60 * 60 * 1000})`;
43});
44
45export const acceptWebhookDedupe = Effect.fn("ScheduledTaskService.acceptWebhookDedupe")(function* (
46 task: ScheduledTask,
47 request: WebhookRequest,
48 deliveryId: string,
49 now: number,
50 reserve: Effect.Effect<void, "first_rejected" | "rejected">,
51 reserveCoalescedLog: Effect.Effect<boolean>,
52 checkCapacity: Effect.Effect<void, "queue_full" | ScheduledTaskError>,
53 replaceDeliveryRequest: (ownerId: string) => Effect.Effect<void, ScheduledTaskError>,
54) {
55 const sql = yield* SqlClient.SqlClient;
56 const dedupe = task.schedule.type === "webhook" ? task.schedule.dedupe : undefined;
57 if (dedupe === undefined) return "run" as const;
58 const replayKey = key(dedupe.replayKey, request);
59 if (yield* hasReplayKey(task.id, replayKey, now)) return "duplicate" as const;
60 const coalesceKey = key(dedupe.coalesceKey, request);
61 if (coalesceKey === "") return "run" as const;
62 const pending = yield* sql<{
63 delivery_id: string;
64 }>`SELECT delivery_id FROM scheduled_task_webhook_pending
65 WHERE task_id = ${task.id} AND coalesce_key = ${coalesceKey} AND state = 'pending' AND due_at > ${now}`;
66 const logCoalesced = pending.length > 0 ? yield* reserveCoalescedLog : true;
67 if (pending.length === 0) {
68 yield* checkCapacity;
69 yield* reserve;
70 }
71 // Coalesced requests retain their validation slot, bounding writes without dropping accepted payloads.
72 yield* claimReplayKey(task.id, replayKey, pending[0]?.delivery_id ?? deliveryId, now);
73 const requestJson = yield* encodePendingRequest({
74 method: request.method,
75 path: request.path,
76 query: request.query,
77 headers: request.headers,
78 bodyText: request.bodyText,
79 });
80 if (pending.length > 0) {
81 yield* sql`UPDATE scheduled_task_webhook_pending SET request_json = ${requestJson},
82 delivery_ids_json = json_insert(delivery_ids_json, '$[#]', ${deliveryId}) WHERE delivery_id = ${pending[0]!.delivery_id}`;
83 yield* replaceDeliveryRequest(pending[0]!.delivery_id);
84 return logCoalesced ? ("coalesced" as const) : ("coalesced_unlogged" as const);
85 }
86 // Expired windows are immutable, even if the scheduler has not dispatched them yet.
87 yield* sql`UPDATE scheduled_task_webhook_pending SET state = 'dispatching'
88 WHERE task_id = ${task.id} AND coalesce_key = ${coalesceKey} AND state = 'pending' AND due_at <= ${now}`;
89 yield* sql`INSERT INTO scheduled_task_webhook_pending (delivery_id, task_id, coalesce_key, due_at, task_json, request_json, delivery_ids_json)
90 VALUES (${deliveryId}, ${task.id}, ${coalesceKey}, ${now + (dedupe.coalesceWindowSeconds ?? 120) * 1000}, ${yield* encodePendingTask(task)}, ${requestJson}, json_array(${deliveryId}))`;
91 return "pending" as const;
92});
93
94export const acceptImmediateWebhookDedupe = Effect.fn(
95 "ScheduledTaskService.acceptImmediateWebhookDedupe",
96)(function* (
97 task: ScheduledTask,
98 request: WebhookRequest,
99 deliveryId: string,
100 now: number,
101 reserve: Effect.Effect<void, "first_rejected" | "rejected" | "queue_full" | ScheduledTaskError>,
102) {
103 const replayKey = key(
104 task.schedule.type === "webhook" ? task.schedule.dedupe?.replayKey : undefined,
105 request,
106 );
107 if (yield* hasReplayKey(task.id, replayKey, now)) return "duplicate" as const;
108 yield* reserve;
109 yield* claimReplayKey(task.id, replayKey, deliveryId, now);
110 return "run" as const;
111});
112
113const PendingTaskJson = Schema.fromJsonString(ScheduledTask);
114export const encodePendingTask = Schema.encodeEffect(PendingTaskJson);
115export const decodePendingTask = Schema.decodeUnknownEffect(PendingTaskJson);
116const PendingRequestJson = Schema.fromJsonString(
117 Schema.Struct({
118 method: Schema.String,
119 path: Schema.String,
120 query: Schema.String,
121 headers: Schema.Record(Schema.String, Schema.String),
122 bodyText: Schema.String,
123 }),
124);
125const encodePendingRequest = Schema.encodeEffect(PendingRequestJson);
126export const decodePendingRequest = Schema.decodeUnknownEffect(PendingRequestJson);
127export const dedupeError = (cause: unknown) =>
128 new ScheduledTaskError({ message: "Could not persist webhook dedupe state.", cause });