import { sha1Hex } from "./hash.js";
import type { createPinboard } from "./server.js";
import {
  parseWorkerRecord,
  type Keys,
  type Takeover,
  type WorkerRecord,
} from "./keys.js";
import {
  releaseScript,
  resolveScript,
  windDownScript,
} from "./scripts/placement.js";
import { nsWorkersScript } from "./scripts/registry.js";
import type { Script } from "./scripts/twins.js";

type RedisKV = createPinboard.RedisKV;

export type Resolution =
  | {
      kind: "forward";
      worker: WorkerRecord;
      /** Placement epoch (decimal string): rides every forward as the
       * Pinboard-Epoch header and quotes releases. */
      epoch: string;
      listed: boolean;
      /** True when this call created the placement row (nothing activated on
       * the worker yet). */
      fresh: boolean;
      carriedExpiry?: number;
    }
  | { kind: "unknown-ns" }
  | { kind: "no-backends" }
  | { kind: "quarantine"; retryAfterMs: number }
  | {
      kind: "takeover-pending";
      owner: WorkerRecord;
      /** True when this call issued the wind-down fence. */
      initiated: boolean;
      fenceExpiry: number;
    }
  | { kind: "takeover-denied" };

/** The request's placement world: hostname (compute) and env (data);
 * SINGLE_WORLD ("*") for both in single-hostname mode. */
export type World = { hostname: string; env: string; takeover: Takeover };

const shaCache = new Map<string, string>();

export const runScript = async (
  redis: RedisKV,
  script: Script.Def,
  keys: string[],
  argv: string[],
): Promise<unknown> => {
  let sha = shaCache.get(script.lua);
  if (sha === undefined) {
    sha = sha1Hex(script.lua);
    shaCache.set(script.lua, sha);
  }
  try {
    return await redis.evalsha(sha, keys, argv);
  } catch (err) {
    if (!String(err).includes("NOSCRIPT")) throw err;
    await redis.scriptLoad(script.lua);
    return redis.evalsha(sha, keys, argv);
  }
};

/** Sets (or clears, atMs = null) the wind-down expiry on each placed id in
 * one bucket's placements hash; returns the (id, owning sessionId) pairs. */
export const windDownPlacements = async (
  redis: RedisKV,
  placementsKey: string,
  ids: string[],
  atMs: number | null,
): Promise<{ id: string; sessionId: string }[]> => {
  const result = await runScript(
    redis,
    windDownScript,
    [placementsKey],
    [atMs === null ? "" : String(atMs), ...ids],
  );
  if (!Array.isArray(result) || result.length % 2 !== 0) {
    throw new Error(`malformed wind_down result: ${JSON.stringify(result)}`);
  }
  const owners: { id: string; sessionId: string }[] = [];
  for (let i = 0; i < result.length; i += 2) {
    owners.push({ id: String(result[i]), sessionId: String(result[i + 1]) });
  }
  return owners;
};

/** Single-id wind_down; returns the owning sessionId, or null when unplaced. */
export const windDownPlacement = async (
  redis: RedisKV,
  keys: Keys,
  env: string,
  ns: string,
  id: string,
  atMs: number | null,
): Promise<string | null> => {
  const owners = await windDownPlacements(
    redis,
    keys.placements(keys.bucket(id), env, ns),
    [id],
    atMs,
  );
  return owners[0]?.sessionId ?? null;
};

/** Compare-and-delete on (sessionId, epoch); an entry without an epoch quotes
 * an epoch-less row. Returns the number of rows freed. */
export const releasePlacements = async (
  redis: RedisKV,
  keys: Keys,
  env: string,
  ns: string,
  sessionId: string,
  entries: { id: string; epoch?: string | undefined }[],
): Promise<number> => {
  const byBucket = new Map<number, string[]>();
  for (const { id, epoch } of entries) {
    const bucket = keys.bucket(id);
    const list = byBucket.get(bucket) ?? [];
    if (list.length === 0) byBucket.set(bucket, list);
    list.push(id, epoch ?? "");
  }
  let released = 0;
  for (const [bucket, pairs] of byBucket) {
    const count = await runScript(
      redis,
      releaseScript,
      [keys.placements(bucket, env, ns)],
      [sessionId, ...pairs],
    );
    released += Number(count);
  }
  return released;
};

/** Atomic read-modify-write of one ns_workers field, pruning dead sessions. */
export const updateNsWorkers = async (
  redis: RedisKV,
  keys: Keys,
  bucket: number,
  field: string,
  sessionId: string,
  op: "add" | "remove",
  now: number,
): Promise<void> => {
  await runScript(
    redis,
    nsWorkersScript,
    [keys.nsWorkers(bucket), keys.workers(bucket)],
    [field, sessionId, op, String(now)],
  );
};

export namespace createPlacer {
  export type Options = {
    leaseTtlMs: number;
    /** Bounded-load factor c: a preference is honored only while the
     * preferred worker's headroom stays >= mean(headrooms) / c. Default 1.25. */
    balanceFactor?: number | undefined;
    now?: (() => number) | undefined;
    random?: (() => number) | undefined;
    /** Outlier-ejected sessionIds, withheld from new placements while any
     * other candidate remains. */
    ejected?: (() => readonly string[]) | undefined;
  };

  export type Preference = {
    /** Exact co-location target; wins over affinityKey. */
    preferredSessionId?: string | undefined;
    /** Rendezvous-hashed over live candidates to derive a preferred worker. */
    affinityKey?: string | undefined;
  };

  export type Instance = {
    resolve(
      world: World,
      ns: string,
      id: string,
      preference?: Preference,
      /** Never place on this session (in-band activation NACK re-place). */
      exclude?: string,
    ): Promise<Resolution>;
  };
}

/**
 * Placement path: one atomic read-or-allocate script per resolve. The script
 * stamps the bucket's boot_time (SET NX, no TTL) and refuses new placements
 * while now < boot_time + lease window, so a wiped bucket self-quarantines
 * for one lease window while surviving buckets keep serving.
 */
export const createPlacer = (
  redis: RedisKV,
  keys: Keys,
  options: createPlacer.Options,
): createPlacer.Instance => {
  const now = options.now ?? (() => Date.now());
  const random = options.random ?? Math.random;
  const balanceFactor = options.balanceFactor ?? 1.25;
  if (!(balanceFactor > 1)) {
    throw new Error(
      `balanceFactor must be > 1: ${String(options.balanceFactor)}`,
    );
  }

  const ejectedArg = (): string => {
    const ejected = options.ejected?.() ?? [];
    return ejected.length === 0 ? "" : JSON.stringify(ejected);
  };

  const resolve = async (
    world: World,
    ns: string,
    id: string,
    preference?: createPlacer.Preference,
    exclude?: string,
  ): Promise<Resolution> => {
    const bucket = keys.bucket(id);
    const result = await runScript(
      redis,
      resolveScript,
      [
        keys.placements(bucket, world.env, ns),
        keys.workers(bucket),
        keys.nsWorkers(bucket),
        keys.bootTime(bucket),
        keys.overwritten(bucket),
        keys.epochFloor(bucket),
      ],
      [
        id,
        String(now()),
        String(random()),
        String(random()),
        String(options.leaseTtlMs),
        ns,
        preference?.preferredSessionId ?? "",
        preference?.affinityKey ?? "",
        String(balanceFactor),
        world.hostname,
        world.env,
        world.takeover,
        exclude ?? "",
        ejectedArg(),
      ],
    );
    if (!Array.isArray(result) || result.length !== 6) {
      throw new Error(`malformed resolve result: ${JSON.stringify(result)}`);
    }
    const [status, workerRaw, aux, carriedRaw, epoch, freshRaw] = result as [
      string,
      string,
      string,
      string,
      string,
      string,
    ];
    if (status === "forward") {
      const worker = parseWorkerRecord(workerRaw);
      if (epoch === "") {
        throw new Error(`placement row has no epoch: ${ns}/${id}`);
      }
      return {
        kind: "forward",
        worker,
        epoch,
        listed: aux === "1",
        fresh: freshRaw === "1",
        ...(carriedRaw !== "" && { carriedExpiry: Number(carriedRaw) }),
      };
    }
    if (status === "quarantine") {
      return { kind: "quarantine", retryAfterMs: Math.max(1, Number(aux)) };
    }
    if (status === "takeover-pending") {
      const owner = parseWorkerRecord(workerRaw);
      return {
        kind: "takeover-pending",
        owner,
        initiated: aux === "1",
        fenceExpiry: Number(carriedRaw),
      };
    }
    if (status === "takeover-denied") return { kind: "takeover-denied" };
    if (status === "unknown-ns") return { kind: "unknown-ns" };
    if (status === "no-backends") return { kind: "no-backends" };
    throw new Error(`unknown resolve status: ${status}`);
  };

  return { resolve };
};