feat(canvas): migrate canvas + cooldown storage to DO SQLite

Move pixel state and rate-limit cooldowns out of Upstash Redis and into
the existing CanvasRoom Durable Object's SQLite-backed storage. Worker
becomes a thin validation/proxy; the DO does atomic cooldown check +
pixel write + WS broadcast in one in-memory step.

Why: eliminate external dependency, keep $0/month free-tier forever,
exploit single-threaded actor for strong consistency without round-trips.

Architecture:
- canvas_chunks: 256 BLOB rows × 64 KB; CHUNK_COUNT derived from
  CANVAS_WIDTH × CANVAS_HEIGHT / CHUNK_BYTES so resize is config-only.
- cooldowns: user_id → expires_at, 1% sample-rate lazy GC.
- Worker forwards /api/canvas, /api/place, /api/ws to DO endpoints.
- POST /admin/migrate-from-upstash: token-gated one-shot importer.

Phases 1-3 code complete; Upstash dependency stays until prod migration
runs and 7-day rollback window passes (Phase 4).

Plan: plans/260509-2309-canvas-on-do-storage/

src/admin/migrate-from-upstash.js

@@ -0,0 +1,100 @@
+import { getFullCanvas } from '../lib/canvas-storage.js';
+import { TOTAL_PIXELS, CANVAS_WIDTH } from '../lib/constants.js';
+
+/**
+ * One-shot Upstash → DO migration. Reads the full canvas via the legacy
+ * Upstash REST path, ships the raw bytes to the DO `/import` endpoint,
+ * then verifies a few sample coordinates round-trip correctly.
+ *
+ * Removed entirely in Phase 4 of the canvas-on-do storage plan, along
+ * with the @upstash/redis dependency.
+ *
+ * @param {object} env - Worker env (Upstash creds + CANVAS_ROOM binding)
+ * @param {DurableObjectStub} roomStub
+ * @param {{force?: boolean}} opts
+ * @returns {Promise<Response>}
+ */
+export async function migrateFromUpstash(env, roomStub, { force = false } = {}) {
+  let upstashBytes;
+  try {
+    upstashBytes = await getFullCanvas(env);
+  } catch (err) {
+    return Response.json({ error: 'upstash_read_failed', message: String(err) }, { status: 500 });
+  }
+
+  if (upstashBytes.length !== TOTAL_PIXELS) {
+    return Response.json(
+      {
+        error: 'upstash_size_mismatch',
+        expected: TOTAL_PIXELS,
+        got: upstashBytes.length,
+      },
+      { status: 500 },
+    );
+  }
+
+  const importUrl = force ? 'http://do/import?force=1' : 'http://do/import';
+  const importRes = await roomStub.fetch(importUrl, {
+    method: 'POST',
+    headers: { 'Content-Type': 'application/octet-stream' },
+    body: upstashBytes,
+  });
+
+  if (!importRes.ok) {
+    const text = await importRes.text().catch(() => '');
+    return Response.json(
+      { error: 'do_import_failed', status: importRes.status, body: text },
+      { status: 502 },
+    );
+  }
+
+  // Round-trip verification: pull canvas back from the DO and compare a
+  // handful of sampled bytes. Catches subtle byte-order or chunking bugs.
+  const verifyRes = await roomStub.fetch('http://do/canvas');
+  if (!verifyRes.ok) {
+    return Response.json({ error: 'do_verify_read_failed' }, { status: 502 });
+  }
+  const doBytes = new Uint8Array(await verifyRes.arrayBuffer());
+
+  const samples = pickSampleOffsets(upstashBytes);
+  const mismatches = [];
+  for (const offset of samples) {
+    if (doBytes[offset] !== upstashBytes[offset]) {
+      mismatches.push({ offset, upstash: upstashBytes[offset], do: doBytes[offset] });
+    }
+  }
+
+  return Response.json({
+    ok: mismatches.length === 0,
+    bytes_imported: upstashBytes.length,
+    samples_checked: samples.length,
+    mismatches,
+  });
+}
+
+/**
+ * Pick byte offsets to verify post-migration. Includes corners, midpoints,
+ * and (preferentially) up to 5 offsets where the source has a non-zero
+ * value — those catch byte-order bugs that all-zero samples would miss.
+ */
+function pickSampleOffsets(srcBytes) {
+  const offsets = new Set([
+    0,                                               // (0, 0)
+    CANVAS_WIDTH - 1,                                // top-right
+    (CANVAS_WIDTH * CANVAS_WIDTH) - 1,               // bottom-right (square canvas)
+    Math.floor(TOTAL_PIXELS / 2),                    // middle
+    Math.floor(TOTAL_PIXELS / 2) + CANVAS_WIDTH + 1, // off-middle
+  ]);
+
+  // Add up to 5 non-zero offsets so we don't only check empty pixels.
+  let found = 0;
+  const stride = Math.max(1, Math.floor(TOTAL_PIXELS / 1000));
+  for (let i = 0; i < TOTAL_PIXELS && found < 5; i += stride) {
+    if (srcBytes[i] !== 0) {
+      offsets.add(i);
+      found++;
+    }
+  }
+
+  return [...offsets];
+}

src/durable-objects/canvas-room.js

@@ -1,47 +1,145 @@
+import { CANVAS_WIDTH, CANVAS_HEIGHT, MAX_COLORS, MAX_BATCH_SIZE, TOTAL_PIXELS } from '../lib/constants.js';
+import { init as initSchema } from './lib/schema.js';
+import { readAllChunks, writePixels, importFullCanvas } from './lib/chunk-storage.js';
+import { tryAcquire } from './lib/cooldown-store.js';
+
 /**
- * Durable Object for WebSocket broadcast room.
- * Uses Hibernation API so connections survive DO eviction.
+ * CanvasRoom: single Durable Object that owns
+ *  - canvas pixel state (SQLite chunk_blob rows)
+ *  - per-user rate-limit cooldowns (SQLite TTL rows)
+ *  - the WebSocket broadcast hub (Hibernation API)
+ *
+ * The Worker is a thin proxy that validates input and forwards to one of
+ * the four internal endpoints below.
  */
 export class CanvasRoom {
   constructor(state, env) {
     this.state = state;
     this.env = env;
+    this.sql = state.storage.sql;
+    initSchema(this.sql);
   }
 
   async fetch(request) {
     const url = new URL(request.url);
+    switch (url.pathname) {
+      case '/canvas':  return this.#handleGetCanvas();
+      case '/place':   return this.#handlePlace(request);
+      case '/import':  return this.#handleImport(request);
+      case '/ws':      return this.#handleWsUpgrade();
+      default:         return new Response('not found', { status: 404 });
+    }
+  }
 
-    // Internal broadcast from worker
-    if (url.pathname === '/broadcast' && request.method === 'POST') {
-      const pixels = await request.json();
-      const message = JSON.stringify({ type: 'pixels', pixels });
-      for (const ws of this.state.getWebSockets()) {
-        try {
-          ws.send(message);
-        } catch (err) {
-          console.warn('WS send failed, closing socket:', err?.message || err);
-          ws.close(1011, 'send failed');
-        }
-      }
-      return new Response('ok');
+  #handleGetCanvas() {
+    const buffer = readAllChunks(this.sql);
+    return new Response(buffer, {
+      headers: {
+        'Content-Type': 'application/octet-stream',
+        'Cache-Control': 'public, max-age=10, s-maxage=10, stale-while-revalidate=30',
+      },
+    });
+  }
+
+  async #handlePlace(request) {
+    let body;
+    try {
+      body = await request.json();
+    } catch {
+      return Response.json({ error: 'invalid_json' }, { status: 400 });
     }
 
-    // WebSocket upgrade
+    const { userId, pixels } = body || {};
+    if (typeof userId !== 'string' || !userId) {
+      return Response.json({ error: 'invalid_user' }, { status: 400 });
+    }
+    if (!Array.isArray(pixels) || pixels.length === 0) {
+      return Response.json({ error: 'pixels_required' }, { status: 400 });
+    }
+    if (pixels.length > MAX_BATCH_SIZE) {
+      return Response.json({ error: 'batch_too_large', max: MAX_BATCH_SIZE }, { status: 400 });
+    }
+    for (const p of pixels) {
+      if (
+        !Number.isInteger(p?.x) || !Number.isInteger(p?.y) || !Number.isInteger(p?.color) ||
+        p.x < 0 || p.x >= CANVAS_WIDTH ||
+        p.y < 0 || p.y >= CANVAS_HEIGHT ||
+        p.color < 0 || p.color >= MAX_COLORS
+      ) {
+        return Response.json({ error: 'invalid_pixel', pixel: p }, { status: 400 });
+      }
+    }
+
+    // Rate-limit then write+broadcast atomically (DO single-threaded).
+    const { allowed, retryAfter } = tryAcquire(this.sql, userId);
+    if (!allowed) {
+      return Response.json({ error: 'rate_limited', retryAfter }, { status: 429 });
+    }
+
+    try {
+      writePixels(this.sql, pixels);
+    } catch (err) {
+      console.error('writePixels failed:', err);
+      return Response.json({ error: 'storage_failed', message: String(err) }, { status: 500 });
+    }
+
+    this.#broadcastPixels(pixels);
+    return Response.json({ ok: true });
+  }
+
+  /**
+   * One-shot Upstash → DO migration target. Body is the full raw canvas
+   * (TOTAL_PIXELS bytes). Token-gated by the worker; this DO endpoint is
+   * not internet-reachable except via that gate.
+   */
+  async #handleImport(request) {
+    const url = new URL(request.url);
+    const force = url.searchParams.get('force') === '1';
+    const buf = new Uint8Array(await request.arrayBuffer());
+    if (buf.length !== TOTAL_PIXELS) {
+      return Response.json(
+        { error: 'size_mismatch', expected: TOTAL_PIXELS, got: buf.length },
+        { status: 400 },
+      );
+    }
+    let result;
+    try {
+      result = importFullCanvas(this.sql, buf, force);
+    } catch (err) {
+      return Response.json({ error: 'import_failed', message: String(err) }, { status: 500 });
+    }
+    if (result.skipped) {
+      return Response.json({ error: 'already_populated', hint: 'pass ?force=1 to overwrite' }, { status: 409 });
+    }
+    return Response.json({ ok: true, chunks_written: result.imported });
+  }
+
+  #handleWsUpgrade() {
     const pair = new WebSocketPair();
     const [client, server] = Object.values(pair);
-
     this.state.acceptWebSocket(server);
-
     return new Response(null, { status: 101, webSocket: client });
   }
 
-  /** Called when a WebSocket receives a message (required by Hibernation API).
-   * Clients aren't expected to send anything in this protocol; close defensively. */
+  #broadcastPixels(pixels) {
+    const message = JSON.stringify({ type: 'pixels', pixels });
+    for (const ws of this.state.getWebSockets()) {
+      try {
+        ws.send(message);
+      } catch (err) {
+        console.warn('WS send failed, closing socket:', err?.message || err);
+        try { ws.close(1011, 'send failed'); } catch { /* already closed */ }
+      }
+    }
+  }
+
+  /** Hibernation-API callbacks. */
+
   webSocketMessage(ws) {
+    // Protocol is broadcast-only; reject any inbound payload.
     ws.close(1003, 'unexpected client message');
   }
 
-  /** Called when a WebSocket is closed. */
   webSocketClose(ws, code, reason, wasClean) {
     if (!wasClean) {
       console.warn(`WS unclean close: code=${code} reason=${reason || '<none>'}`);
@@ -50,7 +148,6 @@ export class CanvasRoom {
     ws.close(code, reason);
   }
 
-  /** Called on WebSocket error. */
   webSocketError(ws, error) {
     console.error('WS error:', error?.message || error);
     ws.close(1011, 'error');

src/durable-objects/lib/chunk-storage.js

@@ -0,0 +1,134 @@
+import { CANVAS_WIDTH, TOTAL_PIXELS, CHUNK_BYTES, CHUNK_COUNT } from '../../lib/constants.js';
+
+/**
+ * Canvas pixel storage as fixed-size BLOB chunks in DO SQLite.
+ *
+ * Layout: linear byte stream (y * CANVAS_WIDTH + x), partitioned into
+ * CHUNK_COUNT rows of CHUNK_BYTES bytes each. Missing rows read as zeros,
+ * which is exactly the "uninitialized canvas" semantic.
+ */
+
+/** Map a pixel coordinate to its chunk and byte offset within that chunk. */
+function pixelToChunk(x, y) {
+  const offset = y * CANVAS_WIDTH + x;
+  return {
+    chunkId: Math.floor(offset / CHUNK_BYTES),
+    byteOffset: offset % CHUNK_BYTES,
+  };
+}
+
+/** Number of bytes the chunk at this id should hold. The last chunk may be
+ *  short if TOTAL_PIXELS isn't a multiple of CHUNK_BYTES. */
+function chunkSize(chunkId) {
+  const start = chunkId * CHUNK_BYTES;
+  return Math.min(CHUNK_BYTES, TOTAL_PIXELS - start);
+}
+
+/**
+ * Read one chunk's bytes. Returns a zero-filled buffer of the correct length
+ * if the row doesn't exist yet (lazy initialization).
+ */
+export function readChunk(sql, chunkId) {
+  const cursor = sql.exec('SELECT bytes FROM canvas_chunks WHERE chunk_id = ?', chunkId);
+  const rows = cursor.toArray();
+  if (rows.length === 0) {
+    return new Uint8Array(chunkSize(chunkId));
+  }
+  const blob = rows[0].bytes;
+  // CF DO returns BLOBs as ArrayBuffer; normalize to Uint8Array.
+  return blob instanceof Uint8Array ? blob : new Uint8Array(blob);
+}
+
+/**
+ * Read all chunks concatenated into a single TOTAL_PIXELS-sized buffer.
+ * Used to serve GET /api/canvas.
+ */
+export function readAllChunks(sql) {
+  const out = new Uint8Array(TOTAL_PIXELS);
+  const cursor = sql.exec('SELECT chunk_id, bytes FROM canvas_chunks');
+  for (const row of cursor) {
+    const chunkId = row.chunk_id;
+    const blob = row.bytes;
+    const view = blob instanceof Uint8Array ? blob : new Uint8Array(blob);
+    out.set(view, chunkId * CHUNK_BYTES);
+  }
+  return out;
+}
+
+/**
+ * Apply a batch of pixel writes. Groups by chunk so each touched chunk
+ * incurs at most one read and one write. Single transaction so the batch
+ * is atomic across all touched chunks.
+ *
+ * @param {SqlStorage} sql
+ * @param {Array<{x:number, y:number, color:number}>} pixels
+ */
+export function writePixels(sql, pixels) {
+  if (pixels.length === 0) return;
+
+  // Group by chunk_id; each group is a list of {byteOffset, color}.
+  /** @type {Map<number, Array<{byteOffset:number, color:number}>>} */
+  const groups = new Map();
+  for (const p of pixels) {
+    const { chunkId, byteOffset } = pixelToChunk(p.x, p.y);
+    let bucket = groups.get(chunkId);
+    if (!bucket) {
+      bucket = [];
+      groups.set(chunkId, bucket);
+    }
+    bucket.push({ byteOffset, color: p.color });
+  }
+
+  // For each touched chunk: read current bytes (or zero-fill), apply edits,
+  // write back. INSERT OR REPLACE upserts the row.
+  // Note: CF DO transactions are implicit per-fetch handler invocation —
+  // multiple sql.exec calls within the same handler are atomic.
+  for (const [chunkId, edits] of groups) {
+    const buf = readChunk(sql, chunkId);
+    // readChunk returns a fresh Uint8Array (or wraps a buffer); writes must
+    // not alias persisted state, so copy to be safe.
+    const next = new Uint8Array(buf);
+    for (const { byteOffset, color } of edits) {
+      next[byteOffset] = color;
+    }
+    sql.exec(
+      'INSERT INTO canvas_chunks (chunk_id, bytes) VALUES (?, ?) ' +
+        'ON CONFLICT(chunk_id) DO UPDATE SET bytes = excluded.bytes',
+      chunkId,
+      next,
+    );
+  }
+}
+
+/**
+ * Bulk replace the entire canvas. Used by the one-shot Upstash migration.
+ * Refuses to run if the canvas already has data, unless `force` is true.
+ *
+ * @param {SqlStorage} sql
+ * @param {Uint8Array} fullCanvas - exactly TOTAL_PIXELS bytes
+ * @param {boolean} force - overwrite even if rows already exist
+ * @returns {{imported: number, skipped: boolean}}
+ */
+export function importFullCanvas(sql, fullCanvas, force = false) {
+  if (fullCanvas.length !== TOTAL_PIXELS) {
+    throw new Error(`expected ${TOTAL_PIXELS} bytes, got ${fullCanvas.length}`);
+  }
+  if (!force) {
+    const existing = sql.exec('SELECT COUNT(*) AS n FROM canvas_chunks').one().n;
+    if (existing > 0) {
+      return { imported: 0, skipped: true };
+    }
+  }
+  for (let chunkId = 0; chunkId < CHUNK_COUNT; chunkId++) {
+    const start = chunkId * CHUNK_BYTES;
+    const end = Math.min(start + CHUNK_BYTES, TOTAL_PIXELS);
+    const slice = fullCanvas.slice(start, end);
+    sql.exec(
+      'INSERT INTO canvas_chunks (chunk_id, bytes) VALUES (?, ?) ' +
+        'ON CONFLICT(chunk_id) DO UPDATE SET bytes = excluded.bytes',
+      chunkId,
+      slice,
+    );
+  }
+  return { imported: CHUNK_COUNT, skipped: false };
+}

src/durable-objects/lib/cooldown-store.js

@@ -0,0 +1,58 @@
+import { REQUEST_COOLDOWN_SEC } from '../../lib/constants.js';
+
+const TTL_MS = REQUEST_COOLDOWN_SEC * 1000;
+
+// Probability of running a GC sweep on each tryAcquire. Cheap insurance
+// against unbounded table growth without paying for it on every call.
+const GC_SAMPLE_RATE = 0.01;
+
+/**
+ * Atomically claim a 1-second cooldown window for a user. Returns
+ * { allowed:true } on success, { allowed:false, retryAfter } if the user
+ * already holds an unexpired claim.
+ *
+ * Implementation: UPDATE existing row only if expired; if no rows changed,
+ * try INSERT. If INSERT fails (race or fresh non-expired row), the user is
+ * blocked.
+ *
+ * @param {SqlStorage} sql
+ * @param {string} userId
+ * @param {number} now - Date.now()
+ */
+export function tryAcquire(sql, userId, now = Date.now()) {
+  const expiresAt = now + TTL_MS;
+
+  // Update only if existing row is expired. CF DO sql.exec returns a cursor
+  // with a `rowsWritten` property after execution.
+  const updateCursor = sql.exec(
+    'UPDATE cooldowns SET expires_at = ? WHERE user_id = ? AND expires_at <= ?',
+    expiresAt,
+    userId,
+    now,
+  );
+  // Drain the cursor so rowsWritten is finalized.
+  updateCursor.toArray();
+  if (updateCursor.rowsWritten > 0) {
+    if (Math.random() < GC_SAMPLE_RATE) gc(sql, now);
+    return { allowed: true, retryAfter: 0 };
+  }
+
+  // No expired row to update. Either the user has never been seen (insert
+  // succeeds) or they hold an active claim (insert fails on PK conflict).
+  try {
+    sql.exec(
+      'INSERT INTO cooldowns (user_id, expires_at) VALUES (?, ?)',
+      userId,
+      expiresAt,
+    );
+    if (Math.random() < GC_SAMPLE_RATE) gc(sql, now);
+    return { allowed: true, retryAfter: 0 };
+  } catch {
+    return { allowed: false, retryAfter: REQUEST_COOLDOWN_SEC };
+  }
+}
+
+/** Delete all expired cooldown rows. Cheap with the expires_at index. */
+export function gc(sql, now = Date.now()) {
+  sql.exec('DELETE FROM cooldowns WHERE expires_at <= ?', now).toArray();
+}

src/durable-objects/lib/schema.js

@@ -0,0 +1,33 @@
+/**
+ * SQLite schema for the canvas Durable Object.
+ *
+ * Idempotent: safe to run on every DO construction. Cloudflare DOs persist
+ * across evictions, so the first-ever run creates the tables; later runs are
+ * no-ops because of IF NOT EXISTS.
+ *
+ * @param {SqlStorage} sql - state.storage.sql (CF DO SQLite handle)
+ */
+export function init(sql) {
+  // Canvas pixel bytes, sharded into fixed-size BLOB rows. chunk_id =
+  // floor(byteOffset / CHUNK_BYTES). Missing rows are zero-filled on read,
+  // so growing the canvas never requires a migration — just a redeploy.
+  sql.exec(`
+    CREATE TABLE IF NOT EXISTS canvas_chunks (
+      chunk_id INTEGER PRIMARY KEY,
+      bytes    BLOB NOT NULL
+    )
+  `);
+
+  // Per-user request cooldown (1s rate-limit). Lazy GC: rows are deleted
+  // opportunistically on read; the index keeps the GC sweep cheap if/when
+  // we need to run it.
+  sql.exec(`
+    CREATE TABLE IF NOT EXISTS cooldowns (
+      user_id    TEXT PRIMARY KEY,
+      expires_at INTEGER NOT NULL
+    )
+  `);
+  sql.exec(`
+    CREATE INDEX IF NOT EXISTS idx_cooldowns_expires ON cooldowns(expires_at)
+  `);
+}

src/lib/constants.js

@@ -3,7 +3,7 @@ export const CANVAS_WIDTH = 4096;
 export const CANVAS_HEIGHT = 4096;
 export const TOTAL_PIXELS = CANVAS_WIDTH * CANVAS_HEIGHT;
 
-/** 1 byte per pixel, 256 palette entries — the raw Redis bytes are directly
+/** 1 byte per pixel, 256 palette entries — the raw bytes are directly
  *  the pixel indices, no bit-level decode. */
 export const MAX_COLORS = 256;
 
@@ -11,9 +11,16 @@ export const MAX_COLORS = 256;
 export const REQUEST_COOLDOWN_SEC = 1;
 export const MAX_BATCH_SIZE = 2048;
 
-/** Redis keys. Key is versioned (":v2") so the old 32-color / 2048² canvas
- *  key (`rplace:canvas`) is ignored after this rollout — old data stays in
- *  Redis harmlessly until an operator deletes it. */
+/** Canvas chunked storage layout (DO SQLite). 64 KB chunks → 256 chunks for
+ *  the 16 MB / 4096² canvas. CHUNK_COUNT is derived: bumping CANVAS_WIDTH or
+ *  CANVAS_HEIGHT and redeploying transparently allocates more chunks (lazy-
+ *  initialized to zero on first read). */
+export const CHUNK_BYTES = 65536;
+export const CHUNK_COUNT = Math.ceil(TOTAL_PIXELS / CHUNK_BYTES);
+
+/** Legacy Upstash keys — used by the one-shot migration endpoint only.
+ *  Removed after migration verification (see Phase 4 of the canvas-on-do
+ *  storage plan). */
 export const REDIS_KEY_PREFIX = 'rplace:';
 export const REDIS_CANVAS_KEY = `${REDIS_KEY_PREFIX}canvas:v2`;
 

src/worker.js

@@ -1,8 +1,7 @@
 import { Hono } from 'hono';
-import { getFullCanvas, setPixels } from './lib/canvas-storage.js';
 import { getUserId } from './lib/get-user-id.js';
-import { checkRateLimit } from './lib/rate-limiter.js';
 import { CANVAS_WIDTH, CANVAS_HEIGHT, MAX_COLORS, MAX_BATCH_SIZE } from './lib/constants.js';
+import { migrateFromUpstash } from './admin/migrate-from-upstash.js';
 
 export { CanvasRoom } from './durable-objects/canvas-room.js';
 
@@ -11,26 +10,17 @@ const app = new Hono();
 // ~64 bytes is generous per pixel JSON object {"x":2047,"y":2047,"color":31}
 const MAX_BODY_BYTES = MAX_BATCH_SIZE * 64;
 
-/** GET /api/canvas — full canvas as binary.
- * Cloudflare's edge auto-compresses compressible content; we don't set
- * Content-Encoding manually (caused double-encoding / undecoded blobs
- * through wrangler dev + vite proxy during testing). */
+/** Resolve the singleton CanvasRoom DO stub. */
+function room(env) {
+  return env.CANVAS_ROOM.get(env.CANVAS_ROOM.idFromName('main'));
+}
+
+/** GET /api/canvas — full canvas binary, served by the DO directly. */
 app.get('/api/canvas', async (c) => {
-  try {
-    const buffer = await getFullCanvas(c.env);
-    return new Response(buffer, {
-      headers: {
-        'Content-Type': 'application/octet-stream',
-        'Cache-Control': 'public, max-age=10, s-maxage=10, stale-while-revalidate=30',
-      },
-    });
-  } catch (err) {
-    console.error('Canvas read failed:', err);
-    return c.json({ error: 'canvas_read_failed', message: String(err) }, 500);
-  }
+  return room(c.env).fetch('http://do/canvas');
 });
 
-/** POST /api/place — batch pixel placement */
+/** POST /api/place — validate at the edge, forward to the DO. */
 app.post('/api/place', async (c) => {
   const contentLength = parseInt(c.req.header('content-length') || '0', 10);
   if (contentLength > MAX_BODY_BYTES) {
@@ -51,78 +41,48 @@ app.post('/api/place', async (c) => {
   if (pixels.length > MAX_BATCH_SIZE) {
     return c.json({ error: 'batch_too_large', max: MAX_BATCH_SIZE }, 400);
   }
-
-  // Validate each pixel
   for (const p of pixels) {
     if (
-      typeof p.x !== 'number' || typeof p.y !== 'number' ||
-      typeof p.color !== 'number' ||
+      !Number.isInteger(p?.x) || !Number.isInteger(p?.y) || !Number.isInteger(p?.color) ||
       p.x < 0 || p.x >= CANVAS_WIDTH ||
       p.y < 0 || p.y >= CANVAS_HEIGHT ||
-      p.color < 0 || p.color >= MAX_COLORS ||
-      !Number.isInteger(p.x) || !Number.isInteger(p.y) ||
-      !Number.isInteger(p.color)
+      p.color < 0 || p.color >= MAX_COLORS
     ) {
       return c.json({ error: 'invalid_pixel', pixel: p }, 400);
     }
   }
 
-  // Rate limiting — 1 request per second per user, regardless of batch size.
   const userId = await getUserId(c.req.raw);
-  const { allowed, retryAfter } = await checkRateLimit(c.env, userId);
-  if (!allowed) {
-    return c.json({ error: 'rate_limited', retryAfter }, 429);
-  }
 
-  // Persist pixels (must succeed before broadcast)
-  try {
-    await setPixels(c.env, pixels);
-  } catch (err) {
-    console.error('Redis write failed:', err);
-    return c.json({ error: 'storage_failed', message: String(err) }, 500);
-  }
-
-  // Broadcast in background — don't block the user response on DO fetch.
-  // In non-CF runtimes (tests), executionCtx is unavailable; fall back to fire-and-forget.
-  const broadcastTask = broadcastPixels(c.env, pixels);
-  let ctx = null;
-  try { ctx = c.executionCtx; } catch { /* no-op */ }
-  if (ctx) {
-    ctx.waitUntil(broadcastTask);
-  } else {
-    broadcastTask.catch((err) => console.error('Broadcast:', err));
-  }
-
-  return c.json({ ok: true });
+  // DO does cooldown check + pixel write + broadcast in one atomic step.
+  return room(c.env).fetch('http://do/place', {
+    method: 'POST',
+    headers: { 'Content-Type': 'application/json' },
+    body: JSON.stringify({ userId, pixels }),
+  });
 });
 
-async function broadcastPixels(env, pixels) {
-  try {
-    const roomId = env.CANVAS_ROOM.idFromName('main');
-    const room = env.CANVAS_ROOM.get(roomId);
-    const r = await room.fetch(new Request('http://internal/broadcast', {
-      method: 'POST',
-      body: JSON.stringify(pixels),
-    }));
-    if (!r.ok) {
-      const text = await r.text().catch(() => '');
-      console.error('Broadcast non-OK:', r.status, text);
-    }
-  } catch (err) {
-    console.error('Broadcast threw:', err);
-  }
-}
-
-/** WebSocket upgrade — delegate to Durable Object */
+/** GET /api/ws — WebSocket upgrade routed to the DO. */
 app.get('/api/ws', async (c) => {
   const upgradeHeader = c.req.header('Upgrade');
   if (upgradeHeader !== 'websocket') {
     return c.text('Expected WebSocket', 426);
   }
+  return room(c.env).fetch(c.req.raw);
+});
 
-  const roomId = c.env.CANVAS_ROOM.idFromName('main');
-  const room = c.env.CANVAS_ROOM.get(roomId);
-  return room.fetch(c.req.raw);
+/**
+ * POST /admin/migrate-from-upstash — one-shot Upstash → DO canvas import.
+ * Token-gated; deleted in Phase 4 of the canvas-on-do storage plan.
+ */
+app.post('/admin/migrate-from-upstash', async (c) => {
+  const auth = c.req.header('Authorization') || '';
+  const expected = `Bearer ${c.env.MIGRATION_TOKEN || ''}`;
+  if (!c.env.MIGRATION_TOKEN || auth !== expected) {
+    return c.json({ error: 'forbidden' }, 403);
+  }
+  const force = c.req.query('force') === '1';
+  return migrateFromUpstash(c.env, room(c.env), { force });
 });
 
 export default app;