Bindings: KV, Queues, D1, and R2

import { App, DurableObject, stopAll, sleep, type DurableObjectNamespace } from "./celld_nb.ts";
import { DatabaseSync } from "node:sqlite";

interface KVNamespace {
  get(key: string, type?: "text" | "json" | { cacheTtl?: number }): Promise<unknown>;
  getWithMetadata(key: string, opts?: { cacheTtl?: number }): Promise<unknown>;
  put(key: string, value: string, opts?: { expirationTtl?: number; expiration?: number; metadata?: unknown }): Promise<void>;
  list(opts?: { prefix?: string; limit?: number }): Promise<unknown>;
  delete(key: string): Promise<void>;
}
interface D1Result { success: boolean; results: Record<string, unknown>[]; meta: Record<string, unknown> }
interface D1Statement {
  bind(...values: unknown[]): D1Statement;
  all(): Promise<D1Result>;
  first(column?: string): Promise<unknown>;
  run(): Promise<D1Result>;
  raw(): Promise<unknown[][]>;
}
interface D1Database {
  prepare(sql: string): D1Statement;
  batch(statements: D1Statement[]): Promise<D1Result[]>;
  exec(sql: string): Promise<{ count: number; duration: number }>;
  dump(): Promise<ArrayBuffer>;
}
interface R2Object {
  key: string; size: number; version: string; etag: string; httpEtag: string; uploaded: Date;
  httpMetadata?: { contentType?: string }; customMetadata?: Record<string, string>;
  checksums: { toJSON(): Record<string, string> };
  text(): Promise<string>; json(): Promise<unknown>;
}
interface R2Bucket {
  put(key: string, body: string, opts?: Record<string, unknown>): Promise<R2Object | null>;
  get(key: string): Promise<R2Object | null>;
  head(key: string): Promise<R2Object | null>;
  list(opts?: { prefix?: string; include?: string[] }): Promise<{ objects: R2Object[]; truncated: boolean }>;
  delete(key: string): Promise<void>;
}
interface QueueMessage { id: string; attempts: number; body: Record<string, unknown>; ack(): void; retry(opts?: { delaySeconds?: number }): void }
interface MessageBatch { queue: string; messages: QueueMessage[]; ackAll(): void; retryAll(): void }
interface Queue { send(body: unknown): Promise<void>; sendBatch(messages: { body: unknown }[]): Promise<void> }

interface Env {
  SESSIONS: KVNamespace;
  DB: D1Database;
  FILES: R2Bucket;
  OUTBOX: Queue;
  PROFILE: DurableObjectNamespace;
}

const app = new App("bindings", 9902);
console.log("project dir:", app.dir);

const wrangler: Record<string, unknown> = {
  kv_namespaces: [{ binding: "SESSIONS", id: "sessions-prod" }],
  d1_databases: [{ binding: "DB", database_name: "ledger" }],
  r2_buckets: [{ binding: "FILES", bucket_name: "files" }],
  queues: { producers: [{ binding: "OUTBOX", queue: "outbox" }] },
};

const envWorker = {
  async fetch(_request: Request, env: Record<string, unknown>): Promise<Response> {
    const shape: Record<string, string> = {};
    for (const [name, binding] of Object.entries(env)) shape[name] = binding?.constructor?.name ?? typeof binding;
    return Response.json(shape);
  },
};

await app.ship({ worker: envWorker, wrangler });
await app.start({ clean: true, logs: true });   // --logs: the node's INFO lines show each service's cell being born
console.log(app.logs({ grep: /^(Binding|env\.)/ }));
console.log(await app.json("/"));

app.clearLogs();
const current = JSON.parse(await Deno.readTextFile(`${app.dir}/wrangler.jsonc`));
await Deno.writeTextFile(`${app.dir}/wrangler.jsonc`, JSON.stringify({ ...current, routes: ["example.com/*"] }, null, 2));
await sleep(3000);
console.log(app.logs({ grep: /reload/ }));
console.log("still serving the last good deployment:", (await app.fetch("/")).status);
await app.ship({ worker: envWorker, wrangler });   // put the good configuration back
console.log("restored:", (await app.fetch("/")).status);

class Profile extends DurableObject<Env> {
  async fetch(request: Request): Promise<Response> {
    const token = new URL(request.url).searchParams.get("token") ?? "";
    const session = await this.env.SESSIONS.get(`session:${token}`, "json");   // shared data: the namespace
    const visits = ((await this.ctx.storage.get<number>("visits")) ?? 0) + 1;  // per-entity data: this cell
    await this.ctx.storage.put("visits", visits);
    return Response.json({ cell: this.ctx.id.name, session, visits });
  }
}

const kvWorker = {
  async fetch(request: Request, env: Env): Promise<Response> {
    const url = new URL(request.url);
    const q = (name: string) => url.searchParams.get(name);
    const key = q("key") ?? "";
    try {
      switch (url.pathname) {
        case "/put": {
          const opts: { expirationTtl?: number; metadata?: unknown } = {};
          if (q("ttl")) opts.expirationTtl = Number(q("ttl"));
          if (q("meta")) opts.metadata = JSON.parse(q("meta") ?? "null");
          await env.SESSIONS.put(key, q("value") ?? "", opts);
          return Response.json({ put: key, ...opts });
        }
        case "/get":
          return Response.json(await env.SESSIONS.getWithMetadata(key, { cacheTtl: 3600 }));   // cacheTtl: no effect
        case "/list":
          return Response.json(await env.SESSIONS.list({ prefix: q("prefix") ?? "" }));
        case "/delete":
          await env.SESSIONS.delete(key);
          return Response.json({ deleted: key });
        case "/profile":
          return env.PROFILE.get(env.PROFILE.idFromName(q("name") ?? "anon")).fetch(request);
      }
      return new Response("not found", { status: 404 });
    } catch (e) {
      return Response.json({ error: String(e) }, { status: 500 });
    }
  },
};

wrangler.durable_objects = { bindings: [{ name: "PROFILE", class_name: "Profile" }] };
wrangler.migrations = [{ tag: "v1", new_sqlite_classes: ["Profile"] }];
await app.ship({ classes: [Profile], worker: kvWorker, wrangler });

const claims = encodeURIComponent(JSON.stringify({ user: "ada", role: "admin" }));
console.log(await app.json(`/put?key=session:abc&value=${claims}&ttl=3600&meta=${encodeURIComponent('{"ua":"notebook"}')}`));
console.log(await app.json("/put?key=session:def&value=plain&ttl=3600"));
console.log(await app.json("/put?key=flag:beta&value=on"));
console.log("get     →", await app.json("/get?key=session:abc"));
console.log("list    →", await app.json("/list?prefix=session:"));
console.log("cell    →", await app.json("/profile?name=ada&token=abc"));
console.log("cell    →", await app.json("/profile?name=ada&token=nope"));
console.log("delete  →", await app.json("/delete?key=flag:beta"), await app.json("/get?key=flag:beta"));
console.log("ttl<60  →", await app.json("/put?key=short&value=x&ttl=5"));
console.log(app.logs({ grep: /cell_isolate_startup_timing.*__KvNamespace/, last: 1 }).replace(/^.*scope=/, "the namespace's cell: scope=").replace(/ node=.*$/, ""));

const initSql = `CREATE TABLE IF NOT EXISTS entries (
  id INTEGER PRIMARY KEY,
  account TEXT NOT NULL,
  amount INTEGER NOT NULL,
  at INTEGER NOT NULL
);
`;

const d1Worker = {
  async fetch(request: Request, env: Env): Promise<Response> {
    const url = new URL(request.url);
    try {
      switch (`${request.method} ${url.pathname}`) {
        case "POST /admin/sql":   // local stand-in for `celld d1 execute --file`
          return Response.json(await env.DB.exec(await request.text()));
        case "GET /tables": {
          const { results } = await env.DB.prepare("SELECT name FROM sqlite_schema WHERE type = 'table' AND name NOT LIKE '\\_%' ESCAPE '\\'").all();
          return Response.json(results.map((r) => r.name));
        }
        case "POST /entries": {
          const { account, amount } = await request.json() as { account: string; amount: number };
          const r = await env.DB.prepare("INSERT INTO entries (account, amount, at) VALUES (?, ?, ?)").bind(account, amount, Date.now()).run();
          return Response.json(r.meta);
        }
        case "GET /balance":
          return Response.json(await env.DB
            .prepare("SELECT account, SUM(amount) AS balance FROM entries WHERE account = ? GROUP BY account")
            .bind(url.searchParams.get("account")).first());
        case "GET /summary": {
          const [count, byAccount] = await env.DB.batch([
            env.DB.prepare("SELECT COUNT(*) AS n FROM entries"),
            env.DB.prepare("SELECT account, SUM(amount) AS balance FROM entries GROUP BY account ORDER BY account"),
          ]);
          return Response.json({ count: count.results[0], byAccount: byAccount.results });
        }
        case "GET /dump":
          return Response.json({ bytes: (await env.DB.dump()).byteLength });
      }
      return new Response("not found", { status: 404 });
    } catch (e) {
      return Response.json({ error: String(e) }, { status: 500 });
    }
  },
};

await app.ship({ classes: [Profile], worker: d1Worker, wrangler, files: { "migrations/0001_init.sql": initSql } });
console.log("tables after deploy:", await app.json("/tables"));
console.log("log lines mentioning migrations:", JSON.stringify(app.logs({ grep: /migrat/i })));

// Apply the same file through the Worker — the local stand-in for `celld d1 execute ledger --file migrations/0001_init.sql`.
console.log("exec →", await app.json("/admin/sql", { method: "POST", body: await Deno.readTextFile(`${app.dir}/migrations/0001_init.sql`) }));
console.log("tables now:", await app.json("/tables"));

const entry = (account: string, amount: number) =>
  app.json("/entries", { method: "POST", body: JSON.stringify({ account, amount }) });

console.log("run   →", await entry("ada", 120));
await entry("ada", -45);
await entry("bob", 30);
console.log("first →", await app.json("/balance?account=ada"));
console.log("batch →", await app.json("/summary"));
console.log("dump  →", await app.json("/dump"));
console.log("bad   →", await app.json("/admin/sql", { method: "POST", body: "SELECT * FROM nowhere;" }));

const r2Worker = {
  async fetch(request: Request, env: Env): Promise<Response> {
    const url = new URL(request.url);
    const key = url.searchParams.get("key") ?? "";
    const describe = (o: R2Object | null) => o && {
      key: o.key, size: o.size, version: o.version, etag: o.etag, httpEtag: o.httpEtag,
      checksums: o.checksums.toJSON(), contentType: o.httpMetadata?.contentType, customMetadata: o.customMetadata,
    };
    try {
      switch (`${request.method} ${url.pathname}`) {
        case "PUT /object":
          return Response.json(describe(await env.FILES.put(key, await request.text(), {
            httpMetadata: { contentType: request.headers.get("content-type") ?? "application/octet-stream" },
            customMetadata: { by: "notebook" },
          })));
        case "GET /object": {
          const o = await env.FILES.get(key);
          if (!o) return new Response("no such object", { status: 404 });
          return new Response(await o.text(), { headers: { etag: o.httpEtag, "content-type": o.httpMetadata?.contentType ?? "application/octet-stream" } });
        }
        case "GET /head":
          return Response.json(describe(await env.FILES.head(key)));
        case "GET /list": {
          const l = await env.FILES.list({ prefix: url.searchParams.get("prefix") ?? "", include: ["httpMetadata", "customMetadata"] });
          return Response.json({ truncated: l.truncated, objects: l.objects.map(describe) });
        }
        case "DELETE /object":
          await env.FILES.delete(key);
          return Response.json({ deleted: key });
      }
      return new Response("not found", { status: 404 });
    } catch (e) {
      return Response.json({ error: String(e) }, { status: 500 });
    }
  },
};
await app.ship({ classes: [Profile], worker: r2Worker, wrangler });

const put = (key: string, body: string) =>
  app.json(`/object?key=${key}`, { method: "PUT", body, headers: { "content-type": "application/json" } });
console.log("a →", await put("reports/a.json", '{"total":1}'));
console.log("b →", await put("reports/b.json", '{"total":1}'));   // the same bytes as a
console.log("c →", await put("reports/c.json", '{"total":2}'));
console.log("get  →", await app.text("/object?key=reports/a.json"));
console.log("head →", await app.json("/head?key=reports/c.json"));
console.log("list →", (await app.json<{ objects: { key: string; version: string }[] }>("/list?prefix=reports/")).objects.map((o) => `${o.key} @ ${o.version}`));
console.log("delete →", await app.json("/object?key=reports/c.json", { method: "DELETE" }), "→", await app.text("/object?key=reports/c.json"));

const store = new DatabaseSync(`${app.dir}/.celld/dev/objects.sqlite3`, { readOnly: true });
console.table(store.prepare("SELECT key, etag, length(body) AS bytes FROM objects WHERE key LIKE 'r2/%' ORDER BY key").all().map((r) => ({ ...r })));
store.close();

const queueWorker = {
  async fetch(request: Request, env: Env): Promise<Response> {
    const url = new URL(request.url);
    switch (url.pathname) {
      case "/send": {
        const n = Number(url.searchParams.get("n") ?? 1);
        await env.OUTBOX.send({ kind: "email", to: "ada@example.com", poison: url.searchParams.get("poison") === "1" });
        if (n > 1) await env.OUTBOX.sendBatch(Array.from({ length: n - 1 }, (_, i) => ({ body: { kind: "email", to: `user${i}@example.com` } })));
        return Response.json({ sent: n });
      }
      case "/delivered":
        return Response.json((await env.SESSIONS.get("outbox:batches", "json")) ?? []);
    }
    return new Response("not found", { status: 404 });
  },

  async queue(batch: MessageBatch, env: Env): Promise<void> {
    const log = ((await env.SESSIONS.get("outbox:batches", "json")) ?? []) as unknown[];
    log.push({ queue: batch.queue, size: batch.messages.length,
      messages: batch.messages.map((m) => `${m.body.to} (attempt ${m.attempts}${m.body.poison ? ", poison" : ""})`) });
    await env.SESSIONS.put("outbox:batches", JSON.stringify(log));
    for (const msg of batch.messages) {
      if (msg.body.poison && msg.attempts < 2) { msg.retry({ delaySeconds: 1 }); continue; }   // redelivered after delaySeconds
      msg.ack();
    }
  },
};

wrangler.queues = {
  producers: [{ binding: "OUTBOX", queue: "outbox" }],
  consumers: [{ queue: "outbox", max_batch_size: 5, max_batch_timeout: 1 }],
};
app.clearLogs();
await app.ship({ classes: [Profile], worker: queueWorker, wrangler });
console.log(app.logs({ grep: /^env\.OUTBOX/, last: 1 }));

console.log(await app.json("/send?n=7&poison=1"));
// With a consumer attached, every reload is a full restart and the port blinks for a few seconds — poll, don't sleep.
let batches: unknown[] = [];
for (let i = 0; i < 60 && batches.length < 3; i++) {
  await sleep(500);
  batches = await app.json<unknown[]>("/delivered").catch(() => batches);
}
for (const b of batches) console.log(JSON.stringify(b));
console.log("restarts during this cell:", app.logs({ grep: /restarting the application/ }).split("\n").filter(Boolean).length);
console.log(app.logs({ grep: /cell_isolate_startup_timing.*__Queue/, last: 1 }).replace(/^.*scope=/, "the queue's cell: scope=").replace(/ node=.*$/, ""));

app.clearLogs();
await Deno.writeTextFile(`${app.dir}/src/index.js`, "export default { fetch() { return new Response('ingress only'); } };\n");
await sleep(4000);
console.log(app.logs({ grep: /restarting|Error|Caused|queue consumer|exited/ }));
console.log("node reachable?", await app.fetch("/delivered").then((r) => r.status, (e) => String(e).split("\n")[0]));

await app.stop();
await app.ship({ classes: [Profile], worker: queueWorker, wrangler });
await app.start({ logs: true });
console.log("delivery log survived the restart:", (await app.json<unknown[]>("/delivered")).length, "batches");

const store2 = new DatabaseSync(`${app.dir}/.celld/dev/objects.sqlite3`, { readOnly: true });
console.log("cell kinds in the local store:",
  store2.prepare("SELECT DISTINCT substr(key, 7, instr(substr(key, 7), ':') - 1) AS kind FROM objects WHERE key LIKE 'cells/%'").all().map((r) => r.kind));
store2.close();

const auditSql = "CREATE TABLE IF NOT EXISTS audit (id INTEGER PRIMARY KEY, user TEXT NOT NULL, event TEXT NOT NULL, at INTEGER NOT NULL);\n";

const sessionWorker = {
  async fetch(request: Request, env: Env): Promise<Response> {
    const url = new URL(request.url);
    try {
      switch (`${request.method} ${url.pathname}`) {
        case "POST /admin/sql":
          return Response.json(await env.DB.exec(await request.text()));
        case "POST /login": {
          const user = url.searchParams.get("user") ?? "";
          // TODO: token = crypto.randomUUID(); SESSIONS.put(`session:${token}`, …, { expirationTtl: 3600 });
          //       DB.prepare("INSERT INTO audit …").bind(user, "login", Date.now()).run(); answer { token }
          return new Response("not implemented", { status: 501 });
        }
        case "GET /whoami": {
          const token = request.headers.get("x-session") ?? "";
          // TODO: SESSIONS.get(`session:${token}`, "json") → { user } or 401
          return new Response("not implemented", { status: 501 });
        }
        case "GET /audit":
          // TODO: DB.prepare("SELECT user, event, at FROM audit ORDER BY id").all() → results
          return new Response("not implemented", { status: 501 });
      }
      return new Response("not found", { status: 404 });
    } catch (e) {
      return Response.json({ error: String(e) }, { status: 500 });
    }
  },
};

// ── checker (celld only: these bindings have no kernel-side fake) ──────────
wrangler.queues = { producers: [{ binding: "OUTBOX", queue: "outbox" }] };   // no queue() handler here → no consumer (Part 5)
await app.ship({ classes: [Profile], worker: sessionWorker, wrangler, files: { "migrations/0002_audit.sql": auditSql } });
await app.json("/admin/sql", { method: "POST", body: await Deno.readTextFile(`${app.dir}/migrations/0002_audit.sql`) });

const check = (label: string, ok: boolean, detail: unknown) => console.log(ok ? "✓" : "✗", label, ok ? "" : `— got ${JSON.stringify(detail)}`);

const login = await app.fetch("/login?user=ada", { method: "POST" });
const { token } = (await login.json().catch(() => ({}))) as { token?: string };
check("POST /login answers 200 with a token", login.status === 200 && typeof token === "string" && token.length > 0, login.status);

const who = await app.fetch("/whoami", { headers: { "x-session": token ?? "" } });
const whoBody = await who.json().catch(() => null) as { user?: string } | null;
check("GET /whoami resolves the token to ada", who.status === 200 && whoBody?.user === "ada", [who.status, whoBody]);

const bad = await app.fetch("/whoami", { headers: { "x-session": "not-a-token" } });
check("GET /whoami with an unknown token is 401", bad.status === 401, bad.status);

const audit = await app.fetch("/audit");
const rows = (await audit.json().catch(() => [])) as { user: string; event: string; at: number }[];
check("GET /audit has ada's login row", audit.status === 200 && rows.some((r) => r.user === "ada" && r.event === "login" && typeof r.at === "number"), [audit.status, rows]);

await stopAll();