import { execFileSync } from 'node:child_process';
import { createHash, randomUUID } from 'node:crypto';
import { hostname } from 'node:os';
import {
  LaneQueue,
  defaultLaneDbPath,
  type AcquireResult,
  type LaneId,
  type LaneStatus,
  type LaneTicket,
} from './lane-queue.js';

export interface LaneCoordinatorPort {
  readonly coordinator: 'local-sqlite' | 'cloudflare-durable-object';
  acquire(params: {
    lane: LaneId;
    project: string;
    scene?: number | null;
    promptHash?: string | null;
    limit: number | null;
    ttlSec?: number;
    holder?: string;
  }): AcquireResult;
  status(lane: LaneId, limit?: number | null): LaneStatus;
  heartbeat(ticketId: string, ttlSec?: number): boolean;
  release(ticketId: string, lane?: LaneId, limit?: number | null, outcome?: LaneReleaseOutcome): boolean;
  recordReceipt?(receipt: SharedLaneReceipt): { stored: boolean; deduped: boolean };
  /** The render log (remote coordinator only): done tickets newest first. */
  history?(lane: LaneId, options?: { limit?: number; project?: string }): LaneHistory;
}

/** What the driver saw when it gave the slot back — the render log's one word per
 *  draw (2026-09-04). `expired` is the coordinator's own verdict, never sent. */
export const LANE_OUTCOMES = ['completed', 'failed', 'timeout', 'lease-lost', 'not-submitted', 'slot-busy', 'abandoned'] as const;
export type LaneOutcome = (typeof LANE_OUTCOMES)[number];
export interface LaneReleaseOutcome {
  outcome: LaneOutcome;
  providerJobId?: string;
  note?: string;
}
export interface LaneHistory {
  lane: LaneId;
  tickets: LaneTicket[];
  terminalRetained: number;
  coordinator: string;
}

export type SharedLaneReceiptPhase =
  | 'lease-acquired'
  | 'references-applied'
  | 'provider-submitted'
  | 'provider-terminal';

export interface SharedLaneReceipt {
  lane: string;
  ticketId: string;
  receiptId: string;
  contentHash: string;
  phase: SharedLaneReceiptPhase;
  payload: Record<string, unknown>;
}

export interface RemoteLaneCoordinatorOptions {
  url: string;
  token: string;
  machineId: string;
  timeoutMs?: number;
  request?: RemoteLaneRequest;
}

export type RemoteLaneRequest = (input: {
  url: string;
  token: string;
  timeoutMs: number;
  body: Record<string, unknown>;
}) => { status: number; text: string };

export interface CreateLaneCoordinatorOptions {
  env?: NodeJS.ProcessEnv;
  dbPath?: string;
}

const PROTOCOL_VERSION = 1;
const DEFAULT_TTL_SECONDS = 1800;
const REMOTE_HELPER = String.raw`
let raw = '';
process.stdin.setEncoding('utf8');
process.stdin.on('data', (chunk) => { raw += chunk; });
process.stdin.on('end', async () => {
  let input;
  try { input = JSON.parse(raw); } catch (error) {
    process.stderr.write('invalid coordinator helper input: ' + String(error));
    process.exitCode = 2;
    return;
  }
  const controller = new AbortController();
  const timer = setTimeout(() => controller.abort(), input.timeoutMs);
  try {
    const response = await fetch(input.url, {
      method: input.method,
      headers: {
        authorization: 'Bearer ' + input.token,
        'content-type': 'application/json',
        accept: 'application/json',
      },
      body: input.body === null ? undefined : JSON.stringify(input.body),
      signal: controller.signal,
    });
    const text = await response.text();
    process.stdout.write(JSON.stringify({ status: response.status, text }));
  } catch (error) {
    process.stderr.write(error instanceof Error ? error.message : String(error));
    process.exitCode = 2;
  } finally {
    clearTimeout(timer);
  }
});
`;

function requiredText(value: string | undefined, name: string): string {
  const result = value?.trim();
  if (!result) throw new Error(`${name} is required for the shared lane coordinator`);
  return result;
}

function validateRemoteUrl(value: string): string {
  let url: URL;
  try {
    url = new URL(value);
  } catch {
    throw new Error('VCLAW_SHARED_QUEUE_URL must be an absolute URL');
  }
  const local = ['localhost', '127.0.0.1', '::1'].includes(url.hostname);
  if (url.protocol !== 'https:' && !(local && url.protocol === 'http:')) {
    throw new Error('VCLAW_SHARED_QUEUE_URL must use HTTPS (HTTP is allowed only for localhost)');
  }
  url.pathname = url.pathname.replace(/\/$/, '');
  url.search = '';
  url.hash = '';
  return url.toString().replace(/\/$/, '');
}

function object(value: unknown, label: string): Record<string, unknown> {
  if (!value || typeof value !== 'object' || Array.isArray(value)) {
    throw new Error(`shared lane coordinator returned malformed ${label}`);
  }
  return value as Record<string, unknown>;
}

function defaultRemoteRequest(input: Parameters<RemoteLaneRequest>[0]): ReturnType<RemoteLaneRequest> {
  const raw = execFileSync(process.execPath, ['--input-type=module', '--eval', REMOTE_HELPER], {
    input: JSON.stringify({ ...input, method: 'POST' }),
    encoding: 'utf8',
    timeout: input.timeoutMs + 2_000,
    maxBuffer: 512 * 1024,
    windowsHide: true,
  });
  return object(JSON.parse(raw) as unknown, 'HTTP envelope') as unknown as ReturnType<RemoteLaneRequest>;
}

function stableJson(value: unknown): string {
  if (Array.isArray(value)) return `[${value.map(stableJson).join(',')}]`;
  if (value && typeof value === 'object') {
    return `{${Object.entries(value as Record<string, unknown>)
      .sort(([left], [right]) => left.localeCompare(right))
      .map(([key, entry]) => `${JSON.stringify(key)}:${stableJson(entry)}`)
      .join(',')}}`;
  }
  return JSON.stringify(value);
}

export function sharedLaneReceiptContentHash(payload: Record<string, unknown>): string {
  return `sha256:${createHash('sha256').update(stableJson(payload)).digest('hex')}`;
}

function numberOrNull(value: unknown): number | null {
  return value === null || value === undefined ? null : Number(value);
}

function coordinatorRetryDelay(milliseconds: number): void {
  Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, milliseconds);
}

export class RemoteLaneCoordinator implements LaneCoordinatorPort {
  readonly coordinator = 'cloudflare-durable-object' as const;
  private readonly baseUrl: string;
  private readonly token: string;
  private readonly machineId: string;
  private readonly timeoutMs: number;
  private readonly request: RemoteLaneRequest;
  // One claim per DRIVER, not per process. lane_render spawns a fresh `vclaw video lane`
  // process for every call, so a per-process claim could never match its own ticket on
  // heartbeat; the driver exports VCLAW_LANE_CLAIM once and every call carries it.
  // Acquire always names a claim (the ledger stores one per ticket); heartbeat and
  // release name it ONLY when the driver set VCLAW_LANE_CLAIM. A hand
  // `vclaw video lane release --ticket X` from the same machine therefore still
  // unwedges a SIGKILLed driver's ticket under the machine rule — a throwaway UUID
  // there would leave the slot dead until the TTL.
  private readonly claimId = process.env.VCLAW_LANE_CLAIM?.trim() || randomUUID();
  private readonly namedClaim: string | undefined = process.env.VCLAW_LANE_CLAIM?.trim() || undefined;

  constructor(options: RemoteLaneCoordinatorOptions) {
    this.baseUrl = validateRemoteUrl(requiredText(options.url, 'VCLAW_SHARED_QUEUE_URL'));
    this.token = requiredText(options.token, 'VCLAW_SHARED_QUEUE_TOKEN');
    this.machineId = requiredText(options.machineId, 'VCLAW_MACHINE_ID');
    this.timeoutMs = options.timeoutMs ?? 10_000;
    this.request = options.request ?? defaultRemoteRequest;
    if (!Number.isInteger(this.timeoutMs) || this.timeoutMs < 500 || this.timeoutMs > 60_000) {
      throw new Error('shared lane coordinator timeout must be 500-60000ms');
    }
  }

  private call(path: string, payload: Record<string, unknown>): Record<string, unknown> {
    let envelope: Record<string, unknown> | null = null;
    const request = {
      url: `${this.baseUrl}${path}`,
      token: this.token,
      timeoutMs: this.timeoutMs,
      body: { protocolVersion: PROTOCOL_VERSION, machineId: this.machineId, ...payload },
    };
    let lastError: unknown;
    for (let attempt = 0; attempt < 3; attempt += 1) {
      try {
        envelope = object(this.request(request), 'HTTP envelope');
        const responseStatus = Number(envelope.status);
        if (![401, 429].includes(responseStatus) && responseStatus < 500) break;
        lastError = new Error(`HTTP ${responseStatus}`);
      } catch (error) {
        lastError = error;
      }
      // Queue operations are idempotent by stable request/ticket/receipt ID.
      // A bounded retry covers Worker cold start, secret propagation and edge
      // overload; it never falls back to SQLite or retries provider submission.
      if (attempt < 2) coordinatorRetryDelay(attempt === 0 ? 250 : 750);
    }
    if (!envelope) {
      const detail = lastError instanceof Error ? lastError.message : String(lastError);
      throw new Error(`shared lane coordinator is unavailable; provider submission is blocked (${detail})`);
    }
    const status = Number(envelope.status);
    const responseText = String(envelope.text ?? '');
    let response: Record<string, unknown>;
    try {
      response = object(JSON.parse(responseText) as unknown, 'JSON response');
    } catch {
      throw new Error(`shared lane coordinator returned HTTP ${status} with a malformed JSON response`);
    }
    if (!Number.isInteger(status) || status < 200 || status >= 300) {
      throw new Error(`shared lane coordinator rejected the request with HTTP ${status}: ${String(response.error ?? 'unknown error')}`);
    }
    if (response.coordinator !== this.coordinator) {
      throw new Error('shared lane coordinator response is missing its Cloudflare authority marker');
    }
    return response;
  }

  acquire(params: Parameters<LaneCoordinatorPort['acquire']>[0]): AcquireResult {
    if (params.limit === null) {
      return {
        status: 'granted', ticketId: `unlimited-${Date.now()}`, position: 0,
        etaSeconds: 0, lane: params.lane, deduped: false, coordinator: this.coordinator,
      };
    }
    const promptHash = params.promptHash?.trim();
    if (!promptHash) throw new Error(`lane ${params.lane} requires a stable promptHash/request identity`);
    const response = this.call('/v1/lanes/acquire', {
      lane: params.lane,
      project: params.project,
      scene: params.scene ?? null,
      promptHash,
      limit: params.limit,
      ttlSeconds: params.ttlSec ?? DEFAULT_TTL_SECONDS,
      claimId: this.claimId,
    });
    const status = response.status;
    if (status !== 'granted' && status !== 'queued') throw new Error('shared lane coordinator returned an invalid acquire status');
    return {
      status,
      ticketId: requiredText(typeof response.ticketId === 'string' ? response.ticketId : undefined, 'ticketId'),
      position: Number(response.position),
      etaSeconds: numberOrNull(response.etaSeconds),
      lane: requiredText(typeof response.lane === 'string' ? response.lane : undefined, 'lane'),
      deduped: response.deduped === true,
      coordinator: this.coordinator,
    };
  }

  status(lane: LaneId, limit: number | null = null): LaneStatus {
    if (limit === null) {
      return { lane, limit, held: [], queued: [], medianJobSeconds: null, terminalRetained: 0, coordinator: this.coordinator };
    }
    const response = this.call('/v1/lanes/status', { lane, limit });
    if (!Array.isArray(response.held) || !Array.isArray(response.queued)) {
      throw new Error('shared lane coordinator returned malformed lane status');
    }
    return {
      lane: requiredText(typeof response.lane === 'string' ? response.lane : undefined, 'lane'),
      limit: Number(response.limit),
      held: response.held as LaneStatus['held'],
      queued: response.queued as LaneStatus['queued'],
      medianJobSeconds: numberOrNull(response.medianJobSeconds),
      terminalRetained: Number(response.terminalRetained),
      coordinator: this.coordinator,
    };
  }

  heartbeat(ticketId: string, ttlSec = DEFAULT_TTL_SECONDS): boolean {
    if (ticketId.startsWith('unlimited-')) return true;
    const lane = ticketId.slice(0, ticketId.lastIndexOf(':'));
    if (!lane) throw new Error('remote lane heartbeat requires a Cloudflare ticket containing its lane');
    return this.call('/v1/lanes/heartbeat', { lane, ticketId, ttlSeconds: ttlSec, ...(this.namedClaim ? { claimId: this.namedClaim } : {}) }).refreshed === true;
  }

  release(ticketId: string, lane?: LaneId, limit: number | null = null, outcome?: LaneReleaseOutcome): boolean {
    if (ticketId.startsWith('unlimited-')) return true;
    const exactLane = lane ?? ticketId.slice(0, ticketId.lastIndexOf(':'));
    if (!exactLane) throw new Error('remote lane release requires the exact lane');
    return this.call('/v1/lanes/release', {
      lane: exactLane,
      ticketId,
      ...(this.namedClaim ? { claimId: this.namedClaim } : {}),
      ...(limit === null ? {} : { limit }),
      ...(outcome ? { outcome: outcome.outcome } : {}),
      ...(outcome?.providerJobId ? { providerJobId: outcome.providerJobId.slice(0, 256) } : {}),
      ...(outcome?.note ? { note: outcome.note.slice(0, 512) } : {}),
    }).released === true;
  }

  history(lane: LaneId, options: { limit?: number; project?: string } = {}): LaneHistory {
    const response = this.call('/v1/lanes/history', {
      lane,
      ...(options.limit === undefined ? {} : { limit: options.limit }),
      ...(options.project === undefined ? {} : { project: options.project }),
    });
    return {
      lane,
      tickets: (response.tickets ?? []) as LaneTicket[],
      terminalRetained: Number(response.terminalRetained ?? 0),
      coordinator: this.coordinator,
    };
  }

  recordReceipt(receipt: SharedLaneReceipt): { stored: boolean; deduped: boolean } {
    const response = this.call('/v1/lanes/receipt', { ...receipt });
    return { stored: response.stored === true, deduped: response.deduped === true };
  }

  verify(lane: string, limit: number): LaneStatus {
    return this.status(lane, limit);
  }
}

export function sharedLaneCoordinatorRequired(env: NodeJS.ProcessEnv = process.env): boolean {
  return env.VCLAW_SHARED_QUEUE_REQUIRED === '1';
}

export function createLaneCoordinator(options: CreateLaneCoordinatorOptions = {}): LaneCoordinatorPort {
  const env = options.env ?? process.env;
  const url = env.VCLAW_SHARED_QUEUE_URL?.trim();
  const required = sharedLaneCoordinatorRequired(env);
  if (url || required) {
    if (!url) throw new Error('VCLAW_SHARED_QUEUE_REQUIRED=1 but VCLAW_SHARED_QUEUE_URL is missing');
    const configuredMachine = env.VCLAW_MACHINE_ID?.trim();
    if (required && !configuredMachine) {
      throw new Error('VCLAW_SHARED_QUEUE_REQUIRED=1 but VCLAW_MACHINE_ID is missing; assign a distinct stable ID on each computer');
    }
    return new RemoteLaneCoordinator({
      url,
      token: requiredText(env.VCLAW_SHARED_QUEUE_TOKEN, 'VCLAW_SHARED_QUEUE_TOKEN'),
      machineId: configuredMachine || hostname(),
      timeoutMs: Number(env.VCLAW_SHARED_QUEUE_TIMEOUT_MS ?? 10_000),
    });
  }
  return new LaneQueue({ dbPath: options.dbPath ?? defaultLaneDbPath(env) });
}