import {
  getCompiledPipelineSubsteps,
  type PlaySheetContract,
  type PlayStaticPipeline,
  type PlayStaticSubstep,
} from '../plays/static-pipeline';
import { normalizeTableNamespace } from '../plays/row-identity';
import {
  RUNTIME_SHEET_ROWS_LOGICAL_TABLE,
  RUNTIME_WORK_RECEIPT_LOGICAL_TABLE,
  RUNTIME_WORK_RECEIPT_TABLE_NAMESPACE,
  type DbLogicalTable,
  type DbSessionLimits,
  type DbSessionOperation,
} from './db-session';

export type RuntimeDbSessionRequirement = {
  tableNamespace: string;
  logicalTable: DbLogicalTable;
  operations: DbSessionOperation[];
  limits?: DbSessionLimits;
  sheetContract?: PlaySheetContract | null;
};

export type RuntimeSheetDbSessionPlanInspection = {
  datasetCount: number;
  datasetNamespaces: string[];
  plannedNamespaces: string[];
  errors: string[];
};

type DatasetSessionContract = {
  tableNamespace: string;
  sheetContract: PlaySheetContract | null;
  error?: string;
};

function currentExecutionScopeSubsteps(
  substeps: readonly PlayStaticSubstep[],
): PlayStaticSubstep[] {
  const flattened: PlayStaticSubstep[] = [];
  for (const substep of substeps) {
    flattened.push(substep);
    if (
      (substep.type === 'dataset' ||
        substep.type === 'step_suite' ||
        substep.type === 'control_flow') &&
      substep.steps?.length
    ) {
      flattened.push(...currentExecutionScopeSubsteps(substep.steps));
    }
    // A child play with a dataset owns a separate run and session scope. Its
    // nested pipeline must never reserve a table under the parent play token.
  }
  return flattened;
}

function datasetSessionContracts(
  pipeline: PlayStaticPipeline | null,
): DatasetSessionContract[] {
  if (!pipeline) return [];
  const rootNamespace = pipeline.tableNamespace?.trim();
  let normalizedRootNamespace: string | null = null;
  if (rootNamespace) {
    try {
      normalizedRootNamespace = normalizeTableNamespace(rootNamespace);
    } catch {
      normalizedRootNamespace = null;
    }
  }
  const rootContract = pipeline.sheetContract ?? null;
  return currentExecutionScopeSubsteps(
    getCompiledPipelineSubsteps(pipeline),
  ).flatMap<DatasetSessionContract>((substep): DatasetSessionContract[] => {
    if (substep.type !== 'dataset') return [];
    const rawNamespace = (substep.tableNamespace ?? substep.field ?? '').trim();
    if (!rawNamespace) {
      return [{ tableNamespace: '', sheetContract: null }];
    }
    let tableNamespace: string;
    try {
      tableNamespace = normalizeTableNamespace(rawNamespace);
    } catch (error) {
      return [
        {
          tableNamespace: rawNamespace,
          sheetContract: null,
          error: error instanceof Error ? error.message : String(error),
        },
      ];
    }
    const sheetContract =
      substep.sheetContract ??
      (normalizedRootNamespace === tableNamespace ? rootContract : null);
    return [{ tableNamespace, sheetContract }];
  });
}

export function inspectRuntimeSheetDbSessionPlan(
  pipeline: PlayStaticPipeline | null,
): RuntimeSheetDbSessionPlanInspection {
  const datasets = datasetSessionContracts(pipeline);
  const datasetNamespaces: string[] = [];
  const plannedNamespaces: string[] = [];
  const errors = [...(pipeline?.sheetContractErrors ?? [])];
  const seen = new Set<string>();
  for (const dataset of datasets) {
    if (dataset.error) {
      errors.push(dataset.error);
      continue;
    }
    if (!dataset.tableNamespace) {
      errors.push('Dataset is missing a Runtime Sheet namespace.');
      continue;
    }
    datasetNamespaces.push(dataset.tableNamespace);
    if (seen.has(dataset.tableNamespace)) {
      errors.push(
        `Dataset "${dataset.tableNamespace}" is declared more than once in the same play execution scope.`,
      );
      continue;
    }
    seen.add(dataset.tableNamespace);
    if (!dataset.sheetContract) {
      errors.push(
        `Dataset "${dataset.tableNamespace}" is missing a compiled Runtime Sheet contract.`,
      );
      continue;
    }
    let contractNamespace: string;
    try {
      contractNamespace = normalizeTableNamespace(
        dataset.sheetContract.tableNamespace,
      );
    } catch (error) {
      errors.push(error instanceof Error ? error.message : String(error));
      continue;
    }
    if (contractNamespace !== dataset.tableNamespace) {
      errors.push(
        `Dataset "${dataset.tableNamespace}" has a Runtime Sheet contract for "${contractNamespace}".`,
      );
      continue;
    }
    plannedNamespaces.push(dataset.tableNamespace);
  }
  return {
    datasetCount: datasets.length,
    datasetNamespaces,
    plannedNamespaces,
    errors,
  };
}

export function planRuntimeSheetDbSessionRequirements(
  pipeline: PlayStaticPipeline | null,
): RuntimeDbSessionRequirement[] {
  const byNamespace = new Map<string, PlaySheetContract>();
  for (const dataset of datasetSessionContracts(pipeline)) {
    if (!dataset.tableNamespace || !dataset.sheetContract) continue;
    byNamespace.set(dataset.tableNamespace, dataset.sheetContract);
  }
  return [...byNamespace.entries()].map(([tableNamespace, sheetContract]) => ({
    tableNamespace,
    logicalTable: RUNTIME_SHEET_ROWS_LOGICAL_TABLE,
    operations: ['rows.read', 'rows.upsert'],
    sheetContract,
  }));
}

export function runtimeWorkReceiptDbSessionRequirement(): RuntimeDbSessionRequirement {
  return {
    tableNamespace: RUNTIME_WORK_RECEIPT_TABLE_NAMESPACE,
    logicalTable: RUNTIME_WORK_RECEIPT_LOGICAL_TABLE,
    operations: ['rows.read', 'rows.upsert'],
  };
}

export function planRuntimeDbSessionRequirements(
  pipeline: PlayStaticPipeline | null,
): RuntimeDbSessionRequirement[] {
  return [
    ...planRuntimeSheetDbSessionRequirements(pipeline),
    runtimeWorkReceiptDbSessionRequirement(),
  ];
}