KV-optional: run WITH and WITHOUT external Hanzo KV (#4)

* KV-optional: in-process backend when KV_URL unset

Make dataroom run WITH and WITHOUT external Hanzo KV, uniform with commerce
infra/kv.go. KV_URL unset -> in-process MemoryKV (single-replica correct: cache,
rate-limit, tus upload locks, export/download job stores, digest queues all work
with no external datastore). KV_URL set -> external Hanzo KV over ioredis
(multi-replica HA). Malformed KV_URL fails CLOSED (throws) -- never a silent
in-process fallback.

- lib/kv/url.ts: resolveKvUrl -- fail-closed parse, kv:// brand scheme -> redis://
  wire scheme, password redaction in errors.
- lib/kv/memory-kv.ts: in-process ioredis-compatible store (strings+TTL, zset,
  set, hash, list, pipeline) -- the exact surface dataroom uses.
- lib/kv/select.ts: the one backend selector (mirrors NewKVClient/FromURL).
- lib/redis.ts: use createKvClient() for redis + lockerRedisClient; drop the
  redis://localhost:6379 silent default (the WITHOUT-KV break) and vestigial
  REDIS_URL; Upstash-compat shims unchanged.
- Tests (node --test, dep-free): 25 pass proving selection + in-process
  correctness incl NX lock mutual-exclusion, TTL eviction, zset ordering,
  pipeline shape, fail-closed URL, cross-instance isolation (multi-replica
  split-brain contract).

* kv: warn once when falling back to in-process (non-silent split-brain)

Emit a one-time startup warning when KV_URL is unset so the WITHOUT-KV mode is
never silent — a multi-replica deploy with KV_URL unset splits sessions,
rate-limit windows and tus locks across replicas (single-replica only).

---------

Co-authored-by: Hanzo AI <ai@hanzo.ai>

.env.example

@@ -73,8 +73,19 @@ 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).
+#   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=
 

lib/kv/memory-kv.test.ts

@@ -0,0 +1,170 @@
+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("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 (the getdel shim relies on results[0][1])", 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);
+});

lib/kv/memory-kv.ts

@@ -0,0 +1,452 @@
+/**
+ * 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>();
+
+  // ── 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;
+  }
+
+  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> {
+    const v = await this.get(key);
+    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);
+  }
+
+  // ── 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,24 @@
 /**
  * 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 { 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 = new IORedis(kvUrl, {
-  maxRetriesPerRequest: 3,
-  lazyConnect: true,
-});
+export const lockerRedisClient = createKvClient();
 
 // Upstash-compatible set() shim — translates options-object calls to ioredis positional args
 const originalSet = redis.set.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"]
 }