import { resolveTimingWindow } from './live-events';
import {
  LOG_TAIL_LIMIT,
  normalizePlayRunLedgerSnapshot,
  summarizeRunRowOutcomes,
  type PlayRunLedgerSnapshot,
  type PlayRunLedgerDatasetSnapshot,
  type PlayRunLedgerStepSnapshot,
  type PlayRunRowOutcomeSummary,
} from './run-ledger';
import {
  isActivePlayRunLifecycleStatus,
  isTerminalPlayRunLifecycleStatus,
  normalizePlayRunLifecycleStatus,
  type PlayRunLifecycleStatus,
} from './run-lifecycle-policy';
import type { PlayDocflowNodeIoState } from './docflow-node-io';
import type { PlayVisualNodeUsageSnapshot } from './live-state-contract';
import {
  projectPlayRunActivity,
  type PlayActivityObservation,
  type PlayRunActivityProjection,
} from './activity-observation';

/**
 * Run Snapshot Stream.
 *
 * The canonical client-side projection of a persisted Play Run document into
 * the live snapshot shape, plus the differ that turns successive snapshots
 * into incremental `play.*` live events. This module is shared between the
 * Vercel app (legacy SSE shim, dashboard) and the SDK subscription transport
 * (`sdk/src/runs/observe-transport.ts`), so both watchers render identical
 * event streams from the same Convex Run Snapshot. See ADR-0008.
 */

export type PlayRunLiveStatus = PlayRunLifecycleStatus;

export type PlayRunStreamNodeProgress = {
  completed?: number;
  total?: number;
  failed?: number;
  startedRows?: number;
  activeRows?: number;
  waitingRows?: number;
  completedRows?: number;
  message?: string;
  updatedAt?: number | null;
  startedAt?: number | null;
  completedAt?: number | null;
  artifactTableNamespace?: string | null;
  nodeIo?: PlayDocflowNodeIoState;
};

export type PlayRunStreamNodeState = {
  nodeId: string;
  status: 'idle' | 'running' | 'completed' | 'failed' | 'skipped';
  error?: string | null;
  artifactTableNamespace?: string | null;
  progress?: PlayRunStreamNodeProgress | null;
  startedAt?: number | null;
  completedAt?: number | null;
  updatedAt?: number | null;
  nodeIo?: PlayDocflowNodeIoState;
  /**
   * Observed provider usage (ADR 0018). Attached by the read path, not by the
   * ledger projection — settlement lags finalize, so this is always computed
   * at read time.
   */
  usage?: PlayVisualNodeUsageSnapshot;
};

export type PlayRunLiveSnapshot = {
  runId: string;
  status: PlayRunLiveStatus;
  createdAt?: number | null;
  startedAt?: number | null;
  finishedAt?: number | null;
  durationMs?: number | null;
  updatedAt: number | null;
  /**
   * Rotating log tail (the ledger snapshot's bounded `logTail`; the wire name
   * stays `logs` for installed clients). `totalLogCount` is the run's
   * cumulative ingested-line count (monotonic), so stream differs can cursor
   * on absolute sequence numbers instead of indexes into the rotated tail.
   * Full retention lives in the Run Log Stream (`GET /api/v2/runs/:id/logs`).
   */
  logs: string[];
  totalLogCount: number;
  /**
   * True once the Run Log Stream stopped storing log bodies because the run
   * crossed the retention cap (ADR-0009). Additive/optional on the wire.
   */
  logsTruncated?: boolean;
  activeArtifactTableNamespace: string | null;
  resultTableNamespace: string | null;
  datasets?: PlayRunLedgerDatasetSnapshot[];
  /** Explicit typed activity facts; legacy steps/datasets remain alongside them. */
  activities?: PlayActivityObservation[];
  activeActivity?: PlayActivityObservation | null;
  /** Canonical current activity, including compatibility projection. */
  activity?: PlayRunActivityProjection | null;
  nodeStates: PlayRunStreamNodeState[];
  activeNodeId: string | null;
  /**
   * Additive, terminal-only aggregate of row outcomes across all steps. Present
   * once the run is terminal so readers can render a truthful "completed with
   * failures" surface (row-failure isolation persists failed rows for retry, so
   * a run legitimately reports `status: 'completed'` with `rowOutcomes.failedRows
   * > 0`). Absent on non-terminal snapshots and on runs with no row/map steps.
   */
  rowOutcomes?: PlayRunRowOutcomeSummary;
};

export function normalizePlayRunLiveStatus(value: unknown): PlayRunLiveStatus {
  return normalizePlayRunLifecycleStatus(value);
}

export function isTerminalPlayRunLiveStatus(
  status: PlayRunLiveStatus,
): boolean {
  return isTerminalPlayRunLifecycleStatus(status);
}

export function isActivePlayRunStatus(status: unknown): boolean {
  return isActivePlayRunLifecycleStatus(status);
}

/**
 * The minimal persisted Play Run document shape required to project the live
 * snapshot. This is a structural subset of the Convex `playRuns` doc and of
 * the run-observer projection returned by
 * `convex/runObservers.getPlayRunSnapshotForObserver`.
 */
export type LedgerBackedRunLike = {
  workflowId: string;
  status: string;
  name?: string | null;
  createdAt?: number | null;
  startedAt?: number | null;
  finishedAt?: number | null;
  updatedAt?: number | null;
  runSnapshot?: unknown;
  result?: unknown;
  wait?: { kind?: string; eventKey?: string; boundaryId?: string } | null;
  waitKind?: string | null;
  waitUntil?: number | null;
  activeBoundaryId?: string | null;
  runtimeBackend?: string | null;
};

function isRecord(value: unknown): value is Record<string, unknown> {
  return Boolean(value && typeof value === 'object' && !Array.isArray(value));
}

function finiteNumber(value: unknown): number | null {
  return typeof value === 'number' && Number.isFinite(value) ? value : null;
}

function extractTerminalRunLogTail(result: unknown): {
  logTail: string[];
  totalLogCount: number;
} | null {
  if (!isRecord(result) || !isRecord(result._metadata)) {
    return null;
  }
  const runLogTail = result._metadata.runLogTail;
  if (!isRecord(runLogTail) || !Array.isArray(runLogTail.tail)) {
    return null;
  }
  const logTail = runLogTail.tail.filter(
    (line): line is string =>
      typeof line === 'string' && line.trim().length > 0,
  );
  if (logTail.length === 0) {
    return null;
  }
  const totalLogCount = Math.max(
    finiteNumber(runLogTail.totalLogCount) ?? logTail.length,
    logTail.length,
  );
  return {
    logTail: logTail.slice(-LOG_TAIL_LIMIT),
    totalLogCount,
  };
}

function buildSnapshotFromLedger(
  snapshot: PlayRunLedgerSnapshot,
): PlayRunLiveSnapshot {
  const nodeStates = snapshot.orderedStepIds
    .map((stepId) => snapshot.stepsById[stepId])
    .filter((step): step is PlayRunLedgerStepSnapshot => Boolean(step))
    .map((step) => ({
      nodeId: step.stepId,
      status: step.status,
      error: step.error ?? null,
      artifactTableNamespace: step.artifactTableNamespace ?? null,
      progress: step.progress
        ? {
            completed: step.progress.completed,
            total: step.progress.total,
            failed: step.progress.failed,
            startedRows: step.progress.startedRows,
            activeRows: step.progress.activeRows,
            waitingRows: step.progress.waitingRows,
            completedRows: step.progress.completedRows,
            message: step.progress.message,
            artifactTableNamespace:
              step.progress.artifactTableNamespace ??
              step.artifactTableNamespace ??
              null,
            startedAt: step.startedAt ?? null,
            completedAt: step.completedAt ?? null,
            updatedAt: step.progress.updatedAt ?? step.updatedAt ?? null,
            nodeIo: step.progress.nodeIo,
          }
        : null,
      startedAt: step.startedAt ?? null,
      completedAt: step.completedAt ?? null,
      updatedAt: step.updatedAt ?? null,
      ...(step.progress?.nodeIo ? { nodeIo: step.progress.nodeIo } : {}),
    }));
  const liveStatus = normalizePlayRunLiveStatus(snapshot.status);
  // Only surface the row-outcome aggregate once the run is terminal and it
  // actually observed row/map steps — mid-flight `failed` counts churn, and a
  // no-map play has no meaningful row outcome to report.
  const rowOutcomes =
    isTerminalPlayRunLiveStatus(liveStatus) &&
    Object.keys(snapshot.stepsById).length > 0
      ? summarizeRunRowOutcomes(snapshot)
      : null;
  return {
    runId: snapshot.runId,
    status: liveStatus,
    createdAt: snapshot.createdAt ?? null,
    startedAt: snapshot.startedAt ?? null,
    finishedAt: snapshot.finishedAt ?? null,
    durationMs: snapshot.durationMs ?? null,
    updatedAt:
      snapshot.updatedAt ?? snapshot.finishedAt ?? snapshot.startedAt ?? null,
    logs: snapshot.logTail,
    totalLogCount: snapshot.totalLogCount,
    ...(snapshot.logsTruncated ? { logsTruncated: true } : {}),
    activeArtifactTableNamespace: snapshot.activeArtifactTableNamespace ?? null,
    resultTableNamespace: snapshot.resultTableNamespace ?? null,
    datasets: snapshot.orderedDatasetIds
      .map((datasetId) => snapshot.datasetsById[datasetId])
      .filter((dataset): dataset is PlayRunLedgerDatasetSnapshot =>
        Boolean(dataset),
      ),
    activities: snapshot.orderedActivityIds
      .map((activityId) => snapshot.activitiesById[activityId])
      .filter((activity): activity is PlayActivityObservation =>
        Boolean(activity),
      ),
    activeActivity: snapshot.activeActivityId
      ? (snapshot.activitiesById[snapshot.activeActivityId] ?? null)
      : null,
    nodeStates,
    activeNodeId: snapshot.activeStepId ?? null,
    ...(rowOutcomes ? { rowOutcomes } : {}),
  };
}

export function buildPlayRunStatusSnapshot(input: {
  run: LedgerBackedRunLike;
}): PlayRunLiveSnapshot {
  const ledgerSnapshot = normalizePlayRunLedgerSnapshot(input.run.runSnapshot, {
    runId: input.run.workflowId,
    playName: input.run.name ?? null,
    status: input.run.status,
    createdAt: input.run.createdAt ?? null,
    startedAt: input.run.startedAt ?? null,
    updatedAt: input.run.updatedAt ?? null,
    finishedAt: input.run.finishedAt ?? null,
  });
  const terminalRunLogTail = extractTerminalRunLogTail(input.run.result);
  const projectedSnapshot =
    terminalRunLogTail &&
    terminalRunLogTail.totalLogCount > ledgerSnapshot.totalLogCount
      ? buildSnapshotFromLedger({
          ...ledgerSnapshot,
          logTail: terminalRunLogTail.logTail,
          totalLogCount: terminalRunLogTail.totalLogCount,
        })
      : buildSnapshotFromLedger(ledgerSnapshot);
  const activity = projectPlayRunActivity({
    runId: input.run.workflowId,
    playName: input.run.name,
    status: projectedSnapshot.status,
    updatedAt: projectedSnapshot.updatedAt,
    waitKind: input.run.waitKind ?? input.run.wait?.kind ?? null,
    waitUntil: input.run.waitUntil,
    eventKey: input.run.wait?.eventKey ?? null,
    runtimeBackend: input.run.runtimeBackend,
    activeNodeId: input.run.activeBoundaryId ?? projectedSnapshot.activeNodeId,
    nodeStates: projectedSnapshot.nodeStates,
    datasets: projectedSnapshot.datasets,
    explicit: projectedSnapshot.activities,
  });
  return {
    ...projectedSnapshot,
    activity,
  };
}

/** Generic live-event envelope shape shared with the SSE protocol. */
export type RunStreamEventEnvelope<TPayload, TType extends string> = {
  cursor: string;
  streamId: string;
  scope: 'play';
  type: TType;
  at: string;
  payload: TPayload;
};

export type PlayRunStreamEvent =
  | RunStreamEventEnvelope<
      {
        runId: string;
        status: PlayRunLiveStatus;
        updatedAt: number | null;
      },
      'play.run.status'
    >
  | RunStreamEventEnvelope<PlayRunLiveSnapshot, 'play.run.snapshot'>
  | RunStreamEventEnvelope<
      {
        runId: string;
        stepId: string;
        status: PlayRunStreamNodeState['status'];
        artifactTableNamespace: string | null;
        startedAt?: number | null;
        completedAt?: number | null;
        updatedAt?: number | null;
      },
      'play.step.status'
    >
  | RunStreamEventEnvelope<
      {
        runId: string;
        stepId: string;
        completed?: number;
        total?: number;
        failed?: number;
        message?: string;
        artifactTableNamespace: string | null;
        startedAt?: number | null;
        completedAt?: number | null;
        updatedAt?: number | null;
      },
      'play.step.progress'
    >
  | RunStreamEventEnvelope<PlayRunLogStreamPayload, 'play.run.log'>;

/**
 * Log lines for one `play.run.log` event. `firstSeq`/`totalLogCount` are
 * additive (ADR-0009): when `firstSeq` is present, `lines` is a contiguous
 * run of log lines whose absolute (1-based) sequences are
 * `firstSeq .. firstSeq + lines.length - 1`, letting clients append by seq
 * instead of text-deduping or replacing from snapshots. When absent (gap
 * marker payloads), clients append the lines verbatim.
 */
export type PlayRunLogStreamPayload = {
  runId: string;
  lines: string[];
  source: string;
  firstSeq?: number;
  totalLogCount?: number;
};

function makeRunStreamEvent<TPayload, TType extends string>(input: {
  cursor: string;
  streamId: string;
  type: TType;
  payload: TPayload;
  at?: string;
}): RunStreamEventEnvelope<TPayload, TType> {
  return {
    ...input,
    scope: 'play',
    at: input.at ?? new Date().toISOString(),
  };
}

export type PlayRunStreamDiffState = {
  runSignature: string;
  snapshotSignature: string;
  stepStatusSignature: string;
  stepProgressSignature: string;
  /**
   * Absolute sequence number (1-based, monotonic per run) of the last log
   * line emitted to this stream. Cursors on `snapshot.totalLogCount`, not on
   * indexes into the rotated log tail.
   */
  lastLogSeq: number;
};

export const EMPTY_PLAY_RUN_STREAM_DIFF_STATE: PlayRunStreamDiffState = {
  runSignature: '',
  snapshotSignature: '',
  stepStatusSignature: '',
  stepProgressSignature: '',
  lastLogSeq: 0,
};

function getSnapshotCursor(snapshot: PlayRunLiveSnapshot): string {
  return String(snapshot.updatedAt ?? Date.now());
}

function getRunSignature(snapshot: PlayRunLiveSnapshot): string {
  return [snapshot.runId, snapshot.status, snapshot.updatedAt ?? 0].join(':');
}

function getStepStatusSignature(snapshot: PlayRunLiveSnapshot): string {
  return snapshot.nodeStates
    .map((state) =>
      [
        state.nodeId,
        state.status,
        state.artifactTableNamespace ?? '',
        state.startedAt ?? '',
        state.completedAt ?? '',
        state.progress?.startedAt ?? '',
        state.progress?.completedAt ?? '',
        state.updatedAt ?? '',
        state.progress?.updatedAt ?? '',
      ].join(':'),
    )
    .join('|');
}

function getStepProgressSignature(snapshot: PlayRunLiveSnapshot): string {
  return snapshot.nodeStates
    .map((state) =>
      [
        state.nodeId,
        state.progress?.completed ?? '',
        state.progress?.total ?? '',
        state.progress?.failed ?? '',
        state.progress?.artifactTableNamespace ?? '',
        state.progress?.startedAt ?? '',
        state.progress?.completedAt ?? '',
        state.progress?.updatedAt ?? '',
        state.progress?.message ?? '',
      ].join(':'),
    )
    .join('|');
}

function getSnapshotSignature(snapshot: PlayRunLiveSnapshot): string {
  return JSON.stringify(snapshot);
}

export type PlayRunLogGap = {
  /** Number of log lines that fell out of the retained tail window. */
  missingCount: number;
  /** Absolute (1-based) sequence of the first retained tail line. */
  tailFirstSeq: number;
};

/**
 * Resolve whether the differ would have to skip log lines because the cursor
 * fell behind the retained tail window. Transports can use this to backfill
 * the gap from the durable run-event ledger before running the differ.
 */
export function resolvePlayRunLogGap(
  snapshot: PlayRunLiveSnapshot,
  lastLogSeq: number,
): PlayRunLogGap | null {
  if (snapshot.totalLogCount <= lastLogSeq) {
    return null;
  }
  const tailFirstSeq = snapshot.totalLogCount - snapshot.logs.length + 1;
  if (lastLogSeq + 1 >= tailFirstSeq) {
    return null;
  }
  return {
    missingCount: tailFirstSeq - 1 - lastLogSeq,
    tailFirstSeq,
  };
}

/**
 * Resolve which log lines this stream still has to emit, cursoring on
 * absolute per-run sequence numbers. The snapshot only retains a rotated
 * tail, so when the cursor has fallen behind the retained window the gap is
 * surfaced as one loud marker line followed by the whole retained tail.
 * Transports that can read the durable Run Log Stream resolve the gap with
 * `resolvePlayRunLogGap` + a log-page read BEFORE diffing, so this marker
 * path remains only for first-connects to in-flight runs and failed
 * page reads (ADR-0009).
 *
 * `firstSeq` is set only when `lines` is a contiguous seq run (no marker).
 */
function diffLogLines(input: {
  snapshot: PlayRunLiveSnapshot;
  lastLogSeq: number;
}): { lines: string[]; lastLogSeq: number; firstSeq: number | null } {
  const { logs, totalLogCount } = input.snapshot;
  if (totalLogCount <= input.lastLogSeq) {
    return { lines: [], lastLogSeq: input.lastLogSeq, firstSeq: null };
  }
  // Absolute sequence (1-based) of the first retained tail line.
  const tailFirstSeq = totalLogCount - logs.length + 1;
  if (input.lastLogSeq + 1 >= tailFirstSeq) {
    return {
      lines: logs.slice(input.lastLogSeq + 1 - tailFirstSeq),
      lastLogSeq: totalLogCount,
      firstSeq: input.lastLogSeq + 1,
    };
  }
  const missingCount = tailFirstSeq - 1 - input.lastLogSeq;
  return {
    lines: [
      `[stream] ${missingCount} log lines not retained in the live window; full logs via runs logs`,
      ...logs,
    ],
    lastLogSeq: totalLogCount,
    firstSeq: null,
  };
}

export function diffPlayRunStreamEvents(input: {
  streamId: string;
  snapshot: PlayRunLiveSnapshot;
  previous: PlayRunStreamDiffState;
}): {
  events: PlayRunStreamEvent[];
  next: PlayRunStreamDiffState;
} {
  const { snapshot, streamId, previous } = input;
  const cursor = getSnapshotCursor(snapshot);
  const logDiff = diffLogLines({
    snapshot,
    lastLogSeq: previous.lastLogSeq,
  });
  const next: PlayRunStreamDiffState = {
    runSignature: getRunSignature(snapshot),
    stepStatusSignature: getStepStatusSignature(snapshot),
    stepProgressSignature: getStepProgressSignature(snapshot),
    snapshotSignature: getSnapshotSignature(snapshot),
    lastLogSeq: logDiff.lastLogSeq,
  };
  const events: PlayRunStreamEvent[] = [];

  if (next.stepStatusSignature !== previous.stepStatusSignature) {
    for (const state of snapshot.nodeStates) {
      if (state.status === 'idle') {
        continue;
      }
      const persistedStartedAt =
        state.startedAt ?? state.progress?.startedAt ?? null;
      const persistedCompletedAt =
        state.completedAt ?? state.progress?.completedAt ?? null;

      events.push(
        makeRunStreamEvent({
          cursor,
          streamId,
          type: 'play.step.status',
          payload: {
            runId: snapshot.runId,
            stepId: state.nodeId,
            status: state.status,
            artifactTableNamespace: state.artifactTableNamespace ?? null,
            ...resolveTimingWindow({
              startedAt: persistedStartedAt,
              completedAt: persistedCompletedAt,
              updatedAt:
                state.updatedAt ??
                state.progress?.updatedAt ??
                snapshot.updatedAt ??
                null,
            }),
          },
        }),
      );
    }
  }

  if (next.stepProgressSignature !== previous.stepProgressSignature) {
    for (const state of snapshot.nodeStates) {
      if (!state.progress) {
        continue;
      }
      events.push(
        makeRunStreamEvent({
          cursor: String(
            state.progress.updatedAt ?? snapshot.updatedAt ?? Date.now(),
          ),
          streamId,
          type: 'play.step.progress',
          payload: {
            runId: snapshot.runId,
            stepId: state.nodeId,
            completed: state.progress.completed,
            total: state.progress.total,
            failed: state.progress.failed,
            startedRows: state.progress.startedRows,
            activeRows: state.progress.activeRows,
            waitingRows: state.progress.waitingRows,
            completedRows: state.progress.completedRows,
            message: state.progress.message,
            artifactTableNamespace:
              state.progress.artifactTableNamespace ??
              state.artifactTableNamespace ??
              null,
            ...resolveTimingWindow({
              startedAt: state.startedAt ?? state.progress.startedAt ?? null,
              completedAt:
                state.completedAt ?? state.progress.completedAt ?? null,
              updatedAt: state.progress.updatedAt ?? snapshot.updatedAt ?? null,
            }),
          },
        }),
      );
    }
  }

  if (logDiff.lines.length > 0) {
    events.push(
      makeRunStreamEvent({
        cursor,
        streamId,
        type: 'play.run.log',
        payload: {
          runId: snapshot.runId,
          lines: logDiff.lines,
          source: 'worker',
          ...(logDiff.firstSeq !== null ? { firstSeq: logDiff.firstSeq } : {}),
          totalLogCount: snapshot.totalLogCount,
        },
      }),
    );
  }

  if (next.snapshotSignature !== previous.snapshotSignature) {
    const enrichedNodeStates = snapshot.nodeStates.map((state) => {
      const timing = resolveTimingWindow({
        startedAt: state.startedAt ?? state.progress?.startedAt ?? null,
        completedAt: state.completedAt ?? state.progress?.completedAt ?? null,
        updatedAt:
          state.updatedAt ??
          state.progress?.updatedAt ??
          snapshot.updatedAt ??
          null,
      });
      return {
        ...state,
        ...timing,
        progress: state.progress
          ? {
              ...state.progress,
              ...resolveTimingWindow({
                startedAt: state.progress.startedAt ?? state.startedAt ?? null,
                completedAt:
                  state.progress.completedAt ?? state.completedAt ?? null,
                updatedAt:
                  state.progress.updatedAt ??
                  state.updatedAt ??
                  snapshot.updatedAt ??
                  null,
              }),
            }
          : state.progress,
      };
    });
    events.push(
      makeRunStreamEvent({
        cursor,
        streamId,
        type: 'play.run.snapshot',
        payload: { ...snapshot, nodeStates: enrichedNodeStates },
      }),
    );
  }

  if (next.runSignature !== previous.runSignature) {
    events.push(
      makeRunStreamEvent({
        cursor,
        streamId,
        type: 'play.run.status',
        payload: {
          runId: snapshot.runId,
          status: snapshot.status,
          updatedAt: snapshot.updatedAt,
          ...(snapshot.activity ? { activity: snapshot.activity } : {}),
        },
      }),
    );
  }

  return { events, next };
}