1import { Bucket } from "@/Neon/Bucket";
2import { BucketEventSource } from "@/Neon/BucketEventSource";
3import { BucketEventSourceHttp } from "@/Neon/BucketEventSourceHttp";
4import { CronEventSource } from "@/Neon/CronEventSource";
5import { CronEventSourceHttp } from "@/Neon/CronEventSourceHttp";
6import { Function } from "@/Neon/Function";
7import { Project } from "@/Neon/Project";
8import { WriteBucket } from "@/Neon/WriteBucket";
9import { WriteBucketHttp } from "@/Neon/WriteBucketHttp";
10import { Postgres } from "@/SQL/Postgres";
11import * as Effect from "effect/Effect";
12import * as Layer from "effect/Layer";
13import * as Redacted from "effect/Redacted";
14import { HttpServerRequest } from "effect/http/HttpServerRequest";
15import * as HttpServerResponse from "effect/http/HttpServerResponse";
16
17export const project = Project("EventProject", { region: "aws-us-east-2" });
18export const bucket = Bucket(
19 "EventBucket",
20 Effect.gen(function* () {
21 return { project: yield* project, forceDestroy: true };
22 }),
23);
24
25export default class EventFunction extends Function<EventFunction>()(
26 "EventFunction",
27 Effect.gen(function* () {
28 return { project: yield* project, main: import.meta.url };
29 }),
30 Effect.gen(function* () {
31 const uploads = yield* bucket;
32 const files = yield* WriteBucket(uploads);
33 const sql = yield* Postgres({
34 url: Effect.sync(() => Redacted.make(process.env.DATABASE_URL!)),
35 });
36 const prepare = sql`CREATE TABLE IF NOT EXISTS alchemy_function_events (id text PRIMARY KEY, kind text NOT NULL, object_key text)`;
37 yield* CronEventSource("Minute", { cron: "* * * * *" }, (event) =>
38 prepare.pipe(
39 Effect.andThen(
40 sql`INSERT INTO alchemy_function_events (id, kind) VALUES (${event.invocationId}, 'schedule') ON CONFLICT DO NOTHING`,
41 ),
42 Effect.asVoid,
43 ),
44 );
45 yield* BucketEventSource(
46 uploads,
47 { name: "Uploads", prefix: "incoming/" },
48 (event) =>
49 prepare.pipe(
50 Effect.andThen(
51 sql`INSERT INTO alchemy_function_events (id, kind, object_key) VALUES (${event.invocationId}, 'upload', ${event.objectKey}) ON CONFLICT DO NOTHING`,
52 ),
53 Effect.asVoid,
54 ),
55 );
56 return {
57 fetch: Effect.gen(function* () {
58 const request = yield* HttpServerRequest;
59 const path = new URL(request.url, "https://function.test").pathname;
60 yield* prepare.pipe(Effect.orDie);
61 if (path === "/upload") {
62 yield* files.put("incoming/test.txt", "uploaded").pipe(Effect.orDie);
63 yield* files.put("outside.txt", "not-matched").pipe(Effect.orDie);
64 return HttpServerResponse.empty({ status: 204 });
65 }
66 const events =
67 yield* sql`SELECT id, kind, object_key FROM alchemy_function_events ORDER BY id`.pipe(
68 Effect.orDie,
69 );
70 return yield* HttpServerResponse.json(events);
71 }),
72 };
73 }).pipe(
74 Effect.provide(
75 Layer.mergeAll(
76 WriteBucketHttp,
77 CronEventSourceHttp,
78 BucketEventSourceHttp,
79 ),
80 ),
81 ),
82) {}