import { sha256Hex } from "./hash.js";
import {
  arrayOf,
  boolean,
  finiteNumber,
  literal,
  nonEmpty,
  object,
  optional,
  recordOf,
  string,
} from "./schema.js";

export const RedisKVMarker: unique symbol = Symbol.for("pinboard.RedisKV");

/** Heartbeat lease granted to workers: the worker record's expiry horizon and
 * the re-place deadline, surfaced in acks as expiresInMs. */
export const DEFAULT_LEASE_TTL_MS = 15_000;

export const DEFAULT_KEY_PREFIX = "pinboard:";

/** Hostname and env in single-hostname mode; never a valid registered value. */
export const SINGLE_WORLD = "*";

export type Takeover = "wind_down" | "deny";

export interface WorkerRecord {
  sessionId: string;
  clientId: string;
  /** Passport account that registered this client, when registered through HTTP. */
  account?: string;
  advertiseUrl: string;
  secret: string;
  namespaces: string[];
  hostname: string;
  /** Per-namespace data identity: every registered namespace has an env. */
  envs: Record<string, string>;
  takeover: Takeover;
  headroom: number;
  draining: boolean;
  /** Epoch ms the reverse challenge probe started failing; present = degraded:
   * leased and routable, no new placements. */
  degradedSince?: number;
  /** Epoch ms liveness horizon; the record counts as dead past it. */
  expiry: number;
}

export interface PlacementRow {
  sessionId: string;
  /** Placement epoch, decimal string: routing generation, release CAS value,
   * and lease fencing token. Absent only on ownerless placeholder rows. */
  epoch?: string;
  /** Epoch ms wind-down deadline; the row counts as free past it. */
  expiry?: number;
  /** Resurrection target on ownerless placeholder rows. */
  hostname?: string;
}

export interface ClientConfig {
  /** null = cordon (drain without deadline); number = forced release at T. */
  drainDeadline: number | null;
}

interface ClientMetadata {
  account: string;
  hostname: string;
  namespaces: string[];
  envs: Record<string, string>;
}

export interface Keys {
  shards: number;
  bucket(id: string): number;
  /** Per-bucket cold-start stamp: SET NX inside the resolve script. */
  bootTime(bucket: number): string;
  /** Re-claimed-with-a-dead-owner ids awaiting a proven-alive 200. */
  overwritten(bucket: number): string;
  /** Bucket-wide high-water mark of every minted placement epoch. */
  epochFloor(bucket: number): string;
  /** Serving license: JSON {schema, shards, hostnameAware} of the fleet allowed to serve. */
  config: string;
  /** Cluster members: nodeId -> JSON {c: config, expiry}; joiners must match every live member. */
  members: string;
  /** Single-sweeper lease: the sweeping node's random token. */
  sweeper: string;
  placements(bucket: number, env: string, ns: string): string;
  workers(bucket: number): string;
  nsWorkers(bucket: number): string;
  /** World agreement: field `h:${hostname}${ns}` -> env, `e:${env}` -> takeover. */
  bindings(bucket: number): string;
  /** Binding epochs sidecar: fields mirror `bindings`, value JSON
   * {epoch, quarantineUntil?}; a missing field is epoch 0. */
  bindepochs(bucket: number): string;
  client(clientId: string): string;
  /** Durable account and scope for a client, independent of drain expiry. */
  clientMeta(clientId: string): string;
  /** Pending control messages, delivered in heartbeat acks and trimmed once
   * the worker acks their msgIds. */
  controlQueue(sessionId: string): string;
}

export const createKeys = (
  prefix: string = DEFAULT_KEY_PREFIX,
  shards = 1,
): Keys => {
  if (prefix.length === 0 || /\s/.test(prefix)) {
    throw new Error(`invalid key prefix: ${JSON.stringify(prefix)}`);
  }
  if (!Number.isInteger(shards) || shards < 1) {
    throw new Error(`shards must be a positive integer: ${shards}`);
  }
  return {
    shards,
    bucket: (id) => parseInt(sha256Hex(id).slice(0, 8), 16) % shards,
    bootTime: (bucket) => `${prefix}{${bucket}}:boot_time`,
    overwritten: (bucket) => `${prefix}{${bucket}}:overwritten`,
    epochFloor: (bucket) => `${prefix}{${bucket}}:epoch_floor`,
    config: `${prefix}{0}:config`,
    members: `${prefix}{0}:members`,
    sweeper: `${prefix}{0}:sweeper`,
    placements: (bucket, env, ns) =>
      `${prefix}{${bucket}}:${env}:${ns}:placements`,
    workers: (bucket) => `${prefix}{${bucket}}:workers`,
    nsWorkers: (bucket) => `${prefix}{${bucket}}:ns_workers`,
    bindings: (bucket) => `${prefix}{${bucket}}:bindings`,
    bindepochs: (bucket) => `${prefix}{${bucket}}:bindepochs`,
    client: (clientId) => `${prefix}{0}:client:${clientId}`,
    clientMeta: (clientId) => `${prefix}{0}:client_meta:${clientId}`,
    controlQueue: (sessionId) => `${prefix}ctl:${sessionId}`,
  };
};

// Hostnames hold no "/", namespaces start with one — the field is unambiguous.
export const nsWorkersField = (hostname: string, ns: string): string =>
  `${hostname}${ns}`;

export const parseNsWorkersField = (
  field: string,
): { hostname: string; ns: string } | null => {
  const sep = field.indexOf("/");
  if (sep === -1) return null;
  return { hostname: field.slice(0, sep), ns: field.slice(sep) };
};

// These records are pinboard's own durable writes: corruption is a bug,
// never coerced to defaults.
const parseJson = <T>(
  raw: string,
  schema: (value: unknown) => value is T,
  what: string,
): T => {
  let parsed: unknown;
  try {
    parsed = JSON.parse(raw);
  } catch {
    throw new Error(`malformed ${what}: ${raw}`);
  }
  if (!schema(parsed)) throw new Error(`malformed ${what}: ${raw}`);
  return parsed;
};

const workerRecordSchema = object<WorkerRecord>({
  sessionId: nonEmpty(string),
  clientId: nonEmpty(string),
  account: optional(nonEmpty(string)),
  advertiseUrl: string,
  secret: nonEmpty(string),
  namespaces: nonEmpty(arrayOf(string)),
  hostname: nonEmpty(string),
  envs: recordOf(nonEmpty(string)),
  takeover: literal("wind_down", "deny"),
  headroom: finiteNumber,
  draining: boolean,
  degradedSince: optional(finiteNumber),
  expiry: finiteNumber,
});

export const parseWorkerRecord = (raw: string): WorkerRecord =>
  parseJson(raw, workerRecordSchema, "worker record");

const placementRowSchema = object<PlacementRow>({
  sessionId: nonEmpty(string),
  epoch: optional(nonEmpty(string)),
  expiry: optional(finiteNumber),
  hostname: optional(nonEmpty(string)),
});

export const parsePlacementRow = (raw: string): PlacementRow =>
  parseJson(raw, placementRowSchema, "placement row");

const clientConfigSchema = object<ClientConfig>({
  drainDeadline: (value): value is number | null =>
    value === null || finiteNumber(value),
});

export const parseClientConfig = (raw: string): ClientConfig =>
  parseJson(raw, clientConfigSchema, "client config");

const clientMetadataSchema = object<ClientMetadata>({
  account: nonEmpty(string),
  hostname: nonEmpty(string),
  namespaces: nonEmpty(arrayOf(string)),
  envs: recordOf(nonEmpty(string)),
});

export const parseClientMetadata = (raw: string): ClientMetadata =>
  parseJson(raw, clientMetadataSchema, "client metadata");

export interface BindingEpoch {
  epoch: number;
  /** Epoch ms: new placements in the displaced (hostname, ns) refuse until then. */
  quarantineUntil?: number;
}

const bindingEpochSchema = object<BindingEpoch>({
  epoch: finiteNumber,
  quarantineUntil: optional(finiteNumber),
});

export const parseBindingEpoch = (raw: string): BindingEpoch =>
  parseJson(raw, bindingEpochSchema, "binding epoch");

// Field encoding in the overwritten hash; env, hostname, and ids never
// contain "\n".
export const overwrittenField = (
  env: string,
  hostname: string,
  ns: string,
  id: string,
): string => `${env}\n${hostname}\n${ns}\n${id}`;

export const parseOverwrittenField = (
  field: string,
): { env: string; hostname: string; ns: string; id: string } | null => {
  const parts = field.split("\n");
  if (parts.length !== 4 || parts.some((part) => part === "")) return null;
  const [env, hostname, ns, id] = parts as [string, string, string, string];
  return { env, hostname, ns, id };
};

export const parseSessionIds = (raw: string): string[] =>
  parseJson(raw, arrayOf(string), "ns_workers session list");