| 1 | import * as GCP from "alchemy/GCP"; |
| 2 | import * as Effect from "effect/Effect"; |
| 3 | import { HttpServerRequest } from "effect/http/HttpServerRequest"; |
| 4 | import * as HttpServerResponse from "effect/http/HttpServerResponse"; |
| 5 | import { OrderEvents, type OrderEvent } from "./resources.ts"; |
| 6 | |
| 7 | const badRequest = (error: string) => |
| 8 | HttpServerResponse.json({ error }, { status: 400 }); |
| 9 | |
| 10 | /** |
| 11 | * The public orders API — the producer. |
| 12 | * |
| 13 | * It publishes one event per state change and returns. It knows nothing |
| 14 | * about who consumes the events: adding a consumer is a new subscription |
| 15 | * on the topic, not a change here. |
| 16 | * |
| 17 | * - `POST /orders` — `{ email, total }` → publishes `order.created`, |
| 18 | * returns `202 { orderId, eventId }`. |
| 19 | * - `POST /orders/:id/cancel` — `{ email }` → publishes `order.cancelled`, |
| 20 | * returns `202 { orderId, eventId }`. |
| 21 | */ |
| 22 | export default class Orders extends GCP.Function<Orders>()( |
| 23 | "Orders", |
| 24 | { |
| 25 | main: import.meta.url, |
| 26 | invokerIamDisabled: true, |
| 27 | }, |
| 28 | Effect.gen(function* () { |
| 29 | const events = yield* GCP.PubSub.WriteTopic(OrderEvents); |
| 30 | |
| 31 | const publish = (event: OrderEvent) => |
| 32 | events |
| 33 | .publish({ |
| 34 | data: JSON.stringify(event), |
| 35 | // Subscription filters match attributes, never the body, so the |
| 36 | // type goes here as well. |
| 37 | attributes: { type: event.type }, |
| 38 | }) |
| 39 | .pipe(Effect.orDie); |
| 40 | |
| 41 | const accepted = (event: OrderEvent) => |
| 42 | HttpServerResponse.json( |
| 43 | { orderId: event.orderId, eventId: event.eventId }, |
| 44 | { status: 202 }, |
| 45 | ); |
| 46 | |
| 47 | return { |
| 48 | fetch: Effect.gen(function* () { |
| 49 | const request = yield* HttpServerRequest; |
| 50 | const url = new URL(request.originalUrl); |
| 51 | const segments = url.pathname.split("/").filter(Boolean); |
| 52 | |
| 53 | if (request.method === "GET" && segments.length === 0) { |
| 54 | return HttpServerResponse.text("ok"); |
| 55 | } |
| 56 | |
| 57 | if (request.method !== "POST" || segments[0] !== "orders") { |
| 58 | return yield* HttpServerResponse.json( |
| 59 | { error: "not found" }, |
| 60 | { status: 404 }, |
| 61 | ); |
| 62 | } |
| 63 | |
| 64 | const body = (yield* request.json.pipe( |
| 65 | Effect.orElseSucceed(() => ({})), |
| 66 | )) as { email?: unknown; total?: unknown }; |
| 67 | if (typeof body.email !== "string" || !body.email.includes("@")) { |
| 68 | return yield* badRequest("email is required"); |
| 69 | } |
| 70 | |
| 71 | if (segments.length === 1) { |
| 72 | if (typeof body.total !== "number" || body.total <= 0) { |
| 73 | return yield* badRequest("total must be a positive number"); |
| 74 | } |
| 75 | const event: OrderEvent = { |
| 76 | eventId: crypto.randomUUID(), |
| 77 | type: "order.created", |
| 78 | orderId: crypto.randomUUID(), |
| 79 | email: body.email, |
| 80 | total: body.total, |
| 81 | occurredAt: new Date().toISOString(), |
| 82 | }; |
| 83 | yield* publish(event); |
| 84 | return yield* accepted(event); |
| 85 | } |
| 86 | |
| 87 | if (segments.length === 3 && segments[2] === "cancel") { |
| 88 | const event: OrderEvent = { |
| 89 | eventId: crypto.randomUUID(), |
| 90 | type: "order.cancelled", |
| 91 | orderId: segments[1]!, |
| 92 | email: body.email, |
| 93 | total: 0, |
| 94 | occurredAt: new Date().toISOString(), |
| 95 | }; |
| 96 | yield* publish(event); |
| 97 | return yield* accepted(event); |
| 98 | } |
| 99 | |
| 100 | return yield* HttpServerResponse.json( |
| 101 | { error: "not found" }, |
| 102 | { status: 404 }, |
| 103 | ); |
| 104 | }), |
| 105 | }; |
| 106 | }).pipe(Effect.provide(GCP.PubSub.WriteTopicHttp)), |
| 107 | ) {} |