import type { DaytonaSandbox } from './daytona-lifecycle';
import { RUNTIME_RELIABILITY_POLICY } from '@shared_libs/play-runtime/runtime-reliability-policy';

export type DetachedDaytonaStartupDiagnostic = {
  schedulerReadinessObserved: boolean;
  schedulerReadinessResponses: number;
  commandFound: boolean;
  exitCode: number | null;
  sessionFound: boolean;
  sessionCommandCount: number | null;
  exitCodeFile: number | null;
  processSummary: {
    total: number;
    node: number;
    states: Record<string, number>;
  } | null;
  runnerStartup: {
    phase: 'booted' | 'heartbeat_transport';
    outcome?:
      | 'ok'
      | 'terminal'
      | 'redirect'
      | 'gateway_4xx'
      | 'gateway_5xx'
      | 'protocol_error'
      | 'timeout'
      | 'network_error';
    status?: number;
    errorCode?: string;
    transportAttemptId?: string;
  } | null;
  errors: string[];
};

function parseRunnerStartupDiagnostic(
  value: Buffer,
): DetachedDaytonaStartupDiagnostic['runnerStartup'] {
  try {
    const parsed = JSON.parse(value.toString('utf-8')) as Record<
      string,
      unknown
    >;
    if (
      parsed.schemaVersion !== 1 ||
      (parsed.phase !== 'booted' && parsed.phase !== 'heartbeat_transport')
    ) {
      return null;
    }
    if (parsed.phase === 'booted') return { phase: 'booted' };
    const event = parsed.event;
    if (!event || typeof event !== 'object') return null;
    const record = event as Record<string, unknown>;
    const allowedOutcomes = new Set([
      'ok',
      'terminal',
      'redirect',
      'gateway_4xx',
      'gateway_5xx',
      'protocol_error',
      'timeout',
      'network_error',
    ]);
    if (
      typeof record.outcome !== 'string' ||
      !allowedOutcomes.has(record.outcome)
    )
      return null;
    const diagnostic: NonNullable<
      DetachedDaytonaStartupDiagnostic['runnerStartup']
    > = {
      phase: 'heartbeat_transport',
      outcome: record.outcome as NonNullable<
        DetachedDaytonaStartupDiagnostic['runnerStartup']
      >['outcome'],
    };
    if (
      typeof record.status === 'number' &&
      Number.isSafeInteger(record.status) &&
      record.status >= 100 &&
      record.status <= 599
    )
      diagnostic.status = record.status;
    if (
      typeof record.errorCode === 'string' &&
      /^[A-Za-z0-9_]{1,64}$/.test(record.errorCode)
    )
      diagnostic.errorCode = record.errorCode;
    if (
      typeof record.transportAttemptId === 'string' &&
      /^[a-f0-9-]{36}$/i.test(record.transportAttemptId)
    )
      diagnostic.transportAttemptId = record.transportAttemptId;
    return diagnostic;
  } catch {
    return null;
  }
}

function summarizeProcesses(value: unknown): {
  total: number;
  node: number;
  states: Record<string, number>;
} | null {
  if (typeof value !== 'string' || !value.trim()) return null;
  const rows = value
    .split('\n')
    .map((line) => line.trim().split(/\s+/, 2))
    .filter(([state, command]) => Boolean(state && command));
  const states: Record<string, number> = {};
  let node = 0;
  for (const [state, command] of rows) {
    const normalizedState = state![0]?.toUpperCase() || '?';
    states[normalizedState] = (states[normalizedState] ?? 0) + 1;
    if (/^node(?:-|$)/i.test(command!)) node += 1;
  }
  return { total: rows.length, node, states };
}

/**
 * Inspect an accepted detached command without exposing its command line or
 * environment. Daytona's command model has no queued/running state, so the
 * combination of session membership, a numeric exit marker, and a
 * aggregate process counts are the narrowest useful startup diagnostic.
 * Process names are reduced to a Node count before logging; runner logs,
 * output, and arbitrary process names are deliberately excluded because no
 * heuristic redactor can guarantee removal of arbitrary customer secrets.
 */
export async function inspectDetachedDaytonaStartup(input: {
  sandbox: DaytonaSandbox;
  sessionId: string;
  cmdId: string;
  exitCodePath: string;
  startupDiagnosticPath?: string;
}): Promise<DetachedDaytonaStartupDiagnostic> {
  const errors: string[] = [];
  const diagnostic: DetachedDaytonaStartupDiagnostic = {
    schedulerReadinessObserved: false,
    schedulerReadinessResponses: 0,
    commandFound: false,
    exitCode: null,
    sessionFound: false,
    sessionCommandCount: null,
    exitCodeFile: null,
    processSummary: null,
    runnerStartup: null,
    errors,
  };

  await Promise.all([
    input.sandbox.process
      .getSessionCommand(input.sessionId, input.cmdId)
      .then((command) => {
        diagnostic.commandFound = true;
        diagnostic.exitCode =
          typeof command.exitCode === 'number' ? command.exitCode : null;
      })
      .catch(() => errors.push('get_session_command_failed')),
    input.sandbox.process
      .getSession(input.sessionId)
      .then((session) => {
        diagnostic.sessionFound = true;
        diagnostic.sessionCommandCount = session.commands.length;
      })
      .catch(() => errors.push('get_session_failed')),
    input.sandbox.fs
      .downloadFile(input.exitCodePath, 5)
      .then((file) => {
        const value = Number.parseInt(file.toString('utf-8').trim(), 10);
        diagnostic.exitCodeFile = Number.isFinite(value) ? value : null;
      })
      .catch(() => errors.push('exit_code_file_unavailable')),
    ...(input.startupDiagnosticPath
      ? [
          input.sandbox.fs
            .downloadFile(input.startupDiagnosticPath, 5)
            .then((file) => {
              diagnostic.runnerStartup = parseRunnerStartupDiagnostic(file);
              if (!diagnostic.runnerStartup)
                errors.push('startup_diagnostic_invalid');
            })
            .catch(() => errors.push('startup_diagnostic_unavailable')),
        ]
      : []),
    input.sandbox.process
      .executeCommand('ps -eo stat=,comm=', undefined, {}, 5)
      .then((result) => {
        diagnostic.processSummary = summarizeProcesses(result.result);
      })
      .catch(() => errors.push('process_snapshot_failed')),
  ]);

  return diagnostic;
}

/**
 * Detached Daytona command start (push-execution B2-final).
 *
 * The runner command is started DETACHED via a Daytona session
 * (`createSession` + `executeSessionCommand({ runAsync: true })`, which returns
 * a cmdId immediately) and the worker PARKS on a `detached_runner` suspension —
 * nothing holds a socket or polls for completion. The parked attempt wakes on
 * the runner's pushed terminal (receipt-gateway wake event) or on the ceiling
 * timeout, where `inspectDetachedDaytonaRunner` (daytona.ts) reads the session
 * command's exit code and salvages the captured output file as the fallback
 * verification.
 */

export type DetachedDaytonaRunnerState =
  | { phase: 'preparing'; sessionId: string }
  | {
      phase: 'command_accepted';
      sessionId: string;
      cmdId: string;
      baselineHeartbeatAt: string | null;
    }
  | {
      phase: 'ready';
      sessionId: string;
      cmdId: string;
      baselineHeartbeatAt: string | null;
    };

export type DetachedDaytonaRunnerSupervisor = {
  start(input: {
    command: string;
    exitCodePath: string;
    startupDiagnosticPath?: string;
  }): Promise<Extract<DetachedDaytonaRunnerState, { phase: 'ready' }>>;
};

// The durable heartbeat proves runner-process liveness plus gateway reachability.
// Keep the handoff bounded while allowing a cold gateway to recover.
export const DAYTONA_RUNNER_READY_TIMEOUT_MS =
  RUNTIME_RELIABILITY_POLICY.sandbox.runnerReadyTimeoutMs;
const DAYTONA_RUNNER_READY_POLL_MS = 50;

export type DetachedRunnerReadinessPort = (
  baselineHeartbeatAt?: string | null,
) => Promise<RunnerReadinessResponse | null>;

type RunnerReadinessResponse = {
  ready: boolean;
  heartbeatAt: string | null;
};

async function requestRunnerReadiness(input: {
  readiness: DetachedRunnerReadinessPort;
  baselineHeartbeatAt?: string | null;
  timeoutMs: number;
}): Promise<{
  response: RunnerReadinessResponse | null;
  diagnosis: string;
  timedOut: boolean;
}> {
  const timedOut = Symbol('readiness_timeout');
  let timeout: ReturnType<typeof setTimeout> | undefined;
  try {
    // Bound the only in-flight scheduler-store read by the remaining startup
    // budget. If it times out, callers stop polling instead of layering a new
    // query over the abandoned one and amplifying a degraded connection pool.
    const response = await Promise.race([
      input.readiness(input.baselineHeartbeatAt),
      new Promise<typeof timedOut>((resolve) => {
        timeout = setTimeout(
          () => resolve(timedOut),
          Math.max(1, input.timeoutMs),
        );
      }),
    ]);
    if (response === timedOut) {
      return {
        response: null,
        diagnosis: 'readiness_timeout',
        timedOut: true,
      };
    }
    if (!response)
      return {
        response: null,
        diagnosis: 'readiness_unavailable',
        timedOut: false,
      };
    return {
      response,
      diagnosis: 'ok',
      timedOut: false,
    };
  } catch {
    return {
      response: null,
      diagnosis: 'readiness_unavailable',
      timedOut: false,
    };
  } finally {
    if (timeout) clearTimeout(timeout);
  }
}

/**
 * Capture the attempt heartbeat using the scheduler's Postgres clock before
 * launching customer code. The post-launch readiness poll waits for this exact
 * database value to advance, avoiding both a gateway protocol dependency and
 * any comparison with the worker host clock.
 */
export async function captureDetachedDaytonaRunnerReadinessBaseline(input: {
  readiness: DetachedRunnerReadinessPort;
  timeoutMs?: number;
  pollMs?: number;
  now?: () => number;
  sleep?: (ms: number) => Promise<void>;
}): Promise<string | null> {
  const now = input.now ?? Date.now;
  const sleep =
    input.sleep ??
    ((ms: number) => new Promise<void>((resolve) => setTimeout(resolve, ms)));
  const timeoutMs = Math.max(
    1,
    input.timeoutMs ?? DAYTONA_RUNNER_READY_TIMEOUT_MS,
  );
  const pollMs = Math.max(1, input.pollMs ?? DAYTONA_RUNNER_READY_POLL_MS);
  const deadline = now() + timeoutMs;
  let diagnosis = 'no_readiness_response';
  let retryDelayMs = pollMs;
  while (now() < deadline) {
    const read = await requestRunnerReadiness({
      readiness: input.readiness,
      timeoutMs: Math.max(1, deadline - now()),
    });
    diagnosis = read.diagnosis;
    if (read.response) return read.response.heartbeatAt;
    if (read.timedOut) break;
    await sleep(Math.min(retryDelayMs, Math.max(1, deadline - now())));
    retryDelayMs = Math.min(1_000, retryDelayMs * 2);
  }
  throw new Error(
    `Daytona readiness baseline could not reach the scheduler store before command launch within ${timeoutMs}ms (${diagnosis}). Customer play code was not started.`,
  );
}

export class DaytonaRunnerInitializationError extends Error {
  constructor(
    message: string,
    readonly startupDiagnostic: DetachedDaytonaStartupDiagnostic,
  ) {
    super(message);
    this.name = 'DaytonaRunnerInitializationError';
  }
}

/**
 * Begin the independent control-plane work needed by a detached runner while
 * its payload uploads. `start` crosses the customer-code side-effect boundary:
 * it accepts one command, then returns only after the scheduler has observed
 * durable runner liveness.
 */
export function prepareDetachedDaytonaRunner(input: {
  sandbox: DaytonaSandbox;
  sessionId: string;
  readiness: DetachedRunnerReadinessPort;
  cancellation?: Promise<never>;
}): DetachedDaytonaRunnerSupervisor {
  const cancellation = input.cancellation ?? new Promise<never>(() => {});
  const sessionCreation = input.sandbox.process
    .createSession(input.sessionId)
    .then(
      () => ({ ok: true as const }),
      (error: unknown) => ({ ok: false as const, error }),
    );
  const readinessBaseline = captureDetachedDaytonaRunnerReadinessBaseline({
    readiness: input.readiness,
  }).then(
    (heartbeatAt) => ({ ok: true as const, heartbeatAt }),
    (error: unknown) => ({ ok: false as const, error }),
  );
  let state: DetachedDaytonaRunnerState = {
    phase: 'preparing',
    sessionId: input.sessionId,
  };
  let started = false;

  return {
    async start({ command, exitCodePath, startupDiagnosticPath }) {
      if (started) {
        throw new Error(
          `Detached Daytona runner session ${input.sessionId} was already started.`,
        );
      }
      started = true;
      const session = await Promise.race([sessionCreation, cancellation]);
      if (session.ok === false) throw session.error;
      const baseline = await Promise.race([readinessBaseline, cancellation]);
      if (baseline.ok === false) throw baseline.error;
      const response = await Promise.race([
        input.sandbox.process.executeSessionCommand(input.sessionId, {
          command,
          runAsync: true,
        }),
        cancellation,
      ]);
      const cmdId = response?.cmdId;
      if (!cmdId) {
        throw new Error(
          `Daytona executeSessionCommand did not return a cmdId for session ${input.sessionId}.`,
        );
      }
      state = {
        phase: 'command_accepted',
        sessionId: input.sessionId,
        cmdId,
        baselineHeartbeatAt: baseline.heartbeatAt,
      };
      await Promise.race([
        confirmDetachedDaytonaRunnerReady({
          sandbox: input.sandbox,
          sessionId: state.sessionId,
          cmdId: state.cmdId,
          exitCodePath,
          startupDiagnosticPath,
          readiness: input.readiness,
          baselineHeartbeatAt: state.baselineHeartbeatAt,
        }),
        cancellation,
      ]);
      state = { ...state, phase: 'ready' };
      return state;
    },
  };
}

/**
 * A detached command id only proves that Daytona accepted the request. Before
 * the worker parks, require a marker written only after the initialized runner
 * materializes its artifact and the receipt gateway accepts its first
 * heartbeat. This closes the unobservable handoff where a sandbox or command
 * disappears after executeSessionCommand but before durable runner liveness.
 */
export async function confirmDetachedDaytonaRunnerReady(input: {
  sandbox: DaytonaSandbox;
  sessionId: string;
  cmdId: string;
  exitCodePath: string;
  startupDiagnosticPath?: string;
  readiness: DetachedRunnerReadinessPort;
  baselineHeartbeatAt: string | null;
  timeoutMs?: number;
  pollMs?: number;
  now?: () => number;
  sleep?: (ms: number) => Promise<void>;
}): Promise<void> {
  const now = input.now ?? Date.now;
  const sleep =
    input.sleep ??
    ((ms: number) => new Promise<void>((resolve) => setTimeout(resolve, ms)));
  const timeoutMs = Math.max(
    1,
    input.timeoutMs ?? DAYTONA_RUNNER_READY_TIMEOUT_MS,
  );
  const pollMs = Math.max(1, input.pollMs ?? DAYTONA_RUNNER_READY_POLL_MS);
  const deadline = now() + timeoutMs;
  let schedulerDiagnosis = 'no_readiness_response';
  let schedulerReadinessResponses = 0;
  let retryDelayMs = pollMs;

  while (now() < deadline) {
    const read = await requestRunnerReadiness({
      readiness: input.readiness,
      baselineHeartbeatAt: input.baselineHeartbeatAt,
      timeoutMs: Math.max(1, deadline - now()),
    });
    schedulerDiagnosis = read.diagnosis;
    if (read.response) {
      schedulerReadinessResponses += 1;
      if (read.response.ready) return;
      schedulerDiagnosis = 'runner_heartbeat_not_observed';
    }
    if (read.timedOut) break;
    await sleep(Math.min(retryDelayMs, Math.max(1, deadline - now())));
    retryDelayMs = Math.min(1_000, retryDelayMs * 2);
  }

  const diagnostic = await inspectDetachedDaytonaStartup({
    sandbox: input.sandbox,
    sessionId: input.sessionId,
    cmdId: input.cmdId,
    exitCodePath: input.exitCodePath,
    startupDiagnosticPath: input.startupDiagnosticPath,
  });
  diagnostic.schedulerReadinessObserved = false;
  diagnostic.schedulerReadinessResponses = schedulerReadinessResponses;
  const oomIndicated =
    diagnostic.exitCode === 137 || diagnostic.exitCodeFile === 137;
  throw new DaytonaRunnerInitializationError(
    `RUNTIME_SANDBOX_START_FAILED: Daytona accepted detached command ${input.cmdId} in sandbox ${input.sandbox.id}, but the scheduler did not observe play-runner liveness within ${timeoutMs}ms. Scheduler diagnosis: ${schedulerDiagnosis}. Daytona diagnosis: ${JSON.stringify(diagnostic)}. OOM is ${oomIndicated ? 'indicated by exit code 137' : 'not confirmed by the available Daytona evidence'}. The worker will not park this runner; the failed runner was stopped and was not retried in place.`,
    diagnostic,
  );
}