1import * as Effect from "effect/Effect";
2import * as Layer from "effect/Layer";
3import * as Schedule from "effect/Schedule";
4import * as Stream from "effect/Stream";
5
6import * as CloudManagedEndpointRuntime from "../cloud/ManagedEndpointRuntime.ts";
7import { readHoldWebhooksWhileOffline, readRelayConnection } from "../cloud/config.ts";
8import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
9import { makeRelayEnvironmentClient } from "./relayEnvironmentClient.ts";
10
11/**
12 * Tells T3 Connect this environment is reachable again, so the relay delivers
13 * the webhook requests it held while we were offline now rather than at its
14 * next backoff step. Nothing happens unless the environment opted in.
15 */
16const wakeHeldHooks = Effect.fn("HeldHooksWaker.wake")(function* () {
17 if (!(yield* readHoldWebhooksWhileOffline)) return false;
18 const connection = yield* readRelayConnection;
19 if (connection === null) return false;
20 const environmentId = yield* (yield* ServerEnvironment.ServerEnvironment).getEnvironmentId;
21 const client = yield* makeRelayEnvironmentClient(connection);
22 const { pending } = yield* client.server.wakeHeldHooks({ params: { environmentId } });
23 yield* Effect.annotateCurrentSpan({ "relay.inbox.pending": pending });
24 return pending;
25});
26
27/** Wakes held webhooks each time the managed tunnel connects. */
28export const layer = Layer.effectDiscard(
29 Effect.gen(function* () {
30 const runtime = yield* CloudManagedEndpointRuntime.CloudManagedEndpointRuntime;
31 const wake = wakeHeldHooks().pipe(
32 Effect.timeout("10 seconds"),
33 // The relay retries on its own schedule too, so a few tries are enough.
34 Effect.retry({ schedule: Schedule.exponential("2 seconds"), times: 3 }),
35 Effect.tap((pending) =>
36 pending ? Effect.logInfo("T3 Connect is delivering held webhook requests") : Effect.void,
37 ),
38 Effect.catchCause((cause) =>
39 Effect.logWarning("Could not ask T3 Connect to deliver held webhook requests", { cause }),
40 ),
41 );
42 yield* runtime.tunnelConnected.pipe(
43 // cloudflared registers several connections per (re)connect within a
44 // few seconds; they are one wake.
45 Stream.debounce("3 seconds"),
46 Stream.runForEach(() => wake),
47 Effect.forkScoped,
48 );
49 }),
50);