openharness

/**
 * Cross-instance data bus over Redis pub/sub.
 *
 * Behind a load balancer a given agent's web socket(s) and its single manager socket can land on
 * DIFFERENT backend instances. Redis bridges that gap (this is the ONLY cross-instance state — see
 * docs/plans/2026-07-09_invert-ws-transport-backend-hub-redis.md):
 *
 *  - up:{machineId}   — node→client events. Published by the backend holding the manager socket (B_m);
 *                     delivered by every backend with ≥1 local client for the agent (≤ M subscribers).
 *  - down:{machineId} — client→node messages. Published by any client-holding backend; delivered by the
 *                     single backend holding the manager socket (exactly 1 subscriber, B_m).
 *  - mgr:{managerId} — control/provisioning commands to a specific manager (Phase 3), delivered by the
 *                     backend holding that manager's socket.
 *
 * ioredis requires a connection in subscribe mode to be dedicated (it can't run other commands), so
 * we keep one `sub` connection for SUBSCRIBE and one `pub` connection for PUBLISH/keys.
 */
import { Redis } from 'ioredis'
import { env } from '../config/env.js'
import { logger } from '../utils/logger.js'
import type { Frame, UpBusMsg, DownBusMsg } from './tunnel.js'

// A connection in subscriber mode must never fail its (re)SUBSCRIBE commands, so the sub side keeps
// ioredis' "retry forever" setting.
const subOpts = { maxRetriesPerRequest: null as null, lazyConnect: false }
// The pub/KV side is the opposite: a command that cannot reach Redis must FAIL FAST, not wait. With the
// default offline queue, every `publish` issued during a Redis outage (every up-frame from every manager
// socket, every terminal packet) was stringified and parked in this process' heap until reconnect, and
// every `await pub.set(...)` in an attach path hung — pinning its closure — for the whole outage.
const pubOpts = {
  maxRetriesPerRequest: 1,
  enableOfflineQueue: false,
  commandTimeout: 5_000,
  lazyConnect: false,
}

export const pub = new Redis(env.REDIS_URL, pubOpts)
const sub = new Redis(env.REDIS_URL, subOpts)
const terminalSub = new Redis(env.REDIS_URL, subOpts)
// App-proxy replies (`appup:{streamId}`) can be multi-MiB base64 bodies. Redis cuts a subscriber that
// exceeds `client-output-buffer-limit pubsub`; on the shared connection that took every chat / presence
// subscription of this instance down with it (ioredis re-subscribes, but everything in between is lost).
// Isolated here, a heavy app stream can only hurt app streams.
const appSub = new Redis(env.REDIS_URL, subOpts)

/** Every fan-out publish is best-effort: Redis unreachable → dropped with a warning, 0 subscribers.
 *  Callers already read 0 as "nobody holds this machine right now", which is exactly the truth. */
async function safePublish(channel: string, payload: string | Buffer): Promise<number> {
  try {
    return await pub.publish(channel, payload)
  } catch (err) {
    logger.warn('[bus] publish dropped', { channel, error: err instanceof Error ? err.message : String(err) })
    return 0
  }
}

pub.on('error', (err: Error) => logger.error('[bus] pub redis error', err))
sub.on('error', (err: Error) => logger.error('[bus] sub redis error', err))
terminalSub.on('error', (err: Error) => logger.error('[bus] terminal sub redis error', err))
appSub.on('error', (err: Error) => logger.error('[bus] app sub redis error', err))

export async function closeBus(): Promise<void> {
  await Promise.allSettled([pub.quit(), sub.quit(), terminalSub.quit(), appSub.quit()])
}

type Cb = (msg: unknown) => void

/** One refcounted JSON subscriber registry per subscribe-mode connection. */
interface JsonSubscriber {
  conn: Redis
  channelSubs: Map<string, Set<Cb>>
  channelSubscribePromises: Map<string, Promise<void>>
}

function createJsonSubscriber(conn: Redis, label: string): JsonSubscriber {
  const registry: JsonSubscriber = { conn, channelSubs: new Map(), channelSubscribePromises: new Map() }
  conn.on('message', (channel: string, payload: string) => {
    const set = registry.channelSubs.get(channel)
    if (!set || set.size === 0) return
    let msg: unknown
    try {
      msg = JSON.parse(payload)
    } catch (err) {
      logger.error(`[bus] bad JSON on channel (${label})`, err, { channel })
      return
    }
    for (const cb of set) {
      try {
        cb(msg)
      } catch (err) {
        logger.error(`[bus] subscriber callback threw (${label})`, err, { channel })
      }
    }
  })
  return registry
}

const mainSubscriber = createJsonSubscriber(sub, 'main')
const appSubscriber = createJsonSubscriber(appSub, 'app')

/** Refcounted subscribe: SUBSCRIBEs on the first callback for a channel, UNSUBSCRIBEs on the last. */
async function addSubOn(reg: JsonSubscriber, channel: string, cb: Cb): Promise<() => void> {
  const { conn, channelSubs, channelSubscribePromises } = reg
  let set = channelSubs.get(channel)
  let subscribePromise = channelSubscribePromises.get(channel)
  if (!set) {
    set = new Set()
    channelSubs.set(channel, set)
    subscribePromise = conn.subscribe(channel)
      .then(() => undefined)
      .catch((err) => {
        logger.error('[bus] subscribe failed', err, { channel })
        if (channelSubs.get(channel) === set) channelSubs.delete(channel)
        throw err
      })
      .finally(() => {
        if (channelSubscribePromises.get(channel) === subscribePromise) channelSubscribePromises.delete(channel)
      })
    channelSubscribePromises.set(channel, subscribePromise)
  }
  set.add(cb)
  if (subscribePromise) {
    try {
      await subscribePromise
    } catch (err) {
      const s = channelSubs.get(channel)
      if (s === set) {
        s.delete(cb)
        if (s.size === 0) channelSubs.delete(channel)
      }
      throw err
    }
  }
  return () => {
    const s = channelSubs.get(channel)
    if (!s) return
    s.delete(cb)
    if (s.size === 0) {
      channelSubs.delete(channel)
      conn.unsubscribe(channel).catch(() => { /* ignore */ })
    }
  }
}

const addSub = (channel: string, cb: Cb): Promise<() => void> => addSubOn(mainSubscriber, channel, cb)

const upChannel = (machineId: string): string => `up:${machineId}`
// Transition-only mirror of `up:`. The machine-list watchers (webWs.watchAgents / deviceWs.watchMachines)
// used to subscribe `up:{machineId}` for EVERY machine the user owns just to pick out node_status —
// which meant parsing and dispatching the full turn/delta stream of machines nobody was looking at.
const statusChannel = (machineId: string): string => `status:${machineId}`
const STATUS_TYPES = new Set(['node_status', 'machine_app_status'])
const downChannel = (machineId: string): string => `down:${machineId}`
const mgrChannel = (managerId: string): string => `mgr:${managerId}`
const appDownChannel = (machineId: string): string => `appdown:${machineId}`
const appUpChannel = (streamId: string): string => `appup:${streamId}`
const presenceKey = (machineId: string): string => `machine:${machineId}:mgr`
const terminalUpChannel = (machineId: string): string => `termup:${machineId}`
const terminalDownChannel = (machineId: string): string => `termdown:${machineId}`

type BinaryCb = (payload: Buffer) => void
const binarySubs = new Map<string, Set<BinaryCb>>()
terminalSub.on('messageBuffer', (channelRaw: Buffer, payload: Buffer) => {
  const callbacks = binarySubs.get(channelRaw.toString('utf8'))
  if (!callbacks) return
  for (const callback of callbacks) {
    try { callback(Buffer.from(payload)) } catch (err) { logger.error('[bus] binary subscriber callback threw', err) }
  }
})

async function addBinarySub(channel: string, callback: BinaryCb): Promise<() => void> {
  let callbacks = binarySubs.get(channel)
  if (!callbacks) {
    callbacks = new Set()
    binarySubs.set(channel, callbacks)
    await terminalSub.subscribe(channel)
  }
  callbacks.add(callback)
  return () => {
    const current = binarySubs.get(channel)
    if (!current) return
    current.delete(callback)
    if (current.size === 0) {
      binarySubs.delete(channel)
      void terminalSub.unsubscribe(channel)
    }
  }
}

export function subscribeTerminalUp(machineId: string, callback: BinaryCb): Promise<() => void> {
  return addBinarySub(terminalUpChannel(machineId), callback)
}
export function subscribeTerminalDown(machineId: string, callback: BinaryCb): Promise<() => void> {
  return addBinarySub(terminalDownChannel(machineId), callback)
}
export function publishTerminalUp(machineId: string, payload: Uint8Array): Promise<number> {
  return safePublish(terminalUpChannel(machineId), Buffer.from(payload))
}
export function publishTerminalDown(machineId: string, payload: Uint8Array): Promise<number> {
  return safePublish(terminalDownChannel(machineId), Buffer.from(payload))
}

// ── up / down data channels ──────────────────────────────────────────────────────────────────────

export function subscribeUp(machineId: string, cb: (msg: UpBusMsg) => void): Promise<() => void> {
  return addSub(upChannel(machineId), cb as Cb)
}

export function subscribeDown(machineId: string, cb: (msg: DownBusMsg) => void): Promise<() => void> {
  return addSub(downChannel(machineId), cb as Cb)
}

/** A grant change invalidates observer sockets across backend instances. No content on this channel. */
export function subscribeShareChanged(id: string, cb: () => void): Promise<() => void> {
  return addSub(`harness-share:${id}`, cb)
}
export function publishShareChanged(id: string): Promise<number> {
  return safePublish(`harness-share:${id}`, '{}')
}

export function publishUp(machineId: string, msg: UpBusMsg): Promise<number> {
  const type = (msg.frame as { type?: unknown } | undefined)?.type
  if (typeof type === 'string' && STATUS_TYPES.has(type)) void safePublish(statusChannel(machineId), JSON.stringify(msg.frame))
  return safePublish(upChannel(machineId), JSON.stringify(msg))
}

/** node_status / machine_app_status frames only (the bare frame, not the UpBusMsg envelope). */
export function subscribeStatus(machineId: string, cb: (frame: Frame) => void): Promise<() => void> {
  return addSub(statusChannel(machineId), cb as Cb)
}

export function publishDown(machineId: string, msg: DownBusMsg): Promise<number> {
  return safePublish(downChannel(machineId), JSON.stringify(msg))
}

// ── app-proxy tunnel channels ────────────────────────────────────────────────────────────────────
// App traffic is routed by MACHINEID (not managerId): an app belongs to a machine, and the machine's
// node manager socket has exactly ONE backend subscriber (B_m), so appdown:{machineId} has a single
// delivery path (no duplication under clustering) and rides the machine's sharded pool socket. Responses flow back
// on a per-STREAM channel so they reach the exact backend instance terminating the public request.

/** client→app frames (app_req, app_body, app_ws_open/msg/close, app_abort), delivered by B_m. */
export function subscribeAppDown(machineId: string, cb: (msg: unknown) => void): Promise<() => void> {
  return addSub(appDownChannel(machineId), cb)
}
export function publishAppDown(machineId: string, msg: unknown): Promise<number> {
  return safePublish(appDownChannel(machineId), JSON.stringify(msg))
}

/** app→client frames (app_res, app_res_body, app_ws_msg/close, app_abort) → the origin instance. */
export function subscribeAppUp(streamId: string, cb: (msg: unknown) => void): Promise<() => void> {
  return addSubOn(appSubscriber, appUpChannel(streamId), cb)
}
export function publishAppUp(streamId: string, msg: unknown): Promise<number> {
  return safePublish(appUpChannel(streamId), JSON.stringify(msg))
}

// ── manager command channel (Phase 3) ──────────────────────────────────────────────────────────

export function subscribeMgr(managerId: string, cb: (msg: unknown) => void): Promise<() => void> {
  return addSub(mgrChannel(managerId), cb)
}

export function publishMgr(managerId: string, msg: unknown): Promise<number> {
  return safePublish(mgrChannel(managerId), JSON.stringify(msg))
}

// Per-request reply channel for the provisioning RPC (backend → mgr:{managerId} → manager → reply).
const replyChannel = (requestId: string): string => `mgrreply:${requestId}`

export function subscribeReply(requestId: string, cb: (msg: unknown) => void): Promise<() => void> {
  return addSub(replyChannel(requestId), cb)
}

export function publishReply(requestId: string, msg: unknown): Promise<number> {
  return safePublish(replyChannel(requestId), JSON.stringify(msg))
}

// ── agent→manager presence (which manager currently owns an agent's node) ─────────────────────────

/** Set/refresh the presence key. Called on register + on each manager ping. */
export async function setAgentPresence(machineId: string, managerId: string, ttlSec = 30): Promise<void> {
  try {
    await pub.set(presenceKey(machineId), managerId, 'EX', ttlSec)
  } catch (err) {
    logger.error('[bus] setAgentPresence failed', err, { machineId })
  }
}

export async function getAgentPresence(machineId: string): Promise<string | null> {
  return (await presenceValue(machineId)) ?? null
}

/** The presence key's value, or undefined when the store could not be read (logged). */
async function presenceValue(machineId: string): Promise<string | null | undefined> {
  try {
    return await pub.get(presenceKey(machineId))
  } catch (err) {
    logger.error('[bus] presence read failed', err, { machineId })
    return undefined
  }
}

/**
 * Whether the machine's daemon is connected: true present, false absent — and null when the presence store
 * could not be READ, which [getAgentPresence] folds into "absent". A reader that turns absence into a
 * verdict (the machine list's `offline`, which a daemon labels models "seems offline" by) must not also
 * turn an outage of this store into one.
 */
export async function readAgentPresence(machineId: string): Promise<boolean | null> {
  const value = await presenceValue(machineId)
  return value === undefined ? null : !!value
}

/** One MGET for a whole machine list (the watchers seed N machines at once). Missing/failed → null. */
export async function getAgentPresenceMany(machineIds: string[]): Promise<Map<string, string | null>> {
  const out = new Map<string, string | null>()
  if (machineIds.length === 0) return out
  try {
    const vals = await pub.mget(...machineIds.map(presenceKey))
    machineIds.forEach((id, i) => out.set(id, vals[i] ?? null))
  } catch (err) {
    logger.error('[bus] getAgentPresenceMany failed', err, { count: machineIds.length })
    for (const id of machineIds) out.set(id, null)
  }
  return out
}

// A device's last-selected machine, so the backend can restore the active machine after a reconnect. A
// multi_machine device renders agents by attaching to every machine WITHOUT sending machine_select, so on a
// fresh connection `activeMachineId` would otherwise stay null and voice/RPCs fail with "no machine selected".
const deviceLastMachineKey = (deviceId: string): string => `devlastmachine:${deviceId}`
export async function setDeviceLastMachine(deviceId: string, machineId: string, ttlSec = 30 * 24 * 3600): Promise<void> {
  try { await pub.set(deviceLastMachineKey(deviceId), machineId, 'EX', ttlSec) } catch (err) { logger.error('[bus] setDeviceLastMachine failed', err, { deviceId }) }
}
export async function getDeviceLastMachine(deviceId: string): Promise<string | null> {
  try { return await pub.get(deviceLastMachineKey(deviceId)) } catch { return null }
}

// ── provider-machine recap cache (the device's tile text) ─────────────────────────────────────────
// Redis rather than a process Map because the backend runs FOUR cluster workers: the worker that ran
// the turn is usually not the one the device's `agent_recent` lands on, so an in-heap cache would
// answer "no recap" most of the time and every worker restart would blank the tiles. Only the
// provider path writes here — every other machine kind persists its own recap node-side.
// A whole small array under one key, not LPUSH/LTRIM: `n` is capped at 5 by the schema, so the value
// is tiny, and every other value in this file is a plain JSON `set`. Two turns finishing on the SAME
// agent at once can lose an entry in the read-modify-write; that is no worse than `activeTasks`,
// which is already one-per-machine, and a lost tile self-heals on the next turn.
const machineRecapKey = (machineId: string, agentId: string): string => `machine:${machineId}:recap:${agentId}`
const MACHINE_RECAP_KEEP = 5

export async function pushMachineRecap<T>(machineId: string, agentId: string, entry: T, ttlSec = 30 * 24 * 3600): Promise<void> {
  if (!agentId) return
  try {
    const existing = await getMachineRecaps<T>(machineId, agentId, MACHINE_RECAP_KEEP)
    const next = [entry, ...existing].slice(0, MACHINE_RECAP_KEEP)
    await pub.set(machineRecapKey(machineId, agentId), JSON.stringify(next), 'EX', ttlSec)
  } catch (err) {
    // Best-effort: a recap that fails to cache costs a tile after a device reboot, never a turn.
    logger.error('[bus] pushMachineRecap failed', err, { machineId, agentId })
  }
}

export async function getMachineRecaps<T>(machineId: string, agentId: string, n: number): Promise<T[]> {
  if (!agentId) return []
  try {
    const raw = await pub.get(machineRecapKey(machineId, agentId))
    if (!raw) return []
    const parsed = JSON.parse(raw) as unknown
    return Array.isArray(parsed) ? (parsed.slice(0, n) as T[]) : []
  } catch {
    return []
  }
}

// ── Resumable voice upload: park a partial live-stream so it survives a WS drop + reconnect ──────────
// Streaming voice has no local device copy; a WS blip mid-recording used to lose the whole (up to ~10-min)
// ramble. On a drop the backend PARKS the accumulated audio here keyed by the STABLE deviceId (not the
// per-connection closure), so a reconnect — which may land on a DIFFERENT backend instance (prod is
// multi-instance, non-sticky) — can `resumeVoiceUpload` and continue appending. Grace TTL ~60s. Written
// once per DROP (not per chunk). meta is the commit marker (set AFTER the pcm), so a torn write reads as
// absent. See docs/plans resumable-voice-upload.
export interface ParkedVoiceMeta {
  uploadId: string
  mode: 'turn' | 'ask'
  route: boolean
  goal: boolean
  loop: boolean
  autonomy: 'plan' | 'auto'
  agentId: string | null
  sessionId: string | null
  requestId: string | null
  sr: number
  lang: string
  size: number
  machineId: string | null
}
const voiceUploadPcmKey = (deviceId: string): string => `voiceupload:${deviceId}:pcm`
const voiceUploadMetaKey = (deviceId: string): string => `voiceupload:${deviceId}:meta`
export async function parkVoiceUpload(deviceId: string, meta: ParkedVoiceMeta, pcm: Buffer, ttlSec = 60): Promise<void> {
  try {
    await pub.set(voiceUploadPcmKey(deviceId), pcm, 'EX', ttlSec)            // binary value (ioredis Buffer)
    await pub.set(voiceUploadMetaKey(deviceId), JSON.stringify(meta), 'EX', ttlSec)  // commit marker last
  } catch (err) { logger.error('[bus] parkVoiceUpload failed', err, { deviceId, size: pcm.length }) }
}
export async function resumeVoiceUpload(deviceId: string): Promise<{ meta: ParkedVoiceMeta; pcm: Buffer } | null> {
  try {
    const metaStr = await pub.get(voiceUploadMetaKey(deviceId))
    if (!metaStr) return null
    const pcm = await pub.getBuffer(voiceUploadPcmKey(deviceId))
    if (!pcm) return null
    return { meta: JSON.parse(metaStr) as ParkedVoiceMeta, pcm }
  } catch (err) { logger.error('[bus] resumeVoiceUpload failed', err, { deviceId }); return null }
}
export async function evictVoiceUpload(deviceId: string): Promise<void> {
  try { await pub.del(voiceUploadMetaKey(deviceId), voiceUploadPcmKey(deviceId)) } catch { /* best effort */ }
}

export async function clearAgentPresence(machineId: string): Promise<void> {
  try {
    await pub.del(presenceKey(machineId))
  } catch (err) {
    logger.error('[bus] clearAgentPresence failed', err, { machineId })
  }
}

// ── one-computer-per-machine ownership ────────────────────────────────────────────────────────────
// A SEPARATE key from presence (which stays `machine:{id}:mgr` = 'remote' for the online/status path):
// `machine:{id}:owner` = the connecting COMPUTER's stable id. Atomic compare-and-set so a SECOND
// computer (different computerId) is rejected while the first holds the machine, but the SAME computer
// reconnecting after a blip/restart (its 30s key may still be alive) always reclaims. The value is the
// computerId — not a shared token — so the delete on teardown is conditional (a superseded/other socket
// can't free the real owner's claim). Mirrors the device-presence compare-and-delete pattern below.
const ownerKey = (machineId: string): string => `machine:${machineId}:owner`

/** Claim or refresh ownership. Returns true when this computer owns the machine afterwards (it was free
 *  or already ours), false when a DIFFERENT computer currently holds it. Also used as the refresh call. */
export async function claimMachineOwner(machineId: string, computerId: string, ttlSec = 30): Promise<boolean> {
  try {
    const res = await pub.eval(
      "local cur = redis.call('get', KEYS[1]) " +
      "if (not cur) or (cur == ARGV[1]) then redis.call('set', KEYS[1], ARGV[1], 'EX', ARGV[2]) return 1 else return 0 end",
      1, ownerKey(machineId), computerId, String(ttlSec),
    )
    return res === 1
  } catch (err) {
    // On a Redis error, fail OPEN (allow) — losing the backend link should never lock a user out of
    // their own machine; single-computer enforcement is best-effort liveness, not a security boundary.
    logger.error('[bus] claimMachineOwner failed', err, { machineId })
    return true
  }
}

/** Conditional delete: only frees the claim if it still holds THIS computerId (compare-and-delete). */
export async function releaseMachineOwner(machineId: string, computerId: string): Promise<void> {
  try {
    await pub.eval(
      "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end",
      1, ownerKey(machineId), computerId,
    )
  } catch (err) {
    logger.error('[bus] releaseMachineOwner failed', err, { machineId })
  }
}

// ── desktop-app state on a remote machine ─────────────────────────────────────────────────────────
// `machine:{id}:app` = "<engine>:<state>" — which desktop app this COMPUTER drives and whether it is
// running right now (see docs/specs/machine-adapter-cursor-desktop-integration.md). Written by the
// adapter socket, refreshed on the same 15s heartbeat that renews presence, and given the SAME 30s
// TTL as `machine:{id}:mgr` so the claim can never outlive the socket that made it: if a worker is
// killed without running cleanup, both keys expire together and the machine reads offline rather
// than "Cursor open" forever. Deliberately NOT a Prisma column — the value's useful life is 30s and
// a DB row would survive a crashed worker and lie.
const appStateKey = (machineId: string): string => `machine:${machineId}:app`

export async function setMachineAppState(
  machineId: string, engine: string, state: string, ttlSec = 30,
): Promise<void> {
  try {
    await pub.set(appStateKey(machineId), `${engine}:${state}`, 'EX', ttlSec)
  } catch (err) {
    logger.error('[bus] setMachineAppState failed', err, { machineId })
  }
}

/** null when unknown (no adapter, legacy adapter, or the key expired). */
export async function getMachineAppState(
  machineId: string,
): Promise<{ engine: string; state: string } | null> {
  try {
    const raw = await pub.get(appStateKey(machineId))
    if (!raw) return null
    const sep = raw.indexOf(':')
    if (sep <= 0 || sep === raw.length - 1) return null
    return { engine: raw.slice(0, sep), state: raw.slice(sep + 1) }
  } catch (err) {
    logger.error('[bus] getMachineAppState failed', err, { machineId })
    return null
  }
}

function parseAppState(raw: string | null): { engine: string; state: string } | null {
  if (!raw) return null
  const sep = raw.indexOf(':')
  if (sep <= 0 || sep === raw.length - 1) return null
  return { engine: raw.slice(0, sep), state: raw.slice(sep + 1) }
}

/** MGET variant of getMachineAppState for the list seeds. */
export async function getMachineAppStateMany(machineIds: string[]): Promise<Map<string, { engine: string; state: string } | null>> {
  const out = new Map<string, { engine: string; state: string } | null>()
  if (machineIds.length === 0) return out
  try {
    const vals = await pub.mget(...machineIds.map(appStateKey))
    machineIds.forEach((id, i) => out.set(id, parseAppState(vals[i] ?? null)))
  } catch (err) {
    logger.error('[bus] getMachineAppStateMany failed', err, { count: machineIds.length })
    for (const id of machineIds) out.set(id, null)
  }
  return out
}

export async function clearMachineAppState(machineId: string): Promise<void> {
  try {
    await pub.del(appStateKey(machineId))
  } catch (err) {
    logger.error('[bus] clearMachineAppState failed', err, { machineId })
  }
}

// ── new-id quota (how fast one account may mint machine / device rows) ────────────────────────────────
// Both are resolved-or-created from a SELF-DECLARED computer id, so a valid token plus a loop of fresh
// ids would otherwise mint rows without bound. Fixed windows, one hour and one day, counted in Redis
// because every replica must share them. Called only on the CREATE path — a reconnect never spends one.
const NEW_ID_WINDOWS = [
  { suffix: 'h', ms: 60 * 60_000 },
  { suffix: 'd', ms: 24 * 60 * 60_000 },
] as const

// INCR every window, arming its expiry on first use; return the counts in KEYS order. One round trip,
// and atomic, so two replicas cannot both read "under the limit" off the same count.
const NEW_ID_QUOTA_SCRIPT = `
local out = {}
for i, key in ipairs(KEYS) do
  local n = redis.call('incr', key)
  if n == 1 then redis.call('pexpire', key, ARGV[i]) end
  out[i] = n
end
return out`

export type NewIdKind = 'machine' | 'device'

function newIdLimits(kind: NewIdKind): [number, number] {
  return kind === 'machine'
    ? [env.HARNESS_NEW_MACHINE_PER_HOUR, env.HARNESS_NEW_MACHINE_PER_DAY]
    : [env.HARNESS_NEW_DEVICE_PER_HOUR, env.HARNESS_NEW_DEVICE_PER_DAY]
}

/**
 * Spend one new-id for `userId`; false when any window is over its limit (0 = window disabled).
 * Fails OPEN on a Redis error: the per-account row ceilings still hold, and refusing every first login
 * during a Redis blip would be worse than a few extra rows.
 */
export async function consumeNewIdQuota(kind: NewIdKind, userId: string): Promise<boolean> {
  const limits = newIdLimits(kind)
  if (limits.every((l) => l <= 0)) return true
  try {
    const keys = NEW_ID_WINDOWS.map((w) => `newid:${kind}:${userId}:${w.suffix}`)
    const counts = await pub.eval(NEW_ID_QUOTA_SCRIPT, keys.length, ...keys, ...NEW_ID_WINDOWS.map((w) => w.ms)) as number[]
    return counts.every((n, i) => limits[i]! <= 0 || n <= limits[i]!)
  } catch (err) {
    logger.error('[bus] consumeNewIdQuota failed — allowing', err, { kind, userId })
    return true
  }
}

// ── device presence (is a paired device's socket currently connected?) ─────────────────────────────
// One key PER DEVICE (`device:{deviceId}:conn`) — a user with many devices has one independent key,
// refresh loop and supersede scope per device. The VALUE is the connection's own token so that when
// a device reconnects (new socket, possibly on another worker) the OLD socket's teardown cannot
// delete the NEW socket's key: the delete is conditional on the token still being ours.
const devicePresenceKey = (deviceId: string): string => `device:${deviceId}:conn`
const deviceStatusChannel = (userId: string): string => `devstatus:${userId}`
const deviceMachineListChannel = (userId: string): string => `devmachines:${userId}`
const deviceE2eePairChannel = (userId: string): string => `deve2eepair:${userId}`

/** Set/refresh this connection's presence. Called on device-ws open + every ~15s while connected. */
export async function setDevicePresence(deviceId: string, connToken: string, ttlSec = 45): Promise<void> {
  try {
    await pub.set(devicePresenceKey(deviceId), connToken, 'EX', ttlSec)
  } catch (err) {
    logger.error('[bus] setDevicePresence failed', err, { deviceId })
  }
}

/**
 * What a conditional clear observed. The caller needs all three apart, because they justify
 * different things (see deviceWs.ts closeBoth):
 *  - `cleared`    the key is gone — this socket really was the device's last one.
 *  - `superseded` the key is held by a NEWER connection: the device is ONLINE on another socket and
 *                 this one no longer speaks for it. Reached whenever a device reconnects before its
 *                 previous socket is reaped, which is ordinary, not exceptional.
 *  - `unknown`    Redis could not be reached, so neither of the above is established.
 */
export type DevicePresenceClear = 'cleared' | 'superseded' | 'unknown'

/** Conditional delete: only removes the key if it still holds THIS connection's token. */
export async function clearDevicePresence(deviceId: string, connToken: string): Promise<DevicePresenceClear> {
  try {
    await pub.eval(
      "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end",
      1, devicePresenceKey(deviceId), connToken,
    )
    return (await pub.get(devicePresenceKey(deviceId))) ? 'superseded' : 'cleared'
  } catch (err) {
    logger.error('[bus] clearDevicePresence failed', err, { deviceId })
    return 'unknown'
  }
}

export async function getDevicePresence(deviceId: string): Promise<boolean> {
  try {
    return !!(await pub.get(devicePresenceKey(deviceId)))
  } catch (err) {
    logger.error('[bus] getDevicePresence failed', err, { deviceId })
    return false
  }
}

export interface DeviceStatusMsg { deviceId: string; online: boolean; lastSeenAt?: string }

/** Per-USER transition channel (one sub per watching web socket); messages are PER-DEVICE. */
export function publishDeviceStatus(userId: string, msg: DeviceStatusMsg): Promise<number> {
  return safePublish(deviceStatusChannel(userId), JSON.stringify(msg))
}

export function subscribeDeviceStatus(userId: string, cb: (msg: DeviceStatusMsg) => void): Promise<() => void> {
  return addSub(deviceStatusChannel(userId), cb as Cb)
}

// Per-USER machine-list invalidation for connected devices (and web machine-list watchers). Machine
// create/delete/rename happens on normal REST workers, while device/web sockets may live on another
// worker, so Redis fans out "re-query the DB now".
export interface DeviceMachineListChangedMsg {
  reason: 'created' | 'deleted' | 'renamed' | 'updated' | 'environment_changed'
  autonomousEnv?: 'prod' | 'stag'
}

export function publishDeviceMachineListChanged(userId: string, msg: DeviceMachineListChangedMsg): Promise<number> {
  return safePublish(deviceMachineListChannel(userId), JSON.stringify(msg))
}

export function subscribeDeviceMachineListChanged(userId: string, cb: (msg: DeviceMachineListChangedMsg) => void): Promise<() => void> {
  return addSub(deviceMachineListChannel(userId), cb as Cb)
}

// Per-USER desk invalidation: the account's tabs changed (routes/desk.ts) on some REST worker; every
// adapter socket of that user — one per computer, possibly on other workers — is told to re-fetch.
export interface DeskChangedMsg { revision: number }

const deskChannel = (userId: string): string => `desk:${userId}`

export function publishDeskChanged(userId: string, msg: DeskChangedMsg): Promise<number> {
  return safePublish(deskChannel(userId), JSON.stringify(msg))
}

export function subscribeDeskChanged(userId: string, cb: (msg: DeskChangedMsg) => void): Promise<() => void> {
  return addSub(deskChannel(userId), cb as Cb)
}

export interface DeviceE2eePairMsg {
  kind: 'pending' | 'cleared'
  machineId: string
  machineName?: string | null
  label?: string
  pairId: string
  expiresAt?: number
  computerFingerprint?: string | null
  result?: 'paired' | 'failed' | 'cancelled'
}

export function publishDeviceE2eePair(userId: string, msg: DeviceE2eePairMsg): Promise<number> {
  return safePublish(deviceE2eePairChannel(userId), JSON.stringify(msg))
}

export function subscribeDeviceE2eePair(userId: string, cb: (msg: DeviceE2eePairMsg) => void): Promise<() => void> {
  return addSub(deviceE2eePairChannel(userId), cb as Cb)
}

// ── per-device control channel (backend → the one connected device socket, any worker) ────────────
// Unlike pushDeviceRevoked (rides up:{machineId}, reaches only a hub-ATTACHED commander conn — i.e. a
// device that has selected a machine), this reaches the device-ws relay itself, so a device parked on
// the machine PICKER is covered too.
const deviceControlChannel = (deviceId: string): string => `devctl:${deviceId}`

export interface DeviceControlMsg { action: 'revoked' | 'superseded'; by?: string }

export function publishDeviceControl(deviceId: string, msg: DeviceControlMsg): Promise<number> {
  return safePublish(deviceControlChannel(deviceId), JSON.stringify(msg))
}

export function subscribeDeviceControl(deviceId: string, cb: (msg: DeviceControlMsg) => void): Promise<() => void> {
  return addSub(deviceControlChannel(deviceId), cb as Cb)
}

// ── instance-affinity for the data-plane mesh (Phase 2) ────────────────────────────────────────────
// machine:{machineId}:appinst → the backend INSTANCE id holding that agent's app-pool manager socket.
// inst:{instanceId}:mesh   → that instance's internal mesh endpoint (host:port). Both TTL-refreshed.
const appInstKey = (machineId: string): string => `machine:${machineId}:appinst`
const meshKey = (instanceId: string): string => `inst:${instanceId}:mesh`

export async function setAppInstance(machineId: string, instanceId: string, ttlSec = 30): Promise<void> {
  try { await pub.set(appInstKey(machineId), instanceId, 'EX', ttlSec) } catch (err) { logger.error('[bus] setAppInstance failed', err, { machineId }) }
}
export async function getAppInstance(machineId: string): Promise<string | null> {
  try { return await pub.get(appInstKey(machineId)) } catch (err) { logger.error('[bus] getAppInstance failed', err, { machineId }); return null }
}
export async function clearAppInstance(machineId: string): Promise<void> {
  try { await pub.del(appInstKey(machineId)) } catch (err) { logger.error('[bus] clearAppInstance failed', err, { machineId }) }
}
export async function setMeshEndpoint(instanceId: string, endpoint: string, ttlSec = 30): Promise<void> {
  try { await pub.set(meshKey(instanceId), endpoint, 'EX', ttlSec) } catch (err) { logger.error('[bus] setMeshEndpoint failed', err, { instanceId }) }
}
export async function getMeshEndpoint(instanceId: string): Promise<string | null> {
  try { return await pub.get(meshKey(instanceId)) } catch (err) { logger.error('[bus] getMeshEndpoint failed', err, { instanceId }); return null }
}

// ── global per-agent client-count registry ─────────────────────────────────────────────────────────
// Drives the node's `__clients` aggregate lifecycle. `registry.clientCounts` is PER-PROCESS, but the
// node needs the CROSS-INSTANCE total (web on B1 + device on B2 must both count). We keep a HASH per
// agent, field = backend INSTANCE_ID → "ui,commander". Each field carries a per-field TTL (HEXPIRE,
// Redis ≥7.4) so a CRASHED instance's contribution ages out on its own — else the node would keep a
// commander/web aggregate warm for a dead client-holder. Live instances refresh on the hub heartbeat
// (25s < 45s TTL). Only B_m reads the sum and emits the frame (see hub.recomputeAndSendClients).
const clientsKey = (machineId: string): string => `machine:${machineId}:clients`
const commanderJoinGenerationKey = (machineId: string): string => `machine:${machineId}:commander-join-generation`
const CLIENT_COUNT_TTL_SEC = 45

/** Write THIS instance's local {ui,commander,commanderActive} for an agent (HDEL when zero), with a fresh
 *  per-field TTL. Field format is `ui,commander,commanderActive`; the trailing field is optional so an old
 *  instance's `ui,commander` still parses (missing → 0). */
export async function setAgentClientCount(machineId: string, instanceId: string, ui: number, commander: number, commanderActive = 0): Promise<void> {
  const key = clientsKey(machineId)
  try {
    if (ui === 0 && commander === 0) { await pub.hdel(key, instanceId); return }
    await pub.hset(key, instanceId, `${ui},${commander},${commanderActive}`)
  } catch (err) {
    logger.error('[bus] setAgentClientCount failed', err, { machineId, instanceId })
    return
  }
  // Per-field expiry (Redis 7.4+): FIELDS 1 <field>. Refreshed each heartbeat; a dead instance's field
  // simply expires. `pub.call` avoids depending on ioredis' typed hexpire signature. On Redis < 7.4
  // (HEXPIRE unknown) fall back to a whole-key TTL — coarser (any live instance's heartbeat keeps the
  // whole key warm), but enough that a crashed sole instance still ages out. The count was already
  // written above, so a TTL failure must NOT surface as a "client count failed" error.
  try {
    await pub.call('HEXPIRE', key, String(CLIENT_COUNT_TTL_SEC), 'FIELDS', '1', instanceId)
  } catch {
    try { await pub.expire(key, CLIENT_COUNT_TTL_SEC) } catch { /* best-effort */ }
  }
}

/** Heartbeat variant: every local agent's counts in ONE pipeline instead of 2 round trips per agent.
 *  Same field format and per-field TTL as setAgentClientCount; HEXPIRE failures fall back to a key TTL. */
export async function setAgentClientCountsBatch(
  instanceId: string,
  rows: Array<{ machineId: string; ui: number; commander: number; commanderActive: number }>,
): Promise<void> {
  if (rows.length === 0) return
  const p = pub.pipeline()
  // Remember which pipeline result index each HEXPIRE lands at, so the fallback only re-expires the keys
  // whose HEXPIRE actually failed (rather than every key on any single failure).
  const hexpireByKey: Array<{ key: string; index: number }> = []
  let cmdIndex = 0
  for (const r of rows) {
    const key = clientsKey(r.machineId)
    if (r.ui === 0 && r.commander === 0) { p.hdel(key, instanceId); cmdIndex += 1; continue }
    p.hset(key, instanceId, `${r.ui},${r.commander},${r.commanderActive}`)
    p.call('HEXPIRE', key, String(CLIENT_COUNT_TTL_SEC), 'FIELDS', '1', instanceId)
    hexpireByKey.push({ key, index: cmdIndex + 1 }) // hset is cmdIndex, HEXPIRE is the next command
    cmdIndex += 2
  }
  let results: Array<[Error | null, unknown]> | null
  try {
    results = await p.exec()
  } catch (err) {
    logger.error('[bus] setAgentClientCountsBatch failed', err, { count: rows.length })
    return
  }
  // A HEXPIRE error (Redis < 7.4) → coarser whole-key TTL, exactly like the single-row path. Only the
  // keys whose own HEXPIRE errored need it; a null result set (shouldn't happen post-exec) falls back for all.
  const failedKeys = results
    ? hexpireByKey.filter(({ index }) => results![index]?.[0]).map(({ key }) => key)
    : hexpireByKey.map(({ key }) => key)
  if (failedKeys.length > 0) {
    const fb = pub.pipeline()
    for (const key of failedKeys) fb.expire(key, CLIENT_COUNT_TTL_SEC)
    try { await fb.exec() } catch { /* best-effort */ }
  }
}

function sumClientFields(h: Record<string, string>): { ui: number; commander: number; commanderActive: number } {
  let ui = 0
  let commander = 0
  let commanderActive = 0
  for (const v of Object.values(h)) {
    const [u, c, a] = String(v).split(',')
    ui += parseInt(u, 10) || 0
    commander += parseInt(c, 10) || 0
    commanderActive += parseInt(a, 10) || 0
  }
  return { ui, commander, commanderActive }
}

/** Sum of every live instance's counts for an agent (expired fields already dropped Redis-side). */
export async function getAgentClientTotals(machineId: string): Promise<{ ui: number; commander: number; commanderActive: number }> {
  try {
    return sumClientFields(await pub.hgetall(clientsKey(machineId)))
  } catch (err) {
    logger.error('[bus] getAgentClientTotals failed', err, { machineId })
    return { ui: 0, commander: 0, commanderActive: 0 }
  }
}

/** Totals + commander join generation in one round trip (the `__clients` recompute reads both). */
export async function getAgentClientState(machineId: string): Promise<{
  totals: { ui: number; commander: number; commanderActive: number }
  commanderJoinGeneration: number | undefined
}> {
  try {
    const res = await pub.pipeline().hgetall(clientsKey(machineId)).get(commanderJoinGenerationKey(machineId)).exec()
    const [hErr, h] = res?.[0] ?? [null, {}]
    const [gErr, g] = res?.[1] ?? [null, null]
    if (hErr) throw hErr
    if (gErr) throw gErr
    const generation = g == null ? undefined : Number(g)
    return {
      totals: sumClientFields((h ?? {}) as Record<string, string>),
      commanderJoinGeneration: generation != null && Number.isSafeInteger(generation) && generation >= 0 ? generation : undefined,
    }
  } catch (err) {
    logger.error('[bus] getAgentClientState failed', err, { machineId })
    return { totals: { ui: 0, commander: 0, commanderActive: 0 }, commanderJoinGeneration: undefined }
  }
}

/** Monotonic machine-wide generation. Unlike the aggregate count, this changes for EVERY commander attach,
 * including a leave/rejoin that is coalesced to the same total before the node sees a count snapshot. */
export async function bumpCommanderJoinGeneration(machineId: string): Promise<number | undefined> {
  try {
    return await pub.incr(commanderJoinGenerationKey(machineId))
  } catch (err) {
    logger.error('[bus] bumpCommanderJoinGeneration failed', err, { machineId })
    return undefined
  }
}

export async function getCommanderJoinGeneration(machineId: string): Promise<number | undefined> {
  try {
    const raw = await pub.get(commanderJoinGenerationKey(machineId))
    if (raw == null) return undefined
    const generation = Number(raw)
    return Number.isSafeInteger(generation) && generation >= 0 ? generation : undefined
  } catch (err) {
    logger.error('[bus] getCommanderJoinGeneration failed', err, { machineId })
    return undefined
  }
}

// ── validated SSO tokens (so N processes ask the profile API once, not N times) ────────────────────
//
// Not try/caught like the presence keys above: the cache treats a throw as a miss, and swallowing it
// here would also swallow the difference between "absent" and "Redis is down" for anyone reading logs.
// The key is a digest of the token — see `ssoProfileCache.ts` — and the value holds no credential.

export const redisSsoProfileStore = {
  get: (key: string): Promise<string | null> => pub.get(key),
  async set(key: string, value: string, ttlMs: number): Promise<void> {
    const px = Math.floor(ttlMs)
    if (px > 0) await pub.set(key, value, 'PX', px)
  },
}