#!/usr/bin/env node

const fs = require('node:fs/promises');
const path = require('node:path');
const { createProgressLogger, parseProgressArgs } = require('../../../lib/progress-log.js');

// Parse, dialect detection, and type sniffing live in the shared adapter lib so
// the bulk-extract reader generated by rp-import-codegen inherits exactly the
// same behavior instead of re-deriving it.
const {
  readHeaderRow,
  readTailSample,
  streamRows,
  detectEmptyPolicy,
  summarizeColumn,
  rowWidthReport,
  normalizeHeaderName,
} = require('../lib/csv-parse.js');
const {
  loadVendorProfiles,
  detectVendor,
  diffProfile,
  prefillColumnMap,
} = require('../lib/csv-fingerprint.js');
const {
  classifyLayout,
  buildGroups,
  columnIndex,
  deriveColumnRoles,
  deriveColumnValues,
  detectDerivedCandidates,
} = require('../lib/csv-layout.js');
const {
  CONCAT_POLICY,
  resolveFileSet,
  inferJoins,
} = require('../lib/csv-fileset.js');

const DEFAULT_HEAD_ROWS = 5;
const DEFAULT_TAIL_ROWS = 3;
const DEFAULT_SCAN_ROWS = 5000;

let progress;

function printUsage() {
  console.log(`Usage:
  node csv-discovery.js --file <path> [--file <path> ...] --out-dir <dir> [options]

Required:
  --file <path>                 Input CSV. Repeatable — pass the whole file set in one run
                                so roles and split files are resolved together.
  --out-dir <dir>               Directory to write the raw capture into

Optional:
  --vendor <name>               User-stated vendor; overrides fingerprint detection.
                                One of: shopify, woocommerce, magento, bigcommerce, custom
  --delimiter <char>            Force the delimiter (default: auto-detect , ; \\t |)
  --encoding <name>             Force the encoding (default: auto-detect UTF-8 / BOM)
  --role <path>:<role>          Force a file's role. Repeatable.
  --file-order stated|natural   Order of split files (default: natural)
  --allow-header-superset       Union split files whose headers differ instead of halting
  --head-rows <n>               Sample rows from the head. Default: ${DEFAULT_HEAD_ROWS}
  --tail-rows <n>               Sample rows from the tail. Default: ${DEFAULT_TAIL_ROWS}
  --scan-rows <n>               Rows retained for inference. Default: ${DEFAULT_SCAN_ROWS}
  --progress-log <path>         Append progress NDJSON records to this file.
  --help                        Show this help text

Example:
  node csv-discovery.js \\
    --file ~/exports/products_export.csv \\
    --out-dir migrations/acme/data/csv-discovery
`);
}

function parseArgs(argv) {
  const args = {
    files: [],
    explicitRoles: {},
    vendor: null,
    delimiter: null,
    encoding: null,
    fileOrder: 'natural',
    allowHeaderSuperset: false,
    headRows: DEFAULT_HEAD_ROWS,
    tailRows: DEFAULT_TAIL_ROWS,
    scanRows: DEFAULT_SCAN_ROWS,
    outDir: null,
    help: false,
  };

  for (let i = 0; i < argv.length; i += 1) {
    const arg = argv[i];
    const next = argv[i + 1];

    switch (arg) {
      case '--help':
      case '-h':
        args.help = true;
        break;
      case '--file':
        args.files.push(next);
        i += 1;
        break;
      case '--out-dir':
        args.outDir = next;
        i += 1;
        break;
      case '--vendor':
        args.vendor = next;
        i += 1;
        break;
      case '--delimiter':
        args.delimiter = next === '\\t' ? '\t' : next;
        i += 1;
        break;
      case '--encoding':
        args.encoding = next;
        i += 1;
        break;
      case '--role': {
        const splitIndex = String(next).lastIndexOf(':');
        if (splitIndex === -1) {
          throw new Error(`Invalid --role value: ${next} (expected <path>:<role>)`);
        }
        args.explicitRoles[String(next).slice(0, splitIndex)] = String(next).slice(splitIndex + 1);
        i += 1;
        break;
      }
      case '--file-order':
        args.fileOrder = next;
        i += 1;
        break;
      case '--allow-header-superset':
        args.allowHeaderSuperset = true;
        break;
      case '--head-rows':
        args.headRows = Number.parseInt(next, 10);
        i += 1;
        break;
      case '--tail-rows':
        args.tailRows = Number.parseInt(next, 10);
        i += 1;
        break;
      case '--scan-rows':
        args.scanRows = Number.parseInt(next, 10);
        i += 1;
        break;
      default:
        if (arg.startsWith('--')) {
          throw new Error(`Unknown argument: ${arg}`);
        }
    }
  }

  if (!args.vendor) {
    args.vendor = process.env.CSV_VENDOR || null;
  }
  if (!args.delimiter) {
    args.delimiter = process.env.CSV_DELIMITER || null;
  }
  if (!args.encoding) {
    args.encoding = process.env.CSV_ENCODING || null;
  }
  if (!args.outDir) {
    args.outDir = process.env.CSV_DISCOVERY_OUT_DIR || null;
  }
  if (!Number.isFinite(args.headRows) || args.headRows < 1) {
    args.headRows = DEFAULT_HEAD_ROWS;
  }
  if (!Number.isFinite(args.tailRows) || args.tailRows < 0) {
    args.tailRows = DEFAULT_TAIL_ROWS;
  }
  if (!Number.isFinite(args.scanRows) || args.scanRows < 1) {
    args.scanRows = DEFAULT_SCAN_ROWS;
  }
  if (args.fileOrder !== 'stated' && args.fileOrder !== 'natural') {
    args.fileOrder = 'natural';
  }

  return args;
}

// One streaming pass per file: full row count, bounded row retention for
// inference, and per-row shape checks. The file is never bulk-loaded.
async function scanFile(filePath, { delimiter, encoding, scanRows, header }) {
  const rows = [];
  let rowCount = 0;
  let raggedRows = 0;
  let index = 0;

  for await (const row of streamRows(filePath, { delimiter, encoding, trackQuoted: true })) {
    index += 1;
    if (index === 1) {
      continue; // header
    }
    rowCount += 1;
    if (rows.length < scanRows) {
      rows.push(row);
    }
    if (rowWidthReport(header, row.values).ragged) {
      raggedRows += 1;
    }
  }

  return { rows, rowCount, raggedRows, truncated: rowCount > rows.length };
}

function summarizeColumns(header, rows) {
  return header.map((name, columnIdx) => {
    const samples = rows.map((row) => ({
      value: row.values[columnIdx] === undefined ? '' : row.values[columnIdx],
      quoted: Boolean(row.quoted && row.quoted[columnIdx]),
    }));
    return summarizeColumn(name, samples);
  });
}

const PRIVATE_HOST = /^(https?:)?\/\/(localhost|127\.0\.0\.1|0\.0\.0\.0|10\.|192\.168\.|\[::1\])/i;

// Wix Media import fetches from Wix servers, so a local path or private URL in
// an image column is not reachable at import time — the same concern the
// WordPress adapter records for localhost sources.
const MEDIA_LOCATION = /[\\/]|\.(jpe?g|png|gif|webp|avif|svg|bmp|tiff?|mp4|mov|pdf)$/i;

function mediaReachability(columns) {
  const notes = [];
  for (const column of columns) {
    const looksLikeMedia = /image|photo|media|picture|asset|file/i.test(column.name);
    if (!looksLikeMedia || column.examples.length === 0) {
      continue;
    }
    // Only values that look like a location can be unreachable; an "Image
    // Position" column of 1/2/3 is not a media reference.
    const locations = column.examples.filter((example) => MEDIA_LOCATION.test(String(example)));
    if (locations.length === 0) {
      continue;
    }
    const unreachable = locations.filter((example) => PRIVATE_HOST.test(example) || !/^https?:\/\//i.test(example));
    if (unreachable.length > 0) {
      notes.push({
        column: column.name,
        example: unreachable[0],
        note: 'not publicly fetchable by Wix; rewrite with CSV_MEDIA_URL_REWRITE_FROM/TO, expose the files publicly, or skip/defer media',
      });
    }
  }
  return notes;
}

function uniqueEntityName(name, taken) {
  if (!taken.has(name)) {
    taken.add(name);
    return name;
  }
  let suffix = 2;
  while (taken.has(`${name}-${suffix}`)) {
    suffix += 1;
  }
  const unique = `${name}-${suffix}`;
  taken.add(unique);
  return unique;
}

function distinctNonBlank(rows, index) {
  const values = new Set();
  for (const row of rows) {
    const value = row.values[index];
    if (value !== undefined && String(value).trim() !== '') {
      values.add(value);
    }
  }
  return values.size;
}

// Turn one resolved stream into entities. A single grouped vendor file yields
// several: the parent, its inline children, any declared collection (images),
// and one per column-values descriptor.
function deriveEntities(stream, taken) {
  const { header, rows, layout, roles, derived, role } = stream;
  const entities = [];
  const relations = [];

  const parentName = uniqueEntityName(
    layout.parentEntity && layout.parentEntity !== 'record' ? layout.parentEntity : role,
    taken,
  );

  const childLevels = new Set((layout.childLevels || []).map((level) => normalizeHeaderName(level)));
  const discriminatorIdx = columnIndex(header, layout.discriminatorColumn);
  const groups = buildGroups(header, rows, layout);

  let parentOrigin;
  let parentCount;
  if (layout.pattern === 'grouped-by-key') {
    parentOrigin = {
      kind: 'row-group',
      file: stream.primary,
      groupKey: layout.groupKey,
      continuation: layout.continuation,
    };
    parentCount = groups.length;
  } else if (layout.pattern === 'sectioned') {
    parentOrigin = {
      kind: 'file-rows',
      file: stream.primary,
      filter: { column: layout.discriminatorColumn, excludeValues: layout.childLevels || [] },
    };
    parentCount = rows.filter((row) => !childLevels.has(normalizeHeaderName(row.values[discriminatorIdx] || ''))).length;
  } else {
    parentOrigin = { kind: 'file-rows', file: stream.primary };
    parentCount = rows.length;
  }

  entities.push({
    name: parentName,
    origin: parentOrigin,
    recordCount: parentCount,
    inUse: parentCount > 0,
    primaryKey: layout.groupKey || layout.primaryKey || null,
    columns: roles.byEntity[layout.parentEntity] || roles.parentColumns,
    relations: [],
  });

  if (roles.childColumns.length > 0 && layout.childEntity) {
    const childName = uniqueEntityName(layout.childEntity, taken);
    let childCount;
    let childOrigin;
    if (layout.pattern === 'sectioned') {
      childCount = rows.filter((row) => childLevels.has(normalizeHeaderName(row.values[discriminatorIdx] || ''))).length;
      childOrigin = {
        kind: 'file-rows',
        file: stream.primary,
        filter: { column: layout.discriminatorColumn, includeValues: layout.childLevels || [] },
        parentRefColumn: layout.parentRefColumn || null,
      };
    } else {
      // Not `rows.length`: in a grouped file a continuation row can carry only an
      // extra image, leaving every child column blank. Counting those as variants
      // inflates the child entity by exactly the number of image-only rows.
      const childIndexes = roles.childColumns.map((column) => columnIndex(header, column)).filter((index) => index !== -1);
      childCount = rows.filter((row) => childIndexes.some((index) => {
        const value = row.values[index];
        return value !== undefined && String(value).trim() !== '';
      })).length;
      childOrigin = {
        kind: 'row-group',
        file: stream.primary,
        groupKey: layout.groupKey,
        columnGroup: layout.childEntity,
      };
    }
    entities.push({
      name: childName,
      origin: childOrigin,
      recordCount: childCount,
      inUse: childCount > 0,
      columns: roles.childColumns,
      relations: [],
    });
    relations.push({
      from: parentName,
      field: childName,
      targetEntity: childName,
      cardinality: 'one-to-many',
      evidence: layout.pattern === 'sectioned'
        ? `layout.discriminatorColumn=${layout.discriminatorColumn}${layout.parentRefColumn ? ` + parentRef=${layout.parentRefColumn}` : ''}`
        : `layout.groupKey=${layout.groupKey}`,
    });
  }

  // Overlay-declared collections (Shopify's image columns) — a third bucket the
  // data alone cannot separate from the variant columns.
  for (const [entity, columns] of Object.entries(roles.byEntity)) {
    if (entity === layout.parentEntity || entity === layout.childEntity || columns.length === 0) {
      continue;
    }
    const anchorIdx = columnIndex(header, columns[0]);
    const count = anchorIdx === -1 ? 0 : distinctNonBlank(rows, anchorIdx);
    const entityName = uniqueEntityName(entity, taken);
    entities.push({
      name: entityName,
      origin: {
        kind: 'row-group',
        file: stream.primary,
        groupKey: layout.groupKey || null,
        columnGroup: entity,
        distinctBy: columns[0],
      },
      recordCount: count,
      inUse: count > 0,
      columns,
      relations: [],
    });
    relations.push({
      from: parentName,
      field: entityName,
      targetEntity: entityName,
      cardinality: 'one-to-many',
      evidence: `layout.columnGroups[${entity}]`,
    });
  }

  // column-values entities: categories and tags are columns, not files.
  for (const descriptor of derived) {
    if (descriptor.status === 'missing-column') {
      continue;
    }
    const result = deriveColumnValues(header, rows, descriptor, { layout });
    if (result.missingColumn || result.records.length === 0) {
      continue;
    }
    const entityName = uniqueEntityName(descriptor.entity, taken);
    entities.push({
      name: entityName,
      origin: { ...result.origin, file: stream.primary },
      recordCount: result.records.length,
      inUse: result.records.length > 0,
      hierarchical: result.hierarchical,
      columns: [descriptor.fromColumn],
      sampleRecords: result.records.slice(0, 10),
      relations: [],
    });
    relations.push({
      from: parentName,
      field: (descriptor.relation && descriptor.relation.field) || entityName,
      targetEntity: entityName,
      cardinality: (descriptor.relation && descriptor.relation.cardinality) || 'many-to-many',
      evidence: `column:${descriptor.fromColumn}`,
    });
  }

  for (const relation of relations) {
    const owner = entities.find((entity) => entity.name === relation.from);
    if (owner) {
      owner.relations.push({
        field: relation.field,
        targetEntity: relation.targetEntity,
        cardinality: relation.cardinality,
        evidence: relation.evidence,
      });
    }
  }

  return entities;
}

function renderEntityFile(stream, entity, columnsByName) {
  const columnRows = (entity.columns || [])
    .map((name) => columnsByName.get(name))
    .filter(Boolean)
    .map((column) => `| \`${column.name}\` | ${column.type} | ${column.required ? 'yes' : 'no'} | ${column.distinctCount}${column.distinctCapped ? '+' : ''} | ${column.blankCount} | ${column.examples.map((example) => `\`${String(example).slice(0, 60)}\``).join(', ') || '—'} |`)
    .join('\n');

  const relationLines = entity.relations.length > 0
    ? entity.relations.map((relation) => `- \`${relation.field}\` → \`${relation.targetEntity}\` (${relation.cardinality}) — evidence: \`${relation.evidence}\``).join('\n')
    : '- None';

  const derivedBlock = entity.sampleRecords
    ? `\n## Derived records (first ${entity.sampleRecords.length})\n\n\`\`\`json\n${JSON.stringify(entity.sampleRecords, null, 2)}\n\`\`\`\n`
    : '';

  return `# ${entity.name}\n\n`
    + `- Source file: \`${path.basename(stream.primary)}\`\n`
    + `- Role: \`${stream.role}\`\n`
    + `- Vendor: \`${stream.vendor.vendor}\`\n`
    + `- Origin: \`${entity.origin.kind}\`${entity.origin.groupKey ? ` (group key \`${entity.origin.groupKey}\`)` : ''}${entity.origin.column ? ` (column \`${entity.origin.column}\`)` : ''}\n`
    + `- Record count: \`${entity.recordCount}\`${stream.scan.truncated ? ` (from the first ${stream.rows.length} scanned rows)` : ''}\n`
    + `- In use: \`${entity.inUse ? 'yes' : 'no'}\`\n`
    + (entity.hierarchical ? '- Hierarchical: `yes` — a hierarchical source taxonomy mapped to a flat Wix target requires a faithfulness-ledger entry\n' : '')
    + `\n## Relations\n${relationLines}\n`
    + `\n## Columns\n\n| Column | Type | Required | Distinct | Blank | Examples |\n| --- | --- | --- | ---: | ---: | --- |\n${columnRows || '| None | - | - | - | - | - |'}\n`
    + derivedBlock;
}

function renderIndexFile(capture) {
  const fileRows = capture.sourceFiles
    .map((entry) => `| \`${path.basename(entry.file)}\` | ${entry.role} | ${entry.vendor || 'custom'} | ${entry.roleSource} | ${entry.partOf ? `\`${path.basename(entry.partOf)}\`` : '—'} |`)
    .join('\n');

  const streamRowsMd = capture.streams
    .map((stream) => `| \`${path.basename(stream.primary)}\` | \`${stream.layout.pattern}\` | ${stream.layout.groupKey ? `\`${stream.layout.groupKey}\`` : (stream.layout.discriminatorColumn ? `\`${stream.layout.discriminatorColumn}\`` : '—')} | ${stream.layout.confidence} | ${stream.layout.source} | ${stream.layout.halt ? '**halt**' : 'ok'} |`)
    .join('\n');

  const entityRows = capture.streams
    .flatMap((stream) => stream.entities.map((entity) => `| ${entity.name} | \`${entity.origin.kind}\` | ${entity.recordCount} | ${entity.inUse ? 'yes' : 'no'} | ${entity.hierarchical ? 'yes' : '—'} | [${entity.rawFile}](./${entity.rawFile}) |`))
    .join('\n');

  const driftLines = capture.streams.map((stream) => {
    const drift = stream.drift;
    return `### \`${path.basename(stream.primary)}\` (${stream.vendor.vendor})\n\n`
      + `- Unmapped columns (${drift.unmappedColumns.length}): ${drift.unmappedColumns.map((column) => `\`${column}\``).join(', ') || 'none'}\n`
      + `- Missing expected columns (${drift.missingExpectedColumns.length}): ${drift.missingExpectedColumns.map((column) => `\`${column}\``).join(', ') || 'none'}\n`;
  }).join('\n');

  const warningLines = capture.warnings.length > 0
    ? capture.warnings.map((warning) => `- **${warning.kind}**${warning.file ? ` (\`${warning.file}\`)` : ''}: ${warning.message}`).join('\n')
    : '- None';

  const haltBanner = capture.halt
    ? '## ⚠️ Discovery halted to a decision\n\nAt least one file could not be resolved deterministically. Do not synthesize `source-schema.json` from this capture until the questions below are answered with the user.\n\n'
    : '';

  return `# CSV Discovery\n\n`
    + `- Generated at: \`${capture.generatedAt}\`\n`
    + `- Adapter: \`rp-source-csv\`\n`
    + `- Input files: ${capture.sourceFiles.length}\n`
    + `- Logical streams: ${capture.streams.length}\n`
    + `- Entities documented: ${capture.streams.reduce((sum, stream) => sum + stream.entities.length, 0)}\n\n`
    + haltBanner
    + `## Files\n\n| File | Role | Vendor | Role source | Part of |\n| --- | --- | --- | --- | --- |\n${fileRows}\n\n`
    + `## Vendor detection\n\n${capture.streams.map((stream) => `- \`${path.basename(stream.primary)}\` → **${stream.vendor.vendor}** (confidence ${stream.vendor.confidence}, ${stream.vendor.source}) — ${stream.vendor.evidence}${stream.vendor.nearMiss ? `\n  - Near miss: ${stream.vendor.nearMiss.vendor} (missing required: ${stream.vendor.nearMiss.missingRequired.join(', ') || 'none'})` : ''}${stream.vendor.conflict ? `\n  - ⚠️ user stated \`${stream.vendor.conflict.stated}\` but the header looks like \`${stream.vendor.conflict.detected}\`` : ''}`).join('\n')}\n\n`
    + `## Layout\n\n| File | Pattern | Key | Confidence | Source | Status |\n| --- | --- | --- | ---: | --- | --- |\n${streamRowsMd}\n\n`
    + `## Entities\n\n| Entity | Origin | Records | In use | Hierarchical | File |\n| --- | --- | ---: | --- | --- | --- |\n${entityRows || '| None | - | - | - | - | - |'}\n\n`
    + `## Profile drift\n\n${driftLines}\n`
    + `## Warnings\n\n${warningLines}\n\n`
    + `## Cross-file joins\n\n${capture.joins.deferred ? `Deferred: ${capture.joins.reason}` : 'None detected'}\n\n`
    + `Machine-readable capture: [fileset.json](./fileset.json), [raw-capture.json](./raw-capture.json)\n`;
}

function filesetPayload(capture) {
  return {
    generatedAt: capture.generatedAt,
    adapter: 'rp-source-csv',
    platform: 'csv',
    csvInputRoot: capture.csvInputRoot,
    halt: capture.halt,
    sourceFiles: capture.sourceFiles.map((entry) => ({
      ...entry,
      file: path.relative(capture.csvInputRoot, entry.file) || path.basename(entry.file),
      partOf: entry.partOf ? path.relative(capture.csvInputRoot, entry.partOf) || path.basename(entry.partOf) : null,
    })),
    streams: capture.streams.map((stream) => ({
      role: stream.role,
      primary: path.relative(capture.csvInputRoot, stream.primary) || path.basename(stream.primary),
      // The header row AS READ. This is the authoritative column list for the whole pipeline —
      // rp-mapper's deterministic resolver joins it against the vendor overlay, so it belongs in
      // the hand-off artifact and not only in raw-capture.json (which is evidence, not contract).
      header: stream.header,
      parts: stream.parts.map((part) => ({
        file: path.relative(capture.csvInputRoot, part.file) || path.basename(part.file),
        partOf: part.partOf ? path.relative(capture.csvInputRoot, part.partOf) || path.basename(part.partOf) : null,
        headerRelation: part.headerRelation,
        rowOffset: part.rowOffset,
        rowCount: part.rowCount,
      })),
      vendor: {
        name: stream.vendor.vendor,
        confidence: stream.vendor.confidence,
        source: stream.vendor.source,
        evidence: stream.vendor.evidence,
        profileVersion: stream.profile ? stream.profile.profileVersion : null,
        nearMiss: stream.vendor.nearMiss,
        conflict: stream.vendor.conflict,
      },
      dialect: stream.dialect,
      layout: {
        pattern: stream.layout.pattern,
        groupKey: stream.layout.groupKey,
        continuation: stream.layout.continuation,
        discriminatorColumn: stream.layout.discriminatorColumn,
        childLevels: stream.layout.childLevels,
        parentRefColumn: stream.layout.parentRefColumn,
        confidence: stream.layout.confidence,
        source: stream.layout.source,
        halt: stream.layout.halt,
        evidence: stream.layout.evidence,
      },
      drift: stream.drift,
      mappingHints: stream.mappingHints,
      quirks: stream.profile ? stream.profile.quirks || [] : [],
      // Layout-level conflicts (an overlay whose declared grouping the rows
      // contradict) and column-level ones (an overlay bucketing a column against
      // the derived role) land in one list, because a reviewer is asking the same
      // question of both: where did this capture stop believing the overlay?
      layoutConflicts: [...(stream.layout.layoutConflicts || []), ...stream.roles.layoutConflicts],
      ambiguousColumns: stream.roles.ambiguousColumns,
      entities: stream.entities.map((entity) => ({
        name: entity.name,
        // Paths stay relative to csvInputRoot so moving the migration folder
        // does not invalidate the capture.
        origin: {
          ...entity.origin,
          file: path.relative(capture.csvInputRoot, entity.origin.file) || path.basename(entity.origin.file),
        },
        recordCount: entity.recordCount,
        inUse: entity.inUse,
        hierarchical: Boolean(entity.hierarchical),
        rawFile: entity.rawFile,
        columns: entity.columns,
        relations: entity.relations,
      })),
    })),
    joins: capture.joins.joins,
    integrity: capture.joins.integrity,
    joinsDeferred: capture.joins.deferred,
    conflicts: capture.conflicts,
    warnings: capture.warnings,
  };
}

function rawCapturePayload(capture) {
  return {
    generatedAt: capture.generatedAt,
    adapter: 'rp-source-csv',
    streams: capture.streams.map((stream) => ({
      primary: path.relative(capture.csvInputRoot, stream.primary) || path.basename(stream.primary),
      role: stream.role,
      header: stream.header,
      dialect: stream.dialect,
      scan: {
        rowCount: stream.scan.rowCount,
        scannedRows: stream.rows.length,
        truncated: stream.scan.truncated,
        raggedRows: stream.scan.raggedRows,
      },
      columns: stream.columns,
      headSample: stream.headSample,
      tailSample: stream.tailSample,
      mediaNotes: stream.mediaNotes,
    })),
  };
}

async function capture(args) {
  const profiles = loadVendorProfiles();
  const generatedAt = new Date().toISOString();
  const resolvedFiles = args.files.map((file) => path.resolve(file));
  const csvInputRoot = process.env.CSV_INPUT_ROOT
    ? path.resolve(process.env.CSV_INPUT_ROOT)
    : (resolvedFiles.length === 1 ? path.dirname(resolvedFiles[0]) : commonDirectory(resolvedFiles));

  // 1. header + dialect + vendor per file
  const scanned = [];
  for (let i = 0; i < resolvedFiles.length; i += 1) {
    const file = resolvedFiles[i];
    progress?.progress(`Reading header of ${path.basename(file)}`, {
      phase: 'discovery',
      step: 'read-header',
      entity: path.basename(file),
      count: i + 1,
      total: resolvedFiles.length,
      unit: 'files',
    });
    const head = await readHeaderRow(file, { delimiter: args.delimiter, encoding: args.encoding });
    const vendor = detectVendor(head.header, profiles, { statedVendor: args.vendor });
    const scan = await scanFile(file, {
      delimiter: head.delimiter,
      encoding: head.encoding,
      scanRows: args.scanRows,
      header: head.header,
    });
    scanned.push({ file, head, vendor, scan });
  }

  // 2. roles, split-file concatenation
  const fileSet = resolveFileSet(
    scanned.map((entry) => ({
      file: entry.file,
      header: entry.head.header,
      vendor: entry.vendor.vendor,
      profile: entry.vendor.profile,
      rowCount: entry.scan.rowCount,
    })),
    {
      explicitRoles: args.explicitRoles,
      policy: args.allowHeaderSuperset ? CONCAT_POLICY.UNION : CONCAT_POLICY.STRICT_SET,
      fileOrder: args.fileOrder,
    },
  );

  const byFile = new Map(scanned.map((entry) => [entry.file, entry]));
  const warnings = [];
  const takenEntityNames = new Set();
  const streams = [];

  for (const stream of fileSet.streams) {
    const primary = byFile.get(stream.primary);
    // Split parts share one logical stream, so their rows are inferred together.
    const rows = stream.parts.flatMap((part) => (byFile.get(part.file) || { scan: { rows: [] } }).scan.rows);
    const header = primary.head.header;
    const profile = primary.vendor.profile;

    progress?.progress(`Resolving layout for ${path.basename(stream.primary)}`, {
      phase: 'discovery',
      step: 'resolve-layout',
      entity: path.basename(stream.primary),
    });

    const layout = classifyLayout(header, rows, { overlayLayout: profile ? profile.layout : null });
    const roles = deriveColumnRoles(header, rows, layout, {
      columnGroups: (profile && profile.layout.columnGroups) || [],
    });
    const derived = detectDerivedCandidates(header, rows, { profile });
    const drift = diffProfile(header, profile);
    const columns = summarizeColumns(header, rows);
    const mediaNotes = mediaReachability(columns);

    const tail = await readTailSample(stream.primary, {
      delimiter: primary.head.delimiter,
      encoding: primary.head.encoding,
      header,
    });

    const resolved = {
      role: stream.role,
      primary: stream.primary,
      parts: stream.parts,
      header,
      rows,
      columns,
      profile,
      vendor: primary.vendor,
      layout,
      roles,
      derived,
      drift,
      mediaNotes,
      scan: {
        rowCount: stream.parts.reduce((sum, part) => sum + (part.rowCount || 0), 0),
        truncated: stream.parts.some((part) => (byFile.get(part.file) || { scan: {} }).scan.truncated),
        raggedRows: stream.parts.reduce((sum, part) => sum + ((byFile.get(part.file) || { scan: {} }).scan.raggedRows || 0), 0),
      },
      dialect: {
        delimiter: primary.head.delimiter,
        delimiterAmbiguous: primary.head.delimiterAmbiguous,
        encoding: primary.head.encoding,
        bom: primary.head.bom,
        lineEnding: primary.head.lineEnding,
        emptyPolicy: detectEmptyPolicy(rows),
      },
      headSample: rows.slice(0, args.headRows).map((row) => row.values),
      tailSample: tail.skipped ? [] : tail.rows.slice(-args.tailRows).map((row) => row.values),
      tailSampleSkipped: tail.skipped ? tail.reason : null,
      mappingHints: prefillColumnMap(header, profile),
    };

    resolved.entities = deriveEntities(resolved, takenEntityNames);
    for (const entity of resolved.entities) {
      entity.rawFile = `${stream.role}--${entity.name}.md`;
    }
    streams.push(resolved);

    if (resolved.scan.rowCount === 0) {
      warnings.push({
        kind: 'empty-file',
        file: path.basename(stream.primary),
        message: 'the file has a header but no data rows; its entities are advertised but unused',
      });
    }
    if (layout.halt) {
      warnings.push({
        kind: 'layout-ambiguous',
        file: path.basename(stream.primary),
        message: `layout could not be resolved deterministically (${layout.evidence.join('; ')}) — ask the user how rows group before synthesizing the schema`,
      });
    }
    for (const conflict of layout.layoutConflicts || []) {
      warnings.push({
        kind: conflict.kind,
        file: path.basename(stream.primary),
        message: `the ${primary.vendor.vendor} overlay's declared layout is not what this file's rows show — ${conflict.resolution}. Details: ${layout.evidence[0]}`,
      });
    }
    if (primary.vendor.nearMiss) {
      warnings.push({
        kind: 'vendor-near-miss',
        file: path.basename(stream.primary),
        message: `header looks like a drifted ${primary.vendor.nearMiss.vendor} export (missing required: ${primary.vendor.nearMiss.missingRequired.join(', ') || 'none'}); confirm with the user or set CSV_VENDOR`,
      });
    }
    if (primary.vendor.conflict) {
      warnings.push({
        kind: 'vendor-conflict',
        file: path.basename(stream.primary),
        message: `user stated "${primary.vendor.conflict.stated}" but the header fingerprints as "${primary.vendor.conflict.detected}"`,
      });
    }
    if (resolved.dialect.delimiterAmbiguous) {
      warnings.push({
        kind: 'delimiter-ambiguous',
        file: path.basename(stream.primary),
        message: 'delimiter could not be detected confidently; set CSV_DELIMITER',
      });
    }
    if (resolved.scan.raggedRows > 0) {
      warnings.push({
        kind: 'ragged-rows',
        file: path.basename(stream.primary),
        message: `${resolved.scan.raggedRows} row(s) do not have the header's column count`,
      });
    }
    if (resolved.tailSampleSkipped) {
      warnings.push({
        kind: 'tail-sample-skipped',
        file: path.basename(stream.primary),
        message: `tail sample skipped (${resolved.tailSampleSkipped})`,
      });
    }
    for (const note of mediaNotes) {
      warnings.push({
        kind: 'media-reachability',
        file: path.basename(stream.primary),
        message: `column "${note.column}" holds values Wix cannot fetch (e.g. ${note.example}): ${note.note}`,
      });
    }
    if (roles.ambiguousColumns.length > 0) {
      warnings.push({
        kind: 'ambiguous-columns',
        file: path.basename(stream.primary),
        message: `could not decide which entity these columns belong to: ${roles.ambiguousColumns.join(', ')}`,
      });
    }
    for (const candidate of derived) {
      if (candidate.status === 'proposed') {
        warnings.push({
          kind: 'derived-entity-proposed',
          file: path.basename(stream.primary),
          message: `column "${candidate.fromColumn}" may be a "${candidate.entity}" entity (confidence ${candidate.confidence}); confirm with the user before treating it as one`,
        });
      }
      if (candidate.status === 'missing-column') {
        warnings.push({
          kind: 'derived-column-missing',
          file: path.basename(stream.primary),
          message: `the ${primary.vendor.vendor} overlay expects a "${candidate.fromColumn}" column for the "${candidate.entity}" entity, but this file has none`,
        });
      }
    }
  }

  for (const conflict of fileSet.conflicts) {
    warnings.push({ kind: conflict.kind, file: (conflict.files || []).map((file) => path.basename(file)).join(', '), message: conflict.message });
  }

  return {
    generatedAt,
    csvInputRoot,
    sourceFiles: fileSet.sourceFiles,
    streams,
    conflicts: fileSet.conflicts,
    joins: inferJoins(),
    warnings,
    halt: fileSet.halt || streams.some((stream) => stream.layout.halt),
  };
}

function commonDirectory(files) {
  if (files.length === 0) {
    return process.cwd();
  }
  const segments = files.map((file) => path.dirname(file).split(path.sep));
  const common = [];
  for (let i = 0; i < segments[0].length; i += 1) {
    const candidate = segments[0][i];
    if (segments.every((parts) => parts[i] === candidate)) {
      common.push(candidate);
    } else {
      break;
    }
  }
  return common.join(path.sep) || path.sep;
}

async function writeOutputs(outDir, result) {
  await fs.mkdir(outDir, { recursive: true });
  await fs.writeFile(path.join(outDir, 'README.md'), renderIndexFile(result), 'utf8');
  progress?.progress('Wrote CSV discovery index', { phase: 'discovery', step: 'write-artifact', artifact: path.join(outDir, 'README.md') });

  await fs.writeFile(path.join(outDir, 'fileset.json'), `${JSON.stringify(filesetPayload(result), null, 2)}\n`, 'utf8');
  await fs.writeFile(path.join(outDir, 'raw-capture.json'), `${JSON.stringify(rawCapturePayload(result), null, 2)}\n`, 'utf8');
  progress?.progress('Wrote CSV discovery machine capture', { phase: 'discovery', step: 'write-artifact', artifact: path.join(outDir, 'fileset.json') });

  let entityCount = 0;
  for (const stream of result.streams) {
    const columnsByName = new Map(stream.columns.map((column) => [column.name, column]));
    for (const entity of stream.entities) {
      await fs.writeFile(path.join(outDir, entity.rawFile), renderEntityFile(stream, entity, columnsByName), 'utf8');
      entityCount += 1;
      progress?.progress(`Wrote CSV discovery artifact for ${entity.name}`, {
        phase: 'discovery',
        step: 'write-artifact',
        entity: entity.name,
        artifact: path.join(outDir, entity.rawFile),
      });
    }
  }
  return entityCount;
}

async function main() {
  const parsed = parseProgressArgs(process.argv.slice(2));
  progress = createProgressLogger({
    script: 'skills/wix-replatform/resources/rp-source-csv/scripts/csv-discovery.js',
    ...parsed.progress,
  });
  progress.start('CSV discovery started', { phase: 'discovery' });

  const args = parseArgs(parsed.args);
  if (args.help) {
    printUsage();
    progress.complete('CSV discovery help shown', { phase: 'discovery', step: 'help' });
    return;
  }
  if (args.files.length === 0 || !args.outDir) {
    printUsage();
    progress.error('Missing required CSV discovery arguments', { phase: 'discovery' });
    throw new Error('Missing required arguments: at least one --file and --out-dir are required.');
  }

  const result = await capture(args);
  const entityCount = await writeOutputs(path.resolve(args.outDir), result);

  console.log(`Wrote ${entityCount} entity file(s), README.md, fileset.json and raw-capture.json to ${args.outDir}`);
  if (result.halt) {
    console.log('Discovery halted to a decision — see the Warnings section of README.md before synthesizing source-schema.json.');
  }
  progress.complete('CSV discovery completed', {
    phase: 'discovery',
    artifact: args.outDir,
    count: entityCount,
    unit: 'entities',
  });
}

if (require.main === module) {
  main().catch((error) => {
    console.error(error.stack || error.message);
    if (progress) {
      progress.error(error && error.message ? error.message : 'CSV discovery failed', { phase: 'discovery' });
    }
    process.exitCode = 1;
  });
}

module.exports = {
  DEFAULT_HEAD_ROWS,
  DEFAULT_TAIL_ROWS,
  DEFAULT_SCAN_ROWS,
  parseArgs,
  scanFile,
  summarizeColumns,
  mediaReachability,
  deriveEntities,
  renderIndexFile,
  renderEntityFile,
  filesetPayload,
  rawCapturePayload,
  capture,
};