import { timingSafeEqual } from 'node:crypto';
import { WebSocketServer, type WebSocket } from 'ws';
import {
  ExecutorClientFrameSchema,
  PROTOCOL_VERSION,
  parseContinuationState,
  type ExecutorClientFrame,
  type ExecutorServerFrame,
  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';

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

interface ActiveExecutor {
  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: ActiveExecutor | null = null;
  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;
    if (this.active) this.active.socket.close(1001, 'orchestrator shutting down');
    this.active = null;
    const server = this.server;
    this.server = null;
    if (!server) return;
    await new Promise<void>((resolve) => server.close(() => resolve()));
  }

  async dispatchNow(): Promise<void> {
    if (!this.active || this.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, this.active.executorId);
      if (sent.status !== 'sent') continue;
      this.dispatchedThisSession.add(record.command.commandId);
      this.observedWorkers.add(record.command.workerId);
      this.send(this.active.socket, { protocolVersion: PROTOCOL_VERSION, type: 'executor.command', command: record.command });
    }
  }

  private handleConnection(socket: WebSocket): void {
    let authenticated = false;
    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);
          if (this.active && this.active.socket !== socket) this.active.socket.close(1012, 'executor reconnected');
          this.dispatchedThisSession.clear();
          this.active = { socket, executorId: frame.executorId, lastHeartbeatAt: Date.now() };
          this.send(socket, {
            protocolVersion: PROTOCOL_VERSION,
            type: 'executor.ready',
            executorId: frame.executorId,
            heartbeatIntervalMs: this.options.heartbeatIntervalMs,
          });
          await this.dispatchNow();
          for (const workerId of this.observedWorkers) await this.sendWorkerContinuity(socket, workerId);
          return;
        }

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

        switch (frame.type) {
          case 'executor.heartbeat':
            active.lastHeartbeatAt = Date.now();
            break;
          case 'command.ack':
            await this.service.acknowledgeExecutorCommand(frame.commandId, active.executorId);
            break;
          case 'command.result':
            await this.service.completeExecutorCommand(frame.commandId, active.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: active.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':
            this.observedWorkers.add(frame.workerId);
            await this.handleOperatorAction(socket, active.executorId, frame);
            break;
          case 'executor.hello':
            socket.close(1008, 'duplicate executor hello');
            return;
        }
        await this.dispatchNow();
      })().catch(() => socket.close(1011, 'executor frame handling failed'));
    });

    socket.on('close', () => {
      clearTimeout(authenticationTimeout);
      if (this.active?.socket === socket) this.active = null;
    });
  }

  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 async continuityView(workerId: WorkerId): Promise<WorkerContinuityViewFrame> {
    const view = await this.service.getWorkerContinuity(workerId);
    const attempts = await this.service.listExecutionAttempts(workerId);
    const latest = view.latestAttempt;
    const active = latest && ['scheduled', 'submitted', 'generating'].includes(latest.state);
    return {
      workerId,
      disposition: view.continuity.disposition,
      policyMode: view.effectivePolicy.mode,
      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)),
    };
  }

  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 handleOperatorAction(
    socket: WebSocket,
    executorId: string,
    frame: Extract<ExecutorClientFrame, { type: 'operator.action' }>,
  ): Promise<void> {
    try {
      switch (frame.action) {
        case 'pause':
          await this.service.pauseWorker(frame.workerId);
          break;
        case 'continue_now': {
          const result = await this.service.continueNow(frame.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(frame.workerId, 'Cancelled by operator control.');
          break;
        case 'set_auto_resume':
          await this.service.setWorkerContinuationPolicy(frame.workerId, { mode: frame.enabled ? 'auto' : 'manual' });
          break;
      }
      const view = await this.continuityView(frame.workerId);
      this.send(socket, {
        protocolVersion: PROTOCOL_VERSION,
        type: 'operator.result',
        executorId,
        requestId: frame.requestId,
        workerId: frame.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,
        workerId: frame.workerId,
        success: false,
        ...(orchestratorError ? { errorCode: orchestratorError.code } : {}),
        error: error instanceof Error ? error.message : String(error),
      });
    }
  }

  private async tick(): Promise<void> {
    if (!this.active) return;
    if (Date.now() - this.active.lastHeartbeatAt > this.options.staleAfterMs) {
      this.active.socket.close(1001, 'executor heartbeat timeout');
      this.active = null;
      return;
    }
    await this.dispatchNow();
  }

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