import { randomHex } from "./hash.js";
import type { createPinboard } from "./server.js";
import { validateNamespace } from "./router.js";
import {
  releasePlacements,
  runScript,
  updateNsWorkers,
  windDownPlacement,
  windDownPlacements,
} from "./placement.js";
import { reclaimScript } from "./scripts/placement.js";
import {
  heartbeatScript,
  mirrorScript,
  registerScript,
} from "./scripts/registry.js";
import type { Transport } from "./proxy.js";
import { verifyChallenge, type ProbeFailure } from "./challenge.js";
import {
  deregisterSchema,
  drainSchema,
  heartbeatSchema,
  registerSchema,
  releaseItemSchema,
  undrainSchema,
  validateWorldSegment,
  windDownSchema,
} from "./registry-bodies.js";
import {
  DEFAULT_LEASE_TTL_MS,
  SINGLE_WORLD,
  createKeys,
  nsWorkersField,
  parseBindingEpoch,
  parseClientMetadata,
  parseClientConfig,
  parseNsWorkersField,
  parsePlacementRow,
  parseSessionIds,
  parseWorkerRecord,
  type Keys,
  type Takeover,
  type WorkerRecord,
} from "./keys.js";

type RedisKV = createPinboard.RedisKV;

const WIND_DOWN_SCAN_COUNT = 100;

const DEFAULT_VERIFY_TIMEOUT_MS = 2_500;
const DRAIN_KEY_RETENTION_MS = 7 * 24 * 60 * 60 * 1000;

export type RegistryResult<T> =
  | ({ ok: true } & T)
  | {
      ok: false;
      error: string;
      nack?: "reregister";
      conflict?: true;
      /** The advertise URL was unreachable or answered 502/503/504: retry, not a rejection. */
      unavailable?: true;
      forbidden?: true;
    };

type RegistryAuthorization = {
  account?: string;
  namespaceAllowed(namespace: string): boolean;
  worldAllowed(hostname: string, env: string): boolean;
};

const sameAccount = (
  authorization: RegistryAuthorization | undefined,
  owner: { account?: string | undefined },
): boolean =>
  authorization?.account === undefined ||
  owner.account === authorization.account;

// The scope walk: namespace registered (env known), namespace allowed, world allowed.
const inScope = (
  authorization: RegistryAuthorization | undefined,
  hostname: string,
  env: string | undefined,
  namespace: string,
): boolean =>
  env !== undefined &&
  (authorization === undefined ||
    (authorization.namespaceAllowed(namespace) &&
      authorization.worldAllowed(hostname, env)));

export type HeartbeatResult =
  | {
      ok: true;
      expiresInMs: number;
      draining: boolean;
      /** Remaining seconds of a timed admin drain; absent otherwise. */
      drainDeadline?: number;
      /** Set when this heartbeat's reverse probe failed and marked the record. */
      degraded?: ProbeFailure;
      messages: Record<string, unknown>[];
    }
  | {
      ok: false;
      error: string;
      nack?: "reregister" | "superseded";
      forbidden?: true;
      /** Set when a challenge value mismatch voided the registration. */
      voided?: { advertiseUrl: string };
    };

export interface NamespaceEntry {
  id: string;
  sessionId: string;
  /** Epoch ms wind-down deadline on the placement, when set. */
  expiry?: number;
}

export interface NamespaceOverview {
  hostname: string;
  env: string;
  namespace: string;
  sessionIds: string[];
  entries: NamespaceEntry[];
}

export interface Overview {
  workers: Omit<WorkerRecord, "secret">[];
  bindings: {
    /** (hostname, ns) -> env, keyed `${hostname}${ns}`. */
    hostnames: Record<string, string>;
    envs: Record<string, string>;
  };
  namespaces: NamespaceOverview[];
}

export interface WorldNamespace {
  hostname: string;
  env: string;
  ns: string;
}

export class Registry {
  readonly leaseTtlMs: number;
  readonly hostnameAware: boolean;
  private readonly verifyTimeoutMs: number;
  private readonly redis: RedisKV;
  private readonly keys: Keys;
  private readonly now: () => number;
  private readonly fetchImpl: Transport.FetchLike;

  constructor(
    redis: RedisKV,
    keys: Keys = createKeys(),
    leaseTtlMs: number = DEFAULT_LEASE_TTL_MS,
    now: () => number = () => Date.now(),
    hostnameAware = false,
    verifyTimeoutMs: number = DEFAULT_VERIFY_TIMEOUT_MS,
    fetchImpl: Transport.FetchLike = (request) => fetch(request),
  ) {
    this.fetchImpl = fetchImpl;
    this.redis = redis;
    this.keys = keys;
    this.leaseTtlMs = leaseTtlMs;
    this.now = now;
    this.hostnameAware = hostnameAware;
    this.verifyTimeoutMs = verifyTimeoutMs;
  }

  private buckets(): number[] {
    return Array.from({ length: this.keys.shards }, (_, s) => s);
  }

  /** draining, expiry, and the admin deadline per the drain config: a timed
   * drain past its deadline stops the expiry from advancing beyond
   * deadline + lease. `drainDeadline` is the remaining relative seconds of a
   * timed admin drain (0 once passed), null for cordons and self-drains. */
  private async drainState(
    clientId: string,
    selfDraining: boolean,
  ): Promise<{
    draining: boolean;
    expiry: number;
    drainDeadline: number | null;
  }> {
    const raw = await this.redis.get(this.keys.client(clientId));
    const config = raw === null ? null : parseClientConfig(raw);
    const now = this.now();
    const deadline = config?.drainDeadline ?? now;
    return {
      draining: config !== null || selfDraining,
      expiry: Math.min(now, deadline) + this.leaseTtlMs,
      drainDeadline:
        config?.drainDeadline == null
          ? null
          : Math.max(0, (config.drainDeadline - now) / 1000),
    };
  }

  /**
   * Mints the sessionId (the worker's whole lease/auth/placement identity),
   * after proving the advertiseUrl terminates at the registering process via
   * its challenge. A lost session registers fresh; its old placements
   * re-place through dead-owner eviction — except ids submitted in
   * `instances` (ns -> ids still held in the worker's memory), which reclaim
   * on process continuity only:
   * - rows still owned by `prevSessionId` re-bind to the new session,
   *   keeping their epoch; a pending wind-down carries and re-notifies
   * - every other id is untouched, returned in `duplicates` for the worker
   *   to evict locally; its row re-places through resolve on demand
   */
  async register(
    body: unknown,
    account?: string,
  ): Promise<
    RegistryResult<{
      sessionId: string;
      expiresInMs: number;
      draining: boolean;
      /** Remaining seconds of a timed admin drain; absent otherwise. */
      drainDeadline?: number;
      hostname: string;
      envs: Record<string, string>;
      reclaimed: number;
      duplicates: { ns: string; id: string }[];
    }>
  > {
    if (!registerSchema(body)) {
      return {
        ok: false,
        error:
          "expected { clientId, secret, namespaces, advertiseUrl, headroom, draining, challenge, challengeValue, instances?, prevSessionId?" +
          (this.hostnameAware
            ? ", hostname, env?, envs?, takeover?, epoch? }"
            : " }"),
      };
    }
    const msg = body;
    let namespaces: string[];
    let hostname: string;
    let envs: Record<string, string>;
    const reclaims: { ns: string; env: string; ids: string[] }[] = [];
    try {
      namespaces = msg.namespaces.map(validateNamespace);
      if (this.hostnameAware) {
        if (msg.hostname === undefined) {
          throw new Error("hostname is required in hostname-aware mode");
        }
        hostname = validateWorldSegment("hostname", msg.hostname);
        const defaultEnv =
          msg.env === undefined
            ? hostname
            : validateWorldSegment("env", msg.env);
        const overrides = msg.envs ?? {};
        for (const ns of Object.keys(overrides)) {
          if (!namespaces.includes(ns)) {
            throw new Error(`envs key is not a registered namespace: ${ns}`);
          }
        }
        envs = Object.fromEntries(
          namespaces.map((ns) => [
            ns,
            overrides[ns] === undefined
              ? defaultEnv
              : validateWorldSegment("env", overrides[ns]),
          ]),
        );
      } else {
        if (
          msg.hostname !== undefined ||
          msg.env !== undefined ||
          msg.envs !== undefined ||
          msg.takeover !== undefined ||
          (msg.epoch !== undefined && msg.epoch !== 0)
        ) {
          throw new Error(
            "hostname, env, envs, takeover, and nonzero epoch require hostname-aware mode",
          );
        }
        hostname = SINGLE_WORLD;
        envs = Object.fromEntries(namespaces.map((ns) => [ns, SINGLE_WORLD]));
      }
      for (const [rawNs, ids] of Object.entries(msg.instances ?? {})) {
        const ns = validateNamespace(rawNs);
        if (!namespaces.includes(ns)) {
          throw new Error(`instances key is not a registered namespace: ${ns}`);
        }
        if (ids.length > 0) reclaims.push({ ns, env: envs[ns]!, ids });
      }
    } catch (err) {
      return {
        ok: false,
        error: err instanceof Error ? err.message : "invalid registration",
      };
    }
    const takeover: Takeover = msg.takeover ?? "wind_down";
    const advertiseUrl = msg.advertiseUrl.replace(/\/+$/, "");
    const challengeError = await verifyChallenge(
      this.fetchImpl,
      advertiseUrl,
      msg.secret,
      msg.challenge,
      msg.challengeValue,
      this.verifyTimeoutMs,
    );
    if (challengeError !== null) {
      return {
        ok: false,
        error: challengeError.error,
        ...((challengeError.reason === "unreachable" ||
          (challengeError.status !== undefined &&
            [502, 503, 504].includes(challengeError.status))) && {
          unavailable: true as const,
        }),
      };
    }
    const sessionId = randomHex(16);
    const base = {
      sessionId,
      clientId: msg.clientId,
      advertiseUrl,
      secret: msg.secret,
      ...(account !== undefined && { account }),
      namespaces,
      hostname,
      envs,
      takeover,
      headroom: msg.headroom,
    };
    const now = this.now();
    // Bucket 0 computes draining/expiry against the drain config, checks the
    // hostname/env bindings, and returns the final record; the other buckets
    // mirror it byte-for-byte.
    const result = (await runScript(
      this.redis,
      registerScript,
      [
        this.keys.workers(0),
        this.keys.nsWorkers(0),
        this.keys.client(msg.clientId),
        this.keys.bindings(0),
        this.keys.bindepochs(0),
        this.keys.clientMeta(msg.clientId),
      ],
      [
        JSON.stringify(base),
        String(now),
        String(this.leaseTtlMs),
        msg.draining ? "1" : "0",
        String(msg.epoch ?? 0),
      ],
    )) as [string, string, string];
    if (!Array.isArray(result) || result.length < 2) {
      throw new Error(`malformed register result: ${JSON.stringify(result)}`);
    }
    if (result[0] === "conflict") {
      return { ok: false, error: result[1], conflict: true };
    }
    if (result.length !== 3) {
      throw new Error(`malformed register result: ${JSON.stringify(result)}`);
    }
    const raw = result[1];
    for (const bucket of this.buckets().slice(1)) {
      await runScript(
        this.redis,
        mirrorScript,
        [
          this.keys.workers(bucket),
          this.keys.nsWorkers(bucket),
          this.keys.bindings(bucket),
          this.keys.bindepochs(bucket),
        ],
        [raw, String(now), String(this.leaseTtlMs), result[2]],
      );
    }
    const record = parseWorkerRecord(raw);
    let reclaimed = 0;
    const duplicates: { ns: string; id: string }[] = [];
    for (const { ns, env, ids } of reclaims) {
      const byBucket = new Map<number, string[]>();
      for (const id of ids) {
        const bucket = this.keys.bucket(id);
        const list = byBucket.get(bucket) ?? [];
        if (list.length === 0) byBucket.set(bucket, list);
        list.push(id);
      }
      for (const [bucket, bucketIds] of byBucket) {
        const result = await runScript(
          this.redis,
          reclaimScript,
          [this.keys.placements(bucket, env, ns)],
          [
            sessionId,
            msg.prevSessionId ?? "",
            String(this.now()),
            ...bucketIds,
          ],
        );
        if (!Array.isArray(result) || result.length !== 3) {
          throw new Error(
            `malformed reclaim result: ${JSON.stringify(result)}`,
          );
        }
        const [count, dups, carries] = result as [number, string[], string[]];
        reclaimed += Number(count);
        for (const id of dups) duplicates.push({ ns, id: String(id) });
        for (let i = 0; i < carries.length; i += 2) {
          await this.queueWindDown(
            sessionId,
            ns,
            String(carries[i]),
            Number(carries[i + 1]),
          );
        }
      }
    }
    const { drainDeadline } = await this.drainState(msg.clientId, msg.draining);
    return {
      ok: true,
      sessionId,
      expiresInMs: this.leaseTtlMs,
      draining: record.draining,
      ...(record.draining && drainDeadline !== null && { drainDeadline }),
      hostname: record.hostname,
      envs: record.envs,
      reclaimed,
      duplicates,
    };
  }

  /** Live worker record (secret included) for a session, or null. */
  async worker(sessionId: string): Promise<WorkerRecord | null> {
    const result = await this.liveRecord(sessionId);
    return "record" in result ? result.record : null;
  }

  private async liveRecord(
    sessionId: string,
  ): Promise<{ record: WorkerRecord } | { error: string }> {
    const raw = await this.redis.hget(this.keys.workers(0), sessionId);
    const record = raw === null ? null : parseWorkerRecord(raw);
    if (record === null || record.expiry <= this.now()) {
      return { error: "unknown session" };
    }
    return { record };
  }

  private authorized(
    record: WorkerRecord,
    authorization: RegistryAuthorization | undefined,
  ): boolean {
    return (
      sameAccount(authorization, record) &&
      record.namespaces.every((ns) =>
        inScope(authorization, record.hostname, record.envs[ns], ns),
      )
    );
  }

  private scopedRecord(
    record: WorkerRecord,
    authorization: RegistryAuthorization | undefined,
  ): WorkerRecord | null {
    if (authorization === undefined) return record;
    if (!sameAccount(authorization, record)) return null;
    const namespaces = record.namespaces.filter((ns) =>
      inScope(authorization, record.hostname, record.envs[ns], ns),
    );
    if (namespaces.length === 0) return null;
    if (namespaces.length === record.namespaces.length) return record;
    return {
      ...record,
      namespaces,
      envs: Object.fromEntries(namespaces.map((ns) => [ns, record.envs[ns]!])),
    };
  }

  private forbidden<T>(): RegistryResult<T> {
    return { ok: false, error: "forbidden", forbidden: true };
  }

  /** Authenticates by sessionId, re-proves the advertiseUrl via the carried
   * challenge (an unreachable, non-2xx, or body-less probe still renews but
   * marks the record degraded: existing placements keep routing, new
   * placements skip it; a probe answering the wrong value voids the
   * registration — records dropped, 409 reregister), recomputes the drain
   * state, and writes the refreshed record. The ack echoes draining, plus the remaining seconds of
   * a timed admin drain as `drainDeadline`. */
  async heartbeat(
    body: unknown,
    authorization?: RegistryAuthorization,
  ): Promise<HeartbeatResult> {
    if (!heartbeatSchema(body)) {
      return {
        ok: false,
        error:
          "expected { sessionId, headroom, draining, challenge, challengeValue, acks? }",
      };
    }
    const msg = body;
    const found = await this.liveRecord(msg.sessionId);
    if ("error" in found) {
      return { ok: false, error: found.error, nack: "reregister" };
    }
    const prev = found.record;
    if (!this.authorized(prev, authorization)) {
      return { ok: false, error: "forbidden", forbidden: true };
    }
    const superseded = await this.supersededBy(prev);
    if (superseded !== null) {
      await this.dropRecords(prev);
      return { ok: false, error: superseded, nack: "superseded" };
    }
    const probe = await verifyChallenge(
      this.fetchImpl,
      prev.advertiseUrl,
      prev.secret,
      msg.challenge,
      msg.challengeValue,
      this.verifyTimeoutMs,
    );
    if (probe !== null && probe.reason === "mismatch") {
      await this.dropRecords(prev);
      return {
        ok: false,
        error:
          "registration voided: challenge value mismatch on the advertise URL",
        nack: "reregister",
        voided: { advertiseUrl: prev.advertiseUrl },
      };
    }
    const { draining, expiry, drainDeadline } = await this.drainState(
      prev.clientId,
      msg.draining,
    );
    const { degradedSince: _degradedSince, ...base } = prev;
    const record: WorkerRecord = {
      ...base,
      headroom: msg.headroom,
      draining,
      expiry,
      ...(probe !== null && {
        degradedSince: prev.degradedSince ?? this.now(),
      }),
    };
    const raw = JSON.stringify(record);
    const now = this.now();
    // Bucket 0 is the binding source of truth; the refresh mirrors its
    // bindings into buckets a shard-count raise added, first writer wins.
    let entries = "";
    for (const bucket of this.buckets()) {
      const result = await runScript(
        this.redis,
        heartbeatScript,
        [
          this.keys.workers(bucket),
          this.keys.nsWorkers(bucket),
          this.keys.bindings(bucket),
          this.keys.bindepochs(bucket),
        ],
        [raw, String(now), entries],
      );
      if (bucket === 0) {
        if (typeof result !== "string") {
          throw new Error(
            `malformed heartbeat result: ${JSON.stringify(result)}`,
          );
        }
        entries = result;
      }
    }
    if (msg.acks !== undefined && msg.acks.length > 0) {
      await this.ackControl(msg.sessionId, msg.acks);
    }
    const messages = await this.peekControl(msg.sessionId);
    return {
      ok: true,
      expiresInMs: this.leaseTtlMs,
      draining,
      ...(draining && drainDeadline !== null && { drainDeadline }),
      ...(probe !== null && { degraded: probe }),
      messages,
    };
  }

  /** Message naming the displacing binding when a (hostname, ns) of this
   * record is now bound to a different env, else null. */
  private async supersededBy(record: WorkerRecord): Promise<string | null> {
    if (record.hostname === SINGLE_WORLD) return null;
    for (const ns of record.namespaces) {
      const field = `h:${record.hostname}${ns}`;
      const boundEnv = await this.redis.hget(this.keys.bindings(0), field);
      if (boundEnv === null || boundEnv === record.envs[ns]) continue;
      const raw = await this.redis.hget(this.keys.bindepochs(0), field);
      const epoch = (raw === null ? null : parseBindingEpoch(raw))?.epoch ?? 0;
      return `registration superseded: ${record.hostname}${ns} now bound to env ${boundEnv} at epoch ${epoch}`;
    }
    return null;
  }

  /** Deregister-shaped record removal: ns-list entries and worker records
   * across all buckets, freeing the session's placements to dead-owner
   * eviction. */
  private async dropRecords(record: WorkerRecord): Promise<void> {
    const now = this.now();
    for (const bucket of this.buckets()) {
      for (const ns of record.namespaces) {
        await updateNsWorkers(
          this.redis,
          this.keys,
          bucket,
          nsWorkersField(record.hostname, ns),
          record.sessionId,
          "remove",
          now,
        );
      }
      await this.redis.hdel(this.keys.workers(bucket), record.sessionId);
    }
  }

  /** Compare-and-delete, batched: only the placing session frees its own
   * placements, and only at the quoted epoch. */
  async release(
    body: unknown,
    authorization?: RegistryAuthorization,
  ): Promise<RegistryResult<{ released: number }>> {
    if (
      !Array.isArray(body) ||
      body.length === 0 ||
      !body.every(releaseItemSchema)
    ) {
      return {
        ok: false,
        error: "expected [{ sessionId, namespace, id, epoch? }]",
      };
    }
    const groups = new Map<
      string,
      {
        sessionId: string;
        ns: string;
        entries: { id: string; epoch?: string | undefined }[];
      }
    >();
    for (const { sessionId, namespace, id, epoch } of body) {
      let ns: string;
      try {
        ns = validateNamespace(namespace);
      } catch {
        return { ok: false, error: "invalid namespace" };
      }
      const key = `${sessionId}\n${ns}`;
      const group = groups.get(key) ?? { sessionId, ns, entries: [] };
      if (group.entries.length === 0) groups.set(key, group);
      group.entries.push({ id, epoch });
    }
    let released = 0;
    const records = new Map<string, WorkerRecord>();
    const validated: {
      sessionId: string;
      ns: string;
      env: string;
      entries: { id: string; epoch?: string | undefined }[];
    }[] = [];
    for (const { sessionId, ns, entries } of groups.values()) {
      let record = records.get(sessionId);
      if (record === undefined) {
        const found = await this.liveRecord(sessionId);
        if ("error" in found) {
          return { ok: false, error: found.error, nack: "reregister" };
        }
        record = found.record;
        records.set(sessionId, record);
      }
      if (!sameAccount(authorization, record)) return this.forbidden();
      const env = record.envs[ns];
      if (env === undefined) {
        return { ok: false, error: `namespace not registered: ${ns}` };
      }
      if (!inScope(authorization, record.hostname, env, ns)) {
        return this.forbidden();
      }
      validated.push({ sessionId, ns, env, entries });
    }
    for (const { sessionId, ns, env, entries } of validated) {
      released += await releasePlacements(
        this.redis,
        this.keys,
        env,
        ns,
        sessionId,
        entries,
      );
    }
    return { ok: true, released };
  }

  /** Graceful shutdown: removes the session from ns_workers and drops the
   * worker record; the worker batch-releases its placements beforehand. */
  async deregister(
    body: unknown,
    authorization?: RegistryAuthorization,
  ): Promise<RegistryResult<{}>> {
    if (!deregisterSchema(body)) {
      return { ok: false, error: "expected { sessionId }" };
    }
    const found = await this.liveRecord(body.sessionId);
    if ("error" in found) {
      return { ok: false, error: found.error, nack: "reregister" };
    }
    if (!this.authorized(found.record, authorization)) {
      return this.forbidden();
    }
    await this.dropRecords(found.record);
    return { ok: true };
  }

  /** Live worker records including secrets; never serialized to clients. */
  async liveWorkers(
    authorization?: RegistryAuthorization,
  ): Promise<WorkerRecord[]> {
    const now = this.now();
    const workers: WorkerRecord[] = [];
    for (const raw of Object.values(
      await this.redis.hgetall(this.keys.workers(0)),
    )) {
      const record = parseWorkerRecord(raw);
      if (record.expiry <= now) continue;
      const scoped = this.scopedRecord(record, authorization);
      if (scoped !== null) workers.push(scoped);
    }
    return workers.sort(
      (a, b) =>
        a.hostname.localeCompare(b.hostname) ||
        a.sessionId.localeCompare(b.sessionId),
    );
  }

  async listWorkers(
    authorization?: RegistryAuthorization,
  ): Promise<Omit<WorkerRecord, "secret">[]> {
    return (await this.liveWorkers(authorization)).map(
      ({ secret: _secret, account: _account, ...rest }) => rest,
    );
  }

  /** The (hostname, ns) -> env (keyed `${hostname}${ns}`) and
   * env -> takeover binding tables. */
  async bindings(): Promise<Overview["bindings"]> {
    const hostnames: Record<string, string> = {};
    const envs: Record<string, string> = {};
    for (const [field, value] of Object.entries(
      await this.redis.hgetall(this.keys.bindings(0)),
    )) {
      if (field.startsWith("h:")) hostnames[field.slice(2)] = value;
      else if (field.startsWith("e:")) envs[field.slice(2)] = value;
    }
    return { hostnames, envs };
  }

  /** Every (hostname, env, namespace) with at least one registered worker. */
  async worlds(): Promise<WorldNamespace[]> {
    const { hostnames } = await this.bindings();
    const fields = new Set<string>();
    for (const bucket of this.buckets()) {
      for (const field of Object.keys(
        await this.redis.hgetall(this.keys.nsWorkers(bucket)),
      )) {
        fields.add(field);
      }
    }
    const worlds: WorldNamespace[] = [];
    for (const field of [...fields].sort()) {
      const parsed = parseNsWorkersField(field);
      if (parsed === null) continue;
      const env = hostnames[field];
      if (env === undefined) continue;
      worlds.push({ hostname: parsed.hostname, env, ns: parsed.ns });
    }
    return worlds;
  }

  /** Live workers, the binding tables, and per (hostname, namespace) its
   * sessions and placed entries (via HSCAN of the placements hashes). */
  async overview(authorization?: RegistryAuthorization): Promise<Overview> {
    const workers = await this.listWorkers(authorization);
    const live = new Set(workers.map((worker) => worker.sessionId));
    const bindings = await this.bindings();
    if (authorization !== undefined) {
      const hostnames = Object.fromEntries(
        Object.entries(bindings.hostnames).filter(([field, env]) => {
          const world = parseNsWorkersField(field);
          return (
            world !== null &&
            inScope(authorization, world.hostname, env, world.ns)
          );
        }),
      );
      const visibleEnvs = new Set(Object.values(hostnames));
      for (const field of Object.keys(bindings.hostnames)) {
        if (!(field in hostnames)) delete bindings.hostnames[field];
      }
      for (const env of Object.keys(bindings.envs)) {
        if (!visibleEnvs.has(env)) delete bindings.envs[env];
      }
    }
    const namespaces: NamespaceOverview[] = [];
    const scannedEnvNs = new Map<string, NamespaceEntry[]>();
    for (const world of await this.worlds()) {
      if (!inScope(authorization, world.hostname, world.env, world.ns))
        continue;
      const listed = await this.redis.hget(
        this.keys.nsWorkers(0),
        nsWorkersField(world.hostname, world.ns),
      );
      const sessionIds = (listed === null ? [] : parseSessionIds(listed))
        .filter((sessionId) => live.has(sessionId))
        .sort();
      const envNs = `${world.env}\n${world.ns}`;
      let entries = scannedEnvNs.get(envNs);
      if (entries === undefined) {
        entries = [];
        scannedEnvNs.set(envNs, entries);
        for (const bucket of this.buckets()) {
          const key = this.keys.placements(bucket, world.env, world.ns);
          let cursor = "0";
          do {
            const scan = await this.redis.hscan(key, cursor, 100);
            cursor = scan.cursor;
            for (const [id, raw] of scan.entries) {
              const row = parsePlacementRow(raw);
              if (!live.has(row.sessionId)) continue;
              entries.push({
                id,
                sessionId: row.sessionId,
                ...(row.expiry !== undefined && { expiry: row.expiry }),
              });
            }
          } while (cursor !== "0");
        }
        entries.sort((a, b) => a.id.localeCompare(b.id));
      }
      namespaces.push({
        hostname: world.hostname,
        env: world.env,
        namespace: world.ns,
        sessionIds,
        entries,
      });
    }
    return { workers, bindings, namespaces };
  }

  /**
   * Stamps the wind-down expiry on the placement row (never landing before
   * the current lease can expire; null revokes) and tells the owner via its
   * ctl queue so it can comply early. Unplaced id = no-op. Omitting id winds
   * down every placement in the (env, namespace); queued is then the count.
   */
  async windDown(
    body: unknown,
    authorization?: RegistryAuthorization,
  ): Promise<RegistryResult<{ queued: boolean | number }>> {
    if (!windDownSchema(body)) {
      return {
        ok: false,
        error: this.hostnameAware
          ? "expected { namespace, id?, deadline, env }"
          : "expected { namespace, id?, deadline }",
      };
    }
    let ns: string;
    let env: string;
    try {
      ns = validateNamespace(body.namespace);
      if (this.hostnameAware) {
        if (body.env === undefined) {
          throw new Error("env is required in hostname-aware mode");
        }
        env = validateWorldSegment("env", body.env);
      } else {
        if (body.env !== undefined) {
          throw new Error("env requires hostname-aware mode");
        }
        env = SINGLE_WORLD;
      }
    } catch (err) {
      return {
        ok: false,
        error: err instanceof Error ? err.message : "invalid namespace",
      };
    }
    const atMs =
      body.deadline === null
        ? null
        : this.now() + Math.max(body.deadline * 1000, this.leaseTtlMs);
    if (body.id !== undefined) {
      const placementKey = this.keys.placements(
        this.keys.bucket(body.id),
        env,
        ns,
      );
      const placementRaw = await this.redis.hget(placementKey, body.id);
      if (placementRaw === null) return { ok: true, queued: false };
      if (
        authorization !== undefined &&
        (
          await this.scopedPlacements(
            [[body.id, placementRaw]],
            env,
            ns,
            authorization,
          )
        ).length === 0
      ) {
        return this.forbidden();
      }
      const owner = await windDownPlacement(
        this.redis,
        this.keys,
        env,
        ns,
        body.id,
        atMs,
      );
      if (owner === null) return { ok: true, queued: false };
      await this.queueWindDown(owner, ns, body.id, atMs);
      return { ok: true, queued: true };
    }
    let queued = 0;
    for (const bucket of this.buckets()) {
      const key = this.keys.placements(bucket, env, ns);
      let cursor = "0";
      do {
        const scan = await this.redis.hscan(key, cursor, WIND_DOWN_SCAN_COUNT);
        cursor = scan.cursor;
        if (scan.entries.length === 0) continue;
        const allowedEntries =
          authorization === undefined
            ? scan.entries
            : await this.scopedPlacements(scan.entries, env, ns, authorization);
        if (allowedEntries.length === 0) continue;
        const owners = await windDownPlacements(
          this.redis,
          key,
          allowedEntries.map(([id]) => id),
          atMs,
        );
        for (const { id, sessionId } of owners) {
          await this.queueWindDown(sessionId, ns, id, atMs);
        }
        queued += owners.length;
      } while (cursor !== "0");
    }
    return { ok: true, queued };
  }

  private async scopedPlacements(
    entries: [string, string][],
    env: string,
    ns: string,
    authorization: RegistryAuthorization,
  ): Promise<[string, string][]> {
    const allowed: [string, string][] = [];
    for (const entry of entries) {
      const placement = parsePlacementRow(entry[1]);
      const ownerRaw = await this.redis.hget(
        this.keys.workers(0),
        placement.sessionId,
      );
      const owner = ownerRaw === null ? null : parseWorkerRecord(ownerRaw);
      if (
        inScope(
          authorization,
          owner?.hostname ?? placement.hostname ?? SINGLE_WORLD,
          owner?.envs[ns] ?? env,
          ns,
        )
      ) {
        allowed.push(entry);
      }
    }
    return allowed;
  }

  /** Queues a wind_down ctl message; deadlineMs absolute epoch ms, null revokes. */
  async queueWindDown(
    sessionId: string,
    ns: string,
    id: string,
    deadlineMs: number | null,
  ): Promise<void> {
    const queue = this.keys.controlQueue(sessionId);
    await this.redis.rpush(
      queue,
      JSON.stringify({
        msgId: randomHex(8),
        type: "wind_down",
        ns,
        id,
        deadlineMs,
      }),
    );
    // Two leases: one covers a live session's next delivery, one absorbs the
    // enqueue-vs-peek race and Redis-vs-node clock skew.
    await this.redis.pexpire(queue, 2 * this.leaseTtlMs);
  }

  /** Trims ctl messages the worker confirmed applying; unknown ids no-op. */
  private async ackControl(sessionId: string, msgIds: string[]): Promise<void> {
    const queue = this.keys.controlQueue(sessionId);
    const acked = new Set(msgIds);
    for (const raw of await this.redis.lrange(queue, 0, -1)) {
      const { msgId } = JSON.parse(raw) as { msgId?: string };
      if (msgId !== undefined && acked.has(msgId)) {
        await this.redis.lrem(queue, 1, raw);
      }
    }
  }

  /** Peeks ctl messages without deleting — a lost response re-delivers on the
   * next heartbeat until the worker acks the msgIds — converting stored
   * absolute deadlines to the wire's relative seconds at peek time. */
  async peekControl(sessionId: string): Promise<Record<string, unknown>[]> {
    const now = this.now();
    const queue = this.keys.controlQueue(sessionId);
    const raws = await this.redis.lrange(queue, 0, -1);
    if (raws.length > 0) await this.redis.pexpire(queue, 2 * this.leaseTtlMs);
    return raws.map((value) => {
      const { deadlineMs, ...rest } = JSON.parse(value) as Record<
        string,
        unknown
      >;
      if (deadlineMs === undefined) return rest;
      return {
        ...rest,
        deadline:
          deadlineMs === null
            ? null
            : Math.max(0, ((deadlineMs as number) - now) / 1000),
      };
    });
  }

  /**
   * Admin drain: writes the client config (the only home of the drain
   * deadline), so it works while the worker is down and survives
   * re-register. deadline is relative seconds, floored to the lease window;
   * null = cordon. Register and heartbeat acks carry the draining flag and,
   * for timed drains, the remaining seconds as `drainDeadline`.
   */
  async drain(
    body: unknown,
    authorization?: RegistryAuthorization,
  ): Promise<RegistryResult<{ drained: true }>> {
    if (!drainSchema(body)) {
      return { ok: false, error: "expected { clientId, deadline }" };
    }
    if (!(await this.clientAllowed(body.clientId, authorization))) {
      return this.forbidden();
    }
    if (body.deadline === null) {
      await this.redis.set(
        this.keys.client(body.clientId),
        JSON.stringify({ drainDeadline: null }),
      );
    } else {
      const atMs = this.now() + Math.max(body.deadline * 1000, this.leaseTtlMs);
      await this.redis.set(
        this.keys.client(body.clientId),
        JSON.stringify({ drainDeadline: atMs }),
        { pxMs: atMs - this.now() + DRAIN_KEY_RETENTION_MS },
      );
    }
    return { ok: true, drained: true };
  }

  async undrain(
    body: unknown,
    authorization?: RegistryAuthorization,
  ): Promise<RegistryResult<{ revoked: boolean }>> {
    if (!undrainSchema(body)) {
      return { ok: false, error: "expected { clientId }" };
    }
    if (!(await this.clientAllowed(body.clientId, authorization))) {
      return this.forbidden();
    }
    const deleted = await this.redis.del(this.keys.client(body.clientId));
    return { ok: true, revoked: deleted === 1 };
  }

  private async clientAllowed(
    clientId: string,
    authorization: RegistryAuthorization | undefined,
  ): Promise<boolean> {
    if (authorization === undefined) return true;
    const raw = await this.redis.get(this.keys.clientMeta(clientId));
    if (raw === null) return false;
    const metadata = parseClientMetadata(raw);
    return (
      sameAccount(authorization, metadata) &&
      metadata.namespaces.every((ns) =>
        inScope(authorization, metadata.hostname, metadata.envs[ns], ns),
      )
    );
  }
}