import { timingSafeEqual } from 'node:crypto';
import { WebSocketServer, type WebSocket } from 'ws';
import {
  ExecutorClientFrameSchema,
  PROTOCOL_VERSION,
  parseContinuationState,
  type ExecutorClientFrame,
  type ExecutorServerFrame,
  type RunId,
  type WorkerContinuityViewFrame,
  type WorkerId,
} from '@platform-modules/chatgpt-orchestrator-protocol';
import { OrchestratorError, type OrchestratorService } from '@platform-modules/chatgpt-orchestrator-core';
import { commandForExecutionAttempt } from './execution-continuity.js';
import { workerBootstrap } from './bootstrap.js';

export interface ExecutorBridgeOptions {
  host: '127.0.0.1';
  port: number;
  token: string;
  heartbeatIntervalMs?: number;
  staleAfterMs?: number;
}

interface ExecutorSession {
  socket: WebSocket;
  executorId: string;
  lastHeartbeatAt: number;
}

function tokenEqual(expected: string, actual: string): boolean {
  const left = Buffer.from(expected);
  const right = Buffer.from(actual);
  return left.length === right.length && timingSafeEqual(left, right);
}

export class ExecutorBridge {
  private readonly service: OrchestratorService;
  private readonly options: Required<ExecutorBridgeOptions>;
  private server: WebSocketServer | null = null;
  private active: ExecutorSession | null = null;
  private readonly sessions = new Set<ExecutorSession>();
  private timer: NodeJS.Timeout | null = null;
  private readonly dispatchedThisSession = new Set<string>();
  private readonly observedWorkers = new Set<WorkerId>();

  constructor(service: OrchestratorService, options: ExecutorBridgeOptions) {
    this.service = service;
    this.options = {
      heartbeatIntervalMs: 20_000,
      staleAfterMs: 65_000,
      ...options,
    };
  }

  get status(): { connected: boolean; executorId: string | null; lastHeartbeatAt: string | null } {
    const active = this.active;
    return {
      connected: Boolean(active && active.socket.readyState === 1),
      executorId: active?.executorId ?? null,
      lastHeartbeatAt: active ? new Date(active.lastHeartbeatAt).toISOString() : null,
    };
  }

  get address(): string {
    const bound = this.server?.address();
    const port = typeof bound === 'object' && bound ? bound.port : this.options.port;
    return `ws://${this.options.host}:${port}`;
  }

  async start(): Promise<void> {
    if (this.server) return;
    this.server = new WebSocketServer({ host: this.options.host, port: this.options.port });
    this.server.on('connection', (socket) => this.handleConnection(socket));
    await new Promise<void>((resolve, reject) => {
      const server = this.server;
      if (!server) return reject(new Error('Executor bridge server was not created.'));
      if (server.address()) return resolve();
      server.once('listening', resolve);
      server.once('error', reject);
    });
    this.timer = setInterval(() => { void this.tick(); }, Math.min(5_000, this.options.heartbeatIntervalMs));
    this.timer.unref();
  }

  async close(): Promise<void> {
    if (this.timer) clearInterval(this.timer);
    this.timer = null;
    const sessions = [...this.sessions];
    this.sessions.clear();
    this.active = null;
    this.dispatchedThisSession.clear();
    for (const session of sessions) session.socket.close(1001, 'orchestrator shutting down');
    const server = this.server;
    this.server = null;
    if (!server) return;
    await new Promise<void>((resolve) => server.close(() => resolve()));
  }

  async dispatchNow(): Promise<void> {
    const active = this.active;
    if (!active || active.socket.readyState !== 1) return;
    const commands = await this.service.listDispatchableExecutorCommands();
    for (const record of commands) {
      if (Date.parse(record.command.deadline) <= Date.now()) continue;
      if (this.dispatchedThisSession.has(record.command.commandId)) continue;
      const sent = await this.service.markExecutorCommandSent(record.command.commandId, active.executorId);
      if (sent.status !== 'sent') continue;
      this.dispatchedThisSession.add(record.command.commandId);
      this.observedWorkers.add(record.command.workerId);
      this.send(active.socket, { protocolVersion: PROTOCOL_VERSION, type: 'executor.command', command: record.command });
    }
  }

  private ready(session: ExecutorSession): void {
    this.send(session.socket, {
      protocolVersion: PROTOCOL_VERSION,
      type: 'executor.ready',
      executorId: session.executorId,
      heartbeatIntervalMs: this.options.heartbeatIntervalMs,
    });
  }

  private activate(session: ExecutorSession): void {
    this.active = session;
    this.dispatchedThisSession.clear();
    this.ready(session);
    void this.dispatchNow();
  }

  private promoteStandby(): void {
    if (this.active?.socket.readyState === 1) return;
    const standby = [...this.sessions].find((session) => session.socket.readyState === 1) ?? null;
    if (standby) this.activate(standby);
  }

  private handleConnection(socket: WebSocket): void {
    let authenticated = false;
    let session: ExecutorSession | null = null;
    const authenticationTimeout = setTimeout(() => {
      if (!authenticated) socket.close(1008, 'executor hello required');
    }, 5_000);

    socket.on('message', (data) => {
      void (async () => {
        let frame: ExecutorClientFrame;
        try {
          frame = ExecutorClientFrameSchema.parse(JSON.parse(data.toString()));
        } catch {
          socket.close(1008, 'invalid executor frame');
          return;
        }

        if (!authenticated) {
          if (frame.type !== 'executor.hello' || !tokenEqual(this.options.token, frame.token)) {
            socket.close(1008, 'authentication failed');
            return;
          }
          authenticated = true;
          clearTimeout(authenticationTimeout);
          session = { socket, executorId: frame.executorId, lastHeartbeatAt: Date.now() };
          this.sessions.add(session);

          // Firefox/LibreWolf may instantiate multiple MV3 background contexts for
          // one profile. Keep the first healthy session active and retain later
          // authenticated contexts as warm standbys instead of evicting each other.
          if (!this.active || this.active.socket.readyState !== 1) {
            this.activate(session);
            for (const workerId of this.observedWorkers) await this.sendWorkerContinuity(socket, workerId);
          } else {
            this.ready(session);
          }
          return;
        }

        const current = session;
        if (!current || frame.executorId !== current.executorId) {
          socket.close(1008, 'executor identity mismatch');
          return;
        }

        if (frame.type === 'executor.heartbeat') {
          current.lastHeartbeatAt = Date.now();
          return;
        }
        if (frame.type === 'executor.hello') {
          socket.close(1008, 'duplicate executor hello');
          return;
        }

        // Standbys receive executor.ready so they can heartbeat, but only the active
        // session may mutate orchestration state or complete commands. If the active
        // socket disappears, promotion sends a fresh ready frame and command replay
        // resumes through the normal durable executor-command path.
        if (current !== this.active) return;

        switch (frame.type) {
          case 'command.ack':
            await this.service.acknowledgeExecutorCommand(frame.commandId, current.executorId);
            break;
          case 'command.result':
            await this.service.completeExecutorCommand(frame.commandId, current.executorId, frame.success, frame.result, frame.error);
            break;
          case 'conversation.state':
            this.observedWorkers.add(frame.workerId);
            if (frame.state === 'error' && frame.error) {
              if (/^MESSAGE_DELIVERY_TIMEOUT(?:$|:)/.test(frame.error)) {
                const recovery = await this.service.recordDeliveryTimeout(frame.workerId);
                if (recovery) await this.service.enqueueExecutorCommand(commandForExecutionAttempt(recovery));
              } else {
                const [errorCode = 'CHATGPT_PRODUCT_ERROR'] = frame.error.split(':', 1);
                await this.service.recordExecutionError(frame.workerId, { code: errorCode, detail: frame.error });
              }
            }
            await this.service.recordConversationState({
              workerId: frame.workerId,
              executorId: current.executorId,
              state: frame.state,
              ...(frame.tabId === undefined ? {} : { tabId: frame.tabId }),
              ...(frame.conversationId === undefined ? {} : { conversationId: frame.conversationId }),
              ...(frame.url === undefined ? {} : { url: frame.url }),
              ...(frame.error === undefined ? {} : { error: frame.error }),
            });
            if (frame.state === 'idle' && frame.assistantOutputTail && !frame.assistantOutputTruncated) {
              await this.reconcileContinuationOutput(frame.workerId, frame.assistantOutputTail);
            }
            await this.sendWorkerContinuity(socket, frame.workerId);
            break;
          case 'operator.action':
            if (frame.workerId) this.observedWorkers.add(frame.workerId);
            await this.handleOperatorAction(socket, current.executorId, frame);
            break;
          case 'deadman.recovery_request':
            await this.handleDeadmanRecoveryRequest(socket, current.executorId, frame);
            break;
        }
        await this.dispatchNow();
      })().catch(() => socket.close(1011, 'executor frame handling failed'));
    });

    socket.on('close', () => {
      clearTimeout(authenticationTimeout);
      if (!session) return;
      const wasActive = this.active === session;
      this.sessions.delete(session);
      if (wasActive) {
        this.active = null;
        this.promoteStandby();
      }
    });
  }

  private async reconcileContinuationOutput(workerId: WorkerId, outputTail: string): Promise<void> {
    const parsed = parseContinuationState(outputTail);
    if (!parsed.success) return;
    if (parsed.value.status !== 'blocked') return;
    const view = await this.service.getWorkerContinuity(workerId);
    if (['awaiting_human', 'awaiting_dependency', 'paused', 'terminal'].includes(view.continuity.disposition)) return;
    await this.service.awaitDependency(workerId, {
      reason: `Unclassified continuation-state blocker: ${parsed.value.summary}; next: ${parsed.value.next}`,
      dependencies: [],
    });
  }

  private recentEventDetail(payload: Record<string, unknown>): string | null {
    for (const key of ['detail', 'reason', 'errorDetail', 'error', 'state']) {
      const value = payload[key];
      if (typeof value === 'string' && value.trim()) return value.trim().slice(0, 500);
    }
    return null;
  }

  private async continuityView(workerId: WorkerId): Promise<WorkerContinuityViewFrame> {
    const context = await this.service.getWorkerAssignmentContext(workerId);
    const [view, attempts, runPolicy, binding, events] = await Promise.all([
      this.service.getWorkerContinuity(workerId),
      this.service.listExecutionAttempts(workerId),
      this.service.getRunContinuationPolicy(context.run.runId),
      this.service.getConversationBinding(workerId),
      this.service.listEvents(context.run.runId),
    ]);
    const latest = view.latestAttempt;
    const active = latest && ['scheduled', 'submitted', 'generating'].includes(latest.state);
    const blockReason = view.continuity.humanWait
      ? `${view.continuity.humanWait.reason}: ${view.continuity.humanWait.request}`
      : view.continuity.dependencyWait?.reason
        ?? (view.continuity.pauseReason ? `Paused: ${view.continuity.pauseReason}` : null);
    const recentEvents = events
      .filter((event) => event.workerId === workerId)
      .filter((event) => event.type.startsWith('continuation.') || event.type.startsWith('execution.') || event.type === 'worker.progress' || event.type === 'worker.message')
      .slice(-8)
      .map((event) => ({ type: event.type, timestamp: event.timestamp, detail: this.recentEventDetail(event.payload) }));
    return {
      workerId,
      runId: context.run.runId,
      runTitle: context.run.title,
      workerName: context.worker.name,
      ...(context.worker.displayName ? { displayName: context.worker.displayName } : {}),
      assignmentTitle: context.worker.taskSummary ?? context.assignment.objective,
      disposition: view.continuity.disposition,
      policyMode: view.effectivePolicy.mode,
      runPolicyMode: runPolicy.mode,
      policySource: view.policySource,
      workerPolicyOverride: view.continuity.policyOverride !== null,
      deadmanFallbackEnabled: view.effectivePolicy.deadmanFallbackEnabled,
      deadmanThresholdMs: view.effectivePolicy.deadmanThresholdMs,
      attemptNumber: latest?.sequence ?? null,
      autoResumeCount: attempts.filter((attempt) => attempt.reason === 'auto_resume').length,
      lastProgressAt: view.continuity.lastProgressAt,
      needsUser: ['awaiting_human', 'awaiting_dependency'].includes(view.continuity.disposition),
      isWorking: view.continuity.disposition === 'running',
      isResuming: Boolean(active && latest && ['auto_resume', 'manual_resume', 'recovery'].includes(latest.reason)),
      blockReason,
      pauseReason: view.continuity.pauseReason,
      noProgressAttempts: view.continuity.consecutiveExecutionAttemptsWithoutProgress,
      stallWarningAt: view.continuity.stallWarningAt,
      stallPausedAt: view.continuity.stallPausedAt,
      conversationUrl: binding?.url ?? null,
      recentEvents,
    };
  }

  private async continuityViewsForRun(runId: RunId): Promise<WorkerContinuityViewFrame[]> {
    const workers = await this.service.listWorkers(runId);
    return Promise.all(workers.map((worker) => this.continuityView(worker.workerId)));
  }

  private async sendWorkerContinuity(socket: WebSocket, workerId: WorkerId): Promise<void> {
    try {
      const view = await this.continuityView(workerId);
      this.send(socket, {
        protocolVersion: PROTOCOL_VERSION,
        type: 'worker.continuity',
        executorId: this.active?.executorId ?? 'unknown',
        view,
      });
    } catch (error) {
      if (!(error instanceof OrchestratorError && error.code === 'NOT_FOUND')) throw error;
    }
  }

  private async handleDeadmanRecoveryRequest(
    socket: WebSocket,
    executorId: string,
    frame: Extract<ExecutorClientFrame, { type: 'deadman.recovery_request' }>,
  ): Promise<void> {
    const recovery = await this.service.requestDeadmanRecovery(frame.runId, frame.observedAt);
    for (const attempt of recovery.attempts) {
      const context = await this.service.getWorkerAssignmentContext(attempt.workerId);
      const binding = await this.service.getConversationBinding(attempt.workerId);
      const command = commandForExecutionAttempt(attempt, undefined, {
        bootstrap: workerBootstrap(context),
        recoveryRequiresCreate: !binding || ['closed', 'error'].includes(binding.state),
      });
      await this.service.enqueueExecutorCommand(command);
      this.observedWorkers.add(attempt.workerId);
    }
    this.send(socket, {
      protocolVersion: PROTOCOL_VERSION,
      type: 'deadman.recovery_result',
      executorId,
      requestId: frame.requestId,
      runId: frame.runId,
      fired: recovery.fired,
      reason: recovery.reason,
      recoveredWorkerIds: recovery.attempts.map((attempt) => attempt.workerId),
    });
    for (const attempt of recovery.attempts) await this.sendWorkerContinuity(socket, attempt.workerId);
    await this.dispatchNow();
  }

  private async handleOperatorAction(
    socket: WebSocket,
    executorId: string,
    frame: Extract<ExecutorClientFrame, { type: 'operator.action' }>,
  ): Promise<void> {
    try {
      const runAction = ['set_run_auto_resume', 'pause_run', 'resume_run', 'cancel_run'].includes(frame.action);
      if (runAction) {
        const runId = frame.runId!;
        switch (frame.action) {
          case 'set_run_auto_resume':
            await this.service.setRunContinuationPolicy(runId, { mode: frame.enabled ? 'auto' : 'manual' });
            break;
          case 'pause_run': {
            const workers = await this.service.listWorkers(runId);
            for (const worker of workers) {
              if (['completed', 'failed', 'cancelled'].includes(worker.state)) continue;
              await this.service.pauseWorker(worker.workerId);
            }
            break;
          }
          case 'resume_run': {
            const workers = await this.service.listWorkers(runId);
            for (const worker of workers) {
              if (['completed', 'failed', 'cancelled'].includes(worker.state)) continue;
              await this.service.resumeWorker(worker.workerId);
            }
            break;
          }
          case 'cancel_run':
            await this.service.cancelRun(runId, 'Cancelled by operator run control.');
            break;
        }
        const views = await this.continuityViewsForRun(runId);
        this.send(socket, {
          protocolVersion: PROTOCOL_VERSION,
          type: 'operator.result',
          executorId,
          requestId: frame.requestId,
          runId,
          success: true,
          views,
        });
        for (const view of views) {
          this.observedWorkers.add(view.workerId);
          this.send(socket, { protocolVersion: PROTOCOL_VERSION, type: 'worker.continuity', executorId, view });
        }
      } else {
        const workerId = frame.workerId!;
        switch (frame.action) {
          case 'pause':
            await this.service.pauseWorker(workerId);
            break;
          case 'continue_now': {
            const result = await this.service.continueNow(workerId, {
              overrideBlock: frame.overrideBlock ?? false,
              ...(frame.idempotencyKey ? { idempotencyKey: frame.idempotencyKey } : {}),
            });
            if (result.attempt?.state === 'scheduled') await this.service.enqueueExecutorCommand(commandForExecutionAttempt(result.attempt));
            break;
          }
          case 'cancel':
            await this.service.interruptWorker(workerId, 'Cancelled by operator control.');
            break;
          case 'set_auto_resume':
            await this.service.setWorkerContinuationPolicy(workerId, { mode: frame.enabled ? 'auto' : 'manual' });
            break;
        }
        const view = await this.continuityView(workerId);
        this.send(socket, {
          protocolVersion: PROTOCOL_VERSION,
          type: 'operator.result',
          executorId,
          requestId: frame.requestId,
          workerId,
          success: true,
          view,
        });
        this.send(socket, { protocolVersion: PROTOCOL_VERSION, type: 'worker.continuity', executorId, view });
      }
      await this.dispatchNow();
    } catch (error) {
      const orchestratorError = error instanceof OrchestratorError ? error : null;
      this.send(socket, {
        protocolVersion: PROTOCOL_VERSION,
        type: 'operator.result',
        executorId,
        requestId: frame.requestId,
        ...(frame.workerId ? { workerId: frame.workerId } : {}),
        ...(frame.runId ? { runId: frame.runId } : {}),
        success: false,
        ...(orchestratorError ? { errorCode: orchestratorError.code } : {}),
        error: error instanceof Error ? error.message : String(error),
      });
    }
  }

  private async tick(): Promise<void> {
    const cutoff = Date.now() - this.options.staleAfterMs;
    let lostActive = false;
    for (const session of [...this.sessions]) {
      if (session.lastHeartbeatAt >= cutoff) continue;
      this.sessions.delete(session);
      if (this.active === session) {
        this.active = null;
        lostActive = true;
      }
      session.socket.close(1001, 'executor heartbeat timeout');
    }
    if (lostActive || !this.active) this.promoteStandby();
    await this.dispatchNow();
  }

  private send(socket: WebSocket, frame: ExecutorServerFrame): void {
    if (socket.readyState === 1) socket.send(JSON.stringify(frame));
  }
}