import type { Resolution, World } from "./placement.js";
import type { createLease } from "./license.js";
import type { createOutlierDetector } from "./outlier.js";
import type { WorldNamespace } from "./registry.js";
import type { Transport } from "./proxy.js";
import { releasePlacements, runScript } from "./placement.js";
import { gcScript } from "./scripts/registry.js";
import { createWakePass } from "./wake.js";
import { validateNamespace } from "./router.js";
import { logger, serializeError } from "./logger.js";
import type { createPinboard } from "./server.js";
import {
  parseBindingEpoch,
  parseClientConfig,
  parsePlacementRow,
  parseWorkerRecord,
  type Keys,
} from "./keys.js";

type RedisKV = createPinboard.RedisKV;

const DEFAULT_SWEEP_INTERVAL_MS = 5_000;
const DEFAULT_SWEEP_LIMIT = 100;
const DEFAULT_BASE_DELAY_MS = 1_000;
const DEFAULT_MAX_DELAY_MS = 60_000;
const DEFAULT_PARK_AFTER_FAILURES = 10;
const DEFAULT_WAKE_TIMEOUT_MS = 15_000;
const DEFAULT_WAKE_LIMIT = 25;
const DEFAULT_WAKE_CONCURRENCY = 4;
const DEFAULT_PARKED_TTL_MS = 3_600_000;

// Wipe-resync placeholder: a sessionId no worker ever holds, so the row reads
// as dead-owner and re-places through the normal machinery.
const OWNERLESS_SESSION = "-";

export type DurabilitySweep = {
  sweepOnce(): Promise<void>;
  start(): void;
  close(): Promise<void>;
};

export namespace createDurabilitySweep {
  export type Deps = {
    redis: RedisKV;
    keys: Keys;
    resolve(world: World, ns: string, id: string): Promise<Resolution>;
    worlds(): Promise<WorldNamespace[]>;
    leaseTtlMs: number;
    now?: (() => number) | undefined;
    fetch?: Transport.FetchLike | undefined;
    outlier?: createOutlierDetector.Instance | undefined;
  };
}

/**
 * The `{0}:sweeper` lease holder's loop: drains the overwritten lists
 * (`_wake` each entry; a 200 proves the instance alive and drops it),
 * re-claims dead-owner placements through the resolve script (which feeds the
 * lists), GCs dead worker records / ns_workers entries / lapsed drain
 * configs, and pages the DurabilityStore's marked rows into placeholder
 * rows so a wiped Redis re-places every in-flight activity. Without a store
 * there is nothing to resurrect: the overwritten lists are deleted, dead-owner
 * rows are evicted, and only the GC remains.
 */
export const createDurabilitySweep = (
  deps: createDurabilitySweep.Deps,
  options: createPinboard.DurabilityOptions,
  lease?: createLease.Instance,
): DurabilitySweep => {
  const { redis, keys } = deps;
  const now = deps.now ?? (() => Date.now());
  const random = options.random ?? Math.random;
  const fetchImpl = deps.fetch ?? ((request) => fetch(request));
  const { store } = options;
  const intervalMs = options.sweepIntervalMs ?? DEFAULT_SWEEP_INTERVAL_MS;
  const limit = options.sweepLimit ?? DEFAULT_SWEEP_LIMIT;
  const wakeLimit = options.wakeLimit ?? DEFAULT_WAKE_LIMIT;
  const wakeConcurrency = options.wakeConcurrency ?? DEFAULT_WAKE_CONCURRENCY;
  const wakeTimeoutMs = options.wakeTimeoutMs ?? DEFAULT_WAKE_TIMEOUT_MS;
  const parkedTtlMs = options.parkedTtlMs ?? DEFAULT_PARKED_TTL_MS;
  const baseDelayMs = options.backoff?.baseDelayMs ?? DEFAULT_BASE_DELAY_MS;
  const maxDelayMs = options.backoff?.maxDelayMs ?? DEFAULT_MAX_DELAY_MS;
  const parkAfterFailures =
    options.backoff?.parkAfterFailures ?? DEFAULT_PARK_AFTER_FAILURES;
  for (const [name, value] of [
    ["sweepIntervalMs", intervalMs],
    ["wakeTimeoutMs", wakeTimeoutMs],
    ["parkedTtlMs", parkedTtlMs],
    ["baseDelayMs", baseDelayMs],
    ["maxDelayMs", maxDelayMs],
  ] as const) {
    if (!Number.isFinite(value) || value <= 0) {
      throw new Error(`${name} must be a positive finite number`);
    }
  }
  for (const [name, value] of [
    ["sweepLimit", limit],
    ["wakeLimit", wakeLimit],
    ["wakeConcurrency", wakeConcurrency],
    ["parkAfterFailures", parkAfterFailures],
  ] as const) {
    if (!Number.isInteger(value) || value < 1) {
      throw new Error(`${name} must be a positive integer`);
    }
  }

  const buckets = Array.from({ length: keys.shards }, (_, s) => s);

  // A displaced (hostname, ns) binding quarantines new placements for one
  // lease; quarantined ids are left in place and revisited next sweep.
  const bindingQuarantined = async (
    bucket: number,
    hostname: string,
    ns: string,
  ): Promise<boolean> => {
    const raw = await redis.hget(keys.bindepochs(bucket), `h:${hostname}${ns}`);
    const until =
      raw === null ? undefined : parseBindingEpoch(raw).quarantineUntil;
    return until !== undefined && until > now();
  };

  const drainOverwritten = createWakePass({
    redis,
    keys,
    resolve: (world, ns, id) => deps.resolve(world, ns, id),
    fetch: fetchImpl,
    outlier: deps.outlier,
    now,
    random,
    bindingQuarantined,
    hasStore: store !== undefined,
    wakeTimeoutMs,
    wakeLimit,
    wakeConcurrency,
    parkedTtlMs,
    baseDelayMs,
    maxDelayMs,
    parkAfterFailures,
  });

  const sessionState = async (
    bucket: number,
    sessionId: string,
  ): Promise<{ live: boolean; hostname: string | null }> => {
    const raw = await redis.hget(keys.workers(bucket), sessionId);
    const record = raw === null ? null : parseWorkerRecord(raw);
    return {
      live: record !== null && record.expiry > now(),
      hostname: record?.hostname ?? null,
    };
  };

  // With a store, dead-owner rows re-claim through the resolve script, which
  // appends them to the overwritten list; refusals (quarantine, no backends,
  // no live workers under the target hostname) leave the row as the marker
  // for the next pass. Without one they are evicted via a compare-and-delete
  // on the dead owner. Resurrection targets the row's recorded hostname (the
  // dead owner's, or the placeholder's), falling back to a live hostname of
  // the env.
  const reclaimDead = async (): Promise<void> => {
    let reclaims = 0;
    const byEnvNs = new Map<
      string,
      { env: string; ns: string; hostnames: string[] }
    >();
    for (const world of await deps.worlds()) {
      const key = `${world.env}\n${world.ns}`;
      const group = byEnvNs.get(key) ?? {
        env: world.env,
        ns: world.ns,
        hostnames: [],
      };
      if (group.hostnames.length === 0) byEnvNs.set(key, group);
      group.hostnames.push(world.hostname);
    }
    for (const { env, ns, hostnames } of byEnvNs.values()) {
      for (const bucket of buckets) {
        const key = keys.placements(bucket, env, ns);
        let cursor = "0";
        do {
          const scan = await redis.hscan(key, cursor, 100);
          cursor = scan.cursor;
          for (const [id, raw] of scan.entries) {
            const row = parsePlacementRow(raw);
            const owner = await sessionState(bucket, row.sessionId);
            if (owner.live) continue;
            if (reclaims >= limit) return;
            reclaims += 1;
            try {
              if (store === undefined) {
                await releasePlacements(redis, keys, env, ns, row.sessionId, [
                  { id, epoch: row.epoch },
                ]);
              } else {
                const hostname =
                  row.hostname ?? owner.hostname ?? hostnames[0]!;
                if (await bindingQuarantined(bucket, hostname, ns)) continue;
                await deps.resolve({ env, hostname, takeover: "deny" }, ns, id);
              }
            } catch (err) {
              logger.warn(
                {
                  operation: "durability",
                  namespace: ns,
                  env,
                  id,
                  err: serializeError(err),
                },
                "dead-owner sweep failed",
              );
            }
          }
        } while (cursor !== "0");
      }
    }
  };

  const gc = async (): Promise<void> => {
    const clientIds = new Set<string>();
    for (const bucket of buckets) {
      const seen = (await runScript(
        redis,
        gcScript,
        [
          keys.workers(bucket),
          keys.nsWorkers(bucket),
          keys.bindings(bucket),
          keys.bindepochs(bucket),
        ],
        [String(now()), String(deps.leaseTtlMs)],
      )) as string[];
      for (const clientId of seen) clientIds.add(clientId);
    }
    for (const clientId of clientIds) {
      const raw = await redis.get(keys.client(clientId));
      const config = raw === null ? null : parseClientConfig(raw);
      if (
        typeof config?.drainDeadline === "number" &&
        config.drainDeadline + deps.leaseTtlMs < now()
      ) {
        await redis.del(keys.client(clientId));
      }
    }
  };

  // Steady state every activity has a row, so the HSETNX no-ops; after a
  // Redis wipe the placeholders re-enter the dead-owner machinery, riding the
  // boot_time quarantine window.
  // A bad row skips alone; a redis failure leaves the cursor so the page
  // (idempotent HSETNX) re-applies next pass.
  let resyncCursor: string | null = null;
  const resync = async (): Promise<void> => {
    if (store === undefined) return;
    const page = await store.listMarked(resyncCursor, limit);
    for (const { env, hostname, ns, id } of page.entries) {
      let normalized: string;
      try {
        normalized = validateNamespace(ns);
      } catch (err) {
        logger.warn(
          {
            operation: "durability",
            namespace: ns,
            id,
            err: serializeError(err),
          },
          "skipping malformed marked-index row",
        );
        continue;
      }
      await redis.hsetnx(
        keys.placements(keys.bucket(id), env, normalized),
        id,
        JSON.stringify({ sessionId: OWNERLESS_SESSION, hostname }),
      );
    }
    resyncCursor = page.cursor;
  };

  const sweepOnce = async (): Promise<void> => {
    await drainOverwritten();
    await reclaimDead();
    await gc();
    await resync();
  };

  let timer: NodeJS.Timeout | null = null;
  let inflight: Promise<void> | null = null;
  let closed = false;
  const schedule = (): void => {
    if (closed) return;
    timer = setTimeout(() => {
      inflight = (lease?.held() ?? Promise.resolve(true))
        .then((held) => (held ? sweepOnce() : undefined))
        .catch((err) => {
          logger.error(
            { operation: "durability", err: serializeError(err) },
            "durability sweep failed",
          );
        })
        .finally(() => {
          inflight = null;
          schedule();
        });
    }, intervalMs);
    timer.unref?.();
  };

  return {
    sweepOnce,
    start: () => {
      lease?.start();
      schedule();
    },
    close: async () => {
      closed = true;
      if (timer !== null) clearTimeout(timer);
      await lease?.close();
      await inflight;
    },
  };
};