1import * as storage from "@distilled.cloud/gcp/storage_v1";
2import * as Effect from "effect/Effect";
3import * as Layer from "effect/Layer";
4import type { HttpServerRequest } from "effect/http/HttpServerRequest";
5import * as HttpServerResponse from "effect/http/HttpServerResponse";
6import * as Namespace from "../../Namespace.ts";
7import * as Output from "../../Output.ts";
8import * as RemovalPolicy from "../../RemovalPolicy.ts";
9import { GcpEnvironment } from "../Environment.ts";
10import {
11 EventarcEventSource as EventarcEventSourceTag,
12 type CloudEvent,
13 type EventarcEventSourceProps,
14 type EventarcEventSourceService,
15} from "../Eventarc/EventSource.ts";
16import { Trigger, type EventFilter } from "../Eventarc/Trigger.ts";
17import { bindGcpHost } from "../Host.ts";
18import { Member } from "../IAM/Member.ts";
19import {
20 grantSelfInvoker,
21 hostEndpoint,
22 listenForDeliveries,
23 pathSegment,
24 pushHost,
25} from "../PushDelivery.ts";
26
27const lastSegment = (value: string) => value.split("/").pop() ?? value;
28
29/** Default delivery path for an Eventarc subscription. */
30export const eventarcPath = (id: string, props: EventarcEventSourceProps) =>
31 props.path ?? `/__alchemy/eventarc/${pathSegment(id)}`;
32
33const hasTypePrefix = (props: EventarcEventSourceProps, prefix: string) =>
34 props.eventFilters.some(
35 (filter) =>
36 typeof filter === "object" &&
37 "attribute" in filter &&
38 filter.attribute === "type" &&
39 typeof filter.value === "string" &&
40 filter.value.startsWith(prefix),
41 );
42
43const isStorageEvent = (props: EventarcEventSourceProps) =>
44 hasTypePrefix(props, "google.cloud.storage.");
45
46/**
47 * Firestore triggers reject an unset `eventDataContentType`, so default it
48 * to the `application/protobuf` encoding `gcloud` uses.
49 */
50const eventDataContentType = (props: EventarcEventSourceProps) =>
51 props.eventDataContentType ??
52 (hasTypePrefix(props, "google.cloud.firestore.")
53 ? "application/protobuf"
54 : undefined);
55
56/**
57 * Decode a binary-mode CloudEvent delivery. JSON payloads are parsed,
58 * `text/*` payloads stay strings, and anything else (e.g. the
59 * `application/protobuf` Firestore events) is handed over as raw bytes.
60 */
61const toCloudEvent = (
62 request: HttpServerRequest,
63 body: Uint8Array,
64): CloudEvent => {
65 const attributes: Record<string, string> = {};
66 for (const [name, value] of Object.entries(request.headers)) {
67 if (name.startsWith("ce-") && typeof value === "string") {
68 attributes[name.slice(3)] = value;
69 }
70 }
71 const contentType = request.headers["content-type"] ?? "";
72 let data: unknown = body;
73 if (contentType.includes("json") || contentType.startsWith("text/")) {
74 const text = new TextDecoder().decode(body);
75 data = text;
76 if (contentType.includes("json") && text.length > 0) {
77 try {
78 data = JSON.parse(text);
79 } catch {
80 data = text;
81 }
82 }
83 }
84 return {
85 id: attributes.id ?? "",
86 type: attributes.type ?? "",
87 source: attributes.source ?? "",
88 subject: attributes.subject,
89 time: attributes.time,
90 attributes,
91 data,
92 };
93};
94
95/**
96 * HTTP implementation of `GCP.Eventarc.EventSource` for `GCP.Run.Service`
97 * / `GCP.Function` and `GCP.CloudFunctions.Function`.
98 *
99 * Deploy-time: grants the host's runtime service account
100 * `roles/eventarc.eventReceiver` on the project and `roles/run.invoker` on
101 * the host (and, for Cloud Storage events, the Storage service agent
102 * `roles/pubsub.publisher` on the project, which direct Storage events
103 * require), then creates the trigger with that account as its identity.
104 * Each grant and the trigger block until GCP reports them in place and
105 * healthy. Runtime: claims deliveries on the path, verifies the OIDC
106 * token, decodes the binary-mode CloudEvent, and runs the handler; a 2xx
107 * acks, a failed handler answers 500 and Eventarc redelivers.
108 *
109 * @layer
110 * @provides GCP.Eventarc.EventSource
111 * @category Run
112 */
113export const EventarcEventSource = Layer.effect(
114 EventarcEventSourceTag,
115 Effect.gen(function* () {
116 const trigger = yield* Trigger;
117 const member = yield* Member;
118
119 return Effect.fn(function* <Req = never>(
120 id: string,
121 props: EventarcEventSourceProps,
122 process: (event: CloudEvent) => Effect.Effect<void, never, Req>,
123 ) {
124 const host = yield* pushHost("GCP.Eventarc.EventSource");
125 const path = eventarcPath(id, props);
126
127 if (!globalThis.__ALCHEMY_RUNTIME__) {
128 const endpoint = hostEndpoint(host);
129 const attrs = host as unknown as Record<string, Output.Output<string>>;
130 const env = yield* GcpEnvironment.current;
131 // Requested through the host's own bindings, so the host's IAM sync
132 // keeps it on every redeploy instead of revoking it as out of band.
133 yield* bindGcpHost({
134 tag: "GCP.Eventarc.EventSource",
135 resource: { LogicalId: id },
136 iam: [{ role: "roles/eventarc.eventReceiver" }],
137 });
138 yield* Namespace.push(
139 host.LogicalId,
140 Effect.gen(function* () {
141 yield* grantSelfInvoker(host);
142 // Project-wide prerequisite shared by every Storage trigger and
143 // notification in the project: retained, never revoked on destroy.
144 const storageAgent = isStorageEvent(props)
145 ? yield* member(`${id}-StorageAgentPublisher`, {
146 kind: "project",
147 name: env.project,
148 role: "roles/pubsub.publisher",
149 member: `serviceAccount:${
150 (yield* storage.getProjectsServiceAccount({
151 projectId: env.project,
152 })).email_address
153 }`,
154 }).pipe(RemovalPolicy.retain())
155 : undefined;
156 yield* trigger(`${id}-Trigger`, {
157 location: props.location ?? attrs.location,
158 eventFilters: props.eventFilters as EventFilter[],
159 eventDataContentType: eventDataContentType(props),
160 // The host's service account resolves after the host (and its
161 // IAM, including eventReceiver) is in place.
162 serviceAccount:
163 storageAgent === undefined
164 ? endpoint.serviceAccount
165 : Output.map(
166 Output.all(endpoint.serviceAccount, storageAgent.member),
167 ([email]) => email,
168 ),
169 destination: {
170 cloudRun: {
171 service: Output.map(endpoint.invokerService, lastSegment),
172 region: attrs.location!,
173 path,
174 },
175 },
176 });
177 }),
178 );
179 }
180
181 yield* listenForDeliveries(host, path, (request) =>
182 Effect.gen(function* () {
183 const body = yield* request.arrayBuffer.pipe(
184 Effect.map((buffer) => new Uint8Array(buffer)),
185 Effect.orElseSucceed(() => new Uint8Array()),
186 );
187 yield* process(toCloudEvent(request, body)).pipe(Effect.orDie);
188 return HttpServerResponse.empty({ status: 204 });
189 }),
190 );
191 }) as EventarcEventSourceService;
192 }),
193);