1import * as Effect from "effect/Effect";
2import * as Layer from "effect/Layer";
3import * as HttpServerResponse from "effect/http/HttpServerResponse";
4import { AlchemyContext } from "../AlchemyContext.ts";
5import * as Namespace from "../Namespace.ts";
6import { ProviderModePolicy } from "../ProviderMode.ts";
7import { Function } from "./Function.ts";
8import { FunctionRequest } from "./FunctionEnvironment.ts";
9import { FunctionTrigger } from "./FunctionTrigger.ts";
10import type { Bucket } from "./Bucket.ts";
11import {
12 BucketEventSource,
13 type BucketEventSourceProps,
14 type BucketEventSourceService,
15} from "./BucketEventSource.ts";
16import {
17 decodeFunctionTriggerEvent,
18 type BucketEvent,
19} from "./FunctionTriggerEvent.ts";
20
21/**
22 * Neon Function HTTP object-event dispatch and deployment wiring.
23 *
24 * ### Subscribe to uploads
25 * **Example:** Provide the subscription implementation
26 * ```typescript
27 * const application = Effect.gen(function* () {
28 * yield* Neon.BucketEventSource(uploads, { name: "Uploads" }, event =>
29 * Effect.log(event.objectKey),
30 * );
31 * return { fetch: Effect.succeed(HttpServerResponse.text("ok")) };
32 * }).pipe(Effect.provide(Neon.BucketEventSourceHttp));
33 * ```
34 *
35 * @layer
36 * @product Bucket
37 * @provides Neon.BucketEventSource
38 */
39export const BucketEventSourceHttp = Layer.effect(
40 BucketEventSource,
41 Effect.gen(function* () {
42 const host = yield* Function;
43 const Trigger = yield* FunctionTrigger;
44 return Effect.fn(function* (
45 bucket: Bucket,
46 props: BucketEventSourceProps,
47 handler: (event: BucketEvent) => Effect.Effect<void, unknown>,
48 ) {
49 const path = `/__alchemy/neon/bucket/${encodeURIComponent(props.name)}`;
50 const triggerName = `${host.FQN}:${props.name}`;
51 if (!globalThis.__ALCHEMY_RUNTIME__) {
52 const context = yield* AlchemyContext;
53 const remote = yield* ProviderModePolicy;
54 if (!context.dev || remote)
55 yield* Namespace.push(
56 host.LogicalId,
57 Trigger(props.name, {
58 function: host,
59 type: "storage_object_created",
60 name: triggerName,
61 storageObjectCreated: { bucket, prefix: props.prefix },
62 path,
63 enabled: props.enabled,
64 }),
65 );
66 }
67 const bucketName = yield* bucket.bucketName;
68 yield* host.route(
69 path,
70 Effect.gen(function* () {
71 const event = yield* decodeFunctionTriggerEvent(
72 yield* FunctionRequest,
73 );
74 if (
75 event.trigger.type !== "storage_object_created" ||
76 event.trigger.name !== triggerName ||
77 !("bucket_name" in event.data) ||
78 event.data.bucket_name !== (yield* bucketName) ||
79 (props.prefix && !event.data.object_key.startsWith(props.prefix))
80 )
81 return HttpServerResponse.empty({ status: 400 });
82 yield* handler({
83 invocationId: event.invocation_id,
84 triggerId: event.trigger.id,
85 name: props.name,
86 bucketName: event.data.bucket_name,
87 objectKey: event.data.object_key,
88 }).pipe(Effect.orDie);
89 return HttpServerResponse.empty({ status: 204 });
90 }).pipe(
91 Effect.catchTag("FunctionTriggerEventError", (error) =>
92 Effect.succeed(HttpServerResponse.empty({ status: error.status })),
93 ),
94 ),
95 );
96 }) as BucketEventSourceService;
97 }),
98);