1import * as SqliteClient from "@effect/sql-sqlite-do/SqliteClient";
2import type * as Alchemy from "alchemy";
3import * as Cloudflare from "alchemy/Cloudflare";
4import * as Clock from "effect/Clock";
5import * as Effect from "effect/Effect";
6import * as Option from "effect/Option";
7import * as Result from "effect/Result";
8import * as FetchHttpClient from "effect/http/FetchHttpClient";
9
10import * as HookInboxStore from "./HookInboxStore.ts";
11import { sendUpstream, TUNNEL_OFFLINE_STATUS } from "./upstream.ts";
12
13/** Statuses that mean the environment is not there; the whole inbox waits and backs off. */
14const UNREACHABLE_STATUSES = new Set([502, 503, 504, TUNNEL_OFFLINE_STATUS]);
15/**
16 * Statuses that mean the environment is there but did not take this request
17 * yet: its task's queue is full (429) or it failed while handling it (500).
18 * The environment drops a delivery id it has already run, so retrying is safe.
19 */
20const BUSY_STATUSES = new Set([429, 500]);
21/** When a run itself fails, the next one is tried after this long. */
22const RUN_FAILURE_RETRY_MS = 60_000;
23
24type Call<A> = Effect.Effect<A, never, Alchemy.RuntimeContext>;
25
26/**
27 * One per managed endpoint, addressed by endpoint key. Holds webhook requests
28 * the environment could not take, in SQLite, and pushes them back through
29 * its tunnel from the alarm, oldest first, backing off while it stays away.
30 */
31export class HookInboxObject extends Cloudflare.DurableObject<
32 HookInboxObject,
33 {
34 /** Holds a request for `baseUrl`; false when the inbox is full and nothing was stored. */
35 readonly hold: (hook: HookInboxStore.HeldHook, baseUrl: string) => Call<boolean>;
36 /** The environment is back at `baseUrl`: deliver what is waiting now. */
37 readonly wake: (baseUrl: string) => Call<boolean>;
38 readonly clear: () => Call<void>;
39 }
40>()("HookInboxObject") {}
41
42/**
43 * Each call into an inbox is its own trace: the alarm has no parent, and a
44 * hold arrives over Durable Object RPC without the forwarding span. The
45 * object's name is the endpoint key, which the forward spans carry too.
46 */
47const withInboxSpan =
48 (name: string, inboxId: string) =>
49 <A, E, R>(effect: Effect.Effect<A, E, R>) =>
50 effect.pipe(
51 Effect.withSpan(name, { root: true, attributes: { "relay.hook.endpoint_key": inboxId } }),
52 );
53
54const deliver = (baseUrl: string, hook: HookInboxStore.HeldHook) =>
55 sendUpstream(baseUrl, hook).pipe(
56 Effect.result,
57 Effect.map((result) => {
58 // Unreachable or timed out: a timeout may still have run it, and the
59 // environment drops a delivery id it has already seen.
60 if (Result.isFailure(result)) {
61 return { outcome: "unreachable" as const, reason: result.failure._tag };
62 }
63 if (Option.isNone(result.success))
64 return { outcome: "unreachable" as const, reason: "timeout" };
65 const status = result.success.value.status;
66 const upstreamOutcome = result.success.value.outcome;
67 const outcome: HookInboxStore.DeliveryOutcome = UNREACHABLE_STATUSES.has(status)
68 ? "unreachable"
69 : BUSY_STATUSES.has(status)
70 ? "busy"
71 : "delivered";
72 return { outcome, reason: `status ${status}`, upstreamOutcome };
73 }),
74 Effect.tap(({ upstreamOutcome }) =>
75 upstreamOutcome === undefined
76 ? Effect.void
77 : Effect.annotateCurrentSpan({ "relay.hook.upstream_outcome": upstreamOutcome }),
78 ),
79 Effect.tap(({ outcome, reason }) =>
80 outcome === "delivered"
81 ? Effect.void
82 : Effect.logInfo("Held webhook not delivered yet", {
83 outcome,
84 reason,
85 deliveryId: hook.id,
86 }),
87 ),
88 Effect.map(({ outcome }) => outcome),
89 Effect.provide(FetchHttpClient.layer),
90 );
91
92export const layer = HookInboxObject.make(
93 Effect.gen(function* () {
94 const state = yield* Cloudflare.DurableObjectState;
95 // The init phase returns the per-instance Effect, which alchemy runs once
96 // per object; only that inner Effect may touch storage.
97 // @effect-diagnostics-next-line returnEffectInGen:off
98 return Effect.gen(function* () {
99 const layerSql = SqliteClient.layer({ storage: state.raw.storage });
100 // Inboxes are opened by endpoint key, so spans line up with the
101 // forward spans that held their requests.
102 const inboxId = state.raw.id.name ?? state.raw.id.toString();
103 const run = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
104 effect.pipe(Effect.provide(layerSql), Effect.orDie);
105 yield* run(HookInboxStore.migrate);
106
107 /** Moves the alarm to `at`, unless one is already due sooner. */
108 const scheduleBy = (at: number) =>
109 Effect.gen(function* () {
110 const current = yield* state.storage.getAlarm();
111 if (current === null || current > at) yield* state.storage.setAlarm(at);
112 });
113
114 return {
115 hold: (hook: HookInboxStore.HeldHook, baseUrl: string) =>
116 Effect.gen(function* () {
117 const dueAt = yield* run(HookInboxStore.hold(hook, baseUrl));
118 yield* Effect.annotateCurrentSpan({ "relay.inbox.stored": dueAt !== null });
119 if (dueAt === null) return false;
120 yield* scheduleBy(dueAt);
121 return true;
122 }).pipe(withInboxSpan("relay.inbox.hold", inboxId)),
123 wake: (baseUrl: string) =>
124 Effect.gen(function* () {
125 const pending = yield* run(HookInboxStore.wake(baseUrl));
126 yield* Effect.annotateCurrentSpan({ "relay.inbox.pending": pending });
127 if (pending) yield* state.storage.setAlarm(yield* Clock.currentTimeMillis);
128 return pending;
129 }).pipe(withInboxSpan("relay.inbox.wake", inboxId)),
130 clear: () =>
131 Effect.gen(function* () {
132 yield* run(HookInboxStore.clear);
133 yield* state.storage.deleteAlarm();
134 }).pipe(withInboxSpan("relay.inbox.clear", inboxId)),
135 alarm: () =>
136 run(HookInboxStore.deliverDue(deliver)).pipe(
137 // A wake during this run may already have asked for an earlier
138 // run; a backoff must not push it out.
139 Effect.flatMap((nextAt) => (nextAt === null ? Effect.void : scheduleBy(nextAt))),
140 Effect.catchCause((cause) =>
141 Effect.logWarning("Held webhook delivery run failed", { cause }).pipe(
142 Effect.andThen(Effect.annotateCurrentSpan({ "relay.inbox.run_result": "failed" })),
143 Effect.andThen(Clock.currentTimeMillis),
144 Effect.flatMap((now) => scheduleBy(now + RUN_FAILURE_RETRY_MS)),
145 ),
146 ),
147 withInboxSpan("relay.inbox.deliver", inboxId),
148 ),
149 };
150 });
151 }),
152);