Deduplicate events

{
  "durable_objects": {
    "bindings": [
      { "name": "DEDUP", "class_name": "EventDeduplicator" }
    ]
  },
  "migrations": [
    { "tag": "v1", "new_sqlite_classes": ["EventDeduplicator"] }
  ],
  "pipelines": [
    { "binding": "EVENTS", "stream": "accepted_events" }
  ]
}

import { DurableObject } from "cloudflare:workers";

export class EventDeduplicator extends DurableObject {
  constructor(ctx, env) {
    super(ctx, env);
    this.ctx.storage.sql.exec(`
      CREATE TABLE IF NOT EXISTS accepted (
        event_id TEXT PRIMARY KEY,
        accepted_at TEXT NOT NULL
      )
    `);
  }

  async fetch(request) {
    const event = await request.json();
    const existing = this.ctx.storage.sql
      .exec("SELECT event_id FROM accepted WHERE event_id = ?", event.id)
      .toArray();

    if (existing.length) {
      return Response.json({ accepted: true, duplicate: true });
    }

    this.ctx.storage.sql.exec(
      "INSERT INTO accepted (event_id, accepted_at) VALUES (?, ?)",
      event.id,
      new Date().toISOString(),
    );
    await this.env.EVENTS.send([event]);

    return Response.json({ accepted: true, duplicate: false });
  }
}

export default {
  async fetch(request, env) {
    const event = await request.json();
    if (typeof event.id !== "string" || !event.id) {
      return new Response("event.id is required", { status: 400 });
    }

    const id = env.DEDUP.idFromName(event.id);
    return env.DEDUP.get(id).fetch("https://dedup.internal", {
      method: "POST",
      headers: { "content-type": "application/json" },
      body: JSON.stringify(event),
    });
  },
};