1import * as SqliteClient from "@effect/sql-sqlite-do/SqliteClient";
2import type * as Alchemy from "alchemy";
3import * as Cloudflare from "@alchemy.run/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
13const UNREACHABLE_STATUSES = new Set([502, 503, 504, TUNNEL_OFFLINE_STATUS]);
14/** Retry queue-full (429) and handling failures (500); the environment deduplicates delivery ids. */
15const BUSY_STATUSES = new Set([429, 500]);
16const RUN_FAILURE_RETRY_MS = 60_000;
17
18type Call<A> = Effect.Effect<A, never, Alchemy.RuntimeContext>;
19
20export class HookInboxObject extends Cloudflare.DurableObject<
21 HookInboxObject,
22 {
23 /** Holds a request for `baseUrl`; false when the inbox is full and nothing was stored. */
24 readonly hold: (hook: HookInboxStore.HeldHook, baseUrl: string) => Call<boolean>;
25 readonly wake: (baseUrl: string) => Call<boolean>;
26 readonly clear: () => Call<void>;
27 }
28>()("HookInboxObject") {}
29
30const withInboxSpan =
31 (name: string, inboxId: string) =>
32 <A, E, R>(effect: Effect.Effect<A, E, R>) =>
33 effect.pipe(
34 Effect.withSpan(name, { root: true, attributes: { "relay.hook.endpoint_key": inboxId } }),
35 );
36
37const deliver = (baseUrl: string, hook: HookInboxStore.HeldHook) =>
38 sendUpstream(baseUrl, hook).pipe(
39 Effect.result,
40 Effect.map((result) => {
41 // Timeouts may have executed; retries reuse delivery IDs for environment deduplication.
42 if (Result.isFailure(result)) {
43 return { outcome: "unreachable" as const, reason: result.failure._tag };
44 }
45 if (Option.isNone(result.success))
46 return { outcome: "unreachable" as const, reason: "timeout" };
47 const status = result.success.value.status;
48 const upstreamOutcome = result.success.value.outcome;
49 const outcome: HookInboxStore.DeliveryOutcome = UNREACHABLE_STATUSES.has(status)
50 ? "unreachable"
51 : BUSY_STATUSES.has(status)
52 ? "busy"
53 : "delivered";
54 return { outcome, reason: `status ${status}`, upstreamOutcome };
55 }),
56 Effect.tap(({ upstreamOutcome }) =>
57 upstreamOutcome === undefined
58 ? Effect.void
59 : Effect.annotateCurrentSpan({ "relay.hook.upstream_outcome": upstreamOutcome }),
60 ),
61 Effect.tap(({ outcome, reason }) =>
62 outcome === "delivered"
63 ? Effect.void
64 : Effect.logInfo("Held webhook not delivered yet", {
65 outcome,
66 reason,
67 deliveryId: hook.id,
68 }),
69 ),
70 Effect.map(({ outcome }) => outcome),
71 Effect.provide(FetchHttpClient.layer),
72 );
73
74export const layer = HookInboxObject.make(
75 Effect.gen(function* () {
76 const state = yield* Cloudflare.DurableObjectState;
77 // Only the per-object inner Effect may access storage.
78 return Effect.gen(function* () {
79 const layerSql = SqliteClient.layer({ storage: state.raw.storage });
80 const inboxId = state.raw.id.name ?? state.raw.id.toString();
81 const run = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
82 effect.pipe(Effect.provide(layerSql), Effect.orDie);
83 yield* run(HookInboxStore.migrate);
84
85 const scheduleBy = Effect.fn("HookInboxObject.scheduleBy")(function* (at: number) {
86 const current = yield* state.storage.getAlarm();
87 if (current === null || current > at) yield* state.storage.setAlarm(at);
88 });
89
90 // withInboxSpan supplies each method's root span; avoid a second wrapper span.
91 return {
92 hold: Effect.fnUntraced(
93 function* (hook: HookInboxStore.HeldHook, baseUrl: string) {
94 const dueAt = yield* run(HookInboxStore.hold(hook, baseUrl));
95 yield* Effect.annotateCurrentSpan({ "relay.inbox.stored": dueAt !== null });
96 if (dueAt === null) return false;
97 yield* scheduleBy(dueAt);
98 return true;
99 },
100 withInboxSpan("relay.inbox.hold", inboxId),
101 ),
102 wake: Effect.fnUntraced(
103 function* (baseUrl: string) {
104 const pending = yield* run(HookInboxStore.wake(baseUrl));
105 yield* Effect.annotateCurrentSpan({ "relay.inbox.pending": pending });
106 if (pending) yield* state.storage.setAlarm(yield* Clock.currentTimeMillis);
107 return pending;
108 },
109 withInboxSpan("relay.inbox.wake", inboxId),
110 ),
111 clear: Effect.fnUntraced(
112 function* () {
113 yield* run(HookInboxStore.clear);
114 yield* state.storage.deleteAlarm();
115 },
116 withInboxSpan("relay.inbox.clear", inboxId),
117 ),
118 alarm: () =>
119 run(HookInboxStore.deliverDue(deliver)).pipe(
120 // Backoff must not delay an earlier run requested by a concurrent wake.
121 Effect.flatMap((nextAt) => (nextAt === null ? Effect.void : scheduleBy(nextAt))),
122 Effect.catchCause((cause) =>
123 Effect.logWarning("Held webhook delivery run failed", { cause }).pipe(
124 Effect.andThen(Effect.annotateCurrentSpan({ "relay.inbox.run_result": "failed" })),
125 Effect.andThen(Clock.currentTimeMillis),
126 Effect.flatMap((now) => scheduleBy(now + RUN_FAILURE_RETRY_MS)),
127 ),
128 ),
129 withInboxSpan("relay.inbox.deliver", inboxId),
130 ),
131 };
132 });
133 }),
134);