/**
 * @fileoverview Parallel Scheduling Integration Test (P3-07)
 * @description
 * End-to-end tests for Scheduler + Queue + LeaseManager + Recovery
 * Uses real IndexedDB storage (fake-indexeddb) to verify integration.
 */

import { beforeEach, describe, expect, it, vi } from 'vitest';

import type { RunId } from '@/entrypoints/background/record-replay-v3/domain/ids';
import type { RunRecordV3 } from '@/entrypoints/background/record-replay-v3/domain/events';
import type { RunQueueItem } from '@/entrypoints/background/record-replay-v3/engine/queue/queue';
import { DEFAULT_QUEUE_CONFIG } from '@/entrypoints/background/record-replay-v3/engine/queue/queue';
import {
  createLeaseManager,
  generateOwnerId,
} from '@/entrypoints/background/record-replay-v3/engine/queue/leasing';
import {
  createRunScheduler,
  type RunExecutor,
} from '@/entrypoints/background/record-replay-v3/engine/queue/scheduler';
import { InMemoryKeepaliveController } from '@/entrypoints/background/record-replay-v3/engine/keepalive/offscreen-keepalive';
import {
  closeRrV3Db,
  deleteRrV3Db,
} from '@/entrypoints/background/record-replay-v3';
import { createQueueStore } from '@/entrypoints/background/record-replay-v3/storage/queue';
import { createRunsStore } from '@/entrypoints/background/record-replay-v3/storage/runs';
import { recoverFromCrash } from '@/entrypoints/background/record-replay-v3/engine/recovery/recovery-coordinator';

// ==================== Test Utilities ====================

function createMockEventsBus() {
  const events: unknown[] = [];
  return {
    subscribe: vi.fn(() => () => {}),
    append: vi.fn(async (event: unknown) => {
      const fullEvent = { ...(event as object), ts: Date.now(), seq: events.length + 1 };
      events.push(fullEvent);
      return fullEvent;
    }),
    list: vi.fn(async () => []),
    _events: events,
  };
}

function createMockStorage(
  queueStore: ReturnType<typeof createQueueStore>,
  runsStore: ReturnType<typeof createRunsStore>,
) {
  return {
    flows: {} as any,
    runs: runsStore,
    events: {} as any,
    queue: queueStore,
    persistentVars: {} as any,
    triggers: {} as any,
    artifacts: {} as any,
  };
}

function createRunRecord(id: string, status: string): RunRecordV3 {
  return {
    schemaVersion: 3,
    id: id as RunId,
    flowId: 'flow-1' as any,
    status: status as any,
    createdAt: Date.now(),
    updatedAt: Date.now(),
    attempt: 0,
    maxAttempts: 3,
    nextSeq: 0,
  };
}

// ==================== Integration Tests ====================

describe('V3 Scheduler Integration', () => {
  beforeEach(async () => {
    await deleteRrV3Db();
    closeRrV3Db();
  });

  describe('End-to-end scheduling', () => {
    it('scheduler claims from real queue, executes, and marks done', async () => {
      const queue = createQueueStore();
      const keepalive = new InMemoryKeepaliveController();
      const leaseManager = createLeaseManager(queue, DEFAULT_QUEUE_CONFIG);
      const ownerId = generateOwnerId();

      const executed: string[] = [];
      const executor: RunExecutor = async (item) => {
        executed.push(item.id);
        // Simulate short execution
        await new Promise((resolve) => setTimeout(resolve, 10));
      };

      // Enqueue items
      await queue.enqueue({ id: 'run-1' as any, flowId: 'flow-1' as any, priority: 0 });
      await queue.enqueue({ id: 'run-2' as any, flowId: 'flow-1' as any, priority: 0 });

      const scheduler = createRunScheduler({
        queue,
        leaseManager,
        keepalive,
        config: { ...DEFAULT_QUEUE_CONFIG, maxParallelRuns: 1 },
        ownerId,
        execute: executor,
        tuning: { pollIntervalMs: 0, reclaimIntervalMs: 0 },
      });

      scheduler.start();

      // Wait for execution
      await new Promise((resolve) => setTimeout(resolve, 100));

      scheduler.stop();

      // Both runs should be executed
      expect(executed).toContain('run-1');
      expect(executed).toContain('run-2');

      // Queue should be empty
      const remaining = await queue.list();
      expect(remaining).toHaveLength(0);
    });

    it('respects maxParallelRuns with real queue', async () => {
      const queue = createQueueStore();
      const keepalive = new InMemoryKeepaliveController();
      const leaseManager = createLeaseManager(queue, DEFAULT_QUEUE_CONFIG);
      const ownerId = generateOwnerId();

      let concurrentCount = 0;
      let maxConcurrent = 0;
      const executionTimes: Map<string, { start: number; end?: number }> = new Map();

      const executor: RunExecutor = async (item) => {
        concurrentCount++;
        maxConcurrent = Math.max(maxConcurrent, concurrentCount);
        executionTimes.set(item.id, { start: Date.now() });

        await new Promise((resolve) => setTimeout(resolve, 50));

        executionTimes.get(item.id)!.end = Date.now();
        concurrentCount--;
      };

      // Enqueue 5 items
      for (let i = 0; i < 5; i++) {
        await queue.enqueue({ id: `run-${i}` as any, flowId: 'flow-1' as any, priority: 0 });
      }

      const scheduler = createRunScheduler({
        queue,
        leaseManager,
        keepalive,
        config: { ...DEFAULT_QUEUE_CONFIG, maxParallelRuns: 2 },
        ownerId,
        execute: executor,
        tuning: { pollIntervalMs: 10, reclaimIntervalMs: 0 },
      });

      scheduler.start();

      // Wait for all executions
      await new Promise((resolve) => setTimeout(resolve, 500));

      scheduler.stop();

      // Max concurrent should not exceed 2
      expect(maxConcurrent).toBeLessThanOrEqual(2);

      // All runs should complete
      expect(executionTimes.size).toBe(5);
    });

    it('maintains FIFO within same priority', async () => {
      const queue = createQueueStore();
      const keepalive = new InMemoryKeepaliveController();
      const leaseManager = createLeaseManager(queue, DEFAULT_QUEUE_CONFIG);
      const ownerId = generateOwnerId();

      const executionOrder: string[] = [];
      const executor: RunExecutor = async (item) => {
        executionOrder.push(item.id);
        await new Promise((resolve) => setTimeout(resolve, 10));
      };

      // Enqueue in order with same priority
      await queue.enqueue({ id: 'run-1' as any, flowId: 'flow-1' as any, priority: 0 });
      await new Promise((resolve) => setTimeout(resolve, 5)); // Ensure different createdAt
      await queue.enqueue({ id: 'run-2' as any, flowId: 'flow-1' as any, priority: 0 });
      await new Promise((resolve) => setTimeout(resolve, 5));
      await queue.enqueue({ id: 'run-3' as any, flowId: 'flow-1' as any, priority: 0 });

      const scheduler = createRunScheduler({
        queue,
        leaseManager,
        keepalive,
        config: { ...DEFAULT_QUEUE_CONFIG, maxParallelRuns: 1 }, // Serial execution
        ownerId,
        execute: executor,
        tuning: { pollIntervalMs: 0, reclaimIntervalMs: 0 },
      });

      scheduler.start();

      await new Promise((resolve) => setTimeout(resolve, 200));

      scheduler.stop();

      // Should execute in FIFO order
      expect(executionOrder).toEqual(['run-1', 'run-2', 'run-3']);
    });

    it('higher priority runs first', async () => {
      const queue = createQueueStore();
      const keepalive = new InMemoryKeepaliveController();
      const leaseManager = createLeaseManager(queue, DEFAULT_QUEUE_CONFIG);
      const ownerId = generateOwnerId();

      const executionOrder: string[] = [];
      const executor: RunExecutor = async (item) => {
        executionOrder.push(item.id);
        await new Promise((resolve) => setTimeout(resolve, 10));
      };

      // Enqueue with different priorities (low first)
      await queue.enqueue({ id: 'run-low' as any, flowId: 'flow-1' as any, priority: 0 });
      await queue.enqueue({ id: 'run-high' as any, flowId: 'flow-1' as any, priority: 10 });
      await queue.enqueue({ id: 'run-medium' as any, flowId: 'flow-1' as any, priority: 5 });

      const scheduler = createRunScheduler({
        queue,
        leaseManager,
        keepalive,
        config: { ...DEFAULT_QUEUE_CONFIG, maxParallelRuns: 1 },
        ownerId,
        execute: executor,
        tuning: { pollIntervalMs: 0, reclaimIntervalMs: 0 },
      });

      scheduler.start();

      await new Promise((resolve) => setTimeout(resolve, 200));

      scheduler.stop();

      // Should execute in priority order (high -> medium -> low)
      expect(executionOrder).toEqual(['run-high', 'run-medium', 'run-low']);
    });
  });

  describe('Lease management', () => {
    it('heartbeat keeps leases alive during long runs', async () => {
      const queue = createQueueStore();
      const keepalive = new InMemoryKeepaliveController();
      const config = {
        ...DEFAULT_QUEUE_CONFIG,
        leaseTtlMs: 100, // Short TTL for testing
        heartbeatIntervalMs: 30, // Frequent heartbeat
      };
      const leaseManager = createLeaseManager(queue, config);
      const ownerId = generateOwnerId();

      let runningItem: RunQueueItem | null = null;
      const executor: RunExecutor = async (item) => {
        runningItem = item;
        // Run longer than TTL
        await new Promise((resolve) => setTimeout(resolve, 200));
      };

      await queue.enqueue({ id: 'long-run' as any, flowId: 'flow-1' as any });

      const scheduler = createRunScheduler({
        queue,
        leaseManager,
        keepalive,
        config: { ...config, maxParallelRuns: 1 },
        ownerId,
        execute: executor,
        tuning: { pollIntervalMs: 0, reclaimIntervalMs: 0 },
      });

      scheduler.start();

      // Wait for run to be claimed
      await new Promise((resolve) => setTimeout(resolve, 50));
      expect(runningItem).not.toBeNull();

      // Check that lease is being renewed
      const itemMidRun = await queue.get('long-run' as any);
      expect(itemMidRun?.status).toBe('running');
      expect(itemMidRun?.lease?.ownerId).toBe(ownerId);

      // Wait for completion
      await new Promise((resolve) => setTimeout(resolve, 200));

      scheduler.stop();

      // Run should complete successfully
      const remaining = await queue.list();
      expect(remaining).toHaveLength(0);
    });

    it('expired leases are reclaimed by periodic scan', async () => {
      const queue = createQueueStore();
      const keepalive = new InMemoryKeepaliveController();
      const leaseManager = createLeaseManager(queue, DEFAULT_QUEUE_CONFIG);

      // Note: markRunning uses DEFAULT_LEASE_TTL_MS (15s) internally.
      // To simulate an expired lease, we pass a past time that makes
      // expiresAt (pastTime + 15s) be in the past relative to now.
      const pastTime = Date.now() - DEFAULT_QUEUE_CONFIG.leaseTtlMs - 100; // expired 100ms ago
      await queue.enqueue({ id: 'orphan-run' as any, flowId: 'flow-1' as any });
      await queue.markRunning('orphan-run' as any, 'dead-owner', pastTime);

      // Verify lease exists and is expired
      const expiredItem = await queue.get('orphan-run' as any);
      expect(expiredItem?.status).toBe('running');
      expect(expiredItem?.lease?.expiresAt).toBe(pastTime + DEFAULT_QUEUE_CONFIG.leaseTtlMs);
      expect(expiredItem?.lease?.expiresAt).toBeLessThan(Date.now());

      // Manually trigger reclaim to simulate what scheduler does periodically
      const reclaimedIds = await leaseManager.reclaimExpiredLeases(Date.now());
      expect(reclaimedIds).toContain('orphan-run');

      // Now queue item should be back to queued
      const reclaimedItem = await queue.get('orphan-run' as any);
      expect(reclaimedItem?.status).toBe('queued');

      // New scheduler should pick it up
      const ownerId = generateOwnerId();
      const executed: string[] = [];
      const executor: RunExecutor = async (item) => {
        executed.push(item.id);
      };

      const scheduler = createRunScheduler({
        queue,
        leaseManager,
        keepalive,
        config: { ...DEFAULT_QUEUE_CONFIG, maxParallelRuns: 1 },
        ownerId,
        execute: executor,
        tuning: { pollIntervalMs: 0, reclaimIntervalMs: 0 },
      });

      scheduler.start();

      await new Promise((resolve) => setTimeout(resolve, 100));

      scheduler.stop();

      // Orphan run should be executed
      expect(executed).toContain('orphan-run');
    });
  });

  describe('Crash recovery simulation', () => {
    it('recovers orphan running items after restart', async () => {
      const queue = createQueueStore();
      const runsStore = createRunsStore();
      const events = createMockEventsBus();

      // Simulate crash scenario: run was running when SW died
      await queue.enqueue({ id: 'crashed-run' as any, flowId: 'flow-1' as any, maxAttempts: 2 });
      await queue.markRunning('crashed-run' as any, 'old-sw-owner', Date.now());
      await runsStore.save({
        ...createRunRecord('crashed-run', 'running'),
        attempt: 1,
        maxAttempts: 2,
      });

      // Simulate restart with new owner
      const newOwnerId = generateOwnerId();
      const storage = createMockStorage(queue, runsStore);

      const result = await recoverFromCrash({
        storage,
        events: events as any,
        ownerId: newOwnerId,
        now: () => Date.now(),
      });

      // Run should be requeued
      expect(result.requeuedRunning).toContain('crashed-run');

      // Queue item should be back to queued
      const item = await queue.get('crashed-run' as any);
      expect(item?.status).toBe('queued');
      expect(item?.lease).toBeUndefined();

      // RunRecord should be updated
      const run = await runsStore.get('crashed-run' as any);
      expect(run?.status).toBe('queued');

      // Event should be emitted
      expect(events._events.some((e: any) => e.type === 'run.recovered')).toBe(true);
    });

    it('adopts orphan paused items after restart', async () => {
      const queue = createQueueStore();
      const runsStore = createRunsStore();
      const events = createMockEventsBus();

      // Simulate crash scenario: run was paused when SW died
      await queue.enqueue({ id: 'paused-run' as any, flowId: 'flow-1' as any });
      await queue.markPaused('paused-run' as any, 'old-sw-owner', Date.now());
      await runsStore.save(createRunRecord('paused-run', 'paused'));

      // Simulate restart with new owner
      const newOwnerId = generateOwnerId();
      const storage = createMockStorage(queue, runsStore);

      const result = await recoverFromCrash({
        storage,
        events: events as any,
        ownerId: newOwnerId,
        now: () => Date.now(),
      });

      // Run should be adopted (stays paused)
      expect(result.adoptedPaused).toContain('paused-run');

      // Queue item should still be paused with new owner
      const item = await queue.get('paused-run' as any);
      expect(item?.status).toBe('paused');
      expect(item?.lease?.ownerId).toBe(newOwnerId);
    });

    it('preserves attempt count across recovery', async () => {
      const queue = createQueueStore();
      const runsStore = createRunsStore();
      const events = createMockEventsBus();

      // Simulate a run that has already been attempted
      await queue.enqueue({ id: 'retried-run' as any, flowId: 'flow-1' as any, maxAttempts: 2 });
      await queue.claimNext('old-owner', Date.now()); // attempt becomes 1
      await runsStore.save({
        ...createRunRecord('retried-run', 'running'),
        attempt: 1,
        maxAttempts: 2,
      });

      // Simulate restart
      const newOwnerId = generateOwnerId();
      const storage = createMockStorage(queue, runsStore);

      await recoverFromCrash({
        storage,
        events: events as any,
        ownerId: newOwnerId,
        now: () => Date.now(),
      });

      // Queue item should preserve attempt count
      const item = await queue.get('retried-run' as any);
      expect(item?.status).toBe('queued');
      expect(item?.attempt).toBe(1); // Not reset

      // Next claim will increment
      const claimed = await queue.claimNext(newOwnerId, Date.now());
      expect(claimed?.attempt).toBe(2);
    });

    it('cleans terminal runs left in queue due to crash', async () => {
      const queue = createQueueStore();
      const runsStore = createRunsStore();
      const events = createMockEventsBus();

      // Simulate crash scenario: run completed but queue item wasn't removed
      await queue.enqueue({ id: 'completed-run' as any, flowId: 'flow-1' as any });
      await queue.markRunning('completed-run' as any, 'old-owner', Date.now());
      await runsStore.save(createRunRecord('completed-run', 'succeeded'));

      // Simulate restart
      const newOwnerId = generateOwnerId();
      const storage = createMockStorage(queue, runsStore);

      const result = await recoverFromCrash({
        storage,
        events: events as any,
        ownerId: newOwnerId,
        now: () => Date.now(),
      });

      // Run should be cleaned
      expect(result.cleanedTerminal).toContain('completed-run');

      // Queue should be empty
      const remaining = await queue.list();
      expect(remaining).toHaveLength(0);
    });

    it('recovery then scheduler works correctly', async () => {
      const queue = createQueueStore();
      const runsStore = createRunsStore();
      const events = createMockEventsBus();
      const keepalive = new InMemoryKeepaliveController();

      // Simulate crash scenario
      await queue.enqueue({ id: 'recover-run' as any, flowId: 'flow-1' as any, maxAttempts: 2 });
      await queue.markRunning('recover-run' as any, 'old-owner', Date.now());
      await runsStore.save({
        ...createRunRecord('recover-run', 'running'),
        attempt: 1,
        maxAttempts: 2,
      });

      // Recovery
      const newOwnerId = generateOwnerId();
      const storage = createMockStorage(queue, runsStore);

      await recoverFromCrash({
        storage,
        events: events as any,
        ownerId: newOwnerId,
        now: () => Date.now(),
      });

      // Now start scheduler
      const leaseManager = createLeaseManager(queue, DEFAULT_QUEUE_CONFIG);
      const executed: string[] = [];
      const executor: RunExecutor = async (item) => {
        executed.push(item.id);
      };

      const scheduler = createRunScheduler({
        queue,
        leaseManager,
        keepalive,
        config: { ...DEFAULT_QUEUE_CONFIG, maxParallelRuns: 1 },
        ownerId: newOwnerId,
        execute: executor,
        tuning: { pollIntervalMs: 0, reclaimIntervalMs: 0 },
      });

      scheduler.start();
      await new Promise((resolve) => setTimeout(resolve, 100));
      scheduler.stop();

      // Recovered run should be executed
      expect(executed).toContain('recover-run');
    });
  });

  describe('Concurrency', () => {
    it('handles multiple concurrent enqueue/claim cycles', async () => {
      const queue = createQueueStore();
      const keepalive = new InMemoryKeepaliveController();
      const leaseManager = createLeaseManager(queue, DEFAULT_QUEUE_CONFIG);
      const ownerId = generateOwnerId();

      const executed = new Set<string>();
      const executor: RunExecutor = async (item) => {
        executed.add(item.id);
        await new Promise((resolve) => setTimeout(resolve, 20));
      };

      const scheduler = createRunScheduler({
        queue,
        leaseManager,
        keepalive,
        config: { ...DEFAULT_QUEUE_CONFIG, maxParallelRuns: 3 },
        ownerId,
        execute: executor,
        tuning: { pollIntervalMs: 10, reclaimIntervalMs: 0 },
      });

      scheduler.start();

      // Concurrent enqueues while scheduler is running
      const enqueuePromises = [];
      for (let i = 0; i < 10; i++) {
        enqueuePromises.push(
          queue
            .enqueue({
              id: `run-${i}` as any,
              flowId: 'flow-1' as any,
              priority: Math.random() * 10,
            })
            .then(() => scheduler.kick()),
        );
      }

      await Promise.all(enqueuePromises);

      // Wait for all to complete
      await new Promise((resolve) => setTimeout(resolve, 500));

      scheduler.stop();

      // All runs should be executed exactly once
      expect(executed.size).toBe(10);
    });

    it('no double execution under concurrent kicks', async () => {
      const queue = createQueueStore();
      const keepalive = new InMemoryKeepaliveController();
      const leaseManager = createLeaseManager(queue, DEFAULT_QUEUE_CONFIG);
      const ownerId = generateOwnerId();

      const executionCounts = new Map<string, number>();
      const executor: RunExecutor = async (item) => {
        executionCounts.set(item.id, (executionCounts.get(item.id) ?? 0) + 1);
        await new Promise((resolve) => setTimeout(resolve, 50));
      };

      // Pre-enqueue
      for (let i = 0; i < 5; i++) {
        await queue.enqueue({ id: `run-${i}` as any, flowId: 'flow-1' as any });
      }

      const scheduler = createRunScheduler({
        queue,
        leaseManager,
        keepalive,
        config: { ...DEFAULT_QUEUE_CONFIG, maxParallelRuns: 2 },
        ownerId,
        execute: executor,
        tuning: { pollIntervalMs: 0, reclaimIntervalMs: 0 },
      });

      scheduler.start();

      // Hammer with concurrent kicks
      const kickPromises = [];
      for (let i = 0; i < 20; i++) {
        kickPromises.push(scheduler.kick());
      }
      await Promise.all(kickPromises);

      // Wait for completion
      await new Promise((resolve) => setTimeout(resolve, 500));

      scheduler.stop();

      // Each run should execute exactly once
      for (const [runId, count] of executionCounts) {
        expect(count, `${runId} executed ${count} times`).toBe(1);
      }
    });
  });
});
