// Lease and membership scripts; see twins.ts for the Lua/JS twin contract.

import { parseObj, type Script } from "./twins.js";

/** Fleet epoch of a stored value: absent field = 0, non-JSON = null.
 * Arrays count as epoch 0, matching Lua's cjson table decoding. */
export const epochOf = (raw: string | null): number | null => {
  if (raw === null) return null;
  let value: unknown;
  try {
    value = JSON.parse(raw);
  } catch {
    return null;
  }
  if (typeof value !== "object" || value === null) return null;
  const epoch = (value as Record<string, unknown>)["epoch"];
  return typeof epoch === "number" ? epoch : 0;
};

/**
 * KEYS: lease
 * ARGV: value, ttlMs
 * Claims the lease when absent (SET + TTL), extends the TTL when the stored
 * value matches. A mismatched value is overwritten only when both values are
 * JSON objects and the incoming epoch (absent = 0) is higher.
 * Returns 'claimed' | 'serving' | 'mismatch'.
 */
export const leaseScript: Script.Def = {
  lua: `
local function epochOf(raw)
  local ok, value = pcall(cjson.decode, raw)
  if not ok or type(value) ~= 'table' then return nil end
  if type(value.epoch) == 'number' then return value.epoch end
  return 0
end
local raw = redis.call('GET', KEYS[1])
if not raw then
  redis.call('SET', KEYS[1], ARGV[1], 'PX', ARGV[2])
  return 'claimed'
end
if raw == ARGV[1] then
  redis.call('PEXPIRE', KEYS[1], ARGV[2])
  return 'serving'
end
local stored, ours = epochOf(raw), epochOf(ARGV[1])
if stored and ours and ours > stored then
  redis.call('SET', KEYS[1], ARGV[1], 'PX', ARGV[2])
  return 'claimed'
end
return 'mismatch'
`,
  js: (kv, keys, argv) => {
    const leaseKey = keys[0]!;
    const [value, ttlRaw] = [argv[0]!, argv[1]!];
    const raw = kv.get(leaseKey);
    if (raw === null) {
      kv.set(leaseKey, value, Number(ttlRaw));
      return "claimed";
    }
    if (raw === value) {
      kv.pexpire(leaseKey, Number(ttlRaw));
      return "serving";
    }
    const stored = epochOf(raw);
    const ours = epochOf(value);
    if (stored !== null && ours !== null && ours > stored) {
      kv.set(leaseKey, value, Number(ttlRaw));
      return "claimed";
    }
    return "mismatch";
  },
};

/**
 * KEYS: members
 * ARGV: nodeId, config, now(ms), ttlMs, op('join'/'leave')
 * Join: prunes expired members, refuses while any live member's config
 * differs — unless that member's config carries a lower epoch (absent = 0),
 * i.e. it is being displaced. Otherwise upserts self with expiry = now + ttl.
 * Leave: HDELs self.
 * Returns ['ok'] | ['incompatible', otherConfig].
 */
export const membershipScript: Script.Def = {
  lua: `
local function epochOf(raw)
  if type(raw) ~= 'string' then return nil end
  local ok, value = pcall(cjson.decode, raw)
  if not ok or type(value) ~= 'table' then return nil end
  if type(value.epoch) == 'number' then return value.epoch end
  return 0
end
local nodeId, config = ARGV[1], ARGV[2]
local now, ttl = tonumber(ARGV[3]), tonumber(ARGV[4])
if ARGV[5] == 'leave' then
  redis.call('HDEL', KEYS[1], nodeId)
  return { 'ok' }
end
local ours = epochOf(config)
local all = redis.call('HGETALL', KEYS[1])
for i = 1, #all, 2 do
  local ok, rec = pcall(cjson.decode, all[i + 1])
  if not ok or type(rec) ~= 'table' or type(rec.expiry) ~= 'number'
    or rec.expiry <= now then
    redis.call('HDEL', KEYS[1], all[i])
  elseif all[i] ~= nodeId and rec.c ~= config then
    local theirs = epochOf(rec.c)
    if not (theirs and ours and theirs < ours) then
      return { 'incompatible', tostring(rec.c) }
    end
  end
end
redis.call('HSET', KEYS[1], nodeId,
  cjson.encode({ c = config, expiry = now + ttl }))
return { 'ok' }
`,
  js: (kv, keys, argv) => {
    const membersKey = keys[0]!;
    const [nodeId, config] = [argv[0]!, argv[1]!];
    const now = Number(argv[2]!);
    const ttl = Number(argv[3]!);
    if (argv[4] === "leave") {
      kv.hdel(membersKey, nodeId);
      return ["ok"];
    }
    const ours = epochOf(config);
    for (const [member, raw] of Object.entries(kv.hgetall(membersKey))) {
      const rec = parseObj(raw);
      const expiry = rec?.["expiry"];
      if (typeof expiry !== "number" || expiry <= now) {
        kv.hdel(membersKey, member);
      } else if (member !== nodeId && rec!["c"] !== config) {
        const theirConfig = rec!["c"];
        const theirs =
          typeof theirConfig === "string" ? epochOf(theirConfig) : null;
        if (theirs === null || ours === null || theirs >= ours) {
          return ["incompatible", String(theirConfig)];
        }
      }
    }
    kv.hset(
      membersKey,
      nodeId,
      JSON.stringify({ c: config, expiry: now + ttl }),
    );
    return ["ok"];
  },
};

/**
 * KEYS: lease
 * ARGV: value
 * Compare-and-delete: DELs the lease only while the stored value matches.
 * Returns 1 when deleted, 0 otherwise.
 */
export const leaseReleaseScript: Script.Def = {
  lua: `
if redis.call('GET', KEYS[1]) == ARGV[1] then
  return redis.call('DEL', KEYS[1])
end
return 0
`,
  js: (kv, keys, argv) =>
    kv.get(keys[0]!) === argv[0]! ? kv.del(keys[0]!) : 0,
};