1import * as Cloudflare from "alchemy/Cloudflare";
2import * as Config from "effect/Config";
3import * as Effect from "effect/Effect";
4import * as Redacted from "effect/Redacted";
5import * as Stream from "effect/Stream";
6import * as HttpServerRequest from "effect/http/HttpServerRequest";
7import * as HttpServerResponse from "effect/http/HttpServerResponse";
8import { KV } from "./KV.ts";
9import NotifyWorkflow from "./NotifyWorkflow.ts";
10import SandboxDO from "./SandboxDO.ts";
11
12/**
13 * Analytics Engine dataset — a plain binding value, not a cloud resource.
14 * In local dev `writeDataPoint` is accepted and discarded (Miniflare
15 * parity); on a live deploy the points land in the real dataset.
16 */
17export const Events = Cloudflare.AnalyticsEngine.Dataset("Events", {
18 dataset: "cloudflare_dev_events",
19});
20
21/**
22 * Checked-in HMAC key material for the `secret_key` binding (32 bytes,
23 * base64) — never generate key material at deploy time.
24 */
25const HMAC_KEY_BASE64 = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8=";
26
27interface AddInstance {
28 exports: {
29 add(a: number, b: number): number;
30 };
31}
32interface Message {
33 id: string;
34 body: {
35 text: string;
36 sentAt: number;
37 };
38}
39
40export default class EffectWorker extends Cloudflare.Worker<EffectWorker>()(
41 "EffectWorker",
42 {
43 main: import.meta.url,
44 dev: {
45 port: Config.Number("PORT").pipe(Config.withDefault(1338)),
46 },
47 build: {
48 bundleAnalyzer: true,
49 },
50 },
51 Effect.gen(function* () {
52 const publicUrl = yield* Cloudflare.Worker.URL;
53 const kv = yield* Cloudflare.KV.ReadWriteNamespace(KV);
54 const queue = yield* Cloudflare.Queues.Queue("EffectWorkerQueue");
55 const queueBinding = yield* Cloudflare.Queues.WriteQueue(queue);
56 const sandbox = yield* SandboxDO;
57 const queueMessages = yield* QueueMessages;
58 const workflow = yield* NotifyWorkflow;
59 const cronFires = yield* CronFires;
60 const analytics = yield* Cloudflare.AnalyticsEngine.WriteDataset(Events);
61 // `secret_key` binding: workerd imports the key material as a
62 // non-extractable CryptoKey. The accessor is deferred — `yield*` it
63 // again inside the handler to get the CryptoKey.
64 const hmacKey = yield* Cloudflare.Workers.SecretKey("HMAC_KEY", {
65 format: "raw",
66 algorithm: { name: "HMAC", hash: "SHA-256" },
67 usages: ["sign", "verify"],
68 keyBase64: Redacted.make(HMAC_KEY_BASE64),
69 });
70
71 // Cron trigger: registered at init; locally you can fire it on demand
72 // via `POST /cdn-cgi/handler/scheduled?cron=* * * * *&time=<ms>` instead
73 // of waiting for the minute boundary.
74 yield* Cloudflare.Workers.cron("* * * * *", (controller) =>
75 cronFires.getByName("default").record(controller.scheduledTime),
76 );
77
78 yield* Cloudflare.Queues.consumeQueueMessages<Message["body"]>(
79 queue,
80 (stream) =>
81 Stream.runForEach(stream, (msg) =>
82 queueMessages
83 .getByName("global")
84 .put({ id: msg.id, body: msg.body })
85 .pipe(Effect.asVoid),
86 ),
87 );
88
89 return {
90 fetch: Effect.gen(function* () {
91 const request = yield* HttpServerRequest.HttpServerRequest;
92 const url = new URL(request.url, "http://internal");
93 if (url.pathname.startsWith("/sandbox")) {
94 const stub = sandbox.getByName("sandbox-test");
95 return yield* stub.fetch(request).pipe(Effect.orDie);
96 } else if (url.pathname === "/wasm") {
97 const instance = yield* Effect.promise(async () => {
98 // This is dynamically imported so that the WASM import doesn't occur at deploy-time, which works in Bun but fails in Node.
99 const wasm = await import("./modules/wasm-example.wasm");
100 return (await WebAssembly.instantiate(wasm.default)) as AddInstance;
101 });
102 return yield* HttpServerResponse.json({
103 result: instance.exports.add(3, 4),
104 });
105 } else if (url.pathname.startsWith("/workflow/start/")) {
106 const roomId = url.pathname.split("/workflow/start/")[1];
107 if (!roomId) {
108 return yield* HttpServerResponse.json(
109 { error: "roomId is required" },
110 { status: 400 },
111 );
112 }
113 const instance = yield* workflow.create({
114 params: {
115 roomId,
116 message: "hello from workflow",
117 },
118 });
119 return yield* HttpServerResponse.json({ instanceId: instance.id });
120 } else if (url.pathname.startsWith("/workflow/status/")) {
121 const instanceId = url.pathname.split("/workflow/status/")[1];
122 if (!instanceId) {
123 return yield* HttpServerResponse.json(
124 { error: "instanceId is required" },
125 { status: 400 },
126 );
127 }
128 const instance = yield* workflow.get(instanceId);
129 const status = yield* instance.status();
130 return yield* HttpServerResponse.json(status);
131 } else if (url.pathname.startsWith("/queue/send")) {
132 const body = yield* request.json;
133 yield* queueBinding.send(body).pipe(Effect.orDie);
134 return yield* HttpServerResponse.json({ sent: body });
135 } else if (url.pathname.startsWith("/queue/messages")) {
136 const messages = yield* queueMessages.getByName("global").list();
137 return yield* HttpServerResponse.json(messages);
138 } else if (url.pathname.startsWith("/url")) {
139 return yield* HttpServerResponse.json({ url: yield* publicUrl });
140 } else if (url.pathname.startsWith("/cron/times")) {
141 const snapshot = yield* cronFires.getByName("default").snapshot();
142 return yield* HttpServerResponse.json(snapshot);
143 } else if (url.pathname.startsWith("/secret-key")) {
144 const key = yield* hmacKey;
145 const data = new TextEncoder().encode(
146 url.searchParams.get("message") ?? "hello",
147 );
148 const signature = yield* Effect.promise(() =>
149 crypto.subtle.sign("HMAC", key, data),
150 );
151 const verified = yield* Effect.promise(() =>
152 crypto.subtle.verify("HMAC", key, signature, data),
153 );
154 return yield* HttpServerResponse.json({
155 verified,
156 algorithm: key.algorithm.name,
157 // Bound keys are never extractable — local lowering matches.
158 extractable: key.extractable,
159 signatureBase64: btoa(
160 String.fromCharCode(...new Uint8Array(signature)),
161 ),
162 });
163 } else if (url.pathname.startsWith("/analytics")) {
164 // A documented no-op in local dev — the write succeeding (not
165 // throwing) is the observable behavior.
166 yield* analytics
167 .writeDataPoint({
168 indexes: ["example"],
169 blobs: ["visit"],
170 doubles: [1],
171 })
172 .pipe(Effect.orDie);
173 return yield* HttpServerResponse.json({ ok: true });
174 }
175 const value = yield* kv.list().pipe(Effect.orDie);
176 return yield* HttpServerResponse.json(value);
177 }),
178 };
179 }).pipe(
180 Effect.provide([
181 Cloudflare.KV.ReadWriteNamespaceBinding,
182 Cloudflare.Queues.WriteQueueBinding,
183 Cloudflare.Queues.EventSourceLive,
184 Cloudflare.Workers.CronEventSourceLive,
185 Cloudflare.Workers.SecretKeyBinding,
186 Cloudflare.AnalyticsEngine.WriteDatasetBinding,
187 ]),
188 ),
189) {}
190
191/**
192 * Records each `scheduledTime` the cron handler observes; the integ test
193 * fires the trigger route and polls `GET /cron/times` until it shows up.
194 */
195export class CronFires extends Cloudflare.DurableObject<CronFires>()(
196 "CronFires",
197 Effect.succeed(
198 Effect.gen(function* () {
199 const state = yield* Cloudflare.DurableObjectState;
200 return {
201 record: Effect.fn(function* (time: number) {
202 const times = (yield* state.storage.get<number[]>("times")) ?? [];
203 yield* state.storage.put("times", [...times, time]);
204 }),
205 snapshot: Effect.fn(function* () {
206 return { times: (yield* state.storage.get<number[]>("times")) ?? [] };
207 }),
208 };
209 }),
210 ),
211) {}
212
213export class QueueMessages extends Cloudflare.DurableObject<QueueMessages>()(
214 "QueueMessages",
215 Effect.succeed(
216 Effect.gen(function* () {
217 const state = yield* Cloudflare.DurableObjectState;
218 return {
219 put: Effect.fn(function* (message: Message) {
220 yield* state.storage.put(message.id, message);
221 }),
222 list: Effect.fn(function* () {
223 const messages = new Map<string, Message>(
224 state.storage.kv.list<Message>(),
225 );
226 return Array.from(messages.values());
227 }),
228 };
229 }),
230 ),
231) {}