/**
 * @fileoverview Record-Replay V3 service-level E2E test harness
 * @description
 * Assembles a complete V3 runtime (IndexedDB storage + scheduler + runner)
 * and drives it through RpcServer.handleRequest() to avoid Port mocking complexity.
 *
 * Design notes:
 * - Service-level testing: calls internal handler directly, not through Port
 * - Event streaming: reuses RpcServer.broadcastEvent subscription filtering logic
 * - waitForTerminal: uses EventsBus subscription to wait for terminal events, avoiding kick() race
 *
 * WARNING: This harness accesses RpcServer private members (connections/handleRequest/broadcastEvent)
 * via type casting. If RpcServer changes to use ES private fields (#private), these tests will break.
 * All such access is centralized in getRpcServerInternals() for easier maintenance.
 */

import { vi } from 'vitest';
import { z } from 'zod';

import type { JsonObject } from '@/entrypoints/background/record-replay-v3/domain/json';
import type { RunId } from '@/entrypoints/background/record-replay-v3/domain/ids';
import type {
  RunEvent,
  RunRecordV3,
} from '@/entrypoints/background/record-replay-v3/domain/events';
import type {
  RunQueueConfig,
  RunQueueItem,
} from '@/entrypoints/background/record-replay-v3/engine/queue/queue';
import type { StoragePort } from '@/entrypoints/background/record-replay-v3/engine/storage/storage-port';
import type { EventsBus } from '@/entrypoints/background/record-replay-v3/engine/transport/events-bus';
import type {
  RunScheduler,
  RunExecutor,
} from '@/entrypoints/background/record-replay-v3/engine/queue/scheduler';
import type {
  NodeDefinition,
  NodeExecutionResult,
} from '@/entrypoints/background/record-replay-v3/engine/plugins/types';

import { createStoragePort, closeRrV3Db } from '@/entrypoints/background/record-replay-v3';

import { StorageBackedEventsBus } from '@/entrypoints/background/record-replay-v3/engine/transport/events-bus';
import { DEFAULT_QUEUE_CONFIG } from '@/entrypoints/background/record-replay-v3/engine/queue/queue';
import { createLeaseManager } from '@/entrypoints/background/record-replay-v3/engine/queue/leasing';
import { createRunScheduler } from '@/entrypoints/background/record-replay-v3/engine/queue/scheduler';
import { InMemoryKeepaliveController } from '@/entrypoints/background/record-replay-v3/engine/keepalive/offscreen-keepalive';
import { PluginRegistry } from '@/entrypoints/background/record-replay-v3/engine/plugins/registry';
import {
  createRunRunnerFactory,
  type RunRunnerFactory,
} from '@/entrypoints/background/record-replay-v3/engine/kernel/runner';
import {
  createRunnerRegistry,
  type RunnerRegistry,
} from '@/entrypoints/background/record-replay-v3/engine/kernel/debug-controller';
import { createNotImplementedArtifactService } from '@/entrypoints/background/record-replay-v3/engine/kernel/artifacts';
import { RpcServer } from '@/entrypoints/background/record-replay-v3/engine/transport/rpc-server';
import {
  RR_ERROR_CODES,
  createRRError,
  type RRError,
} from '@/entrypoints/background/record-replay-v3/domain/errors';
import { isTerminalStatus } from '@/entrypoints/background/record-replay-v3/domain/events';
import { calculateWorkflowRevision } from '@/entrypoints/background/record-replay-v3/flows/publish';

// ==================== Types ====================

type Logger = Pick<Console, 'debug' | 'info' | 'warn' | 'error'>;

interface TestNodeConfig {
  action: 'succeed' | 'fail';
  outputs?: JsonObject;
  delayMs?: number;
}

/**
 * E2E Harness Configuration options
 */
export interface V3E2EHarnessOptions {
  /** Owner ID（Identifies the scheduler instance) */
  ownerId?: string;
  /** Scheduler configuration override */
  schedulerConfig?: Partial<RunQueueConfig>;
  /** Whether to automatically start the scheduler (default true) */
  autoStartScheduler?: boolean;
  /** Time source (for test injection) */
  now?: () => number;
  /** Logger */
  logger?: Logger;
}

/**
 * RPC client interface
 */
export interface RpcClient {
  /** All messages received */
  readonly messages: unknown[];
  /** Call RPC method */
  call<T = unknown>(method: string, params?: JsonObject): Promise<T>;
  /** Clear messages */
  clearMessages(): void;
  /** Get streaming events */
  getStreamedEvents(): RunEvent[];
}

/**
 * E2E Harness interface
 */
export interface V3E2EHarness {
  readonly ownerId: string;
  readonly storage: StoragePort;
  readonly events: EventsBus;
  readonly scheduler: RunScheduler;
  readonly runners: RunnerRegistry;
  readonly rpcServer: RpcServer;

  /** Create RPC client */
  createClient(): RpcClient;

  /** wait for specific event */
  waitForEvent(
    runId: RunId,
    predicate: (event: RunEvent) => boolean,
    opts?: { timeoutMs?: number },
  ): Promise<RunEvent>;

  /** Wait for Run to reach final state */
  waitForTerminal(runId: RunId, opts?: { timeoutMs?: number }): Promise<RunRecordV3>;

  /** Waiting for queue item to be removed */
  waitForQueueItemGone(runId: RunId, opts?: { timeoutMs?: number }): Promise<void>;

  /** List all events of Run */
  listEvents(runId: RunId): Promise<RunEvent[]>;

  /** Destroy the harness and release resources */
  dispose(): Promise<void>;
}

// ==================== RpcServer Test Internals ====================

/**
 * RpcServer internal access interface for testing.
 * Centralizes all private member access to make maintenance easier.
 */
interface RpcServerInternals {
  connections: Map<string, RpcConnection>;
  handleRequest<T>(req: unknown, conn: RpcConnection): Promise<T>;
  broadcastEvent(event: RunEvent): void;
}

interface RpcConnection {
  port: chrome.runtime.Port;
  subscriptions: Set<RunId | null>;
}

/**
 * Get RpcServer internals for testing.
 * WARNING: This accesses private members via type casting.
 */
function getRpcServerInternals(server: RpcServer): RpcServerInternals {
  const s = server as unknown as {
    connections: Map<string, RpcConnection>;
    handleRequest: <T>(req: unknown, conn: RpcConnection) => Promise<T>;
    broadcastEvent: (event: RunEvent) => void;
  };
  return {
    connections: s.connections,
    handleRequest: s.handleRequest.bind(s),
    broadcastEvent: s.broadcastEvent.bind(s),
  };
}

// ==================== Utilities ====================

function createSilentLogger(): Logger {
  return {
    debug: () => {},
    info: () => {},
    warn: () => {},
    error: () => {},
  };
}

function sleep(ms: number): Promise<void> {
  return new Promise((resolve) => setTimeout(resolve, ms));
}

/**
 * Create a Node definition for testing
 * @description A simple test node that supports success/failure/delay
 */
function createTestNodeDefinition(): NodeDefinition<'test', JsonObject> {
  return {
    kind: 'test',
    schema: z
      .object({
        action: z.enum(['succeed', 'fail']),
        outputs: z.record(z.any()).optional(),
        delayMs: z.number().optional(),
      })
      .passthrough() as unknown as NodeDefinition<'test', JsonObject>['schema'],
    execute: async (_ctx, node): Promise<NodeExecutionResult> => {
      const cfg = node.config as unknown as TestNodeConfig;

      // Analog delay
      if (cfg.delayMs && cfg.delayMs > 0) {
        await sleep(cfg.delayMs);
      }

      if (cfg.action === 'fail') {
        return {
          status: 'failed',
          error: createRRError(RR_ERROR_CODES.TOOL_ERROR, 'Test node intentionally failed'),
        };
      }

      return {
        status: 'succeeded',
        ...(cfg.outputs ? { outputs: cfg.outputs } : {}),
      };
    },
  };
}

// ==================== Factory ====================

/**
 * Create a V3 E2E test harness
 * @description Assemble a complete V3 runtime for integration testing
 */
export function createV3E2EHarness(options: V3E2EHarnessOptions = {}): V3E2EHarness {
  const logger = options.logger ?? createSilentLogger();
  const now = options.now ?? (() => Date.now());
  const ownerId = options.ownerId ?? 'e2e-owner';

  // 1) Storage
  const storage = createStoragePort();

  // 2) EventsBus
  const events = new StorageBackedEventsBus(storage.events);

  // 3) Plugins - Register test node
  const plugins = new PluginRegistry();
  plugins.registerNode(createTestNodeDefinition());

  // 4) RunnerRegistry
  const runners = createRunnerRegistry();

  // 5) RunRunnerFactory
  const runnerFactory = createRunRunnerFactory({
    storage,
    events,
    plugins,
    now,
    artifactService: createNotImplementedArtifactService(),
  });

  // 6) RunExecutor - Connect scheduler and runner
  const execute: RunExecutor = createE2EExecutor({
    storage,
    events,
    runnerFactory,
    runners,
    now,
    logger,
  });

  // 7) Scheduler Configuration
  const config: RunQueueConfig = {
    ...DEFAULT_QUEUE_CONFIG,
    maxParallelRuns: 1,
    ...options.schedulerConfig,
  };

  // 8) Keepalive + LeaseManager + Scheduler
  const keepalive = new InMemoryKeepaliveController();
  const leaseManager = createLeaseManager(storage.queue, config);

  const scheduler = createRunScheduler({
    queue: storage.queue,
    leaseManager,
    keepalive,
    config,
    ownerId,
    execute,
    now,
    tuning: { pollIntervalMs: 0, reclaimIntervalMs: 0 },
    logger,
  });

  // 9) RpcServer
  const rpcServer = new RpcServer({
    storage,
    events,
    scheduler,
    runners,
    now,
  });

  // Get internals via centralized helper
  const rpcInternals = getRpcServerInternals(rpcServer);

  // 10) Forward EventsBus events to RpcServer.broadcastEvent
  const unsubscribeForward = events.subscribe((event) => {
    try {
      rpcInternals.broadcastEvent(event);
    } catch (e) {
      logger.warn('[V3E2EHarness] broadcastEvent failed:', e);
    }
  });

  // 11) Start scheduler if configured
  if (options.autoStartScheduler ?? true) {
    scheduler.start();
  }

  // Client management
  let clientSeq = 0;
  let requestSeq = 0;
  const clientConnIds = new Set<string>();

  function createClient(): RpcClient {
    const connId = `e2e-conn-${++clientSeq}`;
    const messages: unknown[] = [];

    const port = {
      postMessage: (msg: unknown) => {
        messages.push(msg);
      },
      disconnect: vi.fn(),
    } as unknown as chrome.runtime.Port;

    const connection: RpcConnection = {
      port,
      subscriptions: new Set<RunId | null>(),
    };

    // Inject into RpcServer internals so broadcastEvent() can push to this client
    rpcInternals.connections.set(connId, connection);
    clientConnIds.add(connId);

    return {
      messages,
      call: async <T>(method: string, params?: JsonObject): Promise<T> => {
        const req = {
          type: 'rr_v3.request' as const,
          requestId: `e2e-req-${++requestSeq}`,
          method,
          ...(params ? { params } : {}),
        };
        return rpcInternals.handleRequest<T>(req, connection);
      },
      clearMessages: () => {
        messages.splice(0, messages.length);
      },
      getStreamedEvents: () => {
        return messages
          .filter(
            (m): m is { type: 'rr_v3.event'; event: RunEvent } =>
              typeof m === 'object' &&
              m !== null &&
              (m as { type?: string }).type === 'rr_v3.event',
          )
          .map((m) => m.event);
      },
    };
  }

  async function waitForEvent(
    runId: RunId,
    predicate: (event: RunEvent) => boolean,
    opts?: { timeoutMs?: number },
  ): Promise<RunEvent> {
    const timeoutMs = opts?.timeoutMs ?? 5_000;

    // Fast-path: Check for persisted events
    try {
      const existing = await storage.events.list(runId);
      const found = existing.find(predicate);
      if (found) return found;
    } catch {
      // ignore and fall back to subscription
    }

    return new Promise<RunEvent>((resolve, reject) => {
      const timer = setTimeout(() => {
        unsubscribe();
        reject(new Error(`Timed out waiting for event (runId=${runId})`));
      }, timeoutMs);

      const unsubscribe = events.subscribe(
        (event) => {
          if (!predicate(event)) return;
          clearTimeout(timer);
          unsubscribe();
          resolve(event);
        },
        { runId },
      );
    });
  }

  async function waitForTerminal(
    runId: RunId,
    opts?: { timeoutMs?: number },
  ): Promise<RunRecordV3> {
    const timeoutMs = opts?.timeoutMs ?? 10_000;

    // Check the current status first
    const initial = await storage.runs.get(runId);
    if (!initial) {
      throw new Error(`Run "${runId}" not found`);
    }
    if (isTerminalStatus(initial.status)) {
      return initial;
    }

    // Wait for final event
    await waitForEvent(
      runId,
      (e) =>
        e.type === 'run.succeeded' ||
        e.type === 'run.failed' ||
        e.type === 'run.canceled' ||
        e.type === 'run.stopped_at_boundary',
      { timeoutMs },
    );

    const done = await storage.runs.get(runId);
    if (!done) {
      throw new Error(`Run "${runId}" not found after terminal event`);
    }
    return done;
  }

  async function waitForQueueItemGone(runId: RunId, opts?: { timeoutMs?: number }): Promise<void> {
    const timeoutMs = opts?.timeoutMs ?? 5_000;
    const startedAt = Date.now();

    for (;;) {
      const item = await storage.queue.get(runId);
      if (!item) return;

      if (Date.now() - startedAt >= timeoutMs) {
        throw new Error(
          `Timed out waiting for queue item to be removed (runId=${runId}, status=${item.status})`,
        );
      }

      await sleep(10);
    }
  }

  async function listEvents(runId: RunId): Promise<RunEvent[]> {
    return storage.events.list(runId);
  }

  async function dispose(): Promise<void> {
    // Cancel event forwarding
    try {
      unsubscribeForward();
    } catch {
      // ignore
    }

    // Stop scheduler
    try {
      scheduler.dispose();
    } catch {
      // ignore
    }

    // Release lease manager
    try {
      leaseManager.dispose();
    } catch {
      // ignore
    }

    // Remove injected client connections
    for (const connId of clientConnIds) {
      rpcInternals.connections.delete(connId);
    }
    clientConnIds.clear();

    // Close IDB connection
    closeRrV3Db();
  }

  return {
    ownerId,
    storage,
    events,
    scheduler,
    runners,
    rpcServer,
    createClient,
    waitForEvent,
    waitForTerminal,
    waitForQueueItemGone,
    listEvents,
    dispose,
  };
}

// ==================== Internal Helpers ====================

/**
 * Create RunExecutor for E2E testing
 */
function createE2EExecutor(deps: {
  storage: StoragePort;
  events: EventsBus;
  runnerFactory: RunRunnerFactory;
  runners: RunnerRegistry;
  now: () => number;
  logger: Logger;
}): RunExecutor {
  return async (item: RunQueueItem): Promise<void> => {
    const runId = item.id;

    // 1. Get RunRecord
    const run = await deps.storage.runs.get(runId);
    if (!run) {
      deps.logger.warn(`[E2E] RunRecord not found for queue item "${runId}", skipping`);
      return;
    }

    // 2. GetFlow
    const flow = await deps.storage.flows.get(item.flowId);
    if (!flow) {
      await failRun(deps, runId, `Flow "${item.flowId}" not found`);
      return;
    }
    const expectedRevision = run.expectedRevision ?? item.expectedRevision;
    if (expectedRevision) {
      const currentRevision = calculateWorkflowRevision(flow);
      if (currentRevision !== expectedRevision) {
        const message = `Flow "${item.flowId}" changed before execution; expected revision ${expectedRevision}, current ${currentRevision}`;
        await failRun(
          deps,
          runId,
          message,
          createRRError(RR_ERROR_CODES.STALE_WORKFLOW_DESCRIPTOR, message, {
            retryable: true,
            data: {
              flowId: item.flowId,
              expectedRevision,
              currentRevision,
            },
          }),
        );
        return;
      }
    }

    // 3. Synchronize attempt/tabId to RunRecord
    const tabId = item.tabId ?? run.tabId ?? 1;
    try {
      await deps.storage.runs.patch(runId, {
        attempt: item.attempt,
        maxAttempts: item.maxAttempts,
        tabId,
      });
    } catch {
      // ignore
    }

    // 4. Create and run Runner
    const runner = deps.runnerFactory.create(runId, {
      flow,
      tabId,
      args: item.args ?? run.args,
      startNodeId: run.startNodeId,
      stopBeforeNodeId: run.stopBeforeNodeId,
      endNodeId: run.endNodeId,
      debug: item.debug ?? run.debug,
    });

    deps.runners.register(runId, runner);
    try {
      await runner.start();
    } finally {
      deps.runners.unregister(runId);
    }
  };
}

/**
 * Mark Run as failed
 */
async function failRun(
  deps: { storage: StoragePort; events: EventsBus; now: () => number },
  runId: RunId,
  message: string,
  errorOverride?: RRError,
): Promise<void> {
  const t = deps.now();
  const error = errorOverride ?? createRRError(RR_ERROR_CODES.VALIDATION_ERROR, message);

  await deps.storage.runs.patch(runId, {
    status: 'failed',
    finishedAt: t,
    tookMs: 0,
    error,
  });

  await deps.events.append({
    runId,
    type: 'run.failed',
    error,
  });
}
