import { randomUUID } from 'node:crypto';
import { spawn } from 'node:child_process';
import { cp, mkdtemp, mkdir, readFile, rm, writeFile } from 'node:fs/promises';
import { dirname, join } from 'node:path';
import type { SpawnOptionsWithoutStdio } from 'node:child_process';
import {
  setSpanAttributes,
  withActiveSpan,
} from '@shared_libs/observability/tracing';
import type { PlayRunnerBackend } from '../types';
import { buildPlayRunnerBundle } from '../bundle';
import {
  findPlayRunnerResult,
  parsePlayRunnerEventLine,
  parsePlayRunnerEvents,
} from '../runner-events';
import type { PlayRunnerExecutionConfig } from '@shared_libs/play-runtime/protocol';
import { resolveDaytonaSandboxComputeItem } from '@shared_libs/play-runtime/worker-api-types';
import {
  recordComputeBillingItemViaAppRuntime,
  upsertComputeBillingSessionViaAppRuntime,
  type WorkerRuntimeApiContext,
} from '@shared_libs/play-runtime/app-runtime-api';
import {
  PLAY_RUNNER_STARTUP_GRACE_SECONDS,
  PLAY_RUNNER_TERMINAL_GRACE_SECONDS,
} from '@shared_libs/play-runtime/runtime-constants';
import {
  STANDARD_PLAY_SANDBOX_RUNTIME_LIMITS,
  validatePlaySandboxRuntimeLimits,
} from '@shared_libs/play-runtime/sandbox-runtime-limits';
import {
  runtimeLimitExceededCause,
  runtimeLimitExceededFailure,
} from '@shared_libs/play-runtime/run-failure';
import {
  buildDaytonaRuntimeWatchdogSource,
  DAYTONA_RUNTIME_WATCHDOG_EXIT_CODE,
} from './daytona-runtime-watchdog';

interface CommandExecutionResult {
  exitCode: number;
  stdout: string;
  stderr: string;
}

type StreamedCommandCallbacks = {
  onStdoutLine?: (line: string) => void;
  /** Called once when the result event is found — avoids storing full stdout. */
  onResultLine?: (line: string) => void;
  cancellationSignal?: AbortSignal;
};

const LOCAL_PROCESS_COMPUTE_SOURCE = 'local_process';
const LOCAL_PROCESS_COMPUTE_CPU = 1;
const LOCAL_PROCESS_COMPUTE_MEMORY_GIB = 1;
const LOCAL_PROCESS_COMPUTE_DISK_GIB = 3;
export const LOCAL_PROCESS_CANCEL_GRACE_MS = 2_000;
const LOCAL_PROCESS_CANCELLED_ERROR = 'Local play runner cancelled';

export const LOCAL_PROCESS_COMPUTE_PROFILE = {
  source: LOCAL_PROCESS_COMPUTE_SOURCE,
  cpu: LOCAL_PROCESS_COMPUTE_CPU,
  memoryGiB: LOCAL_PROCESS_COMPUTE_MEMORY_GIB,
  diskGiB: LOCAL_PROCESS_COMPUTE_DISK_GIB,
} as const;

function logLocalRunnerPerf(
  workflowId: string | null | undefined,
  phase: string,
  startedAt: number,
  extra: Record<string, unknown> = {},
) {
  console.info('[perf][play-runner.local_process]', {
    workflowId,
    phase,
    ms: Date.now() - startedAt,
    ...extra,
  });
}

async function runCommand(
  command: string,
  args: string[],
  options?: SpawnOptionsWithoutStdio,
  callbacks?: StreamedCommandCallbacks,
): Promise<CommandExecutionResult> {
  return await new Promise((resolve, reject) => {
    if (callbacks?.cancellationSignal?.aborted) {
      reject(new Error(LOCAL_PROCESS_CANCELLED_ERROR));
      return;
    }

    // POSIX: run the command as its own process-group leader so cancellation
    // can signal the entire tree. The watchdog child spawns the runner (which
    // may spawn further descendants) with inherited stdio; killing only the
    // direct child leaves orphans holding our pipes and 'close' never fires.
    const useProcessGroup = process.platform !== 'win32';
    const child = spawn(command, args, {
      stdio: ['ignore', 'pipe', 'pipe'],
      detached: useProcessGroup,
      ...options,
    });

    const killTree = (signal: NodeJS.Signals) => {
      if (useProcessGroup && child.pid) {
        try {
          process.kill(-child.pid, signal);
          return;
        } catch {
          // Group already gone or not yet created — fall through.
        }
      }
      child.kill(signal);
    };

    let stderr = '';
    let stdoutBuffer = '';
    let stdoutLineCount = 0;
    // Only keep the last line that looks like a result event (avoids storing full stdout).
    let resultLine = '';
    let settled = false;
    let cancellationRequested = false;
    let cancellationErrorMessage = LOCAL_PROCESS_CANCELLED_ERROR;
    let cancelKillTimer: NodeJS.Timeout | null = null;

    const cleanupCancellation = () => {
      callbacks?.cancellationSignal?.removeEventListener('abort', onCancel);
      if (cancelKillTimer) {
        clearTimeout(cancelKillTimer);
        cancelKillTimer = null;
      }
    };

    const settleRejected = (error: unknown) => {
      if (settled) {
        return;
      }
      settled = true;
      cleanupCancellation();
      reject(error);
    };

    const settleResolved = (result: CommandExecutionResult) => {
      if (settled) {
        return;
      }
      settled = true;
      cleanupCancellation();
      resolve(result);
    };

    function onCancel() {
      if (settled) {
        return;
      }
      cancellationRequested = true;
      const reason = callbacks?.cancellationSignal?.reason;
      cancellationErrorMessage =
        typeof reason === 'string' && reason.trim()
          ? reason.trim()
          : LOCAL_PROCESS_CANCELLED_ERROR;
      killTree('SIGTERM');
      cancelKillTimer = setTimeout(() => {
        // Kill the whole group even if the direct child already exited: a
        // descendant may have survived it and still hold our stdio pipes.
        killTree('SIGKILL');
        // Backstop: release our read ends so 'close' can fire even if some
        // descendant escaped the group (e.g. via setsid) and never dies.
        child.stdout?.destroy();
        child.stderr?.destroy();
      }, LOCAL_PROCESS_CANCEL_GRACE_MS);
      cancelKillTimer.unref?.();
    }

    callbacks?.cancellationSignal?.addEventListener('abort', onCancel, {
      once: true,
    });

    child.stdout?.on('data', (chunk) => {
      const text = String(chunk);
      stdoutBuffer += text;
      const lines = stdoutBuffer.split(/\r?\n/);
      stdoutBuffer = lines.pop() ?? '';
      for (const line of lines) {
        const trimmed = line.trim();
        if (!trimmed) {
          continue;
        }
        stdoutLineCount++;
        // Capture result lines so we can extract the result without storing all stdout.
        if (trimmed.includes('"type":"result"')) {
          resultLine = trimmed;
        }
        callbacks?.onStdoutLine?.(trimmed);
      }
    });
    child.stdout?.on('error', () => {
      // Absorb pipe errors — we still get partial stdout
    });
    child.stderr?.on('data', (chunk) => {
      stderr += String(chunk);
    });
    child.stderr?.on('error', () => {});

    child.on('error', settleRejected);
    child.on('close', (code) => {
      const trailing = stdoutBuffer.trim();
      if (trailing) {
        stdoutLineCount++;
        if (trailing.includes('"type":"result"')) {
          resultLine = trailing;
        }
        callbacks?.onStdoutLine?.(trailing);
      }
      if (cancellationRequested) {
        settleRejected(new Error(cancellationErrorMessage));
        return;
      }
      settleResolved({
        exitCode: code ?? 0,
        stdout: resultLine,
        stderr,
      });
    });
  });
}

export function resolveLocalProcessRuntimeLimitSeconds(
  context: PlayRunnerExecutionConfig['context'],
): number {
  return validatePlaySandboxRuntimeLimits(
    context.sandboxRuntimeLimits ?? {
      ...STANDARD_PLAY_SANDBOX_RUNTIME_LIMITS,
    },
  ).timeoutSeconds;
}

function createFailedResult(config: PlayRunnerExecutionConfig, error: string) {
  return {
    status: 'failed' as const,
    error,
    logs: [],
    stats: {},
    steps: [],
    checkpoint: config.checkpoint ?? null,
    tableNamespace: null,
  };
}

function runtimeApiContextForConfig(
  config: PlayRunnerExecutionConfig,
): WorkerRuntimeApiContext {
  const baseUrl = config.context.baseUrl?.trim();
  const executorToken = config.context.executorToken?.trim();
  if (!baseUrl || !executorToken) {
    throw new Error(
      'Missing runtime API context for local-process compute billing.',
    );
  }
  return { baseUrl, executorToken };
}

async function prepareLocalProcessWorkspace(
  config: PlayRunnerExecutionConfig,
): Promise<{
  workspaceRoot: string;
  runnerPath: string;
  configPath: string;
  localConfig: PlayRunnerExecutionConfig;
}> {
  const workspaceRoot = await mkdtemp(
    join(config.workspaceRoot, '.deepline-local-play-runner-'),
  );
  const filesRoot = join(workspaceRoot, 'files');
  await mkdir(filesRoot, { recursive: true });

  const localMaterializedFiles: Record<string, string> = {};
  const copiedFiles = new Map<string, string>();

  for (const [logicalPath, localPath] of Object.entries(
    config.materializedFiles,
  )) {
    if (!copiedFiles.has(localPath)) {
      const targetPath = join(filesRoot, randomUUID());
      await mkdir(dirname(targetPath), { recursive: true });
      await cp(localPath, targetPath, { force: true });
      copiedFiles.set(localPath, targetPath);
    }
    localMaterializedFiles[logicalPath] = copiedFiles.get(localPath)!;
  }

  let localCsvSourcePath: string | null = null;
  if (config.csvSourcePath) {
    if (!copiedFiles.has(config.csvSourcePath)) {
      const targetPath = join(filesRoot, randomUUID());
      await mkdir(dirname(targetPath), { recursive: true });
      await cp(config.csvSourcePath, targetPath, { force: true });
      copiedFiles.set(config.csvSourcePath, targetPath);
    }
    localCsvSourcePath = copiedFiles.get(config.csvSourcePath)!;
  }

  const runnerPath = join(workspaceRoot, 'play-runner.cjs');
  const configPath = join(workspaceRoot, 'runner-config.json');
  const localConfig: PlayRunnerExecutionConfig = {
    ...config,
    workspaceRoot,
    csvSourcePath: localCsvSourcePath,
    materializedFiles: localMaterializedFiles,
  };

  return {
    workspaceRoot,
    runnerPath,
    configPath,
    localConfig,
  };
}

export const localProcessPlayRunnerBackend: PlayRunnerBackend = {
  async execute(config, callbacks) {
    return await withActiveSpan(
      'plays.runner.local_process',
      {
        tracer: 'deepline.plays',
        attributes: {
          'plays.workflow_id': config.context.workflowId,
          'plays.play_name': config.context.playName,
          'plays.has_csv_source': Boolean(config.csvSourcePath),
          'plays.materialized_file_count': Object.keys(config.materializedFiles)
            .length,
        },
      },
      async (backendSpan) => {
        const localComputeStartedAt = Date.now();
        await upsertComputeBillingSessionViaAppRuntime(
          runtimeApiContextForConfig(config),
          {
            sessionId:
              config.context.workflowId ?? config.context.runId ?? 'unknown',
            orgId: config.context.orgId ?? 'unknown',
            userId: null,
            operation: 'workflow_run',
            workflowId: config.context.workflowId ?? undefined,
            runId: config.context.runId ?? undefined,
          },
        );
        let phaseStartedAt = Date.now();
        const bundle = await withActiveSpan(
          'plays.runner.bundle_ready',
          {
            tracer: 'deepline.plays',
            attributes: {
              'plays.workflow_id': config.context.workflowId,
              'plays.play_name': config.context.playName,
            },
          },
          async () => await buildPlayRunnerBundle(),
        );
        logLocalRunnerPerf(
          config.context.workflowId,
          'bundle_ready',
          localComputeStartedAt,
          { bundleBytes: bundle.length },
        );
        phaseStartedAt = Date.now();
        const { workspaceRoot, runnerPath, configPath, localConfig } =
          await withActiveSpan(
            'plays.runner.prepare_workspace',
            {
              tracer: 'deepline.plays',
              attributes: {
                'plays.workflow_id': config.context.workflowId,
                'plays.play_name': config.context.playName,
              },
            },
            async (span) => {
              const prepared = await prepareLocalProcessWorkspace(config);
              setSpanAttributes(span, {
                'plays.workspace_materialized_file_count': Object.keys(
                  prepared.localConfig.materializedFiles,
                ).length,
                'plays.workspace_has_csv_source': Boolean(
                  prepared.localConfig.csvSourcePath,
                ),
              });
              return prepared;
            },
          );
        logLocalRunnerPerf(
          config.context.workflowId,
          'prepare_workspace',
          phaseStartedAt,
        );

        try {
          phaseStartedAt = Date.now();
          await withActiveSpan(
            'plays.runner.write_files',
            {
              tracer: 'deepline.plays',
              attributes: {
                'plays.workflow_id': config.context.workflowId,
                'plays.play_name': config.context.playName,
              },
            },
            async () => {
              await writeFile(runnerPath, bundle, 'utf-8');
              await writeFile(configPath, JSON.stringify(localConfig), 'utf-8');
            },
          );
          logLocalRunnerPerf(
            config.context.workflowId,
            'write_files',
            phaseStartedAt,
          );

          phaseStartedAt = Date.now();
          const execution = await withActiveSpan(
            'plays.runner.child_process',
            {
              tracer: 'deepline.plays',
              attributes: {
                'plays.workflow_id': config.context.workflowId,
                'plays.play_name': config.context.playName,
              },
            },
            async (span) => {
              const runtimeLimitSeconds =
                resolveLocalProcessRuntimeLimitSeconds(config.context);
              const runtimeStartedPath = join(
                workspaceRoot,
                'runtime-started.json',
              );
              const runtimeCompletedPath = join(
                workspaceRoot,
                'runtime-completed.json',
              );
              const runtimeLimitMarkerPath = join(
                workspaceRoot,
                'runtime-limit.marker',
              );
              const result = await runCommand(
                'node',
                [
                  '-e',
                  buildDaytonaRuntimeWatchdogSource(),
                  runnerPath,
                  configPath,
                  runtimeStartedPath,
                  runtimeCompletedPath,
                  String(runtimeLimitSeconds * 1_000),
                  String(PLAY_RUNNER_STARTUP_GRACE_SECONDS * 1_000),
                  String(PLAY_RUNNER_TERMINAL_GRACE_SECONDS * 1_000),
                  runtimeLimitMarkerPath,
                ],
                {
                  cwd: workspaceRoot,
                  env: {
                    ...process.env,
                    PWD: workspaceRoot,
                    TMPDIR: workspaceRoot,
                    TEMP: workspaceRoot,
                    TMP: workspaceRoot,
                    DEEPLINE_PLAY_RUNNER_RUNTIME_STARTED_PATH:
                      runtimeStartedPath,
                    DEEPLINE_PLAY_RUNNER_RUNTIME_COMPLETED_PATH:
                      runtimeCompletedPath,
                  },
                },
                {
                  cancellationSignal: callbacks?.cancellationSignal,
                  onStdoutLine: (line) => {
                    const event = parsePlayRunnerEventLine(line);
                    if (!event) {
                      return;
                    }
                    if (event.type === 'log') {
                      callbacks?.onLog?.(event);
                      return;
                    }
                    if (event.type === 'checkpoint') {
                      callbacks?.onCheckpoint?.(event.checkpoint);
                      return;
                    }
                    if (event.type === 'row_update') {
                      callbacks?.onRowUpdate?.(event.update);
                      return;
                    }
                    if (event.type === 'execution_event') {
                      callbacks?.onExecutionEvent?.(event.event);
                    }
                  },
                },
              );
              const runtimeLimitReached =
                result.exitCode === DAYTONA_RUNTIME_WATCHDOG_EXIT_CODE &&
                (await readFile(runtimeLimitMarkerPath, 'utf-8')
                  .then((value) => value.trim() === 'runtime_limit')
                  .catch(() => false));
              setSpanAttributes(span, {
                'plays.child_exit_code': result.exitCode,
                'plays.stdout_bytes': result.stdout.length,
                'plays.stderr_bytes': result.stderr.length,
                'plays.runtime_limit_reached': runtimeLimitReached,
              });
              return { ...result, runtimeLimitReached, runtimeLimitSeconds };
            },
          );
          logLocalRunnerPerf(
            config.context.workflowId,
            'child_process',
            phaseStartedAt,
            {
              exitCode: execution.exitCode,
              stdoutBytes: execution.stdout.length,
              stderrBytes: execution.stderr.length,
            },
          );

          phaseStartedAt = Date.now();
          const events = await withActiveSpan(
            'plays.runner.parse_events',
            {
              tracer: 'deepline.plays',
              attributes: {
                'plays.workflow_id': config.context.workflowId,
                'plays.play_name': config.context.playName,
              },
            },
            async (span) => {
              const parsed = parsePlayRunnerEvents(execution.stdout);
              setSpanAttributes(span, {
                'plays.event_count': parsed.length,
              });
              return parsed;
            },
          );
          logLocalRunnerPerf(
            config.context.workflowId,
            'parse_events',
            phaseStartedAt,
            { eventCount: events.length },
          );
          const result = findPlayRunnerResult(events);
          if (result) {
            setSpanAttributes(backendSpan, {
              'plays.runner_status': result.status,
            });
            return result;
          }

          if (execution.runtimeLimitReached) {
            const error = runtimeLimitExceededCause(
              execution.runtimeLimitSeconds,
            );
            return {
              ...createFailedResult(config, error),
              errors: [
                runtimeLimitExceededFailure(
                  execution.runtimeLimitSeconds,
                  error,
                ),
              ],
            };
          }

          return createFailedResult(
            config,
            execution.exitCode !== 0
              ? `Local play runner exited with code ${execution.exitCode}.${execution.stderr ? ` ${execution.stderr.trim()}` : ''}`.trim()
              : 'Local play runner produced no result.',
          );
        } catch (error) {
          return createFailedResult(
            config,
            error instanceof Error ? error.message : String(error),
          );
        } finally {
          const localComputeEndedAt = Date.now();
          try {
            await recordComputeBillingItemViaAppRuntime(
              runtimeApiContextForConfig(config),
              {
                sessionId:
                  config.context.workflowId ??
                  config.context.runId ??
                  workspaceRoot,
                orgId: config.context.orgId ?? 'unknown',
                userId: null,
                operation: 'workflow_run',
                item: resolveDaytonaSandboxComputeItem({
                  itemId: `${LOCAL_PROCESS_COMPUTE_SOURCE}:${config.context.workflowId ?? workspaceRoot}`,
                  source: LOCAL_PROCESS_COMPUTE_SOURCE,
                  sandboxId: workspaceRoot,
                  wallTimeSeconds:
                    (localComputeEndedAt - localComputeStartedAt) / 1000,
                  cpu: LOCAL_PROCESS_COMPUTE_PROFILE.cpu,
                  memoryGiB: LOCAL_PROCESS_COMPUTE_PROFILE.memoryGiB,
                  diskGiB: LOCAL_PROCESS_COMPUTE_PROFILE.diskGiB,
                  startedAt: localComputeStartedAt,
                  endedAt: localComputeEndedAt,
                }),
              },
            );
          } catch (billingError) {
            console.error(
              '[play-runner.local_process.compute_billing_failed]',
              {
                workflowId: config.context.workflowId,
                workspaceRoot,
                error:
                  billingError instanceof Error
                    ? billingError.message
                    : String(billingError),
              },
            );
          }
          await withActiveSpan(
            'plays.runner.cleanup',
            {
              tracer: 'deepline.plays',
              attributes: {
                'plays.workflow_id': config.context.workflowId,
                'plays.play_name': config.context.playName,
              },
            },
            async () => {
              await rm(workspaceRoot, { recursive: true, force: true });
            },
          );
        }
      },
    );
  },
};