1import * as Effect from "effect/Effect";
2import { HttpServerRequest } from "effect/http/HttpServerRequest";
3import * as HttpServerResponse from "effect/http/HttpServerResponse";
4import type { HttpEffect } from "../Http.ts";
5import * as Http from "../Http.ts";
6import {
7 createHostRuntimeContext,
8 type HostRuntimeContext,
9} from "../Server/Process.ts";
10
11/**
12 * Claims an inbound HTTP request for an event source (a Pub/Sub push
13 * delivery, a Cloud Scheduler call, …). Return `undefined` to pass the
14 * request on to the next listener and finally the user's `fetch`.
15 */
16export type GcpHttpListener = (
17 request: HttpServerRequest,
18) =>
19 | Effect.Effect<HttpServerResponse.HttpServerResponse, never, any>
20 | undefined;
21
22/**
23 * Runtime context of an HTTP-serving GCP host (`GCP.Run.Service`,
24 * `GCP.CloudFunctions.Function`). Extends the process host context with
25 * `listen`, the counterpart of `host.listen` on a Cloudflare Worker or an
26 * AWS Lambda: event sources register listeners that see every request
27 * before the user's `fetch` handler.
28 */
29export interface GcpHostRuntimeContext extends HostRuntimeContext {
30 listen: (listener: GcpHttpListener) => Effect.Effect<void>;
31}
32
33export const isGcpListenHost = (
34 value: unknown,
35): value is GcpHostRuntimeContext & { LogicalId: string; Type: string } =>
36 typeof value === "object" &&
37 value !== null &&
38 typeof (value as { listen?: unknown }).listen === "function" &&
39 typeof (value as { serve?: unknown }).serve === "function";
40
41const notFound = Effect.succeed(
42 HttpServerResponse.text("Not Found", { status: 404 }),
43);
44
45export const createGcpHostRuntimeContext =
46 (type: string) =>
47 (id: string): GcpHostRuntimeContext => {
48 const base = createHostRuntimeContext(type)(id);
49 const listeners: GcpHttpListener[] = [];
50 let served = false;
51
52 // Listeners are read per request, so it does not matter whether an
53 // event source registers before or after the user's `fetch` is served.
54 const route = <Req>(
55 handler: HttpEffect<Req> | Effect.Effect<HttpEffect<Req>>,
56 ): HttpEffect<Req> =>
57 Effect.gen(function* () {
58 const request = yield* HttpServerRequest;
59 for (const listener of listeners) {
60 const claimed = listener(request);
61 if (claimed !== undefined) return yield* claimed;
62 }
63 return yield* Http.safeHttpEffect(handler);
64 }) as HttpEffect<Req>;
65
66 return {
67 ...base,
68 listen: (listener) =>
69 Effect.sync(() => {
70 listeners.push(listener);
71 }),
72 serve: ((handler, options) => {
73 served = true;
74 return base.serve(route(handler as HttpEffect<any>), options);
75 }) as HostRuntimeContext["serve"],
76 // A host whose only job is consuming events has no `fetch`, so the
77 // Platform never calls `serve`; start the server for the listeners.
78 exports: Effect.gen(function* () {
79 if (!served && listeners.length > 0) {
80 served = true;
81 yield* base.serve(route(notFound));
82 }
83 return yield* base.exports;
84 }),
85 };
86 };