1import { eq, sql } from "drizzle-orm";
2import * as Context from "effect/Context";
3import * as Effect from "effect/Effect";
4import * as Layer from "effect/Layer";
5
6import * as RelayDb from "../db.ts";
7import { relayUpstreamApiPauses } from "../persistence/schema.ts";
8
9const CLOUDFLARE_API = "cloudflare";
10
11// Every SqlClient's transaction connection lives under a key with this prefix.
12// Matching by key name instead of `db.$client.transactionService` because the
13// relay's database is an alchemy proxy, whose properties are proxies too, not
14// the real key.
15const SQL_TRANSACTION_KEY_PREFIX = "effect/sql/SqlClient/TransactionConnection/";
16
17const withoutSqlTransaction = (context: Context.Context<never>): Context.Context<never> =>
18 Context.makeUnsafe(
19 new Map(
20 Array.from(context.mapUnsafe).filter(([key]) => !key.startsWith(SQL_TRANSACTION_KEY_PREFIX)),
21 ),
22 );
23
24/**
25 * Cloudflare's account-wide 429 pause is shared across isolates and must fail open.
26 * Run outside provisioning transactions: rollback would lose a recorded pause,
27 * and a failed pause read would abort provisioning's own statements.
28 */
29export class CloudflareApiPause extends Context.Service<
30 CloudflareApiPause,
31 {
32 /** Epoch milliseconds until which calls are paused, or 0. */
33 readonly pausedUntilMillis: Effect.Effect<number>;
34 readonly pauseUntil: (untilMillis: number) => Effect.Effect<void>;
35 }
36>()("t3code-relay/environments/CloudflareApiPause") {}
37
38const make = Effect.gen(function* () {
39 const db = yield* RelayDb.RelayDb;
40 const outsideTransaction = <A, E>(effect: Effect.Effect<A, E>) =>
41 Effect.updateContext(effect, withoutSqlTransaction);
42
43 return CloudflareApiPause.of({
44 pausedUntilMillis: db
45 .select({ pausedUntilMs: relayUpstreamApiPauses.pausedUntilMs })
46 .from(relayUpstreamApiPauses)
47 .where(eq(relayUpstreamApiPauses.api, CLOUDFLARE_API))
48 .limit(1)
49 .pipe(
50 Effect.map((rows) => rows[0]?.pausedUntilMs ?? 0),
51 Effect.catch((cause) =>
52 Effect.logWarning("Could not read the shared Cloudflare API pause", { cause }).pipe(
53 Effect.as(0),
54 ),
55 ),
56 outsideTransaction,
57 Effect.withSpan("relay.cloudflare_api_pause.read"),
58 ),
59 pauseUntil: Effect.fn("relay.cloudflare_api_pause.record")(function (untilMillis) {
60 return db
61 .insert(relayUpstreamApiPauses)
62 .values({ api: CLOUDFLARE_API, pausedUntilMs: untilMillis })
63 .onConflictDoUpdate({
64 target: relayUpstreamApiPauses.api,
65 set: {
66 pausedUntilMs: sql`greatest(${relayUpstreamApiPauses.pausedUntilMs}, excluded.paused_until_ms)`,
67 },
68 })
69 .pipe(
70 Effect.asVoid,
71 Effect.catch((cause) =>
72 Effect.logWarning("Could not record the shared Cloudflare API pause", { cause }),
73 ),
74 outsideTransaction,
75 );
76 }),
77 });
78});
79
80export const layer = Layer.effect(CloudflareApiPause, make);