import { Registry, type Overview } from "./registry.js";
import {
  EPOCH_HEADER,
  LEASE_UNHEALTHY_REFUSAL,
  REFUSAL_HEADER,
  fetchUpgrade,
  forward,
  outboundHeaders,
  type ForwardOutcome,
  type Transport,
} from "./proxy.js";
import { invalidIdReason, parseRoute } from "./router.js";
import { createAffinityKey } from "./affinity.js";
import { createAttestor, depthExceeded } from "./attest.js";
import { createRegistryApi } from "./registry-api.js";
import { attestationHeaders } from "./attestation.js";
import type { Passport } from "@assistant-ui/passport";
import {
  PINBOARD_SCOPE_PREFIX,
  passportRuleFor,
  passportUnauthorizedReason,
  validatePassportRules,
  type PassportRules,
} from "./passport.js";
import {
  createPlacer,
  releasePlacements,
  runScript,
  type World,
} from "./placement.js";
import {
  createResolutionCounts,
  refusalResponse,
  resolutionOutcome,
  type ServeResolution,
} from "./refusals.js";
import { discardResponseBody, json } from "./http.js";
import { createConnectionTracker, createInflightTracker } from "./trackers.js";
import { overwrittenCasScript } from "./scripts/placement.js";
import {
  createKeys,
  overwrittenField,
  parseBindingEpoch,
  RedisKVMarker,
  SINGLE_WORLD,
  type WorkerRecord,
} from "./keys.js";
import { createLease } from "./license.js";
import { createOutlierDetector } from "./outlier.js";
import { createMembership } from "./membership.js";
import { createDurabilitySweep } from "./durability.js";
import { logger, serializeError } from "./logger.js";
import {
  noopMetrics,
  normalizeRegistryRoute,
  type Metrics,
} from "./metrics.js";

/** A resolution older than this at send time is re-resolved, bounding the
 * forward's resolve-to-send flight segment on this node's monotonic clock. */
export const FORWARD_STALENESS_MS = 5_000;

/** Metrics collectors reuse one fleet overview for this long. */
export const OVERVIEW_SNAPSHOT_TTL_MS = 5_000;

export namespace createPinboard {
  export type SetOptions = {
    pxMs?: number;
    nx?: boolean;
  };

  export type RedisKV = {
    readonly [RedisKVMarker]: true;
    get(key: string): Promise<string | null>;
    set(key: string, value: string, opts?: SetOptions): Promise<"OK" | null>;
    del(key: string): Promise<number>;
    pexpire(key: string, ms: number): Promise<number>;
    hget(key: string, field: string): Promise<string | null>;
    hset(key: string, field: string, value: string): Promise<number>;
    hsetnx(key: string, field: string, value: string): Promise<number>;
    hdel(key: string, field: string): Promise<number>;
    hgetall(key: string): Promise<Record<string, string>>;
    hscan(
      key: string,
      cursor: string,
      count: number,
    ): Promise<{ cursor: string; entries: [string, string][] }>;
    rpush(key: string, value: string): Promise<number>;
    lrange(key: string, start: number, stop: number): Promise<string[]>;
    lrem(key: string, count: number, value: string): Promise<number>;
    scriptLoad(script: string): Promise<string>;
    evalsha(sha: string, keys: string[], argv: string[]): Promise<unknown>;
    quit(): Promise<void>;
  };

  export type FleetHealth = {
    resolutions: Record<string, number>;
    connections: number;
    quarantineMs: number;
    uptimeMs: number;
    rssBytes?: number;
  };

  export type DurabilityStore = {
    /** Pages the marked tray-index rows; null cursor starts a pass, a
     * null returned cursor ends it (the next call starts over). */
    listMarked(
      cursor: string | null,
      limit: number,
    ): Promise<{
      entries: { env: string; hostname: string; ns: string; id: string }[];
      cursor: string | null;
    }>;
  };

  export type DurabilityOptions = {
    /** Activities index enabling resurrection; absent, the sweep prunes only. */
    store?: DurabilityStore | undefined;
    sweepIntervalMs?: number | undefined;
    sweepLimit?: number | undefined;
    /** Max overwritten entries woken per sweep pass, default 25. */
    wakeLimit?: number | undefined;
    /** Concurrent wakes within a pass, default 4. */
    wakeConcurrency?: number | undefined;
    /** Per-wake request timeout and the activating re-check delay, default 15000ms. */
    wakeTimeoutMs?: number | undefined;
    /** Parked entries older than this are deleted, default 3600000ms. */
    parkedTtlMs?: number | undefined;
    /** Backoff-jitter rng in [0, 1), default Math.random. */
    random?: (() => number) | undefined;
    backoff?:
      | {
          baseDelayMs?: number | undefined;
          maxDelayMs?: number | undefined;
          parkAfterFailures?: number | undefined;
        }
      | undefined;
  };

  export type ForwardOptions = {
    /** Concurrent forwards per worker awaiting response headers; exceeding answers 503, default 1024. */
    maxInflight?: number | undefined;
    /** Cap on a forward's wait for response headers; exceeding answers 504, default 30000ms. */
    timeoutMs?: number | undefined;
  };

  export type OutlierOptions = {
    /** Consecutive activation failures that eject a worker from placement, default 5. */
    ejectAfterFailures?: number | undefined;
    /** Base ejection cooldown, doubled per consecutive ejection, default 30000ms. */
    baseEjectionMs?: number | undefined;
    /** Ejection cooldown ceiling, default 10 base windows. */
    maxEjectionMs?: number | undefined;
    /** Fraction of live workers ejectable at once (0, 1], default 0.5; one worker always remains placeable. */
    maxEjectionRatio?: number | undefined;
  };

  export type Client = {
    /** Peer IP as observed by the serving transport; feeds affinity and X-Forwarded-For. */
    address?: string | undefined;
  };

  export type Options = {
    redis: RedisKV;
    /** Registry and optional data-plane identity. */
    passport: Passport.Instance;
    /** Data-plane access rules: exact namespace keys plus "*"; unlisted namespaces are denied. */
    namespaces?: Record<string, PassportRules.Rule> | undefined;
    /** Sweep tuning; a store turns the always-on prune sweep into resurrection. */
    durability?: DurabilityOptions | undefined;
    /** Outlier-ejection tuning for placement. */
    outlier?: OutlierOptions | undefined;
    /** Per-worker forward caps. */
    forwards?: ForwardOptions | undefined;
    /** First-placement affinity: bounded-load factor (> 1, default 1.25) and
     * the client-IP source. Unset trustedProxies keys on the observed peer
     * only when no X-Forwarded-For rides the request; "*" trusts XFF
     * outright; a list of proxy IPs/CIDRs takes the rightmost XFF hop not
     * from a trusted proxy. */
    affinity?:
      | {
          balanceFactor?: number | undefined;
          trustedProxies?: string[] | "*" | undefined;
        }
      | undefined;
    /** Mount prefix (e.g. "/v2") stripped from incoming URLs; unprefixed requests get 404. */
    basePath?: string | undefined;
    leaseTtlMs?: number | undefined;
    /** Challenge-probe timeout for register and heartbeat verification, default 2500ms. */
    verifyTimeoutMs?: number | undefined;
    /** Redis key prefix, default "pinboard:". Non-empty, no whitespace. */
    keyPrefix?: string | undefined;
    /** Shard count S this node serves, default 1; the {0}:config license gates which S serves. */
    shards?: number | undefined;
    /** Hostname-aware mode: workers register a hostname/env, requests route
     * by Host. Fenced by the {0}:config license like shards. */
    hostnameAware?: boolean | undefined;
    /** Fleet epoch (integer >= 0, default 0): a higher-epoch fleet displaces
     * the license of a lower-epoch one instead of waiting it out. */
    epoch?: number | undefined;
    /** Fires once when a higher-epoch fleet takes the license; absent, this
     * node keeps running but refuses to serve. */
    onDisplaced?: (() => void) | undefined;
    /** Maps a register-body hostname to a bounded pool label on registration metrics; unset labels pool "". */
    pool?: ((hostname: string) => string) | undefined;
    /** Clock override for tests; defaults to Date.now. */
    now?: (() => number) | undefined;
    /** Placement randomness override for tests; defaults to Math.random. */
    random?: (() => number) | undefined;
    /** Outbound client override; defaults to the global fetch (Workers-ready). */
    transport?: Transport.Options | undefined;
    metrics?: Metrics.Instance | undefined;
  };

  export type Instance = {
    registry: Registry;
    /** WinterTC entry: data plane, registry, upgrades (via the transport seam). */
    fetch(req: Request, client?: Client): Promise<Response>;
    /** One maintenance pass: license refresh plus, under the sweeper lease,
     * a durability sweep. For cron-driven runtimes. */
    sweep(): Promise<void>;
    /** Starts the background license/durability loops; long-lived runtimes only. */
    start(): void;
    close(): Promise<void>;
  };
}

const GATEWAY_STATUSES = new Set([502, 503, 504]);

const isUpgradeRequest = (req: Request): boolean =>
  req.headers.get("upgrade")?.toLowerCase() === "websocket";

export const createPinboard = (
  options: createPinboard.Options,
): createPinboard.Instance => {
  const { basePath, leaseTtlMs } = options;
  const keys = createKeys(options.keyPrefix, options.shards);
  const redis = options.redis;
  if (!(RedisKVMarker in redis)) {
    throw new Error("redis must be a RedisKV (see IoredisKV in ./node)");
  }
  if (
    basePath !== undefined &&
    (!basePath.startsWith("/") || basePath.length < 2 || basePath.endsWith("/"))
  ) {
    throw new Error(
      `basePath must start with "/" and not end with "/": ${JSON.stringify(basePath)}`,
    );
  }
  const stripBasePath = (pathname: string): string | null => {
    if (basePath === undefined) return pathname;
    if (pathname === basePath) return "/";
    if (pathname.startsWith(basePath + "/")) {
      return pathname.slice(basePath.length);
    }
    return null;
  };
  const metrics = options.metrics ?? noopMetrics;
  const now = options.now ?? (() => Date.now());
  const maxInflight = options.forwards?.maxInflight ?? 1024;
  if (!Number.isInteger(maxInflight) || maxInflight < 1) {
    throw new Error(
      `forwards.maxInflight must be a positive integer: ${String(options.forwards?.maxInflight)}`,
    );
  }
  const forwardTimeoutMs = options.forwards?.timeoutMs ?? 30_000;
  if (!Number.isFinite(forwardTimeoutMs) || forwardTimeoutMs <= 0) {
    throw new Error(
      `forwards.timeoutMs must be a positive finite number: ${String(options.forwards?.timeoutMs)}`,
    );
  }
  const inflight = createInflightTracker(maxInflight);
  const transportFetch: Transport.FetchLike =
    options.transport?.fetch ?? ((request) => fetch(request));
  const transportUpgrade: Transport.Upgrade =
    options.transport?.upgrade ?? fetchUpgrade(transportFetch);
  const passport = options.passport;
  const hostnameAware = options.hostnameAware ?? false;
  const passportRules =
    options.namespaces === undefined
      ? null
      : validatePassportRules(options.namespaces);
  const registry = new Registry(
    redis,
    keys,
    leaseTtlMs,
    options.now,
    hostnameAware,
    options.verifyTimeoutMs,
    transportFetch,
  );
  const clientAffinityKey = createAffinityKey(options.affinity?.trustedProxies);
  const outlier = createOutlierDetector(
    {
      workerCount: async () => (await registry.liveWorkers()).length,
      now: options.now,
    },
    options.outlier ?? {},
  );
  const placer = createPlacer(redis, keys, {
    leaseTtlMs: registry.leaseTtlMs,
    balanceFactor: options.affinity?.balanceFactor,
    now: options.now,
    random: options.random,
    ejected: outlier.ejected,
  });
  // A placement that carried a wind-down from a dead owner re-notifies the
  // replacement session with the remaining time; a freshly issued takeover
  // fence notifies the current owner exactly once.
  const resolvePlacement: typeof placer.resolve = async (
    world,
    ns,
    id,
    preference,
    exclude,
  ) => {
    const resolution = await placer.resolve(world, ns, id, preference, exclude);
    if (
      resolution.kind === "forward" &&
      resolution.carriedExpiry !== undefined
    ) {
      await registry.queueWindDown(
        resolution.worker.sessionId,
        ns,
        id,
        resolution.carriedExpiry,
      );
    }
    if (resolution.kind === "takeover-pending") {
      if (resolution.initiated) {
        await registry.queueWindDown(
          resolution.owner.sessionId,
          ns,
          id,
          resolution.fenceExpiry,
        );
        logger.info(
          {
            operation: "takeover",
            namespace: ns,
            id,
            hostname: world.hostname,
            env: world.env,
            ownerSessionId: resolution.owner.sessionId,
            ownerHostname: resolution.owner.hostname,
            deadline: resolution.fenceExpiry,
          },
          "cross-hostname takeover: wind-down issued",
        );
      }
      metrics.takeoversTotal.inc({
        strategy: "wind_down",
        outcome: resolution.initiated ? "initiated" : "pending",
      });
    }
    if (resolution.kind === "takeover-denied") {
      metrics.takeoversTotal.inc({ strategy: "deny", outcome: "denied" });
    }
    return resolution;
  };
  const epoch = options.epoch ?? 0;
  if (!Number.isInteger(epoch) || epoch < 0) {
    throw new Error(`epoch must be an integer >= 0: ${epoch}`);
  }
  const licenseValue = JSON.stringify({
    schema: 3,
    shards: keys.shards,
    hostnameAware,
    epoch,
  });
  const license = createLease(redis, {
    key: keys.config,
    value: licenseValue,
    ttlMs: registry.leaseTtlMs,
    operation: "license",
    releaseOnClose: true,
    now: options.now,
    onDisplaced: options.onDisplaced,
    onMismatch: () => {
      void redis
        .get(keys.config)
        .then((fleet) =>
          logger.warn(
            { operation: "license", fleet, ours: licenseValue },
            "license held by a different fleet configuration; refusing to serve",
          ),
        );
    },
  });
  const membership = createMembership(redis, {
    key: keys.members,
    nodeId: crypto.randomUUID(),
    config: licenseValue,
    ttlMs: registry.leaseTtlMs,
    operation: "membership",
    now: options.now,
    onMismatch: (other) => {
      logger.warn(
        { operation: "membership", member: other, ours: licenseValue },
        "incompatible cluster member; refusing to serve",
      );
    },
  });
  const serving = async (): Promise<boolean> =>
    (await license.held()) && (await membership.ok());
  const sweeperLease = createLease(redis, {
    key: keys.sweeper,
    value: crypto.randomUUID(),
    ttlMs: registry.leaseTtlMs,
    operation: "sweeper",
    now: options.now,
  });
  const durability = createDurabilitySweep(
    {
      redis,
      keys,
      resolve: resolvePlacement,
      worlds: () => registry.worlds(),
      leaseTtlMs: registry.leaseTtlMs,
      now: options.now,
      fetch: transportFetch,
      outlier,
    },
    options.durability ?? {},
    sweeperLease,
  );
  let draining = false;
  const connections = createConnectionTracker(metrics.dataPlaneConnections);
  const startedAt = now();

  const resolutionCounts = createResolutionCounts();
  const resolve = async (
    world: World,
    ns: string,
    id: string,
    preference?: createPlacer.Preference,
    exclude?: string,
  ): Promise<ServeResolution> => {
    let resolution: ServeResolution = (await serving())
      ? await resolvePlacement(world, ns, id, preference, exclude)
      : { kind: "not-serving" };
    // In hostname-aware mode a candidate-less hostname is the routing story.
    if (hostnameAware && resolution.kind === "no-backends") {
      resolution = { kind: "no-hostname-workers" };
    }
    resolutionCounts[resolutionOutcome(resolution.kind)] += 1;
    return resolution;
  };

  const attestCaller = createAttestor({
    keys,
    hget: (key, field) => redis.hget(key, field),
    worker: (sessionId) => registry.worker(sessionId),
  });
  // Compute identity: the Host header names the hostname, port stripped.
  const hostnameOf = (req: Request): string | null => {
    const raw = req.headers.get("host");
    if (raw === null || raw === "") return null;
    const host = raw.startsWith("[")
      ? raw.slice(0, raw.indexOf("]") + 1)
      : raw.replace(/:\d+$/, "");
    return host === "" ? null : host.toLowerCase();
  };
  const worldOf = async (
    req: Request,
    ns: string,
    id: string,
  ): Promise<World | Response> => {
    if (!hostnameAware) {
      return {
        hostname: SINGLE_WORLD,
        env: SINGLE_WORLD,
        takeover: "wind_down",
      };
    }
    const hostname = hostnameOf(req);
    if (hostname === null) return json(400, { error: "missing Host header" });
    const bucket = keys.bucket(id);
    const bindings = keys.bindings(bucket);
    const env = await redis.hget(bindings, `h:${hostname}${ns}`);
    if (env === null) {
      resolutionCounts["no-hostname-workers"] += 1;
      return refusalResponse({ kind: "no-hostname-workers" }, now());
    }
    // A displaced binding quarantines NEW placements for one lease, so every
    // displaced holder is nacked or lapsed before the new env activates them.
    const epochRaw = await redis.hget(
      keys.bindepochs(bucket),
      `h:${hostname}${ns}`,
    );
    const quarantineUntil =
      epochRaw === null
        ? undefined
        : parseBindingEpoch(epochRaw).quarantineUntil;
    if (
      quarantineUntil !== undefined &&
      quarantineUntil > now() &&
      (await redis.hget(keys.placements(bucket, env, ns), id)) === null
    ) {
      resolutionCounts["binding-quarantine"] += 1;
      return refusalResponse(
        { kind: "binding-quarantine", retryAfterMs: quarantineUntil - now() },
        now(),
      );
    }
    const takeover = await redis.hget(bindings, `e:${env}`);
    if (takeover !== "wind_down" && takeover !== "deny") {
      resolutionCounts["env-conflict"] += 1;
      return refusalResponse({ kind: "env-conflict" }, now());
    }
    return { hostname, env, takeover };
  };
  const preferenceOf = (
    req: Request,
    client: createPinboard.Client,
    attested: { sessionId?: string },
  ): createPlacer.Preference => {
    if (attested.sessionId !== undefined) {
      return { preferredSessionId: attested.sessionId };
    }
    const affinityKey = clientAffinityKey(req, client);
    return affinityKey === null ? {} : { affinityKey };
  };
  // One fleet snapshot feeds every metrics collector: a pending overview is
  // shared regardless of age, a settled one serves for the TTL, a failed one
  // is evicted for the next scrape to retry.
  let snapshot: {
    overview: Promise<Overview>;
    freshUntil: number | null;
  } | null = null;
  const fleetOverview = (): Promise<Overview> => {
    if (
      snapshot === null ||
      (snapshot.freshUntil !== null && now() >= snapshot.freshUntil)
    ) {
      const taken: { overview: Promise<Overview>; freshUntil: number | null } =
        { overview: registry.overview(), freshUntil: null };
      snapshot = taken;
      taken.overview.then(
        () => {
          taken.freshUntil = now() + OVERVIEW_SNAPSHOT_TTL_MS;
        },
        () => {
          if (snapshot === taken) snapshot = null;
        },
      );
    }
    return snapshot.overview;
  };
  metrics.setDomainSources({
    workers: async () => (await fleetOverview()).workers.length,
    placements: async () => {
      const seen = new Set<string>();
      let sum = 0;
      for (const ns of (await fleetOverview()).namespaces) {
        const world = `${ns.env}\n${ns.namespace}`;
        if (seen.has(world)) continue;
        seen.add(world);
        sum += ns.entries.length;
      }
      return sum;
    },
    resolutions: () => ({ ...resolutionCounts }),
    worlds: async () => {
      const workers = (await fleetOverview()).workers;
      return {
        hostnames: new Set(workers.map((worker) => worker.hostname)).size,
        envs: new Set(workers.flatMap((worker) => Object.values(worker.envs)))
          .size,
      };
    },
  });

  const healthSnapshot = async (): Promise<createPinboard.FleetHealth> => {
    let quarantineMs = 0;
    for (let bucket = 0; bucket < keys.shards; bucket++) {
      const raw = await redis.get(keys.bootTime(bucket));
      if (raw === null) continue;
      const boot = Number(raw);
      if (!Number.isFinite(boot)) continue;
      const remaining = boot + registry.leaseTtlMs - now();
      if (remaining > quarantineMs) quarantineMs = Math.ceil(remaining);
    }
    return {
      resolutions: { ...resolutionCounts },
      connections: connections.count(),
      quarantineMs,
      uptimeMs: now() - startedAt,
      ...(typeof process !== "undefined" &&
      typeof process.memoryUsage === "function"
        ? { rssBytes: process.memoryUsage().rss }
        : {}),
    };
  };

  const handleRegistry = createRegistryApi({
    registry,
    passport,
    hostnameAware,
    metrics,
    pool: options.pool,
    serving,
    now,
    health: healthSnapshot,
    fetch: transportFetch,
  });

  const stampOf = (
    worker: { sessionId: string; secret: string },
    attested: { caller?: string; depth?: string | undefined },
    account: Passport.Principal | null,
  ): Promise<Record<string, string>> => {
    const publicScopes = account?.scopes.filter(
      (scope) => !scope.startsWith(PINBOARD_SCOPE_PREFIX),
    );
    const publicClaims =
      account === null
        ? null
        : Object.fromEntries(
            Object.entries(account.claims).filter(
              ([key]) => key !== "pinboard",
            ),
          );
    return attestationHeaders(
      worker,
      {
        ...(attested.caller !== undefined && { caller: attested.caller }),
        ...(attested.depth !== undefined && { depth: Number(attested.depth) }),
        ...(account !== null && {
          acct: {
            sub: account.sub,
            ...(publicScopes!.length > 0 && { scopes: publicScopes }),
            claims: publicClaims,
          },
        }),
      },
      now(),
    );
  };

  const outcomeReporter = (
    kind: "http" | "upgrade",
    match: { ns: string; id: string },
    worker: { sessionId: string },
    target: string,
  ): ((outcome: ForwardOutcome) => void) => {
    const release = connections.acquire();
    metrics.forwardsTotal.inc({ kind });
    return (outcome) => {
      release();
      if (outcome === "forwarded") return;
      metrics.forwardFailuresTotal.inc({ outcome });
      logger.warn(
        {
          operation: "forward",
          ...(kind === "upgrade" && { kind }),
          outcome,
          namespace: match.ns,
          id: match.id,
          sessionId: worker.sessionId,
          target,
        },
        kind === "http"
          ? "proxy forward failed"
          : "proxy upgrade/tunnel failed",
      );
    };
  };

  const handleDataPlane = async (
    req: Request,
    client: createPinboard.Client,
    match: { ns: string; id: string; rest: string },
    search: string,
    upgrade: boolean,
  ): Promise<Response> => {
    const invalidReason = invalidIdReason(match.id);
    if (invalidReason !== null) {
      return new Response(invalidReason, {
        status: 400,
        headers: { "content-type": "text/plain" },
      });
    }

    const attested = await attestCaller(req);
    if ("error" in attested) {
      logger.warn(
        {
          operation: "attest",
          namespace: match.ns,
          id: match.id,
          reason: attested.error,
        },
        "caller attestation failed",
      );
      return json(403, { error: attested.error });
    }
    if (depthExceeded(attested, match)) {
      return json(508, { error: "subrequest depth exceeded" });
    }

    // Instance-attested calls (verified Pinboard-From) bypass passport; the
    // target is authorized against the caller's registered scope instead.
    let account: Passport.Principal | null = null;
    if (passportRules !== null && attested.caller === undefined) {
      try {
        account = await passport.validate(req);
      } catch (error) {
        const reason = passportUnauthorizedReason(error);
        if (reason === null) throw error;
        logger.warn(
          {
            operation: "passport",
            namespace: match.ns,
            id: match.id,
            reason,
          },
          "credential rejected",
        );
        return json(401, { error: "invalid credential" });
      }
      const rule = passportRuleFor(passportRules, match.ns);
      if (rule === null) return json(403, { error: "namespace not allowed" });
      if (rule.allow !== "anonymous") {
        if (account === null) {
          return json(401, { error: "authentication required" });
        }
        const held = account.scopes;
        if (
          rule.allow !== "authenticated" &&
          !rule.allow.some((scope) => held.includes(scope))
        ) {
          return json(403, { error: "insufficient scope" });
        }
      }
    }

    const world = await worldOf(req, match.ns, match.id);
    if (world instanceof Response) return world;
    // An attested caller reaches only worlds its own registration covers.
    if (
      attested.worker !== undefined &&
      (attested.worker.hostname !== world.hostname ||
        attested.worker.envs[match.ns] !== world.env)
    ) {
      logger.warn(
        {
          operation: "attest",
          namespace: match.ns,
          id: match.id,
          hostname: world.hostname,
          env: world.env,
          caller: attested.caller,
          sessionId: attested.worker.sessionId,
        },
        "attested target outside the caller's registered scope",
      );
      return json(403, { error: "target outside caller scope" });
    }
    // `delivered` marks a response whose status line came from the worker
    // (the transport had not settled a failure when the headers arrived).
    // A resolution older than FORWARD_STALENESS_MS at send time returns null.
    const dispatch = async (
      placed: { worker: WorkerRecord; epoch: string },
      resolvedAt: number | null,
    ): Promise<{
      response: Response;
      delivered: boolean;
      outcome: ForwardOutcome | null;
    } | null> => {
      const worker = placed.worker;
      // The inflight slot is held while the worker's response headers (or the
      // upgrade's 101) are awaited, bounding the pile-up on a slow worker.
      const admit = inflight.acquire(worker.sessionId);
      if (admit === null) {
        logger.warn(
          {
            operation: "forward",
            namespace: match.ns,
            id: match.id,
            sessionId: worker.sessionId,
          },
          "worker inflight cap exceeded; rejecting forward",
        );
        return {
          response: json(
            503,
            { error: "worker inflight cap exceeded, retry" },
            { "retry-after": "1" },
          ),
          delivered: false,
          outcome: null,
        };
      }
      const target =
        worker.advertiseUrl + match.ns + "/" + match.id + match.rest + search;
      const headers = outboundHeaders(req, client.address, {
        ...(await stampOf(worker, attested, account)),
        [EPOCH_HEADER]: placed.epoch,
      });
      if (resolvedAt !== null && now() - resolvedAt > FORWARD_STALENESS_MS) {
        admit();
        return null;
      }
      if (upgrade) {
        const report = outcomeReporter("upgrade", match, worker, target);
        const settled = { outcome: null as ForwardOutcome | null };
        const reportOnce = (outcome: ForwardOutcome): void => {
          if (settled.outcome !== null) return;
          settled.outcome = outcome;
          report(outcome);
        };
        try {
          const response = await transportUpgrade(
            req,
            { url: target, headers },
            {
              onFailure: reportOnce,
              onClose: () => reportOnce("forwarded"),
            },
          );
          return { response, delivered: false, outcome: settled.outcome };
        } catch (error) {
          reportOnce("failed");
          throw error;
        } finally {
          admit();
        }
      }
      const report = outcomeReporter("http", match, worker, target);
      const settled = { outcome: null as ForwardOutcome | null };
      let response: Response;
      try {
        response = await forward(
          transportFetch,
          req,
          target,
          headers,
          (outcome) => {
            settled.outcome = outcome;
            report(outcome);
          },
          forwardTimeoutMs,
        );
      } finally {
        admit();
      }
      const outcome = settled.outcome;
      return {
        response,
        delivered: outcome === null || outcome === "forwarded",
        outcome,
      };
    };
    // A fresh placement whose forward never reached TCP is released: nothing
    // activated on the worker, so the client's retry re-places instead of
    // pinning to a dead target for the lease.
    const releaseUnreachableFresh = async (
      placed: { worker: WorkerRecord; epoch: string; fresh: boolean },
      outcome: ForwardOutcome | null,
    ): Promise<void> => {
      if (outcome !== "unreachable" || !placed.fresh) return;
      await outlier.report(placed.worker.sessionId, false);
      const released = await releasePlacements(
        redis,
        keys,
        world.env,
        match.ns,
        placed.worker.sessionId,
        [{ id: match.id, epoch: placed.epoch }],
      );
      logger.warn(
        {
          operation: "release",
          namespace: match.ns,
          id: match.id,
          sessionId: placed.worker.sessionId,
          released,
        },
        "fresh placement unreachable; released for re-placement",
      );
    };
    // The refusal marker is proxy-internal; it never reaches the client.
    const refusalOf = (response: Response): string | null => {
      const value = response.headers.get(REFUSAL_HEADER);
      if (value !== null) response.headers.delete(REFUSAL_HEADER);
      return value;
    };
    const nack = async (
      placed: { worker: WorkerRecord; epoch: string },
      response: Response,
    ): Promise<void> => {
      await discardResponseBody(response).catch(() => undefined);
      resolutionCounts.nack += 1;
      await outlier.report(placed.worker.sessionId, false);
      const released = await releasePlacements(
        redis,
        keys,
        world.env,
        match.ns,
        placed.worker.sessionId,
        [{ id: match.id, epoch: placed.epoch }],
      );
      logger.warn(
        {
          operation: "nack",
          namespace: match.ns,
          id: match.id,
          sessionId: placed.worker.sessionId,
          released,
        },
        "worker refused a new activation; re-placing",
      );
    };
    const nackExhausted = (): Response =>
      json(
        503,
        { error: "worker refused activation, retry" },
        { "retry-after": "1" },
      );

    // A stall between resolve and send re-resolves (twice at most); the
    // worker's epoch guard backstops a still-stale forward.
    const sendFresh = async (
      exclude?: string,
    ): Promise<
      | {
          served: Extract<ServeResolution, { kind: "forward" }>;
          sent: NonNullable<Awaited<ReturnType<typeof dispatch>>>;
        }
      | Response
    > => {
      for (let retries = 0; ; retries += 1) {
        const resolvedAt = now();
        const resolution = await resolve(
          world,
          match.ns,
          match.id,
          preferenceOf(req, client, attested),
          exclude,
        );
        if (resolution.kind === "unknown-ns") {
          return json(404, { error: "unknown namespace" });
        }
        if (resolution.kind !== "forward") {
          return refusalResponse(resolution, now());
        }
        if (resolution.worker.sessionId === exclude) return nackExhausted();
        const sent = await dispatch(
          resolution,
          retries < 2 ? resolvedAt : null,
        );
        if (sent !== null) return { served: resolution, sent };
      }
    };

    const first = await sendFresh();
    if (first instanceof Response) return first;
    let { served, sent } = first;
    if (refusalOf(sent.response) === LEASE_UNHEALTHY_REFUSAL) {
      await nack(served, sent.response);
      // A consumed request body cannot be replayed; the client's retry
      // lands on the re-resolved placement.
      if (!upgrade && req.body !== null) return nackExhausted();
      const retried = await sendFresh(served.worker.sessionId);
      if (retried instanceof Response) return retried;
      ({ served, sent } = retried);
      if (refusalOf(sent.response) === LEASE_UNHEALTHY_REFUSAL) {
        await nack(served, sent.response);
        return nackExhausted();
      }
    }
    await releaseUnreachableFresh(served, sent.outcome);
    const response = sent.response;
    if (sent.delivered) {
      await outlier.report(
        served.worker.sessionId,
        !GATEWAY_STATUSES.has(response.status),
      );
    }
    if (
      !upgrade &&
      served.listed &&
      response.status >= 200 &&
      response.status < 300
    ) {
      // A 2xx proves the served epoch alive; the fenced delete leaves a
      // marker re-listed under a newer epoch.
      runScript(
        redis,
        overwrittenCasScript,
        [keys.overwritten(keys.bucket(match.id))],
        [
          overwrittenField(world.env, world.hostname, match.ns, match.id),
          served.epoch,
          "",
        ],
      ).catch(() => undefined);
    }
    return response;
  };

  const handle = async (
    req: Request,
    client: createPinboard.Client,
    track: Metrics.RequestTrack | null,
  ): Promise<Response> => {
    const url = new URL(req.url);
    const pathname = stripBasePath(url.pathname);
    const upgrade = track === null;
    if (pathname === null) return json(404, { error: "not found" });

    if (!upgrade) {
      if (pathname === "/healthz") {
        track.route = "/healthz";
        return json(200, { ok: true });
      }
      if (pathname === "/readyz") {
        track.route = "/readyz";
        const ready = draining
          ? false
          : await serving().catch(() => {
              metrics.internalError("readiness");
              return false;
            });
        return new Response(null, { status: ready ? 200 : 503 });
      }
      if (pathname.startsWith("/registry/")) {
        track.route = normalizeRegistryRoute(pathname);
        return handleRegistry(req, pathname);
      }
    }

    const match = parseRoute(pathname);
    if (match === null) {
      return json(404, { error: upgrade ? "not found" : "unknown namespace" });
    }
    if (track !== null) track.route = "/:namespace/:id";
    return handleDataPlane(req, client, match, url.search, upgrade);
  };

  const instanceFetch = async (
    req: Request,
    client: createPinboard.Client = {},
  ): Promise<Response> => {
    const upgrade = isUpgradeRequest(req);
    const track = upgrade ? null : metrics.trackHttpRequest(req.method);
    try {
      const response = await handle(req, client, track);
      track?.done(response.status);
      return response;
    } catch (err) {
      metrics.internalError(upgrade ? "upgrade" : "request");
      logger.error(
        {
          operation: upgrade ? "upgrade" : "request",
          endpoint: req.url,
          method: req.method,
          err: serializeError(err),
        },
        upgrade ? "upgrade handler failed" : "request handler failed",
      );
      const response = json(500, { error: "internal error" });
      track?.done(response.status);
      return response;
    }
  };

  return {
    registry,
    fetch: instanceFetch,
    sweep: async () => {
      await license.refresh();
      await membership.refresh();
      if (await sweeperLease.held()) {
        await durability.sweepOnce();
      }
    },
    start: () => {
      license.start();
      membership.start();
      durability.start();
    },
    async close() {
      draining = true;
      await license.close();
      await membership.close();
      await durability.close();
      metrics.close();
    },
  };
};