// P2: one broker holds this lifecycle controller after Store acquired its writer lock.
// Every management-list read/modify/write is serialized. Only Credentials owns files.
import { randomBytes, createHash } from 'node:crypto';
import { performance } from 'node:perf_hooks';
import { isDeepStrictEqual } from 'node:util';
import { BusError } from './broker.mjs';
export const REVOKE_TIMEOUT_MS = 5000;
export const REVOKE_POLL_MS = 25;
export const MODEL_STOP_TIMEOUT_MS = 5000;
const keyId = key => createHash('sha256').update(key).digest('hex').slice(0, 12);
const fail = code => { throw new BusError(code); };
// Row 333 R3 (#1831): an id that leaves the record joins the agent's earlier ids, so a daily key
// quota keeps following the agent. Ids only, never a key; at most 64, the oldest dropped first.
// The field stays absent until the first id leaves, as Credentials' format allows.
export const MAX_PREVIOUS_KEY_IDS = 64;
const withPrevious = (record, ids, retiredAt) => {
  const previous = [...(record.previous ?? []), ...ids.map(id => ({ keyId: id, retiredAt }))].slice(-MAX_PREVIOUS_KEY_IDS);
  const { previous: _, ...rest } = record;
  return previous.length ? { ...rest, previous } : rest;
};
export class ModelKeys {
  #credentials; #client; #baseUrl; #prefix; #receipt; #lock; #org; #now; #runs; #stopRuns; #grace;
  #tail = Promise.resolve(); #ready = false; #failure = 'proxy-down'; #confirmed = new Map();
  constructor({ credentials, client, configDir, baseUrl, receipt = () => {}, now = () => Date.now(),
    runs = () => [], stopRuns = async () => {}, rotateGraceMinutes = 120, org = null, lock = null }) {
    this.#credentials = credentials; this.#client = client; this.#baseUrl = baseUrl;
    if ((org === null) !== (lock === null) || (org !== null && !/^[a-z0-9][a-z0-9._-]{0,63}$/.test(org))) fail('proxy-lock-invalid');
    this.#org = org; this.#lock = lock;
    this.#prefix = org === null ? `msk-${keyId(configDir)}-` : `mso-${keyId(configDir)}-${org}-`; this.#receipt = receipt; this.#now = now;
    this.#runs = runs; this.#stopRuns = stopRuns; this.#grace = rotateGraceMinutes * 60_000;
  }
  #exclusive(operation) {
    const result = this.#tail.then(async () => {
      try { return await (this.#lock ? this.#lock.run(operation) : operation()); }
      catch (error) { if (error.code === 'proxy-uncertain') fail('proxy-uncertain'); throw error; }
    });
    this.#tail = result.catch(() => {});
    return result;
  }
  #uncertain() { try { this.#lock.assertCertain(); return false; } catch { return true; } }
  #ownKey(key) { return key.startsWith(this.#prefix + 'key-') && /^[A-Za-z0-9_-]{43}$/.test(key.slice((this.#prefix + 'key-').length)); }
  #instances() { return this.#credentials.status().filter(s => s.service === 'model').map(s => s.instance); }
  #record(instance) { return this.#credentials.modelRecord(instance); }
  hasRecord(instance) { return this.#record(instance) !== null; }
  assertLaunchable(instance) { this.#launchable(instance); }
  #sentinel(keys) {
    const matches = keys.filter(k => k.startsWith(this.#prefix + 'sentinel-') && /^[A-Za-z0-9_-]{43}$/.test(k.slice((this.#prefix + 'sentinel-').length)));
    const registered = this.#org === null ? null : this.#lock.registration(this.#org);
    if (!matches.length || (this.#org !== null && (matches.length !== 1 || (registered && createHash('sha256').update(matches[0]).digest('hex') !== registered)))) {
      this.#ready = false; this.#failure = 'proxy-open'; fail('proxy-open');
    }
  }
  #launchable(instance) {
    try { this.#lock?.assertCertain(); } catch { fail('proxy-uncertain'); }
    if (this.#record(instance)?.state === 'revoking') fail('proxy-key-revoking');
    if (!this.#ready) fail(this.#failure);
  }
  // P5's destructive queue read needs this instance's current identity, not
  // an issued agent credential. A sentinel refusal stays closed until explicit
  // reconcile; telemetry never retries around it or performs reconciliation.
  verifyIdentity() {
    return this.#exclusive(async () => {
      if (!this.#ready) fail(this.#failure);
      let keys;
      try { keys = await this.#client.getKeys(); } catch { fail('proxy-down'); }
      this.#sentinel(keys);
    });
  }
  async #replace(before, after) {
    this.#sentinel(before); this.#sentinel(after);
    if (this.#lock) {
      const latest = await this.#client.getKeys();
      if (!isDeepStrictEqual(latest, before)) fail('proxy-client-response');
      this.#lock.fence({ org: this.#org, op: 'keys.put' });
    }
    await this.#client.putKeys(after);
    const confirmed = await this.#client.getKeys();
    this.#sentinel(confirmed);
    const removed = before.filter(k => !after.includes(k));
    if (after.some(k => !confirmed.includes(k)) || removed.some(k => confirmed.includes(k))) fail('proxy-client-response');
    this.#lock?.confirm();
  }
  async #pollRevoked(keys) {
    const deadline = performance.now() + REVOKE_TIMEOUT_MS;
    for (const key of keys) {
      for (;;) {
        const remaining = deadline - performance.now();
        if (remaining <= 0) fail('credential-revoke-failed');
        let status;
        try { status = await this.#client.probeKey(key, { timeout: remaining }); }
        catch { status = 0; }
        if (status === 401) break;
        const left = deadline - performance.now();
        if (left <= 0) fail('credential-revoke-failed');
        await new Promise(resolve => setTimeout(resolve, Math.min(REVOKE_POLL_MS, left)));
      }
    }
  }
  async #removeKeys(keys, confirm = () => {}) {
    const before = await this.#client.getKeys();
    confirm();
    await this.#replace(before, before.filter(k => !keys.includes(k)));
    await this.#pollRevoked(keys);
  }
  #captureRevocation(instance) {
    const record = this.#record(instance);
    if (record?.state !== 'revoking' || [record.key, ...record.retiring.map(r => r.key)].some(key => this.#org === null ? !key.startsWith(this.#prefix) : !this.#ownKey(key))) fail('credential-revoke-failed');
    return { record, snapshot: this.#credentials.captureModelRevocation(instance) };
  }
  #confirmRevocation(instance, proof) {
    if (!isDeepStrictEqual(this.#record(instance), proof.record)) fail('credential-revoke-failed');
    // A prior tmp removal may leave only final at retry/restart. Neither its
    // absence nor a revoking label proves exit. Fresh views grant NO stop authority.
    if (this.#runs(instance).length) fail('proxy-stop-failed');
  }
  async reconcile() {
    return this.#exclusive(async () => {
      this.#ready = false;
      try {
        let before = await this.#client.getKeys();
        const sentinels = () => before.filter(k => k.startsWith(this.#prefix + 'sentinel-') && /^[A-Za-z0-9_-]{43}$/.test(k.slice((this.#prefix + 'sentinel-').length)));
        const registered = this.#org === null ? null : this.#lock.registration(this.#org);
        if (this.#org !== null && (sentinels().length > 1 || (registered && (sentinels().length !== 1 || createHash('sha256').update(sentinels()[0]).digest('hex') !== registered)))) fail('proxy-open');
        if (this.#org !== null && sentinels().length === 0) {
          const sentinel = this.#prefix + 'sentinel-' + randomBytes(32).toString('base64url');
          const latest = await this.#client.getKeys();
          if (!isDeepStrictEqual(latest, before)) fail('proxy-client-response');
          this.#lock.fence({ org: this.#org, op: 'sentinel.register' });
          await this.#client.putKeys([...before, sentinel]);
          const confirmed = await this.#client.getKeys();
          if (!confirmed.includes(sentinel) || before.some(k => !confirmed.includes(k))) fail('proxy-client-response');
          this.#lock.register(this.#org, sentinel);
          this.#lock.confirm();
          this.#receipt({ action: 'proxy.sentinel.registered', org: this.#org });
          before = confirmed;
        }
        this.#sentinel(before);
        if (this.#org !== null && !registered) this.#lock.register(this.#org, sentinels()[0]);
        const records = this.#instances().map(instance => ({ instance, record: this.#record(instance) })).filter(e => e.record);
        const retained = records.filter(e => e.record.state !== 'revoking').flatMap(e => [e.record.key, ...e.record.retiring.map(r => r.key)]);
        if (this.#org !== null && retained.some(key => !this.#ownKey(key))) fail('proxy-open');
        const revoking = records.filter(e => e.record.state === 'revoking');
        const proofs = new Map(revoking.map(({ instance }) => { this.#confirmed.delete(instance); return [instance, this.#captureRevocation(instance)]; }));
        const confirm = () => { for (const [instance, proof] of proofs) this.#confirmRevocation(instance, proof); };
        confirm();
        const intended = revoking.flatMap(e => [e.record.key, ...e.record.retiring.map(r => r.key)]);
        const removed = before.filter(k => (this.#org === null ? k.startsWith(this.#prefix) && !k.startsWith(this.#prefix + 'sentinel-') : this.#ownKey(k)) && !retained.includes(k));
        const after = [...new Set([...before.filter(k => !removed.includes(k) && !intended.includes(k)), ...retained])];
        const added = after.filter(k => !before.includes(k));
        if (added.length || removed.length || intended.length) await this.#replace(before, after);
        await this.#pollRevoked([...new Set([...removed, ...intended])]);
        confirm();
        for (const { instance } of revoking) {
          const proof = proofs.get(instance);
          this.#credentials.prepareModelCompletion(instance, proof.snapshot);
          this.#confirmed.set(instance, proof);
          this.completeRevoke(instance);
        }
        this.#ready = true;
        const receipt = { action: 'proxy.reconcile', addedKeyIds: added.map(keyId), removedKeyIds: [...new Set([...removed, ...intended])].map(keyId) };
        this.#receipt(receipt); return receipt;
      } catch (error) {
        this.#failure = error.code === 'proxy-uncertain' || (this.#lock && this.#uncertain()) ? 'proxy-uncertain' : error.code === 'proxy-open' ? 'proxy-open' : 'proxy-down';
        this.#receipt({ action: 'proxy.reconcile', code: this.#failure });
        fail(this.#failure);
      }
    });
  }
  async #issue(instance, rotate = false, currentKeys = null) {
    this.#launchable(instance); this.#credentials.registerModel(instance); this.#launchable(instance);
    const old = this.#record(instance);
    let before;
    try { before = currentKeys ?? await this.#client.getKeys(); this.#sentinel(before); }
    catch (error) { if (error.code === 'proxy-open') throw error; fail('proxy-key-issue-failed'); }
    if (this.#org !== null && old && !this.#ownKey(old.key)) fail('proxy-open');
    if (old && !rotate && before.includes(old.key)) return { baseUrl: this.#baseUrl, key: old.key };
    const key = this.#prefix + (this.#org === null ? '' : 'key-') + randomBytes(32).toString('base64url');
    const now = this.#now(), issuedAt = new Date(now).toISOString();
    const date = new Date(now + 30 * 86400_000).toISOString().slice(0, 10);
    const keep = rotate && before.includes(old?.key);
    const retiring = old ? [...old.retiring, ...(keep ? [{ key: old.key, keyId: old.keyId, issuedAt: old.issuedAt, retireAt: new Date(now + this.#grace).toISOString() }] : [])] : [];
    // A re-issue, or a rotation the proxy already lost the key for, drops the old key at once.
    const record = withPrevious({ key, keyId: keyId(key), issuedAt, rotateBy: date, state: 'valid', retiring, previous: old?.previous },
      old && !keep ? [old.keyId] : [], issuedAt);
    try {
      await this.#replace(before, [...new Set([...before, key])]);
      this.#credentials.writeModel(instance, record);
      this.#receipt({ action: rotate ? 'proxy.key.rotated' : 'proxy.key.issued', keyIds: [keyId(key)], instance });
      return { baseUrl: this.#baseUrl, key };
    } catch (error) {
      if (error.code === 'proxy-open' || error.code === 'proxy-uncertain') throw error;
      if (this.#lock && this.#uncertain()) fail('proxy-uncertain');
      fail('proxy-key-issue-failed');
    }
  }
  ensure(instance) { return this.#exclusive(() => this.#issue(instance)); }
  rotate(instance) { return this.#exclusive(() => this.#issue(instance, true)); }
  // Host-only callbacks are the existing synchronous organization path.
  hire(instance, check, publish) {
    return this.#exclusive(async () => {
      await check(); // Fresh validation and captured revision AFTER controller queue.
      const existed = this.hasRecord(instance);
      try {
        await this.#issue(instance);
        await check(); // Recheck captured revision AFTER HTTP, before publication.
        return await publish();
      } catch (error) {
        // A newly issued generation has not escaped to queued admission yet.
        if (!existed && this.hasRecord(instance)) {
          try { await this.#prepareRevoke(instance); this.completeRevoke(instance); } catch {}
        }
        throw error;
      }
    });
  }
  async preflight(instance) {
    // The revoking check must run before even an issue/preflight request.
    this.#launchable(instance);
    const proxy = await this.#exclusive(async () => {
      this.#launchable(instance);
      let keys;
      try { keys = await this.#client.getKeys(); } catch { fail('proxy-down'); }
      this.#sentinel(keys);
      return this.#issue(instance, false, keys);
    });
    let status;
    try { status = await this.#client.probeKey(proxy.key); }
    catch { fail('proxy-down'); }
    if (status < 200 || status >= 300) fail('proxy-key-refused');
    return proxy;
  }
  async #prepareRevoke(instance) {
    const record = this.#record(instance);
    if (!record) { if (this.#confirmed.has(instance)) fail('credential-revoke-failed'); return; }
    this.#confirmed.delete(instance);
    if (record.state !== 'revoking') this.#credentials.writeModel(instance, { ...record, state: 'revoking', revokingAt: new Date(this.#now()).toISOString() });
    try {
      const proof = this.#captureRevocation(instance), confirm = () => this.#confirmRevocation(instance, proof);
      confirm();
      await this.#removeKeys([record.key, ...record.retiring.map(r => r.key)], confirm);
      confirm();
      this.#credentials.prepareModelCompletion(instance, proof.snapshot);
      this.#confirmed.set(instance, proof);
      this.#receipt({ action: 'proxy.key.revoked', keyIds: [record.keyId, ...record.retiring.map(r => r.keyId)], remoteState: 'revoked', localState: 'kept' });
    } catch (error) { if (error.code === 'proxy-uncertain' || (this.#lock && this.#uncertain())) fail('proxy-uncertain'); fail('credential-revoke-failed'); }
  }
  prepareRevoke(instance, beforeRevoke = () => {}) { return this.#exclusive(() => { beforeRevoke(); return this.#prepareRevoke(instance); }); }
  completeRevoke(instance) {
    try { this.#lock?.assertCertain(); } catch { fail('proxy-uncertain'); }
    try {
      const record = this.#record(instance), proof = this.#confirmed.get(instance);
      if (!record && !proof) return; // Ordinary no-record/no-supplied-proof idempotence.
      if (!record || !proof || record.state !== 'revoking') fail('credential-revoke-failed');
      this.#confirmRevocation(instance, proof);
      this.#credentials.finishModelRevocation(instance, proof.snapshot);
      this.#confirmed.delete(instance);
    } catch { fail('credential-revoke-failed'); }
  }
  revisionConflict(instance) {
    const record = this.#record(instance);
    this.#receipt({ action: 'proxy.key.revoke-conflict', keyIds: record ? [record.keyId] : [], remoteState: 'revoked', localState: 'kept' });
    fail('proxy-revoke-revision-conflict');
  }
  sweep() {
    return this.#exclusive(async () => {
      let firstError;
      for (const instance of this.#instances()) {
        try {
        let record = this.#record(instance);
        if (!record || record.state === 'revoking') continue;
        if (this.#now() >= Date.parse(record.rotateBy + 'T00:00:00Z')) {
          await this.#issue(instance, true); record = this.#record(instance);
        }
        for (const old of record.retiring) {
          const runs = this.#runs(instance).filter(r => r.keyId === old.keyId);
          if (runs.length && this.#now() < Date.parse(old.retireAt)) continue;
          if (runs.length) {
            try { await this.#stopRuns(runs); } catch { fail('proxy-stop-failed'); }
          }
          if (this.#runs(instance).some(r => r.keyId === old.keyId)) fail('proxy-stop-failed');
          const confirm = () => {
            if (this.#runs(instance).some(r => r.keyId === old.keyId)) fail('proxy-stop-failed');
          };
          await this.#removeKeys([old.key], confirm);
          confirm();
          const current = this.#record(instance);
          this.#credentials.writeModel(instance, withPrevious({ ...current, retiring: current.retiring.filter(r => r.keyId !== old.keyId) },
            [old.keyId], new Date(this.#now()).toISOString()));
          this.#receipt({ action: 'proxy.key.retired', keyIds: [old.keyId] });
        }
        } catch (error) {
          firstError ??= error;
          this.#receipt({ action: 'proxy.sweep', instance, outcome: 'refused', code: error.code ?? 'proxy-stop-failed' });
        }
      }
      if (firstError) throw firstError;
    });
  }
}

// No new processing-control stop reason/schema: stopSession owns bounded SIGTERM
// and start-time verification. This layer additionally checks agent/key ownership.
export async function stopModelRuns({ runs, sessions, stop, receipt, reason = 'proxy-rotation', timeoutMs = MODEL_STOP_TIMEOUT_MS }) {
  for (const run of runs) {
    let timer;
    const ids = typeof run.keyId === 'string' && /^[a-f0-9]{12}$/.test(run.keyId) ? [run.keyId] : [];
    try {
      if (!ids.length) fail('proxy-stop-failed');
      const entry = sessions().find(s => s.business === run.business && s.run === run.run);
      if (!entry || entry.instance !== run.role || entry.modelKeyId !== run.keyId
          || entry.pid !== run.pid || entry.startTime !== run.startTime) fail('proxy-stop-failed');
      if (entry.live) {
        const result = await Promise.race([
          stop(run.run),
          new Promise((_, reject) => { timer = setTimeout(() => reject(new BusError('proxy-stop-failed')), timeoutMs); }),
        ]);
        if (!result?.stopped || sessions().some(s => s.business === run.business && s.run === run.run && s.live)) fail('proxy-stop-failed');
      }
      receipt({ action: reason, keyIds: ids, business: run.business, agent: run.role, run: run.run, outcome: 'stopped' });
    } catch {
      receipt({ action: reason, code: 'proxy-stop-failed', keyIds: ids, run: run.run });
      fail('proxy-stop-failed');
    } finally { clearTimeout(timer); }
  }
}

// Option C: HTTP is outside the organization lock. This adapter preserves write's
// synchronous contract and passes the captured revision into its existing lock check.
// Neither organization.mjs nor its synchronous settings callers needs modification.
export function modelOrganization({ modelKeys, writer }) {
  const pending = new Map();
  const adapter = {
    ...writer,
    write(business, document, options) {
      const prepare = pending.get(business);
      return writer.write(business, document, prepare ? { ...options, revision: prepare.revision } : options);
    },
  };
  const revoke = original => (business, agent, how, services) => {
    const prepare = pending.get(business), instance = business + '/' + agent;
    if (!prepare || prepare.instance !== instance) fail('credential-revoke-failed');
    // For an already-delisted repeat there is no write callback/lock, but the
    // revision still must match before completing the model credential cleanup.
    if (writer.read(business).revision !== prepare.revision) modelKeys.revisionConflict(instance);
    modelKeys.completeRevoke(instance);
    return original(business, agent, how, services);
  };
  const wrap = invoke => (business, verb, args, actor, context) => {
    const isRevoke = verb === 'instance.fire' || (verb === 'instances.set' && args.op === 'delete');
    if (!isRevoke) return invoke(business, verb, args, actor, context);
    return (async () => {
      if (pending.has(business)) fail('organization-busy');
      const agent = args.name ?? args.instance, instance = business + '/' + agent;
      if (verb === 'instance.fire') invoke(business, 'instance.fireCheck', { name: agent }, actor, context);
      if (verb === 'instances.set') invoke(business, 'instances.check', { instance: agent }, actor, context);
      const { revision } = writer.read(business);
      if (args.revision !== undefined && args.revision !== revision) fail('revision-conflict');
      const prepare = { instance, revision, remoteRevoked: false };
      pending.set(business, prepare);
      const existed = modelKeys.hasRecord(instance);
      try {
        if (isRevoke) { await modelKeys.prepareRevoke(instance, () => context?.beforeWrite?.()); prepare.remoteRevoked = existed; }
        return invoke(business, verb, args, actor, context);
      } catch (error) {
        if (isRevoke && prepare.remoteRevoked && (error.code === 'revision-conflict' || error.code === 'revision-conflict-revoked')) modelKeys.revisionConflict(instance);
        throw error;
      } finally { pending.delete(business); }
    })();
  };
  const hire = async (business, agent, check, publish) => {
    if (pending.has(business)) fail('organization-busy');
    const prepare = { instance: `${business}/${agent}`, revision: writer.read(business).revision, remoteRevoked: false };
    pending.set(business, prepare);
    try { return await modelKeys.hire(prepare.instance, check, publish); }
    finally { pending.delete(business); }
  };
  return { writer: adapter, revoke, wrap, hire };
}