events.ts

import { DurableObject } from "cloudflare:workers";
import { Option, Schema } from "effect";

const createOutboxTable = `
CREATE TABLE IF NOT EXISTS outbox_events (
  id         TEXT    NOT NULL PRIMARY KEY,
  status     TEXT    NOT NULL DEFAULT 'pending',
  data       TEXT    NOT NULL,
  created_at INTEGER NOT NULL DEFAULT (unixepoch()),
  updated_at INTEGER NOT NULL DEFAULT (unixepoch())
);
`;

const processingOutboxEvent = `
UPDATE outbox_events
SET status = 'processing', updated_at = unixepoch()
WHERE id IN (
  SELECT id FROM outbox_events
  WHERE status = 'pending' OR (status = 'processing' AND updated_at <= ?)
  ORDER BY created_at ASC, id ASC
  LIMIT 20
)
RETURNING *;
`;

const finishOutboxEvent = `
UPDATE outbox_events
SET status = ?, updated_at = unixepoch()
WHERE id = ?;
`;

const insertOutboxEvent = `
INSERT INTO outbox_events (id, data) VALUES (?, ?);
`;

export type Subscriber<PayloadSchema extends Schema.Schema<any, any, never>> = (args: {
	payload: Schema.Schema.Type<PayloadSchema>;
	env: CloudflareBindings;
	ctx: DurableObjectState;
}) => { success: boolean } | Promise<{ success: boolean }>;

export type EventDefinitions<Schemas extends Record<string, Schema.Schema<any, any, never>>> = {
	[K in keyof Schemas]: {
		schema: Schemas[K];
		subscriber: Subscriber<Schemas[K]>;
	};
};

const RawOutboxData = Schema.parseJson(
	Schema.Struct({ _tag: Schema.String, payload: Schema.Unknown }),
);

export const declareEvents = <Schemas extends Record<string, Schema.Schema<any, any, never>>>(
	events: EventDefinitions<Schemas>,
) => {
	type Events = typeof events;
	type EventName = keyof Events & string;

	type DispatchArg = {
		[K in EventName]: {
			readonly _tag: K;
			readonly payload: Schema.Schema.Type<Events[K]["schema"]>;
		};
	}[EventName];

	class Actor extends DurableObject<CloudflareBindings> {
		constructor(ctx: DurableObjectState, env: CloudflareBindings) {
			super(ctx, env);
			this.ctx.storage.sql.exec(createOutboxTable);
		}

		dispatch(data: DispatchArg) {
			return this.ctx.storage.sql.exec(
				insertOutboxEvent,
				crypto.randomUUID(),
				JSON.stringify(data),
			);
		}

		async drainOutbox() {
			try {
				const fiveMinAgoSecs = Math.floor((Date.now() - 5 * 60 * 1000) / 1000);
				const evs = this.ctx.storage.sql.exec(processingOutboxEvent, fiveMinAgoSecs);

				for (const ev of evs) {
					const id = ev.id!.toString();

					const rawResult = Schema.decodeUnknownOption(RawOutboxData)(ev.data!.toString());
					if (Option.isNone(rawResult)) {
						this.ctx.storage.sql.exec(finishOutboxEvent, "failed", id);
						continue;
					}

					const { _tag: tag, payload } = rawResult.value;
					if (!(tag in events)) {
						this.ctx.storage.sql.exec(finishOutboxEvent, "failed", id);
						continue;
					}

					const eventName = tag as EventName;
					const decoded = Schema.decodeUnknownOption(events[eventName].schema)(payload);
					if (Option.isNone(decoded)) {
						this.ctx.storage.sql.exec(finishOutboxEvent, "failed", id);
						continue;
					}

					const result = await events[eventName].subscriber({
						ctx: this.ctx,
						env: this.env,
						payload: decoded.value,
					});

					this.ctx.storage.sql.exec(finishOutboxEvent, result.success ? "completed" : "failed", id);
				}

				return true;
			} catch (e) {
				console.error(e);
				return false;
			}
		}
	}

	return Actor;
};