ci(docker): reduce to sync-notice — native pipeline is .hanzo/workflows/deploy.yml

Image build+push+deploy now runs in-cluster (act_runner + BuildKit → operator).
GitHub is a mirror; this workflow is retained only as a manual sync notice.

.env.example

@@ -73,8 +73,25 @@ NEXT_PRIVATE_UPLOAD_DISTRIBUTION_KEY_CONTENTS=
 # Encryption key for document passwords. 
 NEXT_PRIVATE_DOCUMENT_PASSWORD_KEY=my-superstrong-document-secret
 
-# [[REDIS LOCKER CONFIGURATION]]
-# For bulk upload using tus.io, we use a Redis-based locker to prevent corruption of the data.
+# [[HANZO KV]] — OPTIONAL. Leave UNSET to run on the in-process backend
+# (single-replica correct: cache, rate-limit, tus upload locks, export/download
+# job stores and digest queues all work with no external datastore). SET it to
+# an external Hanzo KV instance for multi-replica HA (shared state across pods).
+# Accepts the Hanzo KV brand scheme kv:// (kvs:// for TLS) or a redis:// DSN.
+# A malformed value fails CLOSED (the app throws rather than silently degrade).
+#
+# Re-enabling multi-replica (replicas > 1) REQUIRES KV_URL. dataroom is
+# single-replica-by-construction otherwise: SQLite on a ReadWriteOnce volume plus
+# the in-process KV. Without KV_URL each pod owns its OWN map, so sessions,
+# rate-limit windows and tus upload locks split-brain across replicas — scaling
+# out needs a shared DB AND KV_URL set.
+#   KV_URL=kv://:password@hanzo-kv:6379
+KV_URL=
+
+# [[TUS UPLOAD LOCKER]] — for bulk upload via tus.io, an exclusive locker
+# prevents data corruption. It uses the Hanzo KV client above (KV_URL); with
+# KV_URL unset the lock is in-process (correct for a single replica). The legacy
+# Upstash REST locker below is unused and kept only for reference.
 UPSTASH_REDIS_REST_LOCKER_URL=
 UPSTASH_REDIS_REST_LOCKER_TOKEN=
 

.github/workflows/docker-deploy.yml

@@ -1,24 +1,11 @@
 name: Docker
-
+# Native deploy pipeline is .hanzo/workflows/deploy.yml (Hanzo Git → act_runner →
+# BuildKit → ghcr.io/hanzoai/dataroom:<sha> → operator reconcile → hanzocd).
+# GitHub is a mirror; this workflow is retained only as a manual sync notice.
 on:
   workflow_dispatch:
-  push:
-    branches: [main]
-    tags: ['v*']
-
-permissions:
-  contents: read
-  packages: write
-  id-token: write
-
 jobs:
-  docker:
-    uses: hanzoai/.github/.github/workflows/docker-build.yml@main
-    with:
-      image: ghcr.io/hanzoai/dataroom
-      # amd64-only: DOKS has no arm64 droplets (arm64 build paused).
-      platforms: linux/amd64
-      # Deploy is operator-managed via universe Service CR (services.hanzo.ai/dataroom).
-      # CI only builds+pushes the image; never `kubectl set image` (operator reverts it).
-      deploy: false
-    secrets: inherit
+  notice:
+    runs-on: ubuntu-latest
+    steps:
+      - run: echo "native pipeline is .hanzo/workflows/deploy.yml; GitHub is a mirror"

.hanzo/workflows/deploy.yml

@@ -0,0 +1,24 @@
+name: deploy
+on:
+  push:
+    branches: [main]
+jobs:
+  deploy:
+    runs-on: hanzo-linux-amd64
+    steps:
+      - uses: actions/checkout@v4
+      - name: Build + push image
+        run: |
+          SHA="${GITHUB_SHA::8}"
+          buildctl-daemonless.sh build --frontend=dockerfile.v0 \
+            --opt context="${{ github.server_url }}/${{ github.repository }}.git#${GITHUB_SHA}" \
+            --opt filename=Dockerfile --opt platform=linux/amd64 \
+            --secret id=GIT_AUTH_TOKEN,env=GIT_AUTH_TOKEN \
+            --output "type=image,name=ghcr.io/hanzoai/dataroom:${SHA},push=true" --progress=plain
+        env:
+          GIT_AUTH_TOKEN: ${{ secrets.GIT_CLONE_TOKEN }}
+      - name: Deploy — declare tag to operator
+        run: |
+          for app in dataroom; do
+            kubectl -n hanzo patch app "$app" --type=merge -p "{\"spec\":{\"image\":{\"repository\":\"ghcr.io/hanzoai/dataroom\",\"tag\":\"${GITHUB_SHA::8}\"}}}"
+          done

LLM.md

@@ -5,6 +5,15 @@ Hanzo dataroom service.
 
 **Upstream**: [Papermark](https://github.com/mfts/papermark) (AGPL-3.0). LICENSE retains "Copyright (c) 2023-present Papermark, Inc." Open-source DocSend alternative. This fork is Hanzo Dataroom — single-license AGPL, no `ee/` commercial directory.
 
+## In-process fold (HIP-0106, task #101)
+The production runtime is the **goja bundle** in [`goja/`](goja/): the ESM-free
+port of the API handlers, run in-process inside the unified `hanzoai/cloud` binary
+over Hanzo Base/SQLite (per-tenant) + the cloud object-storage seam. The standalone
+Next.js app + Postgres pod is **retired**; cloud serves `/v1/dataroom/*` itself.
+See [`goja/README.md`](goja/README.md) for the host contract. The Next.js/TS
+sources below remain the reference for the domain model that `goja/bundle.js`
+implements.
+
 ## Tech Stack
 - **Language**: TypeScript/JavaScript
 

goja/README.md

@@ -0,0 +1,75 @@
+# dataroom — goja bundle (HIP-0106 fold, task #101)
+
+This directory holds the **canonical, self-contained business-logic bundle** for
+Hanzo Dataroom. It is the ESM-free port of the Next.js/Prisma API handlers,
+authored to run verbatim inside the `dop251/goja` engine embedded in the unified
+`hanzoai/cloud` binary — the same in-process pattern `@hanzo/plans` and
+`@hanzo/pricing` use.
+
+The standalone Next.js app + Postgres is **retired**: cloud serves the whole
+dataroom surface (`/v1/dataroom/*`) itself, backed by Hanzo Base/SQLite
+(one file per org) and the cloud object-storage seam. No Postgres, no Next.js.
+
+It runs on the **reusable `clients/gojabase` binding** — the RW-Base goja host
+`captable` pilots and `esign` reuses — so the bundle carries only domain logic;
+persistence, per-tenant file selection, and per-request transactions are the
+binding's job.
+
+## What the bundle does
+
+The complete dataroom flow, re-expressed as a route table over injected host
+functions:
+
+- **documents** — metadata rows; the file **bytes** live in object storage (the
+  cloud VFS/S3 data plane), handled by the Go leaf. The bundle only ever
+  stores/reads the opaque storage **key**.
+- **data rooms** + **dataroom_document** — rooms and their document membership.
+- **shareable links** — access controls: email gate + allow-list, bcrypt
+  password, expiry, download toggle, archive.
+- **viewers** + **views** — a viewer authenticates against a link's controls; a
+  View (analytics session) is recorded.
+- **per-page view analytics** — page-by-page tracking rows, aggregated per page
+  (`views`, `totalDuration`, `avgDuration`).
+
+The Prisma models (`Document`, `Dataroom`, `DataroomDocument`, `Link`, `Viewer`,
+`View`, per-page tracking) become the `CREATE TABLE` statements in the bundle's
+`migrate` route — the one source of truth for the Base schema.
+
+## Host contract
+
+`clients/gojabase` injects these globals per dispatch (each dispatch runs inside
+ONE per-tenant SQLite transaction that commits iff `status < 400`) and calls
+`handle` once per request:
+
+```
+globalThis.__db.query(sql, args)   -> [ {col: val, ...}, ... ]   // tenant-scoped
+globalThis.__db.exec(sql, args)    -> { changes, lastId }
+globalThis.__newId()               -> collision-resistant id (crypto/rand)
+globalThis.__now()                 -> unix milliseconds
+globalThis.__bcrypt.hash(pw)       -> bcrypt hash   (leaf HostFn, vetted Go)
+globalThis.__bcrypt.verify(pw, h)  -> bool
+globalThis.handle({ route, params, query, orgId, body }) -> { status, body }
+```
+
+- `__db` is bound **per dispatch** to exactly one tenant's SQLite file (selected
+  from the caller's validated org), so a route can only ever touch its own org's
+  rows.
+- `__bcrypt` is a leaf-injected HostFn (bcrypt in vetted Go) — link passwords are
+  **hashed, never plaintext**.
+- The per-tenant **schema** (DDL) lives in the Go leaf (`clients/dataroom/
+  schema.go`), run by gojabase on first open; this bundle only issues SQL.
+- Admin routes run under the caller's validated cloud principal (org); public
+  viewer routes resolve their org from the link index the leaf maintains.
+
+## Constraints (goja target)
+
+Same as `@hanzo/plans`: ES2020, **no ESM** `import`/`export`, **no Node
+builtins**, no `console` dependency. Authored directly as the artifact goja runs
+— there is exactly one bundle, one way to build it.
+
+## Vendoring
+
+`hanzoai/cloud/clients/dataroom/bundle.js` is a **byte-identical vendored copy**
+that cloud `go:embed`s (the task mandates `go:embed` here rather than importing a
+Go module, since this repo is a TS app, not a Go module). Edit **this** file, then
+re-vendor the copy into cloud.

goja/bundle.js

@@ -0,0 +1,383 @@
+// Hanzo Dataroom — goja bundle (read-WRITE, on clients/gojabase).
+//
+// SELF-CONTAINED, NO ESM, NO node: imports. The complete dataroom business
+// logic (documents, data rooms, shareable links with access controls, viewers,
+// per-page view analytics), authored to run verbatim inside the dop251/goja
+// engine embedded in hanzoai/cloud (HIP-0106, task #101). It is the ESM-free
+// port of the Papermark/Next.js API handlers this fold replaces — the Prisma
+// data model becomes Base/SQLite tables (see the leaf's schema.go), the handlers
+// become the route table below. No Postgres, no Next.js.
+//
+// Host contract (clients/gojabase injects these per dispatch; each dispatch runs
+// inside ONE per-tenant SQLite transaction that commits iff status < 400):
+//   globalThis.__db.query(sql, args)   -> [ {col: val, ...}, ... ]
+//   globalThis.__db.exec(sql, args)    -> { changes, lastId }
+//   globalThis.__newId()               -> collision-resistant id (crypto/rand)
+//   globalThis.__now()                 -> unix milliseconds
+//   globalThis.__bcrypt.hash(pw)       -> bcrypt hash   (leaf HostFn, vetted Go)
+//   globalThis.__bcrypt.verify(pw,h)   -> bool
+//   globalThis.handle({ route, params, query, orgId, body }) -> { status, body }
+//
+// Document BYTES live in object storage (the cloud VFS/S3 seam) and are handled
+// by the Go leaf; this bundle only ever stores/reads the opaque storage KEY.
+//
+// CANONICAL SOURCE: github.com/hanzoai/dataroom → goja/bundle.js. This file is a
+// byte-identical VENDORED copy that clients/dataroom go:embeds (the task mandates
+// go:embed here rather than importing a Go module, since the dataroom repo is a
+// TS app, not a Go module). Edit the canonical copy, then re-vendor here.
+(function () {
+  'use strict';
+
+  // === tiny helpers ==========================================================
+
+  function q(sql, args) { return globalThis.__db.query(sql, args || []); }
+  function e(sql, args) { return globalThis.__db.exec(sql, args || []); }
+  function id(prefix) { return (prefix || '') + globalThis.__newId(); }
+  function now() { return globalThis.__now(); }
+  function one(rows) { return rows && rows.length ? rows[0] : null; }
+
+  function jsonList(v) {
+    if (v == null) return [];
+    try { var a = JSON.parse(v); return Array.isArray(a) ? a : []; }
+    catch (_) { return []; }
+  }
+  function truthy(v) { return v === true || v === 1 || v === '1'; }
+  function asInt(v, dflt) {
+    if (v == null || v === '') return dflt == null ? null : dflt;
+    var n = parseInt(v, 10);
+    return isNaN(n) ? (dflt == null ? null : dflt) : n;
+  }
+
+  // err builds a route result carrying a non-200 status via __status. A >=400
+  // status also rolls back the dispatch transaction (gojabase), so a rejected
+  // request leaves the tenant DB untouched.
+  function err(status, message) { return { __status: status, error: message }; }
+
+  // === shapers (row -> API object) ===========================================
+
+  function documentOut(r) {
+    if (!r) return null;
+    return {
+      id: r.id, name: r.name, fileKey: r.file_key, contentType: r.content_type,
+      type: r.type, numPages: r.num_pages, fileSize: r.file_size,
+      createdAt: r.created_at, updatedAt: r.updated_at,
+    };
+  }
+  function dataroomOut(r) {
+    if (!r) return null;
+    return {
+      id: r.id, pId: r.p_id, name: r.name, description: r.description,
+      createdAt: r.created_at, updatedAt: r.updated_at,
+    };
+  }
+  function linkOut(r) {
+    if (!r) return null;
+    // password_hash is NEVER returned — only whether one is set.
+    return {
+      id: r.id, linkType: r.link_type, dataroomId: r.dataroom_id, documentId: r.document_id,
+      name: r.name, emailProtected: truthy(r.email_protected), hasPassword: !!r.password_hash,
+      allowList: jsonList(r.allow_list), allowDownload: truthy(r.allow_download),
+      expiresAt: r.expires_at, isArchived: truthy(r.is_archived),
+      createdAt: r.created_at, updatedAt: r.updated_at,
+    };
+  }
+
+  // emailAllowed reports whether email passes the link allow list. An empty list
+  // allows anyone (email gate only). Entries may be a full email ("a@b.com"), a
+  // "@domain.com" suffix, or a bare "domain.com".
+  function emailAllowed(email, allowList) {
+    if (!allowList || !allowList.length) return true;
+    if (!email) return false;
+    var lc = String(email).toLowerCase().trim();
+    var at = lc.lastIndexOf('@');
+    var domain = at >= 0 ? lc.slice(at + 1) : '';
+    for (var i = 0; i < allowList.length; i++) {
+      var entry = String(allowList[i]).toLowerCase().trim();
+      if (!entry) continue;
+      if (entry.indexOf('@') === 0) { if (domain === entry.slice(1)) return true; }
+      else if (entry.indexOf('@') > 0) { if (lc === entry) return true; }
+      else if (domain === entry) return true;
+    }
+    return false;
+  }
+
+  function linkExpired(r) {
+    return r.expires_at != null && Number(r.expires_at) > 0 && now() > Number(r.expires_at);
+  }
+
+  // === route handlers ========================================================
+  // Admin routes are org-scoped by the per-tenant DB gojabase selects; the Go
+  // leaf refuses any request without a validated principal before dispatching.
+  // Viewer routes run under the org resolved from the public link id.
+
+  var routes = {
+
+    // ---- documents ----------------------------------------------------------
+    // POST: the Go leaf has already stored the bytes on the object-storage seam
+    // and passes the resulting fileKey; this records the metadata row.
+    'documents.create': function (ctx) {
+      var b = ctx.body || {};
+      if (!b.name || !b.fileKey) return err(400, 'name and fileKey required');
+      var docId = id('doc_');
+      var t = now();
+      e('INSERT INTO document (id,name,file_key,content_type,type,num_pages,file_size,created_at,updated_at) ' +
+        'VALUES (?,?,?,?,?,?,?,?,?)',
+        [docId, String(b.name), String(b.fileKey), b.contentType || null, b.type || null,
+         asInt(b.numPages), asInt(b.fileSize), t, t]);
+      return { document: documentOut(one(q('SELECT * FROM document WHERE id=?', [docId]))) };
+    },
+    'documents.list': function () {
+      return { documents: q('SELECT * FROM document ORDER BY created_at DESC').map(documentOut) };
+    },
+    'documents.get': function (ctx) {
+      var r = one(q('SELECT * FROM document WHERE id=?', [ctx.params.id]));
+      if (!r) return err(404, 'document not found');
+      return { document: documentOut(r) };
+    },
+    // documents.file returns the storage key so the Go leaf can stream the bytes.
+    'documents.file': function (ctx) {
+      var r = one(q('SELECT * FROM document WHERE id=?', [ctx.params.id]));
+      if (!r) return err(404, 'document not found');
+      return { fileKey: r.file_key, contentType: r.content_type, name: r.name };
+    },
+
+    // ---- datarooms ----------------------------------------------------------
+    'datarooms.create': function (ctx) {
+      var b = ctx.body || {};
+      if (!b.name) return err(400, 'name required');
+      var roomId = id('room_');
+      var pId = 'dr_' + globalThis.__newId().slice(1, 13);
+      var t = now();
+      e('INSERT INTO dataroom (id,p_id,name,description,created_at,updated_at) VALUES (?,?,?,?,?,?)',
+        [roomId, pId, String(b.name), b.description || null, t, t]);
+      return { dataroom: dataroomOut(one(q('SELECT * FROM dataroom WHERE id=?', [roomId]))) };
+    },
+    'datarooms.list': function () {
+      return { datarooms: q('SELECT * FROM dataroom ORDER BY created_at DESC').map(dataroomOut) };
+    },
+    'datarooms.get': function (ctx) {
+      var r = one(q('SELECT * FROM dataroom WHERE id=?', [ctx.params.id]));
+      if (!r) return err(404, 'dataroom not found');
+      var docs = q(
+        'SELECT dd.id AS dd_id, dd.order_index, doc.* FROM dataroom_document dd ' +
+        'JOIN document doc ON doc.id = dd.document_id WHERE dd.dataroom_id=? ' +
+        'ORDER BY dd.order_index IS NULL, dd.order_index, dd.created_at', [r.id]);
+      var out = dataroomOut(r);
+      out.documents = docs.map(function (d) {
+        var doc = documentOut(d);
+        doc.dataroomDocumentId = d.dd_id;
+        doc.orderIndex = d.order_index;
+        return doc;
+      });
+      return { dataroom: out };
+    },
+    'datarooms.addDocument': function (ctx) {
+      var b = ctx.body || {};
+      var roomId = ctx.params.id;
+      var docId = b.documentId;
+      if (!docId) return err(400, 'documentId required');
+      if (!one(q('SELECT id FROM dataroom WHERE id=?', [roomId]))) return err(404, 'dataroom not found');
+      if (!one(q('SELECT id FROM document WHERE id=?', [docId]))) return err(404, 'document not found');
+      if (one(q('SELECT id FROM dataroom_document WHERE dataroom_id=? AND document_id=?', [roomId, docId]))) {
+        return err(409, 'document already in dataroom');
+      }
+      var ddId = id('dd_');
+      e('INSERT INTO dataroom_document (id,dataroom_id,document_id,order_index,created_at) VALUES (?,?,?,?,?)',
+        [ddId, roomId, docId, asInt(b.orderIndex), now()]);
+      return { dataroomDocumentId: ddId, dataroomId: roomId, documentId: docId };
+    },
+
+    // ---- links --------------------------------------------------------------
+    'links.create': function (ctx) {
+      var b = ctx.body || {};
+      var roomId = b.dataroomId || null;
+      var docId = b.documentId || null;
+      if (!roomId && !docId) return err(400, 'dataroomId or documentId required');
+      if (roomId && !one(q('SELECT id FROM dataroom WHERE id=?', [roomId]))) return err(404, 'dataroom not found');
+      if (docId && !one(q('SELECT id FROM document WHERE id=?', [docId]))) return err(404, 'document not found');
+      var pwHash = null;
+      if (b.password) pwHash = globalThis.__bcrypt.hash(String(b.password));
+      var allowList = Array.isArray(b.allowList) ? b.allowList : [];
+      var denyList = Array.isArray(b.denyList) ? b.denyList : [];
+      var linkId = id('link_');
+      var t = now();
+      var linkType = docId && !roomId ? 'DOCUMENT_LINK' : 'DATAROOM_LINK';
+      e('INSERT INTO link (id,link_type,dataroom_id,document_id,name,password_hash,email_protected,' +
+        'allow_list,deny_list,allow_download,expires_at,is_archived,created_at,updated_at) ' +
+        'VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)',
+        [linkId, linkType, roomId, docId, b.name || null, pwHash,
+         b.emailProtected === false ? 0 : 1, JSON.stringify(allowList), JSON.stringify(denyList),
+         b.allowDownload ? 1 : 0, asInt(b.expiresAt), 0, t, t]);
+      return { link: linkOut(one(q('SELECT * FROM link WHERE id=?', [linkId]))) };
+    },
+    'links.list': function () {
+      return { links: q('SELECT * FROM link WHERE is_archived=0 ORDER BY created_at DESC').map(linkOut) };
+    },
+
+    // ---- viewer surface (public) -------------------------------------------
+    // view.link returns the pre-auth metadata a visitor sees (what gates apply).
+    'view.link': function (ctx) {
+      var r = one(q('SELECT * FROM link WHERE id=?', [ctx.params.linkId]));
+      if (!r || truthy(r.is_archived)) return err(404, 'link not found');
+      var out = {
+        id: r.id, name: r.name, linkType: r.link_type,
+        emailProtected: truthy(r.email_protected), hasPassword: !!r.password_hash,
+        allowDownload: truthy(r.allow_download), expired: linkExpired(r),
+      };
+      if (r.dataroom_id) {
+        var room = one(q('SELECT * FROM dataroom WHERE id=?', [r.dataroom_id]));
+        if (room) out.dataroom = { id: room.id, pId: room.p_id, name: room.name, description: room.description };
+      }
+      if (r.document_id) {
+        var doc = one(q('SELECT * FROM document WHERE id=?', [r.document_id]));
+        if (doc) out.document = { id: doc.id, name: doc.name, numPages: doc.num_pages };
+      }
+      return { link: out };
+    },
+
+    // view.authenticate enforces the access controls and, on success, records a
+    // Viewer + a View (the analytics session) and returns the viewable payload.
+    'view.authenticate': function (ctx) {
+      var b = ctx.body || {};
+      var r = one(q('SELECT * FROM link WHERE id=?', [ctx.params.linkId]));
+      if (!r || truthy(r.is_archived)) return err(404, 'link not found');
+      if (linkExpired(r)) return err(403, 'link expired');
+
+      var email = b.email ? String(b.email).toLowerCase().trim() : '';
+      if (truthy(r.email_protected) && !email) return err(401, 'email required');
+      if (!emailAllowed(email, jsonList(r.allow_list))) return err(403, 'email not allowed');
+      if (r.password_hash) {
+        if (!b.password || !globalThis.__bcrypt.verify(String(b.password), r.password_hash)) {
+          return err(401, 'invalid password');
+        }
+      }
+
+      var viewerId = null;
+      if (email) {
+        var existing = one(q('SELECT * FROM viewer WHERE email=?', [email]));
+        if (existing) { viewerId = existing.id; }
+        else {
+          viewerId = id('viewer_');
+          var tv = now();
+          e('INSERT INTO viewer (id,email,verified,created_at,updated_at) VALUES (?,?,?,?,?)',
+            [viewerId, email, 0, tv, tv]);
+        }
+      }
+
+      var viewId = id('view_');
+      var viewType = r.dataroom_id ? 'DATAROOM_VIEW' : 'DOCUMENT_VIEW';
+      e('INSERT INTO view (id,link_id,dataroom_id,document_id,viewer_id,viewer_email,view_type,verified,viewed_at) ' +
+        'VALUES (?,?,?,?,?,?,?,?,?)',
+        [viewId, r.id, r.dataroom_id, r.document_id, viewerId, email || null, viewType, 0, now()]);
+
+      var payload = { viewId: viewId, viewerId: viewerId, allowDownload: truthy(r.allow_download) };
+      if (r.dataroom_id) {
+        var docs = q(
+          'SELECT dd.id AS dd_id, doc.* FROM dataroom_document dd ' +
+          'JOIN document doc ON doc.id = dd.document_id WHERE dd.dataroom_id=? ' +
+          'ORDER BY dd.order_index IS NULL, dd.order_index, dd.created_at', [r.dataroom_id]);
+        payload.documents = docs.map(function (d) {
+          return { id: d.id, dataroomDocumentId: d.dd_id, name: d.name, numPages: d.num_pages };
+        });
+      } else if (r.document_id) {
+        var single = one(q('SELECT * FROM document WHERE id=?', [r.document_id]));
+        if (single) payload.document = { id: single.id, name: single.name, numPages: single.num_pages };
+      }
+      return payload;
+    },
+
+    // view.recordPage records ONE per-page analytics event (the page-by-page
+    // tracking) against an existing View.
+    'view.recordPage': function (ctx) {
+      var b = ctx.body || {};
+      var linkId = ctx.params.linkId;
+      var v = one(q('SELECT * FROM view WHERE id=? AND link_id=?', [b.viewId, linkId]));
+      if (!v) return err(404, 'view not found');
+      if (b.pageNumber == null) return err(400, 'pageNumber required');
+      var pvId = id('pv_');
+      e('INSERT INTO page_view (id,view_id,link_id,document_id,dataroom_id,page_number,version_number,duration,viewed_at) ' +
+        'VALUES (?,?,?,?,?,?,?,?,?)',
+        [pvId, v.id, linkId, b.documentId || v.document_id, v.dataroom_id,
+         asInt(b.pageNumber, 0), asInt(b.versionNumber), asInt(b.duration, 0), now()]);
+      return { ok: true, id: pvId };
+    },
+
+    // view.file authorises a viewer download and returns the storage key.
+    'view.file': function (ctx) {
+      var linkId = ctx.params.linkId;
+      var docId = ctx.params.documentId;
+      var viewId = ctx.query.viewId;
+      var r = one(q('SELECT * FROM link WHERE id=?', [linkId]));
+      if (!r || truthy(r.is_archived)) return err(404, 'link not found');
+      if (!one(q('SELECT id FROM view WHERE id=? AND link_id=?', [viewId, linkId]))) return err(403, 'no active view');
+      if (truthy(ctx.query.download) && !truthy(r.allow_download)) return err(403, 'download disabled');
+      // The document must be reachable through this link (dataroom membership or
+      // the link's own document) — a viewer cannot address an unrelated doc.
+      if (r.dataroom_id) {
+        if (!one(q('SELECT id FROM dataroom_document WHERE dataroom_id=? AND document_id=?', [r.dataroom_id, docId]))) {
+          return err(404, 'document not in dataroom');
+        }
+      } else if (r.document_id !== docId) {
+        return err(404, 'document not on link');
+      }
+      var doc = one(q('SELECT * FROM document WHERE id=?', [docId]));
+      if (!doc) return err(404, 'document not found');
+      return { fileKey: doc.file_key, contentType: doc.content_type, name: doc.name };
+    },
+
+    // ---- analytics ----------------------------------------------------------
+    'analytics.link': function (ctx) {
+      var linkId = ctx.params.linkId;
+      if (!one(q('SELECT id FROM link WHERE id=?', [linkId]))) return err(404, 'link not found');
+      return linkAnalytics(linkId);
+    },
+    'analytics.dataroom': function (ctx) {
+      var roomId = ctx.params.dataroomId;
+      if (!one(q('SELECT id FROM dataroom WHERE id=?', [roomId]))) return err(404, 'dataroom not found');
+      var links = q('SELECT id FROM link WHERE dataroom_id=?', [roomId]);
+      var perLink = links.map(function (l) { return linkAnalytics(l.id); });
+      var totalViews = 0, totalPageViews = 0;
+      perLink.forEach(function (a) { totalViews += a.totalViews; totalPageViews += a.totalPageViews; });
+      return { dataroomId: roomId, totalViews: totalViews, totalPageViews: totalPageViews, links: perLink };
+    },
+  };
+
+  function linkAnalytics(linkId) {
+    var totalViews = Number((one(q('SELECT COUNT(*) AS n FROM view WHERE link_id=?', [linkId])) || {}).n || 0);
+    var totalPageViews = Number((one(q('SELECT COUNT(*) AS n FROM page_view WHERE link_id=?', [linkId])) || {}).n || 0);
+    var pages = q(
+      'SELECT page_number, COUNT(*) AS views, SUM(duration) AS total_duration ' +
+      'FROM page_view WHERE link_id=? GROUP BY page_number ORDER BY page_number', [linkId]);
+    return {
+      linkId: linkId, totalViews: totalViews, totalPageViews: totalPageViews,
+      pages: pages.map(function (p) {
+        var v = Number(p.views || 0), d = Number(p.total_duration || 0);
+        return { pageNumber: Number(p.page_number), views: v, totalDuration: d, avgDuration: v ? Math.round(d / v) : 0 };
+      }),
+    };
+  }
+
+  // === dispatch entry (host calls this per request) ==========================
+  globalThis.handle = function (req) {
+    req = req || {};
+    var ctx = {
+      params: req.params || {},
+      query: req.query || {},
+      orgId: req.orgId || '',
+      body: req.body || null,
+    };
+    var fn = routes[req.route];
+    if (!fn) return { status: 404, body: { error: 'unknown dataroom route: ' + req.route } };
+    try {
+      var out = fn(ctx);
+      if (out && out.__status) {
+        var status = out.__status;
+        delete out.__status;
+        return { status: status, body: out };
+      }
+      return { status: 200, body: out };
+    } catch (ex) {
+      return { status: 500, body: { error: String(ex && ex.message || ex) } };
+    }
+  };
+})();

lib/kv/memory-kv.test.ts

@@ -0,0 +1,215 @@
+import assert from "node:assert/strict";
+import { test } from "node:test";
+
+import { MemoryKV } from "./memory-kv.ts";
+
+test("strings: set/get/del with correct del count", async () => {
+  const kv = new MemoryKV();
+  assert.equal(await kv.get("k"), null);
+  assert.equal(await kv.set("k", "v"), "OK");
+  assert.equal(await kv.get("k"), "v");
+  assert.equal(await kv.del("k", "absent"), 1); // only the live key counts
+  assert.equal(await kv.get("k"), null);
+});
+
+test("setex + TTL: expired key reads as absent (lazy eviction, deterministic)", async () => {
+  const kv = new MemoryKV();
+  await kv.set("live", "1", "PX", 60_000);
+  await kv.set("dead", "1", "PXAT", Date.now() - 1); // already past
+  assert.equal(await kv.get("live"), "1");
+  assert.equal(await kv.get("dead"), null);
+  assert.equal(await kv.setex("s", 60, "y"), "OK");
+  assert.equal(await kv.get("s"), "y");
+});
+
+test("SECURITY — NX gives mutual exclusion (tus lock invariant)", async () => {
+  const kv = new MemoryKV();
+  // First acquirer wins.
+  assert.equal(await kv.set("tus-lock-x", "locked", "PX", 30_000, "NX"), "OK");
+  // Second acquirer is refused while the lock is live — the null the
+  // RedisLocker relies on to NOT enter the critical section.
+  assert.equal(await kv.set("tus-lock-x", "stolen", "PX", 30_000, "NX"), null);
+  assert.equal(await kv.get("tus-lock-x"), "locked"); // value untouched
+  // unlock (DEL) then re-acquire succeeds.
+  assert.equal(await kv.del("tus-lock-x"), 1);
+  assert.equal(await kv.set("tus-lock-x", "again", "PX", 30_000, "NX"), "OK");
+});
+
+test("SECURITY — an EXPIRED lock does not block a fresh NX acquire", async () => {
+  const kv = new MemoryKV();
+  await kv.set("tus-lock-y", "old", "PXAT", Date.now() - 1); // expired lock
+  assert.equal(await kv.set("tus-lock-y", "new", "PX", 30_000, "NX"), "OK");
+  assert.equal(await kv.get("tus-lock-y"), "new");
+});
+
+test("incr: absent→1→2 and preserves TTL", async () => {
+  const kv = new MemoryKV();
+  assert.equal(await kv.incr("rl:ip"), 1);
+  assert.equal(await kv.incr("rl:ip"), 2);
+  await kv.set("c", "5", "PX", 60_000);
+  assert.equal(await kv.incr("c"), 6);
+});
+
+test("zset: zadd + zrange asc, zrevrange desc, ties broken lexicographically", async () => {
+  const kv = new MemoryKV();
+  assert.equal(await kv.zadd("z", 3, "c"), 1);
+  assert.equal(await kv.zadd("z", 1, "a"), 1);
+  assert.equal(await kv.zadd("z", 2, "b"), 1);
+  assert.equal(await kv.zadd("z", 1, "a"), 0); // re-add existing member → 0 new
+  assert.deepEqual(await kv.zrange("z", 0, -1), ["a", "b", "c"]);
+  assert.deepEqual(await kv.zrevrange("z", 0, -1), ["c", "b", "a"]);
+  assert.equal(await kv.zcard("z"), 3);
+});
+
+test("zset: job-store pattern — newest-first via score=timestamp + zrevrange", async () => {
+  const kv = new MemoryKV();
+  await kv.zadd("user_jobs:u", 1000, "job1");
+  await kv.zadd("user_jobs:u", 2000, "job2");
+  await kv.zadd("user_jobs:u", 3000, "job3");
+  assert.deepEqual(await kv.zrange("user_jobs:u", 0, 1, ), ["job1", "job2"]);
+  assert.deepEqual(await kv.zrevrange("user_jobs:u", 0, 1), ["job3", "job2"]);
+});
+
+test("zset: zrangebyscore inclusive + zremrangebyscore + zrem", async () => {
+  const kv = new MemoryKV();
+  for (const [s, m] of [[10, "a"], [20, "b"], [30, "c"]] as const) {
+    await kv.zadd("cleanup", s, m);
+  }
+  assert.deepEqual(await kv.zrangebyscore("cleanup", 0, 20), ["a", "b"]);
+  assert.equal(await kv.zremrangebyscore("cleanup", 0, 15), 1); // removes "a"
+  assert.deepEqual(await kv.zrange("cleanup", 0, -1), ["b", "c"]);
+  assert.equal(await kv.zrem("cleanup", "b", "missing"), 1);
+  assert.deepEqual(await kv.zrange("cleanup", 0, -1), ["c"]);
+});
+
+test("sets: sadd dedups, sismember/smembers/srem", async () => {
+  const kv = new MemoryKV();
+  assert.equal(await kv.sadd("report:d", "view1", "view2"), 2);
+  assert.equal(await kv.sadd("report:d", "view1"), 0); // dup → 0 new
+  assert.equal(await kv.sismember("report:d", "view1"), 1);
+  assert.equal(await kv.sismember("report:d", "nope"), 0);
+  assert.deepEqual((await kv.smembers("report:d")).sort(), ["view1", "view2"]);
+  assert.equal(await kv.srem("report:d", "view1"), 1);
+  assert.equal(await kv.sismember("report:d", "view1"), 0);
+});
+
+test("hashes: hset(field,val) + hset(object) + hincrby", async () => {
+  const kv = new MemoryKV();
+  assert.equal(await kv.hset("report:d:details", "view1", "spam"), 1);
+  assert.equal(await kv.hset("report:d:details", { view2: "abuse", view3: "x" }), 2);
+  assert.equal(await kv.hincrby("reportCount", "doc_1", 1), 1);
+  assert.equal(await kv.hincrby("reportCount", "doc_1", 2), 3);
+});
+
+test("lists: rpush + lrange incl negative range (0,-1 = all)", async () => {
+  const kv = new MemoryKV();
+  assert.equal(await kv.rpush("q", "a"), 1);
+  assert.equal(await kv.rpush("q", "b", "c"), 3);
+  assert.deepEqual(await kv.lrange("q", 0, -1), ["a", "b", "c"]);
+  assert.deepEqual(await kv.lrange("q", 0, 0), ["a"]);
+});
+
+test("getdel returns value then removes it", async () => {
+  const kv = new MemoryKV();
+  await kv.set("state", "xyz");
+  assert.equal(await kv.getdel("state"), "xyz");
+  assert.equal(await kv.get("state"), null);
+});
+
+test("SECURITY — getdel is atomic: two racing getdel, exactly one sees the value", async () => {
+  // The one-time login-code guard (lib/emails/send-verification-request.ts,
+  // fetchAndDeleteLoginCodeData) relies on GETDEL atomicity to stop a login code
+  // being used twice. A get-then-delete with an await BETWEEN the read and the
+  // delete lets both racers observe the value (TOCTOU); getdel must read+delete
+  // in one un-yielded step so only one caller ever wins.
+  const kv = new MemoryKV();
+  await kv.set("login_code:once", "grant");
+  const [a, b] = await Promise.all([
+    kv.getdel("login_code:once"),
+    kv.getdel("login_code:once"),
+  ]);
+  const winners = [a, b].filter((v) => v === "grant");
+  assert.equal(winners.length, 1, "exactly one racer may observe the value");
+  assert.equal(await kv.get("login_code:once"), null); // key is gone either way
+});
+
+test("call('SET', ...) dispatches to set (the set-with-options shim path)", async () => {
+  const kv = new MemoryKV();
+  assert.equal(await kv.call("SET", "k", "v", "EX", 60, "NX"), "OK");
+  assert.equal(await kv.call("SET", "k", "v2", "NX"), null); // exists → NX null
+  assert.equal(await kv.get("k"), "v");
+  await assert.rejects(() => kv.call("GET", "k") as Promise<unknown>, /unsupported/);
+});
+
+test("pipeline: ratelimit pattern returns ioredis [err,res] tuples in order", async () => {
+  const kv = new MemoryKV();
+  const now = Date.now();
+  const results = await kv
+    .pipeline()
+    .zremrangebyscore("rl:k", 0, now - 10_000)
+    .zadd("rl:k", now, `${now}:a`)
+    .zcard("rl:k")
+    .pexpire("rl:k", 10_000)
+    .exec();
+  assert.equal(results.length, 4);
+  for (const [err] of results) assert.equal(err, null);
+  assert.equal(results[2][1], 1); // zcard → 1 request in window
+});
+
+test("pipeline: notification-queue pattern (rpush/expire/sadd)", async () => {
+  const kv = new MemoryKV();
+  const res = await kv
+    .pipeline()
+    .rpush("items:v:d", JSON.stringify({ a: 1 }))
+    .expire("items:v:d", 3600)
+    .sadd("viewers:daily", "v:d:t")
+    .exec();
+  assert.equal(res.length, 3);
+  assert.deepEqual(await kv.lrange("items:v:d", 0, -1), ['{"a":1}']);
+  assert.deepEqual(await kv.smembers("viewers:daily"), ["v:d:t"]);
+});
+
+test("pipeline: get+del composes in order (results[0][1] = GET result before DEL)", async () => {
+  const kv = new MemoryKV();
+  await kv.set("g", "val");
+  const res = await kv.pipeline().get("g").del("g").exec();
+  assert.equal(res[0][1], "val");
+  assert.equal(await kv.get("g"), null);
+});
+
+test("SECURITY CONTRACT — two MemoryKV instances DO NOT share state", async () => {
+  // This is the WITHOUT-KV multi-replica split-brain guarantee, made explicit:
+  // each process/replica owns an isolated map. Multi-replica HA requires KV_URL.
+  const a = new MemoryKV();
+  const b = new MemoryKV();
+  await a.set("session:1", "alice");
+  assert.equal(await b.get("session:1"), null);
+});
+
+test("active expiry: sweepExpired reclaims write-once-never-read keys (bounds memory)", async () => {
+  // rl:<ip> rate-limit keys and one-shot session/token keys are written with a
+  // TTL and never read again, so the lazy on-read eviction never fires for them.
+  // Without an active sweep the map grows unbounded until the pod OOMKills.
+  // sweepExpired() reclaims every elapsed-TTL entry in one pass.
+  const kv = new MemoryKV();
+  const N = 100_000;
+  const past = Date.now() - 1; // already expired at write time (deterministic)
+  for (let i = 0; i < N; i++) {
+    await kv.set(`rl:${i}`, "1", "PXAT", past);
+  }
+  assert.equal(kv.size, N); // resident despite expiry — never read, so never lazily evicted
+  const evicted = kv.sweepExpired();
+  assert.equal(evicted, N);
+  assert.equal(kv.size, 0); // resident memory reclaimed
+});
+
+test("active expiry: sweepExpired keeps live keys, drops only expired ones", async () => {
+  const kv = new MemoryKV();
+  await kv.set("live", "1", "PX", 60_000); // far-future TTL
+  await kv.set("nottl", "1"); // no TTL at all
+  await kv.set("dead", "1", "PXAT", Date.now() - 1); // expired
+  assert.equal(kv.sweepExpired(), 1); // only "dead"
+  assert.equal(kv.size, 2);
+  assert.equal(await kv.get("live"), "1");
+  assert.equal(await kv.get("nottl"), "1");
+});

lib/kv/memory-kv.ts

@@ -0,0 +1,509 @@
+/**
+ * MemoryKV — the in-process KV backend (the WITHOUT-KV mode).
+ *
+ * When KV_URL is unset, `lib/redis.ts` binds `redis`/`lockerRedisClient` to a
+ * MemoryKV instead of an external Hanzo KV connection. It implements the exact
+ * ioredis method subset dataroom uses — strings (+TTL), sorted sets, sets,
+ * hashes, lists, and pipelines — with ioredis-faithful signatures and return
+ * types, so every call site and the Upstash-compat shims in `lib/redis.ts` are
+ * unchanged. A single process shares ONE map, which is correct for a
+ * single-replica deployment (dataroom runs replicas: 1).
+ *
+ * SCOPE / SECURITY (read before scaling): MemoryKV is per-process. It is
+ * authoritative ONLY for a single replica. Running dataroom with replicas > 1
+ * and KV_URL unset would give each replica its OWN map — sessions, rate-limit
+ * counters, tus upload locks and digest queues would NOT be shared across
+ * replicas (split-brain). Multi-replica HA REQUIRES KV_URL set (external Hanzo
+ * KV). This is the deliberate WITH/WITHOUT-KV contract, not a bug.
+ *
+ * This module has ZERO imports (pure leaf) so it is unit-testable with
+ * `node --test` and carries no bundling cost into the edge runtime.
+ */
+
+type Zset = Map<string, number>; // member -> score
+type Hash = Map<string, string>; // field -> value
+
+interface Entry {
+  // Exactly one of these is set, per Redis key type.
+  str?: string;
+  zset?: Zset;
+  set?: Set<string>;
+  hash?: Hash;
+  list?: string[];
+  // Absolute expiry in epoch ms; undefined = no expiry.
+  expireAt?: number;
+}
+
+/** A queued pipeline op: run it and yield ioredis's [err, result] tuple. */
+type PipelineOp = () => Promise<[Error | null, unknown]>;
+
+export class MemoryKV {
+  private store = new Map<string, Entry>();
+  private sweepTimer?: ReturnType<typeof setInterval>;
+
+  // ── expiry + typed accessors ──────────────────────────────────────────────
+
+  /** Return the live entry for key, evicting it first if its TTL has elapsed. */
+  private live(key: string): Entry | undefined {
+    const e = this.store.get(key);
+    if (e === undefined) return undefined;
+    if (e.expireAt !== undefined && e.expireAt <= Date.now()) {
+      this.store.delete(key);
+      return undefined;
+    }
+    return e;
+  }
+
+  /** Resident key count (live + not-yet-swept). Redis DBSIZE analogue — used to
+   *  assert active-expiry reclamation. */
+  get size(): number {
+    return this.store.size;
+  }
+
+  /**
+   * sweepExpired — active-expiry pass. Deletes every entry whose TTL has elapsed
+   * and returns the count reclaimed. The lazy on-read eviction in live() never
+   * fires for write-once-never-read keys (rl:<ip> rate-limit counters, one-shot
+   * session/token keys), so without this sweep the map grows unbounded until the
+   * pod OOMKills. One O(n) pass over the in-process map, run off the hot path on
+   * an unref'd timer (see startActiveExpiry). Pure + synchronous → unit-testable
+   * with no timer.
+   */
+  sweepExpired(now: number = Date.now()): number {
+    let evicted = 0;
+    for (const [k, e] of this.store) {
+      if (e.expireAt !== undefined && e.expireAt <= now) {
+        this.store.delete(k);
+        evicted++;
+      }
+    }
+    return evicted;
+  }
+
+  private zsetOf(key: string): Zset {
+    const e = this.live(key);
+    if (e?.zset) return e.zset;
+    const z: Zset = new Map();
+    this.store.set(key, { zset: z, expireAt: e?.expireAt });
+    return z;
+  }
+
+  private setOf(key: string): Set<string> {
+    const e = this.live(key);
+    if (e?.set) return e.set;
+    const s = new Set<string>();
+    this.store.set(key, { set: s, expireAt: e?.expireAt });
+    return s;
+  }
+
+  private hashOf(key: string): Hash {
+    const e = this.live(key);
+    if (e?.hash) return e.hash;
+    const h: Hash = new Map();
+    this.store.set(key, { hash: h, expireAt: e?.expireAt });
+    return h;
+  }
+
+  private listOf(key: string): string[] {
+    const e = this.live(key);
+    if (e?.list) return e.list;
+    const l: string[] = [];
+    this.store.set(key, { list: l, expireAt: e?.expireAt });
+    return l;
+  }
+
+  // ── strings ───────────────────────────────────────────────────────────────
+
+  async get(key: string): Promise<string | null> {
+    const e = this.live(key);
+    return e?.str ?? null;
+  }
+
+  /**
+   * set(key, value, ...opts) — supports the ioredis positional option tokens
+   * dataroom uses: EX <s>, PX <ms>, EXAT <s-epoch>, PXAT <ms-epoch>, NX.
+   * Returns "OK" on write, or null when NX is set and the key already holds a
+   * live value (this null is what the tus RedisLocker relies on for mutual
+   * exclusion — do not change it).
+   */
+  async set(
+    key: string,
+    value: string | number,
+    ...opts: Array<string | number>
+  ): Promise<"OK" | null> {
+    let expireAt: number | undefined;
+    let nx = false;
+    for (let i = 0; i < opts.length; i++) {
+      const tok = String(opts[i]).toUpperCase();
+      switch (tok) {
+        case "EX":
+          expireAt = Date.now() + Number(opts[++i]) * 1000;
+          break;
+        case "PX":
+          expireAt = Date.now() + Number(opts[++i]);
+          break;
+        case "EXAT":
+          expireAt = Number(opts[++i]) * 1000;
+          break;
+        case "PXAT":
+          expireAt = Number(opts[++i]);
+          break;
+        case "NX":
+          nx = true;
+          break;
+        case "XX":
+          // present for completeness; not used by dataroom
+          if (this.live(key) === undefined) return null;
+          break;
+        case "KEEPTTL":
+          expireAt = this.live(key)?.expireAt;
+          break;
+        default:
+          break;
+      }
+    }
+    if (nx && this.live(key) !== undefined) {
+      return null;
+    }
+    this.store.set(key, { str: String(value), expireAt });
+    return "OK";
+  }
+
+  async setex(key: string, seconds: number, value: string | number): Promise<"OK"> {
+    this.store.set(key, {
+      str: String(value),
+      expireAt: Date.now() + Number(seconds) * 1000,
+    });
+    return "OK";
+  }
+
+  async getdel(key: string): Promise<string | null> {
+    // Atomic read+delete: NO await between the read and the delete, so two racing
+    // getdel calls can never both observe the value. The one-time login-code guard
+    // in lib/emails/send-verification-request.ts (fetchAndDeleteLoginCodeData)
+    // depends on this to stop a code being used twice — see memory-kv.test.ts.
+    const e = this.live(key);
+    const v = e?.str ?? null;
+    this.store.delete(key);
+    return v;
+  }
+
+  async del(...keys: string[]): Promise<number> {
+    let n = 0;
+    for (const k of keys.flat()) {
+      // Count only keys that are currently live (matches Redis DEL semantics:
+      // an already-expired key is not counted).
+      if (this.live(k) !== undefined) n++;
+      this.store.delete(k);
+    }
+    return n;
+  }
+
+  async incr(key: string): Promise<number> {
+    const cur = (await this.get(key)) ?? "0";
+    const n = Number.parseInt(cur, 10);
+    if (Number.isNaN(n)) {
+      throw new Error("ERR value is not an integer or out of range");
+    }
+    const next = n + 1;
+    const prev = this.live(key);
+    this.store.set(key, { str: String(next), expireAt: prev?.expireAt });
+    return next;
+  }
+
+  async expire(key: string, seconds: number): Promise<number> {
+    const e = this.live(key);
+    if (e === undefined) return 0;
+    e.expireAt = Date.now() + Number(seconds) * 1000;
+    return 1;
+  }
+
+  async pexpire(key: string, ms: number): Promise<number> {
+    const e = this.live(key);
+    if (e === undefined) return 0;
+    e.expireAt = Date.now() + Number(ms);
+    return 1;
+  }
+
+  // ── sorted sets ────────────────────────────────────────────────────────────
+
+  /** zadd(key, score, member) — single pair (the only form dataroom uses). */
+  async zadd(key: string, score: number | string, member: string): Promise<number> {
+    const z = this.zsetOf(key);
+    const isNew = !z.has(member);
+    z.set(member, Number(score));
+    return isNew ? 1 : 0;
+  }
+
+  /** Members sorted by (score asc, member lexicographic) — Redis ordering. */
+  private sortedMembers(key: string): string[] {
+    const z = this.live(key)?.zset;
+    if (!z) return [];
+    return [...z.entries()]
+      .sort((a, b) => (a[1] - b[1]) || (a[0] < b[0] ? -1 : a[0] > b[0] ? 1 : 0))
+      .map(([m]) => m);
+  }
+
+  private static sliceRange(arr: string[], start: number, stop: number): string[] {
+    const n = arr.length;
+    let s = start < 0 ? n + start : start;
+    let e = stop < 0 ? n + stop : stop;
+    if (s < 0) s = 0;
+    if (e >= n) e = n - 1;
+    if (s > e) return [];
+    return arr.slice(s, e + 1);
+  }
+
+  async zrange(key: string, start: number, stop: number): Promise<string[]> {
+    return MemoryKV.sliceRange(this.sortedMembers(key), start, stop);
+  }
+
+  async zrevrange(key: string, start: number, stop: number): Promise<string[]> {
+    return MemoryKV.sliceRange(this.sortedMembers(key).reverse(), start, stop);
+  }
+
+  async zrangebyscore(
+    key: string,
+    min: number | string,
+    max: number | string,
+  ): Promise<string[]> {
+    const lo = min === "-inf" ? -Infinity : Number(min);
+    const hi = max === "+inf" ? Infinity : Number(max);
+    const z = this.live(key)?.zset;
+    if (!z) return [];
+    return [...z.entries()]
+      .filter(([, s]) => s >= lo && s <= hi)
+      .sort((a, b) => (a[1] - b[1]) || (a[0] < b[0] ? -1 : a[0] > b[0] ? 1 : 0))
+      .map(([m]) => m);
+  }
+
+  async zremrangebyscore(
+    key: string,
+    min: number | string,
+    max: number | string,
+  ): Promise<number> {
+    const lo = min === "-inf" ? -Infinity : Number(min);
+    const hi = max === "+inf" ? Infinity : Number(max);
+    const z = this.live(key)?.zset;
+    if (!z) return 0;
+    let removed = 0;
+    for (const [m, s] of [...z.entries()]) {
+      if (s >= lo && s <= hi) {
+        z.delete(m);
+        removed++;
+      }
+    }
+    return removed;
+  }
+
+  async zrem(key: string, ...members: string[]): Promise<number> {
+    const z = this.live(key)?.zset;
+    if (!z) return 0;
+    let removed = 0;
+    for (const m of members.flat()) {
+      if (z.delete(m)) removed++;
+    }
+    return removed;
+  }
+
+  async zcard(key: string): Promise<number> {
+    return this.live(key)?.zset?.size ?? 0;
+  }
+
+  // ── sets ───────────────────────────────────────────────────────────────────
+
+  async sadd(key: string, ...members: string[]): Promise<number> {
+    const s = this.setOf(key);
+    let added = 0;
+    for (const m of members.flat()) {
+      if (!s.has(m)) {
+        s.add(m);
+        added++;
+      }
+    }
+    return added;
+  }
+
+  async srem(key: string, ...members: string[]): Promise<number> {
+    const s = this.live(key)?.set;
+    if (!s) return 0;
+    let removed = 0;
+    for (const m of members.flat()) {
+      if (s.delete(m)) removed++;
+    }
+    return removed;
+  }
+
+  async smembers(key: string): Promise<string[]> {
+    return [...(this.live(key)?.set ?? [])];
+  }
+
+  async sismember(key: string, member: string): Promise<number> {
+    return this.live(key)?.set?.has(member) ? 1 : 0;
+  }
+
+  // ── hashes ─────────────────────────────────────────────────────────────────
+
+  /** hset(key, field, value) OR hset(key, { field: value, ... }). */
+  async hset(
+    key: string,
+    fieldOrObj: string | Record<string, string | number>,
+    value?: string | number,
+  ): Promise<number> {
+    const h = this.hashOf(key);
+    let added = 0;
+    const put = (f: string, v: string | number) => {
+      if (!h.has(f)) added++;
+      h.set(f, String(v));
+    };
+    if (typeof fieldOrObj === "object") {
+      for (const [f, v] of Object.entries(fieldOrObj)) put(f, v);
+    } else {
+      put(fieldOrObj, value as string | number);
+    }
+    return added;
+  }
+
+  async hincrby(key: string, field: string, increment: number): Promise<number> {
+    const h = this.hashOf(key);
+    const cur = Number.parseInt(h.get(field) ?? "0", 10);
+    if (Number.isNaN(cur)) {
+      throw new Error("ERR hash value is not an integer");
+    }
+    const next = cur + Number(increment);
+    h.set(field, String(next));
+    return next;
+  }
+
+  // ── lists ──────────────────────────────────────────────────────────────────
+
+  async rpush(key: string, ...values: string[]): Promise<number> {
+    const l = this.listOf(key);
+    l.push(...values.flat());
+    return l.length;
+  }
+
+  async lrange(key: string, start: number, stop: number): Promise<string[]> {
+    const l = this.live(key)?.list;
+    if (!l) return [];
+    return MemoryKV.sliceRange(l, start, stop);
+  }
+
+  // ── generic command dispatch (used by the SET-with-options shim) ───────────
+
+  async call(command: string, ...args: Array<string | number>): Promise<unknown> {
+    if (command.toUpperCase() === "SET") {
+      const [key, value, ...opts] = args;
+      return this.set(String(key), value, ...opts);
+    }
+    throw new Error(`MemoryKV: unsupported command '${command}'`);
+  }
+
+  // ── pipeline ───────────────────────────────────────────────────────────────
+
+  pipeline(): MemoryPipeline {
+    return new MemoryPipeline(this);
+  }
+
+  // ── active expiry (bounds resident memory) ─────────────────────────────────
+
+  /**
+   * startActiveExpiry — run sweepExpired() every intervalMs. Idempotent. The
+   * timer handle is .unref()'d so it NEVER holds the Node event loop open (the
+   * process may still exit with a sweep pending). The composition root
+   * (lib/redis.ts) starts this for the long-lived server singleton, gated OUT of
+   * the edge runtime; the pure unit tests never call it, so they stay timer-free.
+   */
+  startActiveExpiry(intervalMs: number = 60_000): void {
+    if (this.sweepTimer !== undefined) return;
+    const t = setInterval(() => this.sweepExpired(), intervalMs);
+    // unref where supported (Node Timeout); harmless no-op under a numeric handle.
+    (t as unknown as { unref?: () => void }).unref?.();
+    this.sweepTimer = t;
+  }
+
+  /** Stop the active-expiry timer (parity with quit/disconnect teardown). */
+  stopActiveExpiry(): void {
+    if (this.sweepTimer !== undefined) {
+      clearInterval(this.sweepTimer);
+      this.sweepTimer = undefined;
+    }
+  }
+
+  // ── lifecycle no-ops (ioredis parity; MemoryKV needs no connection) ────────
+
+  get status(): string {
+    return "ready";
+  }
+  on(): this {
+    return this;
+  }
+  once(): this {
+    return this;
+  }
+  async connect(): Promise<void> {}
+  async quit(): Promise<"OK"> {
+    return "OK";
+  }
+  disconnect(): void {}
+}
+
+/**
+ * MemoryPipeline mirrors ioredis's chainable pipeline. Each method queues the
+ * corresponding MemoryKV op and returns `this`; `exec()` runs them in order and
+ * returns ioredis's `Array<[Error | null, result]>` shape.
+ */
+export class MemoryPipeline {
+  private ops: PipelineOp[] = [];
+  private kv: MemoryKV;
+  constructor(kv: MemoryKV) {
+    this.kv = kv;
+  }
+
+  private queue<T>(fn: () => Promise<T>): this {
+    this.ops.push(async () => {
+      try {
+        return [null, await fn()];
+      } catch (err) {
+        return [err as Error, undefined];
+      }
+    });
+    return this;
+  }
+
+  get(key: string): this {
+    return this.queue(() => this.kv.get(key));
+  }
+  del(...keys: string[]): this {
+    return this.queue(() => this.kv.del(...keys));
+  }
+  expire(key: string, seconds: number): this {
+    return this.queue(() => this.kv.expire(key, seconds));
+  }
+  pexpire(key: string, ms: number): this {
+    return this.queue(() => this.kv.pexpire(key, ms));
+  }
+  zadd(key: string, score: number | string, member: string): this {
+    return this.queue(() => this.kv.zadd(key, score, member));
+  }
+  zcard(key: string): this {
+    return this.queue(() => this.kv.zcard(key));
+  }
+  zremrangebyscore(key: string, min: number | string, max: number | string): this {
+    return this.queue(() => this.kv.zremrangebyscore(key, min, max));
+  }
+  sadd(key: string, ...members: string[]): this {
+    return this.queue(() => this.kv.sadd(key, ...members));
+  }
+  rpush(key: string, ...values: string[]): this {
+    return this.queue(() => this.kv.rpush(key, ...values));
+  }
+
+  async exec(): Promise<Array<[Error | null, unknown]>> {
+    const results: Array<[Error | null, unknown]> = [];
+    for (const op of this.ops) {
+      results.push(await op());
+    }
+    return results;
+  }
+}

lib/kv/select.test.ts

@@ -0,0 +1,54 @@
+import assert from "node:assert/strict";
+import { test } from "node:test";
+
+/**
+ * select.ts statically imports ioredis (a production dependency, parity with the
+ * original lib/redis.ts). In a bare checkout without node_modules the import
+ * can't resolve, so we dynamic-import and SKIP — the selection logic itself is
+ * proven dep-free by url.test.ts (fail-closed + normalization) and memory-kv.test.ts
+ * (in-process semantics). Where ioredis is installed (CI, image), these run.
+ */
+let createKvClient: (() => any) | undefined;
+try {
+  ({ createKvClient } = await import("./select.ts"));
+} catch {
+  // ioredis not installed in this checkout — tests below skip.
+}
+const ready = createKvClient !== undefined;
+
+function withEnv(value: string | undefined, fn: () => void | Promise<void>) {
+  const prev = process.env.KV_URL;
+  if (value === undefined) delete process.env.KV_URL;
+  else process.env.KV_URL = value;
+  const restore = () => {
+    if (prev === undefined) delete process.env.KV_URL;
+    else process.env.KV_URL = prev;
+  };
+  return Promise.resolve()
+    .then(fn)
+    .finally(restore);
+}
+
+test("KV_URL unset ⇒ in-process backend, fully functional", { skip: !ready }, async () => {
+  await withEnv(undefined, async () => {
+    const c = createKvClient!();
+    assert.equal(c.status, "ready"); // MemoryKV lifecycle stub — no connection
+    assert.equal(await c.set("k", "v"), "OK");
+    assert.equal(await c.get("k"), "v");
+  });
+});
+
+test("KV_URL malformed ⇒ THROWS (fail closed, no silent fallback)", { skip: !ready }, async () => {
+  await withEnv("http://:pw@hanzo-kv:6379", () => {
+    assert.throws(() => createKvClient!());
+  });
+});
+
+test("KV_URL set (kv://) ⇒ external ioredis client, lazy (does not connect)", { skip: !ready }, async () => {
+  await withEnv("kv://hanzo-kv:6379", () => {
+    const c = createKvClient!();
+    // ioredis with lazyConnect stays 'wait' until first command; MemoryKV is 'ready'.
+    assert.notEqual(c.status, "ready");
+    if (typeof c.disconnect === "function") c.disconnect();
+  });
+});

lib/kv/select.ts

@@ -0,0 +1,45 @@
+/**
+ * createKvClient — the ONE backend selector, mirroring commerce infra/kv.go's
+ * NewKVClient/NewKVClientFromURL split:
+ *
+ *   KV_URL unset   ⇒ MemoryKV  (in-process, single-replica correct)
+ *   KV_URL set     ⇒ ioredis   (external Hanzo KV, multi-replica HA)
+ *   KV_URL bad     ⇒ THROW     (fail closed — resolveKvUrl never returns a
+ *                               silent in-process fallback for a malformed DSN)
+ *
+ * Kept SYNCHRONOUS so `lib/redis.ts` can export module-level singletons exactly
+ * as before. ioredis is a static import (a declared dependency, present in the
+ * built image); MemoryKV is the pure in-process leaf. The method surface either
+ * backend exposes is identical, so every call site and the Upstash-compat shims
+ * in `lib/redis.ts` are backend-agnostic.
+ */
+import IORedis, { type Redis } from "ioredis";
+
+import { MemoryKV } from "./memory-kv";
+import { resolveKvUrl } from "./url";
+
+const IOREDIS_OPTS = { maxRetriesPerRequest: 3, lazyConnect: true } as const;
+
+let warnedInProcess = false;
+
+export function createKvClient(): Redis {
+  const url = resolveKvUrl(process.env.KV_URL); // throws on malformed → fail closed
+  if (url === null) {
+    // WITHOUT-KV: in-process backend. Warn ONCE per process so this mode is never
+    // SILENT — a multi-replica deploy with KV_URL unset would split sessions,
+    // rate-limit windows and tus locks across replicas (single-replica only).
+    if (!warnedInProcess) {
+      warnedInProcess = true;
+      console.warn(
+        "[kv] KV_URL is unset — using the in-process KV backend. This is correct " +
+          "for a SINGLE replica only; multi-replica HA requires KV_URL (external Hanzo KV).",
+      );
+    }
+    // `unknown as Redis` is a controlled cast — MemoryKV implements the exact
+    // ioredis subset dataroom uses (proven by lib/kv/memory-kv.test.ts), not the
+    // full ioredis type.
+    return new MemoryKV() as unknown as Redis;
+  }
+  // WITH-KV: external Hanzo KV over the RESP wire protocol.
+  return new IORedis(url, IOREDIS_OPTS);
+}

lib/kv/url.test.ts

@@ -0,0 +1,62 @@
+import assert from "node:assert/strict";
+import { test } from "node:test";
+
+import { resolveKvUrl } from "./url.ts";
+
+test("unset / empty / whitespace ⇒ null (in-process backend)", () => {
+  assert.equal(resolveKvUrl(undefined), null);
+  assert.equal(resolveKvUrl(null), null);
+  assert.equal(resolveKvUrl(""), null);
+  assert.equal(resolveKvUrl("   "), null);
+});
+
+test("kv:// brand scheme normalizes to redis:// wire scheme", () => {
+  assert.equal(
+    resolveKvUrl("kv://hanzo-kv.hanzo.svc:6379"),
+    "redis://hanzo-kv.hanzo.svc:6379",
+  );
+  assert.equal(
+    resolveKvUrl("kv://:secret@hanzo-kv:6379"),
+    "redis://:secret@hanzo-kv:6379",
+  );
+});
+
+test("kvs:// (TLS brand scheme) normalizes to rediss://", () => {
+  assert.equal(resolveKvUrl("kvs://hanzo-kv:6380"), "rediss://hanzo-kv:6380");
+});
+
+test("raw redis:// / rediss:// DSN is accepted verbatim (commerce parity)", () => {
+  assert.equal(
+    resolveKvUrl("redis://:pw@hanzo-kv:6379/0"),
+    "redis://:pw@hanzo-kv:6379/0",
+  );
+  assert.equal(resolveKvUrl("rediss://hanzo-kv:6380"), "rediss://hanzo-kv:6380");
+});
+
+test("FAIL CLOSED: unparseable URL throws (never silent in-process fallback)", () => {
+  assert.throws(() => resolveKvUrl("::: not a url :::"), /malformed/);
+});
+
+test("FAIL CLOSED: non-redis scheme throws", () => {
+  // A bare host:port parses as scheme 'localhost:' — must be rejected, not
+  // silently connected to a phantom localhost.
+  assert.throws(() => resolveKvUrl("localhost:6379"), /unsupported scheme|malformed/);
+  assert.throws(() => resolveKvUrl("http://hanzo-kv:6379"), /unsupported scheme/);
+  assert.throws(() => resolveKvUrl("valkey://hanzo-kv:6379"), /unsupported scheme/);
+});
+
+test("FAIL CLOSED: missing host throws", () => {
+  assert.throws(() => resolveKvUrl("redis://"), /missing a host/);
+});
+
+test("error message REDACTS the password (no credential leak in logs)", () => {
+  let msg = "";
+  try {
+    resolveKvUrl("http://:supersecretpw@hanzo-kv:6379");
+  } catch (e) {
+    msg = (e as Error).message;
+  }
+  assert.ok(msg.length > 0, "expected a throw");
+  assert.ok(!msg.includes("supersecretpw"), "password must not appear in error");
+  assert.ok(msg.includes("<redacted>"), "expected redaction marker");
+});

lib/kv/url.ts

@@ -0,0 +1,79 @@
+/**
+ * KV_URL resolution — the ONE place that decides the KV backend.
+ *
+ * Contract (uniform across every Hanzo KV consumer; mirrors commerce infra/kv.go):
+ *   - KV_URL unset/empty  ⇒ returns null ⇒ caller uses the in-process backend
+ *                            (Base/in-memory). This is the WITHOUT-KV mode, which
+ *                            is single-replica correct.
+ *   - KV_URL set          ⇒ returns a normalized `redis://`/`rediss://` DSN for the
+ *                            external Hanzo KV instance (RESP wire protocol).
+ *   - KV_URL malformed     ⇒ THROWS. Fail CLOSED — never silently fall back to the
+ *                            in-process backend, which would mask a misconfigured
+ *                            external KV as a "working" single-replica cache
+ *                            (silent split-brain across replicas).
+ *
+ * Brand law: the env var is `KV_URL` and its canonical scheme is the Hanzo KV
+ * brand scheme `kv://` (`kvs://` for TLS). RESP is the wire protocol, so we map
+ * the brand scheme onto `redis://`/`rediss://` — the ONE allowed "redis" token,
+ * confined to the wire URL. A raw `redis://`/`rediss://` DSN (what commerce's
+ * KMS-synced secret carries) is accepted verbatim for cross-service uniformity.
+ */
+
+/**
+ * resolveKvUrl maps the raw KV_URL env value to either null (⇒ in-process) or a
+ * validated `redis://`/`rediss://` DSN (⇒ external Hanzo KV). Throws on a
+ * malformed URL — the fail-closed contract.
+ */
+export function resolveKvUrl(raw: string | undefined | null): string | null {
+  const value = (raw ?? "").trim();
+  if (value === "") {
+    return null;
+  }
+
+  // Brand scheme kv:// / kvs:// → RESP wire scheme redis:// / rediss://.
+  const normalized = value.startsWith("kv://")
+    ? "redis://" + value.slice("kv://".length)
+    : value.startsWith("kvs://")
+      ? "rediss://" + value.slice("kvs://".length)
+      : value;
+
+  // Parse to validate. `new URL` throws on a malformed URL → propagates as the
+  // fail-closed error (we never swallow it into a null/in-process fallback).
+  let parsed: URL;
+  try {
+    parsed = new URL(normalized);
+  } catch (cause) {
+    throw new Error(
+      `KV_URL is malformed and cannot be parsed as a redis:// DSN (got ${redact(value)})`,
+      { cause },
+    );
+  }
+
+  if (parsed.protocol !== "redis:" && parsed.protocol !== "rediss:") {
+    throw new Error(
+      `KV_URL has unsupported scheme '${parsed.protocol}' — expected kv://, kvs://, redis:// or rediss:// (got ${redact(value)})`,
+    );
+  }
+
+  if (parsed.hostname === "") {
+    throw new Error(
+      `KV_URL is missing a host — expected e.g. redis://:<password>@hanzo-kv:6379 (got ${redact(value)})`,
+    );
+  }
+
+  return normalized;
+}
+
+/**
+ * redact strips any userinfo (password) from a KV_URL before it appears in an
+ * error message or log line — a malformed DSN must never leak its credential.
+ */
+function redact(value: string): string {
+  const at = value.lastIndexOf("@");
+  if (at === -1) {
+    return value;
+  }
+  const schemeEnd = value.indexOf("://");
+  const schemePrefix = schemeEnd === -1 ? "" : value.slice(0, schemeEnd + 3);
+  return `${schemePrefix}<redacted>@${value.slice(at + 1)}`;
+}

lib/redis.ts

@@ -1,24 +1,37 @@
 /**
  * Hanzo KV client
  *
- * Connects to Hanzo KV (Valkey-compatible) via standard wire protocol.
- * Uses ioredis as transport (wire-compatible with Hanzo KV / Valkey / Redis).
+ * Pluggable KV backend, uniform with commerce (infra/kv.go):
+ *   - KV_URL UNSET ⇒ in-process MemoryKV (the WITHOUT-KV mode; single-replica
+ *     correct). dataroom boots and every KV feature works with no external
+ *     datastore — no phantom `redis://localhost:6379` connect that hangs/retries.
+ *   - KV_URL SET   ⇒ external Hanzo KV over the RESP wire protocol via ioredis
+ *     (the WITH-KV mode; multi-replica HA). A malformed KV_URL fails CLOSED
+ *     (throws) — never a silent in-process fallback.
  *
- * Environment: KV_URL (e.g. redis://:password@hanzo-kv.hanzo.svc:6379)
+ * The selected backend exposes an identical method surface, so the Upstash-compat
+ * shims below and every call site are backend-agnostic. See lib/kv/select.ts.
+ *
+ * Environment: KV_URL (e.g. kv://:password@hanzo-kv:6379 or a redis:// DSN).
  */
-import IORedis from "ioredis";
+import { MemoryKV } from "./kv/memory-kv";
+import { createKvClient } from "./kv/select";
 
-const kvUrl = process.env.KV_URL || process.env.REDIS_URL || "redis://localhost:6379";
+export const redis = createKvClient();
 
-export const redis = new IORedis(kvUrl, {
-  maxRetriesPerRequest: 3,
-  lazyConnect: true,
-});
+export const lockerRedisClient = createKvClient();
 
-export const lockerRedisClient = new IORedis(kvUrl, {
-  maxRetriesPerRequest: 3,
-  lazyConnect: true,
-});
+// In-process backend (KV_URL unset): actively expire keys so write-once-never-read
+// keys (rate-limit rl:<ip>, one-shot tokens) can't accrete to an OOM. Server runtime
+// only — the edge runtime has no long-lived process to sweep, and active expiry is
+// moot against the external ioredis client (Hanzo KV expires its own keys). The timer
+// is unref'd (MemoryKV.startActiveExpiry) so it never holds the process open; no test
+// imports this module, so unit runs stay timer-free.
+if (process.env.NEXT_RUNTIME !== "edge") {
+  for (const client of [redis, lockerRedisClient]) {
+    if (client instanceof MemoryKV) client.startActiveExpiry();
+  }
+}
 
 // Upstash-compatible set() shim — translates options-object calls to ioredis positional args
 const originalSet = redis.set.bind(redis);
@@ -58,14 +71,11 @@ const originalZrangebyscore = redis.zrangebyscore.bind(redis);
   return originalZrange(key, start as number, stop as number);
 };
 
-// Upstash-compatible getdel() shim — atomic GET + DEL
-(redis as any).getdel = async function (key: string) {
-  const pipeline = redis.pipeline();
-  pipeline.get(key);
-  pipeline.del(key);
-  const results = await pipeline.exec();
-  return results?.[0]?.[1] ?? null;
-};
+// No getdel() shim: GETDEL is atomic natively on BOTH backends — ioredis maps it
+// to the single-round-trip Redis GETDEL command, and MemoryKV.getdel reads+deletes
+// in one un-yielded step. A get→del pipeline here would NOT be atomic (other
+// commands interleave between GET and DEL) and would reintroduce the one-time
+// login-code double-use race. Do not add one.
 
 // Upstash-compatible get() shim — try to auto-parse JSON
 const originalGet = redis.get.bind(redis);

package.json

@@ -10,6 +10,8 @@
     "build": "next build",
     "start": "next start",
     "lint": "next lint",
+    "test": "node --test --experimental-strip-types \"lib/**/*.test.ts\"",
+    "test:kv": "node --test --experimental-strip-types \"lib/kv/*.test.ts\"",
     "postinstall": "prisma generate",
     "vercel-build": "prisma migrate deploy && next build",
     "email": "email dev --dir ./components/emails --port 3001",

tsconfig.json

@@ -31,5 +31,5 @@
     ".next/types/**/*.ts",
     "trigger.config.ts"
   ],
-  "exclude": ["node_modules"]
+  "exclude": ["node_modules", "**/*.test.ts"]
 }