Open History to see the last commit
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
11const wakeHeldHooks = Effect.fn("HeldHooksWaker.wake")(function* () {
12 if (!(yield* readHoldWebhooksWhileOffline)) return false;
13 const connection = yield* readRelayConnection;
14 if (connection === null) return false;
15 const environmentId = yield* (yield* ServerEnvironment.ServerEnvironment).getEnvironmentId;
16 const client = yield* makeRelayEnvironmentClient(connection);
17 const { pending } = yield* client.server.wakeHeldHooks({ params: { environmentId } });
18 yield* Effect.annotateCurrentSpan({ "relay.inbox.pending": pending });
19 return pending;
20});
21
22export const layer = Layer.effectDiscard(
23 Effect.gen(function* () {
24 const runtime = yield* CloudManagedEndpointRuntime.CloudManagedEndpointRuntime;
25 const wake = wakeHeldHooks().pipe(
26 Effect.timeout("10 seconds"),
27 Effect.retry({ schedule: Schedule.exponential("2 seconds"), times: 3 }),
28 Effect.tap((pending) =>
29 pending ? Effect.logInfo("T3 Connect is delivering held webhook requests") : Effect.void,
30 ),
31 Effect.catchCause((cause) =>
32 Effect.logWarning("Could not ask T3 Connect to deliver held webhook requests", { cause }),
33 ),
34 );
35 yield* runtime.tunnelConnected.pipe(
36 // Coalesce cloudflared's several connections per reconnect into one wake.
37 Stream.debounce("3 seconds"),
38 Stream.runForEach(() => wake),
39 Effect.forkScoped,
40 );
41 }),
42);