Binding a D1 Database to a Worker for AI Request Logging

You cannot answer “why did latency double last Tuesday” from a provider dashboard, because it does not know about your users, your prompt versions or your routing. A per-request table in D1, written by the same Worker that made the call, answers it in one query.

A schema you will not regret

Log the request, not the conversation. Bodies are large, often sensitive, and almost never what you query on; the numbers around them are small and are what every question needs.

-- schema.sql
CREATE TABLE IF NOT EXISTS ai_requests (
  id            TEXT PRIMARY KEY,
  ts            INTEGER NOT NULL,      -- epoch millis
  tenant        TEXT,
  feature       TEXT    NOT NULL,      -- "summarise", "triage", ...
  provider      TEXT    NOT NULL,
  model         TEXT    NOT NULL,
  prompt_version TEXT,
  status        INTEGER NOT NULL,      -- HTTP status from the provider
  finish_reason TEXT,
  ttft_ms       INTEGER,
  latency_ms    INTEGER NOT NULL,
  tokens_in     INTEGER,
  tokens_out    INTEGER,
  cached        INTEGER NOT NULL DEFAULT 0,
  error_code    TEXT
);

CREATE INDEX IF NOT EXISTS ai_requests_ts       ON ai_requests (ts);
CREATE INDEX IF NOT EXISTS ai_requests_feature  ON ai_requests (feature, ts);
CREATE INDEX IF NOT EXISTS ai_requests_tenant   ON ai_requests (tenant, ts);

Three columns earn their place and are usually missing. prompt_version is what lets you attribute a quality or cost change to the deploy that caused it. ttft_ms separately from latency_ms is what distinguishes “the provider was slow to start” from “the answer was long”, which have unrelated fixes. cached keeps a rising cache hit rate from looking like a latency improvement you did not make.

Creating and binding the database

1. Create the database: npx wrangler d1 create ai-logs. The command prints a database_id.
2. Add the binding to your Wrangler configuration, using the id from the previous step.
3. Apply the schema locally first, then remotely: npx wrangler d1 execute ai-logs --local --file=./schema.sql, then the same command with --remote.
4. Confirm from the CLI: npx wrangler d1 execute ai-logs --remote --command="SELECT count(*) FROM ai_requests".

// wrangler.jsonc
{
  "name": "ai-worker",
  "main": "src/index.ts",
  "compatibility_date": "2026-08-11",
  "observability": { "enabled": true },
  "d1_databases": [
    {
      "binding": "LOGS",
      "database_name": "ai-logs",
      "database_id": "<id printed by wrangler d1 create>"
    }
  ]
}

The binding value is the property name on env, so this one becomes env.LOGS. It does not have to match database_name and it is worth keeping short.

Writing off the critical path

The mistake that makes logging unpopular is awaiting the insert before returning the response, which adds a database round trip to every user request in exchange for nothing the user wants. ctx.waitUntil exists exactly for this: it extends the lifetime of the invocation past the response so the write completes without the client waiting for it.

export default {
  async fetch(request: Request, env: Env, ctx: ExecutionContext) {
    const started = Date.now();
    const body = await request.json<{ prompt: string; tenant: string }>();

    const upstream = await fetch(env.GATEWAY_URL, {
      method: "POST",
      headers: {
        "Content-Type": "application/json",
        Authorization: `Bearer ${env.CF_API_TOKEN}`,
      },
      body: JSON.stringify({
        model: "openai/gpt-4.1-mini",
        messages: [{ role: "user", content: body.prompt }],
      }),
    });

    const json = await upstream.json<any>();
    const latency = Date.now() - started;

    ctx.waitUntil(
      env.LOGS.prepare(
        `INSERT INTO ai_requests
           (id, ts, tenant, feature, provider, model, prompt_version,
            status, finish_reason, latency_ms, tokens_in, tokens_out, cached)
         VALUES (?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,?12,?13)`,
      )
        .bind(
          crypto.randomUUID(),
          Date.now(),
          body.tenant,
          "summarise",
          "openai",
          json?.model ?? "unknown",
          "v3",
          upstream.status,
          json?.choices?.[0]?.finish_reason ?? null,
          latency,
          json?.usage?.prompt_tokens ?? null,
          json?.usage?.completion_tokens ?? null,
          upstream.headers.get("cf-aig-cache-status") === "HIT" ? 1 : 0,
        )
        .run()
        .catch((err) => console.error("log write failed", err)),
    );

    return Response.json(json);
  },
};

Two things in there are deliberate. The .catch is not optional: an unhandled rejection inside waitUntil gives you a failing invocation on a request that already succeeded, and a logging table must never be able to fail a user request. And the whole insert is parameterised with ?1 placeholders through .bind() — string-concatenating a model name or a tenant id into SQL is an injection waiting for the first customer whose name contains an apostrophe.

Logging a streamed response

The code above works because it buffers the whole response and reads usage off it. Stream instead, and both halves of that break: the response body is gone by the time you would inspect it, and the token counts are not in the place you were looking.

Two facts to establish before writing any code. First, on an OpenAI-compatible streamed request the usage numbers are not on every chunk — they arrive in a final frame, and only if you asked for them. OpenAI documents a stream_options object with include_usage for exactly this. Without it, a streamed request gives you no token counts at all and your tokens_in column is permanently null. Second, the fastest streaming Worker — returning upstream.body untouched — is the one that sees nothing, so logging a stream means putting something in the path.

A TransformStream is the cheap way in. It observes bytes as they pass without buffering the response, so time to first token is unaffected and memory stays flat:

const started = Date.now();
let ttft: number | null = null;
let tokensIn: number | null = null;
let tokensOut: number | null = null;
let tail = "";

const decoder = new TextDecoder();

const observer = new TransformStream<Uint8Array, Uint8Array>({
  transform(chunk, controller) {
    if (ttft === null) ttft = Date.now() - started;   // first byte, once

    tail += decoder.decode(chunk, { stream: true });
    const lines = tail.split("\n");
    tail = lines.pop() ?? "";

    for (const line of lines) {
      if (!line.startsWith("data: ")) continue;
      const payload = line.slice(6).trim();
      if (payload === "[DONE]") continue;
      try {
        const usage = JSON.parse(payload)?.usage;
        if (usage) {
          tokensIn = usage.prompt_tokens ?? tokensIn;
          tokensOut = usage.completion_tokens ?? tokensOut;
        }
      } catch {
        // a frame we cannot parse must never break the response
      }
    }

    controller.enqueue(chunk);   // pass it through untouched
  },

  flush() {
    // Runs once, after the last chunk. ONE row, not one per chunk.
    ctx.waitUntil(
      env.LOGS.prepare(insertSql)
        .bind(crypto.randomUUID(), Date.now(), tenant, "chat", "openai",
              model, "v3", 200, null, ttft, Date.now() - started,
              tokensIn, tokensOut, 0)
        .run()
        .catch((err) => console.error("log write failed", err)),
    );
  },
});

return new Response(upstream.body!.pipeThrough(observer), {
  headers: { "Content-Type": "text/event-stream", "Cache-Control": "no-cache, no-transform" },
});

The comment on flush is the whole point of this section. The obvious-looking version writes a row, or updates a row, from inside transform — and transform runs once per network chunk, which on a long completion is hundreds of times. That is hundreds of D1 round trips for one request, and it collides directly with the documented limit of 1,000 queries per Worker invocation on Workers Paid and 50 on Free: a Free-plan Worker will start failing partway through its own log writes on any answer longer than a paragraph. Accumulate in local variables, write once in flush.

Two smaller things that matter here. The try/catch around JSON.parse is not defensive clutter — a throw inside transform errors the stream and the user loses their answer to a logging bug, which is the worst possible trade. And a client that disconnects mid-stream may mean flush never runs, so a small fraction of abandoned requests will be missing from the table; if abandonment rate is something you care about, record a row at the start and update it at the end instead of writing only once.

The limits that shape the design

D1 is SQLite with a network in front of it, and Cloudflare’s platform limits page documents the numbers that constrain a log table specifically: a maximum SQL statement length of 100,000 bytes, a maximum of 100 bound parameters per query, a maximum SQL query duration of 30 seconds, and queries per Worker invocation of 1,000 on Workers Paid or 50 on Free. Database size is documented as 10 GB on Workers Paid and 500 MB on Free.

The 100-bound-parameter cap is the one that bites. With thirteen columns you can fit seven rows into a single multi-row insert, not fifty — so if you are batching, batch with env.LOGS.batch(), which sends an array of prepared statements in one round trip and applies the per-statement limits to each statement individually.

await env.LOGS.batch(
  rows.map((r) =>
    stmt.bind(r.id, r.ts, r.tenant, r.feature, r.provider, r.model,
              r.promptVersion, r.status, r.finishReason, r.latencyMs,
              r.tokensIn, r.tokensOut, r.cached),
  ),
);

The size limit is the other planning constraint. A log table grows forever unless something deletes from it; at ten gigabytes it stops accepting writes, and that failure arrives as an error on your logging path rather than as a warning. Schedule a DELETE FROM ai_requests WHERE ts < ?1 on a cron trigger, and roll anything you want to keep for longer into a daily summary table first. If your volume makes per-request rows in D1 implausible, the honest move is to sample — write every error and one in N successes — rather than to log nothing.

The queries this table is for

The reason to build this rather than rely on a dashboard is that you can ask questions in terms of your own product. Percentiles have to be approximated in SQLite, but the shapes below are what you will actually run:

-- Which feature regressed, by day
SELECT date(ts / 1000, 'unixepoch') AS day, feature,
       count(*) AS calls,
       avg(latency_ms) AS avg_ms,
       sum(tokens_out) AS out_tokens
FROM ai_requests
WHERE ts > (unixepoch() - 14 * 86400) * 1000
GROUP BY day, feature
ORDER BY day DESC, out_tokens DESC;

-- Error mix for one tenant
SELECT status, error_code, count(*) AS n
FROM ai_requests
WHERE tenant = ?1 AND ts > (unixepoch() - 86400) * 1000
GROUP BY status, error_code
ORDER BY n DESC;

For what belongs in a log line generally — and what must never — the general treatment is the right companion to this page. The short version: no prompt bodies, no completions, no user text. Store a hash of the prompt if you need to group by it.