fix(relay): bound Firebase authorization and delivery requests

Ryan Hughes committed 2 months ago

Commit ec01afc1 parent 7c1d61f2 files · +63 −2Browse files

Unified · Split

2 files

infra/relay/src/agentActivity/FcmClient.test.ts +51 −0
infra/relay/src/agentActivity/FcmClient.ts +12 −2

infra/relay/src/agentActivity/FcmClient.test.ts+51−0

@@ -1,9 +1,12 @@
1	import * as NodeCrypto from "node:crypto";	1	import * as NodeCrypto from "node:crypto";
2	import { describe, expect, it } from "@effect/vitest";	2	import { describe, expect, it } from "@effect/vitest";
		3	import * as Deferred from "effect/Deferred";
3	import * as Effect from "effect/Effect";	4	import * as Effect from "effect/Effect";
		5	import * as Fiber from "effect/Fiber";
4	import * as Layer from "effect/Layer";	6	import * as Layer from "effect/Layer";
5	import * as Redacted from "effect/Redacted";	7	import * as Redacted from "effect/Redacted";
6	import * as Schema from "effect/Schema";	8	import * as Schema from "effect/Schema";
		9	import * as TestClock from "effect/testing/TestClock";
7	import * as HttpClient from "effect/unstable/http/HttpClient";	10	import * as HttpClient from "effect/unstable/http/HttpClient";
8	import * as HttpClientResponse from "effect/unstable/http/HttpClientResponse";	11	import * as HttpClientResponse from "effect/unstable/http/HttpClientResponse";
9	import type * as HttpClientRequest from "effect/unstable/http/HttpClientRequest";	12	import type * as HttpClientRequest from "effect/unstable/http/HttpClientRequest";
@@ -68,6 +71,54 @@ function testLayer(requests: HttpClientRequest.HttpClientRequest[], responses: R · 58 unmodified lines
68	}	71	}
69		72
70	describe("FCM delivery", () => {	73	describe("FCM delivery", () => {
		74	it.effect.each([
		75	{ operation: "authorize", stage: "headers", status: null },
		76	{ operation: "authorize", stage: "body", status: 200 },
		77	{ operation: "send", stage: "headers", status: null },
		78	{ operation: "send", stage: "body", status: 503 },
		79	] as const)("bounds a stalled $operation $stage and allows the next delivery", (scenario) =>
		80	Effect.gen(function* () {
		81	const started = yield* Deferred.make<void>();
		82	let shouldStall = true;
		83	const http = HttpClient.make((request) => {
		84	const isAuthorization = request.url === "https://oauth2.googleapis.com/token";
		85	const stall = shouldStall && isAuthorization === (scenario.operation === "authorize");
		86	const response = HttpClientResponse.fromWeb(
		87	request,
		88	isAuthorization
		89	? Response.json({ access_token: "access-token" })
		90	: Response.json({}, { status: stall ? 503 : 200 }),
		91	);
		92	if (stall) {
		93	shouldStall = false;
		94	const stalled = Deferred.succeed(started, undefined).pipe(Effect.andThen(Effect.never));
		95	if (scenario.stage === "headers") return stalled;
		96	Object.defineProperty(response, "json", { value: stalled });
		97	}
		98	return Effect.succeed(response);
		99	});
		100	yield* Effect.gen(function* () {
		101	const client = yield* FcmClient;
		102	const delivery = yield* client.send(input).pipe(Effect.flip, Effect.forkChild);
		103	yield* Deferred.await(started);
		104	yield* TestClock.adjust("10 seconds");
		105	expect(yield* Fiber.join(delivery)).toMatchObject({
		106	_tag: "FcmClientError",
		107	operation: scenario.operation,
		108	status: scenario.status,
		109	});
		110	expect(yield* client.send(input)).toEqual({ unregistered: false });
		111	}).pipe(
		112	Effect.provide(
		113	layer.pipe(
		114	Layer.provide(Layer.succeed(RelayConfiguration, config)),
		115	Layer.provide(Layer.succeed(HttpClient.HttpClient, http)),
		116	),
		117	),
		118	);
		119	}),
		120	);
		121
71	it.effect("signs a verifiable Google OAuth assertion scoped to messaging", () =>	122	it.effect("signs a verifiable Google OAuth assertion scoped to messaging", () =>
72	Effect.gen(function* () {	123	Effect.gen(function* () {
73	const assertion = yield* makeFcmAssertion(account, 1000);	124	const assertion = yield* makeFcmAssertion(account, 1000);

infra/relay/src/agentActivity/FcmClient.ts+12−2

 import { RelayConfiguration } from "../Config.ts";
+const FCM_HTTP_STAGE_TIMEOUT = "10 seconds";
 const ServiceAccount = Schema.Struct({
   project_id: Schema.NonEmptyString,
   client_email: Schema.NonEmptyString,
           }),
         ),
       )
-      .pipe(Effect.mapError(() => new FcmClientError({ operation: "authorize", status: null })));
+      .pipe(
+        Effect.timeout(FCM_HTTP_STAGE_TIMEOUT),
+        Effect.mapError(() => new FcmClientError({ operation: "authorize", status: null })),
+      );
     if (response.status !== 200)
       return yield* new FcmClientError({ operation: "authorize", status: response.status });
     return yield* response.json.pipe(
+      Effect.timeout(FCM_HTTP_STAGE_TIMEOUT),
       Effect.flatMap(decodeAccessToken),
       Effect.map((body) => body.access_token),
       Effect.mapError(
           },
         }),
         Effect.flatMap(client.execute),
+        Effect.timeout(FCM_HTTP_STAGE_TIMEOUT),
         Effect.mapError(() => new FcmClientError({ operation: "send", status: null })),
       );
       if (response.status >= 200 && response.status < 300) return { unregistered: false };
       if (response.status === 401) yield* invalidateToken;
-      const body = yield* response.json.pipe(Effect.orElseSucceed(() => null));
+      const body = yield* response.json.pipe(
+        Effect.timeout(FCM_HTTP_STAGE_TIMEOUT),
+        Effect.orElseSucceed(() => null),
+      );
       const decoded = decodeFcmError(body);
       const unregistered =
         Option.isSome(decoded) &&