#!/usr/bin/env -S npx tsx

import { createHash } from 'node:crypto';
import { lstat, readdir, readFile, realpath, stat } from 'node:fs/promises';
import path from 'node:path';
import { fileURLToPath } from 'node:url';

import { config as loadEnvironment } from 'dotenv';
import { Client } from 'pg';
import type { ClientConfig, QueryResultRow } from 'pg';

type JsonObject = Record<string, unknown>;
type DatasetName =
  | 'kv'
  | 'authUsers'
  | 'authIdentities'
  | 'storageBuckets'
  | 'storageObjects';

interface CliOptions {
  exportDirectory?: string;
  replace: boolean;
  help: boolean;
}

interface ManifestFile {
  path: string;
  dataset: DatasetName;
  page: number;
  rowCount: number;
  bytes: number;
  sha256: string;
}

interface DatasetManifest {
  table: string;
  cursorColumn: string;
  columns: string[];
  sensitive: boolean;
  sourceCountBefore: number;
  sourceCountAfter: number;
  rowCount: number;
  pages: number;
}

interface ExportManifest {
  format: 'tonline-supabase-source-export';
  formatVersion: 1;
  export: {
    binaryObjectsDownloaded: boolean;
    rowCount: number;
    fileCount: number;
  };
  datasets: Record<DatasetName, DatasetManifest>;
  files: ManifestFile[];
}

interface DatasetSpec {
  name: DatasetName;
  sourceTable: string;
  targetSchema: string;
  targetTable: string;
  stageTable: string;
  primaryKey: string;
  requiredColumns: readonly string[];
  targetColumns: readonly string[];
  jsonColumns: ReadonlySet<string>;
  hasUpdatedAtTrigger: boolean;
}

interface ValidatedDataset {
  spec: DatasetSpec;
  manifest: DatasetManifest;
  rows: JsonObject[];
  mappedColumns: string[];
  sourceHash: string;
}

interface ValidatedExport {
  directory: string;
  manifest: ExportManifest;
  datasets: Map<DatasetName, ValidatedDataset>;
}

interface TriggerState {
  spec: DatasetSpec;
  state: string;
}

const scriptDirectory = path.dirname(fileURLToPath(import.meta.url));
const backendDirectory = path.resolve(scriptDirectory, '..');
const defaultExportRoot = path.join(backendDirectory, 'data', 'migration');

loadEnvironment({
  path: path.join(backendDirectory, '.env'),
  override: false,
  quiet: true,
});

const userColumns = [
  'instance_id',
  'id',
  'aud',
  'role',
  'email',
  'encrypted_password',
  'email_confirmed_at',
  'invited_at',
  'confirmation_token',
  'confirmation_sent_at',
  'recovery_token',
  'recovery_sent_at',
  'email_change_token_new',
  'email_change',
  'email_change_sent_at',
  'last_sign_in_at',
  'raw_app_meta_data',
  'raw_user_meta_data',
  'is_super_admin',
  'created_at',
  'updated_at',
  'phone',
  'phone_confirmed_at',
  'phone_change',
  'phone_change_token',
  'phone_change_sent_at',
  'confirmed_at',
  'email_change_token_current',
  'email_change_confirm_status',
  'banned_until',
  'reauthentication_token',
  'reauthentication_sent_at',
  'is_sso_user',
  'deleted_at',
  'is_anonymous',
] as const;

const identityColumns = [
  'id',
  'user_id',
  'provider_id',
  'identity_data',
  'provider',
  'email',
  'last_sign_in_at',
  'created_at',
  'updated_at',
] as const;

const bucketColumns = [
  'id',
  'name',
  'owner',
  'owner_id',
  'public',
  'file_size_limit',
  'allowed_mime_types',
  'avif_autodetection',
  'type',
  'created_at',
  'updated_at',
] as const;

const objectColumns = [
  'id',
  'bucket_id',
  'name',
  'owner',
  'owner_id',
  'created_at',
  'updated_at',
  'last_accessed_at',
  'metadata',
  'user_metadata',
  'path_tokens',
  'version',
] as const;

const datasetSpecs: readonly DatasetSpec[] = [
  {
    name: 'kv',
    sourceTable: 'public.kv_store_7249dcd9',
    targetSchema: 'public',
    targetTable: 'kv_store_7249dcd9',
    stageTable: 'import_source_kv',
    primaryKey: 'key',
    requiredColumns: ['key', 'value'],
    targetColumns: ['key', 'value'],
    jsonColumns: new Set(['value']),
    hasUpdatedAtTrigger: false,
  },
  {
    name: 'authUsers',
    sourceTable: 'auth.users',
    targetSchema: 'app_auth',
    targetTable: 'users',
    stageTable: 'import_source_auth_users',
    primaryKey: 'id',
    requiredColumns: ['id', 'created_at', 'updated_at'],
    targetColumns: userColumns,
    jsonColumns: new Set(['raw_app_meta_data', 'raw_user_meta_data']),
    hasUpdatedAtTrigger: true,
  },
  {
    name: 'authIdentities',
    sourceTable: 'auth.identities',
    targetSchema: 'app_auth',
    targetTable: 'identities',
    stageTable: 'import_source_auth_identities',
    primaryKey: 'id',
    requiredColumns: [
      'id',
      'user_id',
      'provider_id',
      'identity_data',
      'provider',
      'created_at',
      'updated_at',
    ],
    targetColumns: identityColumns,
    jsonColumns: new Set(['identity_data']),
    hasUpdatedAtTrigger: true,
  },
  {
    name: 'storageBuckets',
    sourceTable: 'storage.buckets',
    targetSchema: 'local_storage',
    targetTable: 'buckets',
    stageTable: 'import_source_storage_buckets',
    primaryKey: 'id',
    requiredColumns: ['id', 'name', 'created_at', 'updated_at'],
    targetColumns: bucketColumns,
    jsonColumns: new Set(),
    hasUpdatedAtTrigger: true,
  },
  {
    name: 'storageObjects',
    sourceTable: 'storage.objects',
    targetSchema: 'local_storage',
    targetTable: 'objects',
    stageTable: 'import_source_storage_objects',
    primaryKey: 'id',
    requiredColumns: [
      'id',
      'bucket_id',
      'name',
      'created_at',
      'updated_at',
      'last_accessed_at',
    ],
    targetColumns: objectColumns,
    jsonColumns: new Set(['metadata', 'user_metadata']),
    hasUpdatedAtTrigger: true,
  },
];

const specsByName = new Map(datasetSpecs.map((spec) => [spec.name, spec]));

function printHelp(): void {
  console.log(`Usage: npx tsx backend/scripts/import-source.ts [options] [export-directory]

Validates and imports a complete source export into the local PostgreSQL database.
With no directory, the newest complete export in backend/data/migration is used.

Options:
  --export-directory <path>  Import this export directory
  --replace                  Atomically replace the five exported datasets
  --help                     Show this help

By default, the import is refused if any target dataset already contains rows.`);
}

function parseCli(arguments_: string[]): CliOptions {
  const options: CliOptions = { replace: false, help: false };

  for (let index = 0; index < arguments_.length; index += 1) {
    const argument = arguments_[index];
    if (argument === '--help' || argument === '-h') {
      options.help = true;
      continue;
    }
    if (argument === '--replace') {
      options.replace = true;
      continue;
    }
    if (argument === '--export-directory' || argument === '--export') {
      const value = arguments_[index + 1];
      if (value === undefined || value.startsWith('--')) {
        throw new Error(`${argument} requires a path`);
      }
      if (options.exportDirectory !== undefined) {
        throw new Error('Only one export directory may be supplied');
      }
      options.exportDirectory = path.resolve(value);
      index += 1;
      continue;
    }
    if (argument.startsWith('-')) throw new Error(`Unknown argument: ${argument}`);
    if (options.exportDirectory !== undefined) {
      throw new Error('Only one export directory may be supplied');
    }
    options.exportDirectory = path.resolve(argument);
  }

  return options;
}

function nonNegativeInteger(value: unknown, label: string): number {
  if (!Number.isSafeInteger(value) || (value as number) < 0) {
    throw new Error(`${label} must be a non-negative integer`);
  }
  return value as number;
}

function positiveInteger(value: unknown, label: string): number {
  const parsed = nonNegativeInteger(value, label);
  if (parsed === 0) throw new Error(`${label} must be greater than zero`);
  return parsed;
}

function nonEmptyString(value: unknown, label: string): string {
  if (typeof value !== 'string' || value.length === 0) {
    throw new Error(`${label} must be a non-empty string`);
  }
  return value;
}

function isJsonObject(value: unknown): value is JsonObject {
  return value !== null && typeof value === 'object' && !Array.isArray(value);
}

function sha256(contents: Uint8Array | string): string {
  return createHash('sha256').update(contents).digest('hex');
}

function canonicalJson(value: unknown): string {
  if (value === null) return 'null';
  if (typeof value === 'string' || typeof value === 'boolean') {
    return JSON.stringify(value);
  }
  if (typeof value === 'number') {
    if (!Number.isFinite(value)) throw new Error('Canonical JSON cannot contain a non-finite number');
    return JSON.stringify(value);
  }
  if (Array.isArray(value)) {
    return `[${value.map((item) => canonicalJson(item)).join(',')}]`;
  }
  if (isJsonObject(value)) {
    const properties = Object.keys(value)
      .sort()
      .map((key) => `${JSON.stringify(key)}:${canonicalJson(value[key])}`);
    return `{${properties.join(',')}}`;
  }
  throw new Error('Canonical JSON received an unsupported value');
}

function compareCursorRows(left: JsonObject, right: JsonObject, cursorColumn: string): number {
  const leftCursor = left[cursorColumn];
  const rightCursor = right[cursorColumn];
  if (typeof leftCursor !== 'string' || typeof rightCursor !== 'string') {
    throw new Error(`Dataset cursor ${cursorColumn} is not a string`);
  }
  return leftCursor < rightCursor ? -1 : leftCursor > rightCursor ? 1 : 0;
}

function canonicalRowsHash(rows: JsonObject[], cursorColumn: string): string {
  const hash = createHash('sha256');
  const sortedRows = [...rows].sort((left, right) =>
    compareCursorRows(left, right, cursorColumn));
  for (const row of sortedRows) hash.update(canonicalJson(row)).update('\n');
  return hash.digest('hex');
}

function typedMismatchDetails(
  stagedRows: JsonObject[],
  importedRows: JsonObject[],
  primaryKey: string,
): string {
  const sortRows = (rows: JsonObject[]) => [...rows].sort((left, right) =>
    canonicalJson(left[primaryKey]).localeCompare(canonicalJson(right[primaryKey])),
  );
  const staged = sortRows(stagedRows);
  const imported = sortRows(importedRows);
  const rowCount = Math.min(staged.length, imported.length);
  for (let index = 0; index < rowCount; index += 1) {
    const left = staged[index]!;
    const right = imported[index]!;
    const columns = [...new Set([...Object.keys(left), ...Object.keys(right)])].sort();
    const different = columns.filter(
      (column) => canonicalJson(left[column]) !== canonicalJson(right[column]),
    );
    if (different.length > 0) {
      return ` at sorted row ${index + 1}; columns: ${different.join(', ')}`;
    }
  }
  return staged.length === imported.length
    ? ''
    : `; staged rows ${staged.length}, imported rows ${imported.length}`;
}

function parseManifest(payload: unknown): ExportManifest {
  if (!isJsonObject(payload)) throw new Error('manifest.json must contain an object');
  if (payload.format !== 'tonline-supabase-source-export' || payload.formatVersion !== 1) {
    throw new Error('Unsupported export manifest format');
  }
  if (!isJsonObject(payload.export)) throw new Error('Manifest has no export summary');
  if (!isJsonObject(payload.datasets)) throw new Error('Manifest has no datasets');
  if (!Array.isArray(payload.files)) throw new Error('Manifest has no file list');

  nonNegativeInteger(payload.export.rowCount, 'export.rowCount');
  nonNegativeInteger(payload.export.fileCount, 'export.fileCount');
  if (typeof payload.export.binaryObjectsDownloaded !== 'boolean') {
    throw new Error('export.binaryObjectsDownloaded must be boolean');
  }
  if (payload.export.binaryObjectsDownloaded) {
    throw new Error('This importer does not support an export containing storage binaries');
  }

  const datasetNames = Object.keys(payload.datasets);
  const expectedNames = datasetSpecs.map((spec) => spec.name);
  const missingDataset = expectedNames.find((name) => !datasetNames.includes(name));
  const unknownDataset = datasetNames.find((name) => !specsByName.has(name as DatasetName));
  if (missingDataset !== undefined) {
    throw new Error(`Incomplete export: dataset ${missingDataset} is missing`);
  }
  if (unknownDataset !== undefined || datasetNames.length !== expectedNames.length) {
    throw new Error(`Unsupported dataset in manifest: ${unknownDataset ?? 'duplicate dataset entry'}`);
  }

  for (const spec of datasetSpecs) {
    const rawDataset = payload.datasets[spec.name];
    if (!isJsonObject(rawDataset)) throw new Error(`Invalid dataset manifest: ${spec.name}`);
    if (rawDataset.table !== spec.sourceTable) {
      throw new Error(`Unexpected source table for dataset ${spec.name}`);
    }
    if (rawDataset.cursorColumn !== spec.primaryKey) {
      throw new Error(`Unexpected cursor column for dataset ${spec.name}`);
    }
    if (!Array.isArray(rawDataset.columns) || rawDataset.columns.length === 0) {
      throw new Error(`${spec.name}.columns must be a non-empty array`);
    }
    const columns = rawDataset.columns.map((column, index) =>
      nonEmptyString(column, `${spec.name}.columns[${index}]`));
    if (new Set(columns).size !== columns.length) {
      throw new Error(`Duplicate column in dataset ${spec.name}`);
    }
    for (const required of spec.requiredColumns) {
      if (!columns.includes(required)) {
        throw new Error(`Dataset ${spec.name} is missing required column ${required}`);
      }
    }
    if (spec.name === 'kv' &&
      (columns.length !== 2 || !columns.includes('key') || !columns.includes('value'))) {
      throw new Error('The KV dataset contains unsupported columns that cannot be preserved');
    }
    if (typeof rawDataset.sensitive !== 'boolean') {
      throw new Error(`${spec.name}.sensitive must be boolean`);
    }
    nonNegativeInteger(rawDataset.sourceCountBefore, `${spec.name}.sourceCountBefore`);
    nonNegativeInteger(rawDataset.sourceCountAfter, `${spec.name}.sourceCountAfter`);
    nonNegativeInteger(rawDataset.rowCount, `${spec.name}.rowCount`);
    positiveInteger(rawDataset.pages, `${spec.name}.pages`);
  }

  for (const [index, rawFile] of payload.files.entries()) {
    if (!isJsonObject(rawFile)) throw new Error(`Invalid file entry at index ${index}`);
    nonEmptyString(rawFile.path, `files[${index}].path`);
    const dataset = nonEmptyString(rawFile.dataset, `files[${index}].dataset`);
    if (!specsByName.has(dataset as DatasetName)) {
      throw new Error(`File references unsupported dataset: ${dataset}`);
    }
    positiveInteger(rawFile.page, `files[${index}].page`);
    nonNegativeInteger(rawFile.rowCount, `files[${index}].rowCount`);
    nonNegativeInteger(rawFile.bytes, `files[${index}].bytes`);
    if (typeof rawFile.sha256 !== 'string' || !/^[a-f0-9]{64}$/.test(rawFile.sha256)) {
      throw new Error(`Invalid file checksum at index ${index}`);
    }
  }

  return payload as unknown as ExportManifest;
}

function normalizedPathKey(relativePath: string): string {
  return process.platform === 'win32' ? relativePath.toLowerCase() : relativePath;
}

function resolveArtifact(exportDirectory: string, relativePath: string): string {
  if (
    path.isAbsolute(relativePath) ||
    relativePath.includes('\0') ||
    relativePath.includes('\\') ||
    relativePath.split('/').some((part) => part === '' || part === '.' || part === '..')
  ) {
    throw new Error(`Unsafe manifest path: ${relativePath}`);
  }
  const root = path.resolve(exportDirectory);
  const resolved = path.resolve(root, ...relativePath.split('/'));
  if (!resolved.startsWith(`${root}${path.sep}`)) {
    throw new Error(`Manifest path escapes export directory: ${relativePath}`);
  }
  return resolved;
}

async function assertRegularFileWithinRoot(
  exportDirectory: string,
  rootRealPath: string,
  relativePath: string,
): Promise<string> {
  const diskPath = resolveArtifact(exportDirectory, relativePath);
  const fileStats = await lstat(diskPath);
  if (!fileStats.isFile() || fileStats.isSymbolicLink()) {
    throw new Error(`Export artifact is not a regular file: ${relativePath}`);
  }
  const artifactRealPath = await realpath(diskPath);
  if (!artifactRealPath.startsWith(`${rootRealPath}${path.sep}`)) {
    throw new Error(`Export artifact escapes its directory: ${relativePath}`);
  }
  return diskPath;
}

async function listFiles(directory: string, prefix = ''): Promise<string[]> {
  const found: string[] = [];
  for (const entry of await readdir(directory, { withFileTypes: true })) {
    const relativePath = prefix ? path.posix.join(prefix, entry.name) : entry.name;
    const diskPath = path.join(directory, entry.name);
    if (entry.isSymbolicLink()) throw new Error(`Symbolic link in export: ${relativePath}`);
    if (entry.isDirectory()) {
      found.push(...await listFiles(diskPath, relativePath));
    } else if (entry.isFile()) {
      found.push(relativePath);
    } else {
      throw new Error(`Unsupported filesystem entry in export: ${relativePath}`);
    }
  }
  return found.sort();
}

async function newestCompleteExportDirectory(): Promise<string> {
  let entries;
  try {
    entries = await readdir(defaultExportRoot, { withFileTypes: true });
  } catch (error) {
    if ((error as NodeJS.ErrnoException).code === 'ENOENT') {
      throw new Error(`No migration exports found in ${defaultExportRoot}`);
    }
    throw error;
  }

  const candidates: Array<{ directory: string; modified: number }> = [];
  for (const entry of entries) {
    if (!entry.isDirectory() || entry.isSymbolicLink() || entry.name.startsWith('.')) continue;
    const directory = path.join(defaultExportRoot, entry.name);
    try {
      const [manifestStats, checksumStats] = await Promise.all([
        stat(path.join(directory, 'manifest.json')),
        stat(path.join(directory, 'manifest.sha256')),
      ]);
      if (manifestStats.isFile() && checksumStats.isFile()) {
        candidates.push({ directory, modified: manifestStats.mtimeMs });
      }
    } catch {
      // A directory without both final manifest files is an incomplete export.
    }
  }

  candidates.sort((left, right) => right.modified - left.modified);
  const newest = candidates[0];
  if (newest === undefined) {
    throw new Error(`No complete migration exports found in ${defaultExportRoot}`);
  }
  return newest.directory;
}

function sameStringSet(left: string[], right: string[]): boolean {
  if (left.length !== right.length) return false;
  const sortedLeft = [...left].sort();
  const sortedRight = [...right].sort();
  return sortedLeft.every((value, index) => value === sortedRight[index]);
}

async function validateExport(exportDirectory: string): Promise<ValidatedExport> {
  const directory = path.resolve(exportDirectory);
  const directoryStats = await lstat(directory);
  if (!directoryStats.isDirectory()) throw new Error('Export path is not a directory');
  const rootRealPath = await realpath(directory);
  const manifestPath = await assertRegularFileWithinRoot(
    directory,
    rootRealPath,
    'manifest.json',
  );
  const checksumPath = await assertRegularFileWithinRoot(
    directory,
    rootRealPath,
    'manifest.sha256',
  );
  const [manifestContents, checksumContents] = await Promise.all([
    readFile(manifestPath),
    readFile(checksumPath, 'ascii'),
  ]);

  const checksumMatch = /^([a-f0-9]{64})\s+manifest\.json\s*$/i.exec(checksumContents);
  if (checksumMatch?.[1] === undefined) {
    throw new Error('manifest.sha256 has an invalid format');
  }
  if (sha256(manifestContents) !== checksumMatch[1].toLowerCase()) {
    throw new Error('manifest.json checksum does not match manifest.sha256');
  }

  let manifestPayload: unknown;
  try {
    manifestPayload = JSON.parse(manifestContents.toString('utf8'));
  } catch {
    throw new Error('manifest.json is not valid JSON');
  }
  const manifest = parseManifest(manifestPayload);
  if (manifest.files.length !== manifest.export.fileCount) {
    throw new Error('Manifest file count does not match export summary');
  }

  const expectedPaths = new Map<string, string>([
    [normalizedPathKey('manifest.json'), 'manifest.json'],
    [normalizedPathKey('manifest.sha256'), 'manifest.sha256'],
  ]);
  const pageAggregates = new Map<DatasetName, { rows: number; pages: number[] }>();
  const rowsByDataset = new Map<DatasetName, JsonObject[]>(
    datasetSpecs.map((spec) => [spec.name, []]),
  );
  const cursorsByDataset = new Map<DatasetName, Set<string>>(
    datasetSpecs.map((spec) => [spec.name, new Set<string>()]),
  );

  for (const file of manifest.files) {
    const normalizedPath = normalizedPathKey(file.path);
    if (expectedPaths.has(normalizedPath)) {
      throw new Error(`Duplicate or conflicting manifest path: ${file.path}`);
    }
    expectedPaths.set(normalizedPath, file.path);

    const diskPath = await assertRegularFileWithinRoot(
      directory,
      rootRealPath,
      file.path,
    );
    const contents = await readFile(diskPath);
    if (contents.byteLength !== file.bytes) {
      throw new Error(`Byte count mismatch for export artifact: ${file.path}`);
    }
    if (sha256(contents) !== file.sha256) {
      throw new Error(`Checksum mismatch for export artifact: ${file.path}`);
    }

    let pagePayload: unknown;
    try {
      pagePayload = JSON.parse(contents.toString('utf8'));
    } catch {
      throw new Error(`Invalid JSON in export artifact: ${file.path}`);
    }
    if (!isJsonObject(pagePayload)) throw new Error(`Invalid page object: ${file.path}`);
    if (
      pagePayload.formatVersion !== 1 ||
      pagePayload.dataset !== file.dataset ||
      pagePayload.page !== file.page ||
      pagePayload.rowCount !== file.rowCount
    ) {
      throw new Error(`Page metadata mismatch: ${file.path}`);
    }
    if (!Array.isArray(pagePayload.rows) || pagePayload.rows.length !== file.rowCount) {
      throw new Error(`Row count mismatch inside export artifact: ${file.path}`);
    }

    const spec = specsByName.get(file.dataset);
    const datasetManifest = manifest.datasets[file.dataset];
    const datasetRows = rowsByDataset.get(file.dataset);
    const datasetCursors = cursorsByDataset.get(file.dataset);
    if (
      spec === undefined ||
      datasetManifest === undefined ||
      datasetRows === undefined ||
      datasetCursors === undefined
    ) {
      throw new Error(`Unknown dataset in export artifact: ${file.dataset}`);
    }

    for (const [rowIndex, rawRow] of pagePayload.rows.entries()) {
      if (!isJsonObject(rawRow)) {
        throw new Error(`Invalid row in ${file.path} at index ${rowIndex}`);
      }
      if (!sameStringSet(Object.keys(rawRow), datasetManifest.columns)) {
        throw new Error(`Row columns do not match the manifest in ${file.path}`);
      }
      const cursor = rawRow[spec.primaryKey];
      if (typeof cursor !== 'string' || cursor.length === 0) {
        throw new Error(`Invalid row cursor in ${file.path}`);
      }
      if (datasetCursors.has(cursor)) {
        throw new Error(`Duplicate row cursor in dataset ${file.dataset}`);
      }
      datasetCursors.add(cursor);
      datasetRows.push(rawRow);
    }

    const aggregate = pageAggregates.get(file.dataset) ?? { rows: 0, pages: [] };
    aggregate.rows += file.rowCount;
    aggregate.pages.push(file.page);
    pageAggregates.set(file.dataset, aggregate);
  }

  let totalRows = 0;
  const validatedDatasets = new Map<DatasetName, ValidatedDataset>();
  for (const spec of datasetSpecs) {
    const datasetManifest = manifest.datasets[spec.name];
    const rows = rowsByDataset.get(spec.name);
    if (datasetManifest === undefined || rows === undefined) {
      throw new Error(`Incomplete export dataset: ${spec.name}`);
    }
    const aggregate = pageAggregates.get(spec.name) ?? { rows: 0, pages: [] };
    aggregate.pages.sort((left, right) => left - right);
    const expectedPages = Array.from(
      { length: datasetManifest.pages },
      (_value, index) => index + 1,
    );
    if (!aggregate.pages.every((page, index) => page === expectedPages[index]) ||
      aggregate.pages.length !== expectedPages.length) {
      throw new Error(`Non-contiguous page sequence for dataset ${spec.name}`);
    }
    if (aggregate.rows !== datasetManifest.rowCount || rows.length !== datasetManifest.rowCount) {
      throw new Error(`Dataset row count mismatch: ${spec.name}`);
    }
    if (
      datasetManifest.sourceCountBefore !== datasetManifest.sourceCountAfter ||
      datasetManifest.rowCount !== datasetManifest.sourceCountAfter
    ) {
      throw new Error(`Unstable source counts recorded for dataset ${spec.name}`);
    }

    const targetColumns = new Set(spec.targetColumns);
    const mappedColumns = datasetManifest.columns.filter((column) => targetColumns.has(column));
    validatedDatasets.set(spec.name, {
      spec,
      manifest: datasetManifest,
      rows,
      mappedColumns,
      sourceHash: canonicalRowsHash(rows, spec.primaryKey),
    });
    totalRows += rows.length;
  }
  if (totalRows !== manifest.export.rowCount) {
    throw new Error('Dataset totals do not match the export row count');
  }

  const actualPaths = await listFiles(directory);
  const actualPathKeys = new Set(actualPaths.map(normalizedPathKey));
  const missing = [...expectedPaths.keys()].find((entry) => !actualPathKeys.has(entry));
  const unexpected = actualPaths.find((entry) => !expectedPaths.has(normalizedPathKey(entry)));
  if (missing !== undefined) throw new Error(`Missing export artifact: ${expectedPaths.get(missing)}`);
  if (unexpected !== undefined) throw new Error(`Unexpected export artifact: ${unexpected}`);

  return { directory, manifest, datasets: validatedDatasets };
}

function environmentValue(name: string): string | undefined {
  const value = process.env[name];
  return value === undefined || value.trim() === '' ? undefined : value;
}

function parsePort(value: string | undefined): number {
  if (value === undefined) return 5432;
  const port = Number(value);
  if (!Number.isInteger(port) || port < 1 || port > 65_535) {
    throw new Error('DATABASE_PORT must be an integer between 1 and 65535');
  }
  return port;
}

function parseSsl(value: string | undefined): false | { rejectUnauthorized: boolean } {
  switch (value?.trim().toLowerCase()) {
    case undefined:
    case '':
    case 'false':
    case '0':
    case 'no':
    case 'disable':
      return false;
    case 'require':
    case 'no-verify':
      return { rejectUnauthorized: false };
    case 'true':
    case '1':
    case 'yes':
    case 'verify-full':
      return { rejectUnauthorized: true };
    default:
      throw new Error(
        'DATABASE_SSL must be false, true, disable, require, no-verify, or verify-full',
      );
  }
}

function databaseConfig(): ClientConfig {
  const connectionString = environmentValue('DATABASE_URL');
  const ssl = parseSsl(environmentValue('DATABASE_SSL'));
  if (connectionString !== undefined) {
    let parsed: URL;
    try {
      parsed = new URL(connectionString);
    } catch {
      throw new Error('DATABASE_URL is not a valid PostgreSQL URL');
    }
    if (parsed.pathname.replace(/^\//, '') === '') {
      throw new Error('DATABASE_URL must include a database name');
    }
    return {
      connectionString,
      ssl,
      application_name: 'tonline-erp-source-import',
      connectionTimeoutMillis: 10_000,
    };
  }

  const password = environmentValue('DATABASE_PASSWORD');
  if (password === undefined) throw new Error('DATABASE_PASSWORD is not set in backend/.env');
  return {
    host: environmentValue('DATABASE_HOST') ?? environmentValue('PGHOST') ?? '127.0.0.1',
    port: parsePort(environmentValue('DATABASE_PORT') ?? environmentValue('PGPORT')),
    database:
      environmentValue('DATABASE_NAME') ??
      environmentValue('PGDATABASE') ??
      'tonline_erp_local',
    user: environmentValue('DATABASE_USER') ?? environmentValue('PGUSER') ?? 'postgres',
    password,
    ssl,
    application_name: 'tonline-erp-source-import',
    connectionTimeoutMillis: 10_000,
  };
}

function quoteIdentifier(identifier: string): string {
  return `"${identifier.replaceAll('"', '""')}"`;
}

function targetRelation(spec: DatasetSpec): string {
  return `${quoteIdentifier(spec.targetSchema)}.${quoteIdentifier(spec.targetTable)}`;
}

function stageRelation(spec: DatasetSpec): string {
  return `pg_temp.${quoteIdentifier(spec.stageTable)}`;
}

async function assertTargetTablesExist(client: Client): Promise<void> {
  for (const spec of datasetSpecs) {
    const result = await client.query<{ relation: string | null }>(
      'SELECT to_regclass($1) AS relation',
      [`${spec.targetSchema}.${spec.targetTable}`],
    );
    if (result.rows[0]?.relation === null || result.rows[0]?.relation === undefined) {
      throw new Error(
        `Target table ${spec.targetSchema}.${spec.targetTable} does not exist; run migrations first`,
      );
    }
  }
}

function parameterValue(dataset: ValidatedDataset, column: string, row: JsonObject): unknown {
  if (column === 'source_record') return JSON.stringify(row);
  const value = row[column];
  if (value !== null && dataset.spec.jsonColumns.has(column)) return JSON.stringify(value);
  return value;
}

async function createAndLoadStage(client: Client, dataset: ValidatedDataset): Promise<void> {
  const target = targetRelation(dataset.spec);
  const stage = quoteIdentifier(dataset.spec.stageTable);
  await client.query(
    `CREATE TEMP TABLE ${stage} (LIKE ${target} INCLUDING DEFAULTS) ON COMMIT DROP`,
  );

  const columns = dataset.spec.name === 'kv'
    ? dataset.mappedColumns
    : [...dataset.mappedColumns, 'source_record'];
  const batchSize = Math.max(1, Math.min(500, Math.floor(30_000 / columns.length)));
  for (let offset = 0; offset < dataset.rows.length; offset += batchSize) {
    const batch = dataset.rows.slice(offset, offset + batchSize);
    const parameters: unknown[] = [];
    const valueGroups = batch.map((row) => {
      const placeholders = columns.map((column) => {
        parameters.push(parameterValue(dataset, column, row));
        return `$${parameters.length}`;
      });
      return `(${placeholders.join(', ')})`;
    });
    await client.query(
      `INSERT INTO ${stage} (${columns.map(quoteIdentifier).join(', ')}) VALUES ${valueGroups.join(', ')}`,
      parameters,
    );
  }

  const countResult = await client.query<{ count: string }>(
    `SELECT count(*)::text AS count FROM ${stageRelation(dataset.spec)}`,
  );
  if (Number(countResult.rows[0]?.count) !== dataset.rows.length) {
    throw new Error(`Staging row count mismatch for dataset ${dataset.spec.name}`);
  }
}

async function refuseNonEmptyTargets(client: Client): Promise<void> {
  for (const spec of datasetSpecs) {
    const result = await client.query<{ exists: boolean }>(
      `SELECT EXISTS (SELECT 1 FROM ${targetRelation(spec)} LIMIT 1) AS exists`,
    );
    if (result.rows[0]?.exists) {
      throw new Error(
        `Target table ${spec.targetSchema}.${spec.targetTable} is not empty; ` +
          'inspect it and re-run with --replace only when replacement is intended',
      );
    }
  }
}

async function assertNoResult(client: Client, sql: string, message: string): Promise<void> {
  const result = await client.query(sql);
  if (result.rowCount !== 0) throw new Error(message);
}

async function validateStagedRelationships(client: Client): Promise<void> {
  const users = stageRelation(specsByName.get('authUsers')!);
  const identities = stageRelation(specsByName.get('authIdentities')!);
  const buckets = stageRelation(specsByName.get('storageBuckets')!);
  const objects = stageRelation(specsByName.get('storageObjects')!);

  await assertNoResult(
    client,
    `SELECT 1 FROM ${identities} i LEFT JOIN ${users} u ON u.id = i.user_id ` +
      'WHERE u.id IS NULL LIMIT 1',
    'The export contains an identity without its user',
  );
  await assertNoResult(
    client,
    `SELECT 1 FROM ${buckets} b LEFT JOIN ${users} u ON u.id = b.owner ` +
      'WHERE b.owner IS NOT NULL AND u.id IS NULL LIMIT 1',
    'The export contains a storage bucket whose owner is missing',
  );
  await assertNoResult(
    client,
    `SELECT 1 FROM ${objects} o LEFT JOIN ${users} u ON u.id = o.owner ` +
      'WHERE o.owner IS NOT NULL AND u.id IS NULL LIMIT 1',
    'The export contains a storage object whose owner is missing',
  );
  await assertNoResult(
    client,
    `SELECT 1 FROM ${objects} o LEFT JOIN ${buckets} b ON b.id = o.bucket_id ` +
      'WHERE b.id IS NULL LIMIT 1',
    'The export contains a storage object whose bucket is missing',
  );
}

async function protectNonExportedDependencies(client: Client): Promise<void> {
  const users = stageRelation(specsByName.get('authUsers')!);
  await assertNoResult(
    client,
    `SELECT 1 FROM app_auth.sessions s LEFT JOIN ${users} u ON u.id = s.user_id ` +
      'WHERE u.id IS NULL LIMIT 1',
    'Replacement would delete users referenced by non-exported local sessions; refusing',
  );
  await assertNoResult(
    client,
    `SELECT 1 FROM app_auth.refresh_tokens r LEFT JOIN ${users} u ON u.id = r.user_id ` +
      'WHERE u.id IS NULL LIMIT 1',
    'Replacement would delete users referenced by non-exported local refresh tokens; refusing',
  );
}

async function deleteRowsMissingFromStage(client: Client, spec: DatasetSpec): Promise<void> {
  const primaryKey = quoteIdentifier(spec.primaryKey);
  await client.query(
    `DELETE FROM ${targetRelation(spec)} AS target_row ` +
      `WHERE NOT EXISTS (SELECT 1 FROM ${stageRelation(spec)} AS staged ` +
      `WHERE staged.${primaryKey} = target_row.${primaryKey})`,
  );
}

async function disableUpdatedAtTriggers(client: Client): Promise<TriggerState[]> {
  const states: TriggerState[] = [];
  for (const spec of datasetSpecs) {
    if (!spec.hasUpdatedAtTrigger) continue;
    const result = await client.query<{ tgenabled: string }>(
      `SELECT t.tgenabled
       FROM pg_catalog.pg_trigger t
       WHERE t.tgrelid = $1::regclass
         AND t.tgname = 'set_updated_at'
         AND NOT t.tgisinternal`,
      [`${spec.targetSchema}.${spec.targetTable}`],
    );
    const state = result.rows[0]?.tgenabled;
    if (state !== undefined && state !== 'D') {
      states.push({ spec, state });
      await client.query(
        `ALTER TABLE ${targetRelation(spec)} DISABLE TRIGGER ${quoteIdentifier('set_updated_at')}`,
      );
    }
  }
  return states;
}

async function restoreUpdatedAtTriggers(client: Client, states: TriggerState[]): Promise<void> {
  for (const { spec, state } of states) {
    const mode = state === 'A' ? 'ENABLE ALWAYS' : state === 'R' ? 'ENABLE REPLICA' : 'ENABLE';
    await client.query(
      `ALTER TABLE ${targetRelation(spec)} ${mode} TRIGGER ${quoteIdentifier('set_updated_at')}`,
    );
  }
}

async function upsertDataset(client: Client, dataset: ValidatedDataset): Promise<void> {
  const columns = dataset.spec.name === 'kv'
    ? dataset.mappedColumns
    : [...dataset.mappedColumns, 'source_record'];
  const primaryKey = dataset.spec.primaryKey;
  const updateColumns = columns.filter((column) => column !== primaryKey);
  const assignments = updateColumns.map(
    (column) => `${quoteIdentifier(column)} = EXCLUDED.${quoteIdentifier(column)}`,
  );
  const targetValues = updateColumns.map(
    (column) => `target_row.${quoteIdentifier(column)}`,
  );
  const excludedValues = updateColumns.map(
    (column) => `EXCLUDED.${quoteIdentifier(column)}`,
  );
  await client.query(
    `INSERT INTO ${targetRelation(dataset.spec)} AS target_row ` +
      `(${columns.map(quoteIdentifier).join(', ')}) ` +
      `SELECT ${columns.map(quoteIdentifier).join(', ')} FROM ${stageRelation(dataset.spec)} ` +
      `ON CONFLICT (${quoteIdentifier(primaryKey)}) DO UPDATE SET ${assignments.join(', ')} ` +
      `WHERE ROW(${targetValues.join(', ')}) IS DISTINCT FROM ` +
      `ROW(${excludedValues.join(', ')})`,
  );
}

function parseJsonRecord(value: unknown, label: string): JsonObject {
  if (isJsonObject(value)) return value;
  if (typeof value === 'string') {
    try {
      const parsed: unknown = JSON.parse(value);
      if (isJsonObject(parsed)) return parsed;
    } catch {
      // Fall through to the generic validation error below.
    }
  }
  throw new Error(`${label} is not a JSON object`);
}

interface ProjectionRow extends QueryResultRow {
  typed_record: unknown;
  source_record: unknown;
}

async function readProjectedRows(
  client: Client,
  dataset: ValidatedDataset,
  relation: string,
): Promise<{ sourceRows: JsonObject[]; typedRows: JsonObject[] }> {
  if (dataset.spec.name === 'kv') {
    const result = await client.query<ProjectionRow>(
      `SELECT to_jsonb(row_value) AS typed_record, to_jsonb(row_value) AS source_record ` +
        `FROM ${relation} AS row_value`,
    );
    const rows = result.rows.map((row) => parseJsonRecord(row.typed_record, 'KV projection'));
    return { sourceRows: rows, typedRows: rows };
  }

  const result = await client.query<ProjectionRow>(
    `SELECT to_jsonb(row_value) AS typed_record, row_value.source_record AS source_record ` +
      `FROM ${relation} AS row_value`,
  );
  const sourceRows: JsonObject[] = [];
  const typedRows: JsonObject[] = [];
  const mappedColumns = new Set(dataset.mappedColumns);
  for (const row of result.rows) {
    const sourceRecord = parseJsonRecord(row.source_record, 'source_record');
    const typedRecord = parseJsonRecord(row.typed_record, 'typed database record');
    const reconstructed: JsonObject = {};
    for (const column of dataset.manifest.columns) {
      reconstructed[column] = mappedColumns.has(column)
        ? typedRecord[column]
        : sourceRecord[column];
    }
    sourceRows.push(sourceRecord);
    typedRows.push(reconstructed);
  }
  return { sourceRows, typedRows };
}

async function verifyImportedDatasets(
  client: Client,
  validatedExport: ValidatedExport,
): Promise<void> {
  for (const spec of datasetSpecs) {
    const dataset = validatedExport.datasets.get(spec.name);
    if (dataset === undefined) throw new Error(`Missing validated dataset ${spec.name}`);
    const staged = await readProjectedRows(client, dataset, stageRelation(spec));
    const imported = await readProjectedRows(client, dataset, targetRelation(spec));
    if (
      staged.sourceRows.length !== dataset.manifest.rowCount ||
      imported.sourceRows.length !== dataset.manifest.rowCount
    ) {
      throw new Error(`Post-import row count mismatch for dataset ${spec.name}`);
    }

    const stagedSourceHash = canonicalRowsHash(staged.sourceRows, spec.primaryKey);
    const importedSourceHash = canonicalRowsHash(imported.sourceRows, spec.primaryKey);
    if (stagedSourceHash !== dataset.sourceHash || importedSourceHash !== dataset.sourceHash) {
      throw new Error(`Post-import canonical content mismatch for dataset ${spec.name}`);
    }

    const stagedTypedHash = canonicalRowsHash(staged.typedRows, spec.primaryKey);
    const importedTypedHash = canonicalRowsHash(imported.typedRows, spec.primaryKey);
    if (stagedTypedHash !== importedTypedHash) {
      throw new Error(
        `Post-import typed-column mismatch for dataset ${spec.name}` +
          typedMismatchDetails(staged.typedRows, imported.typedRows, spec.primaryKey),
      );
    }
  }
}

async function importExport(validatedExport: ValidatedExport, replace: boolean): Promise<void> {
  const client = new Client(databaseConfig());
  await client.connect();
  try {
    await client.query('BEGIN');
    try {
      await client.query("SET LOCAL TIME ZONE 'UTC'");
      await client.query("SET LOCAL lock_timeout = '30s'");
      await client.query('SELECT pg_advisory_xact_lock($1, $2)', [7249, 2]);
      await assertTargetTablesExist(client);
      await client.query(
        `LOCK TABLE ${datasetSpecs.map(targetRelation).join(', ')} IN EXCLUSIVE MODE`,
      );

      for (const spec of datasetSpecs) {
        const dataset = validatedExport.datasets.get(spec.name);
        if (dataset === undefined) throw new Error(`Missing validated dataset ${spec.name}`);
        await createAndLoadStage(client, dataset);
      }
      await validateStagedRelationships(client);

      if (!replace) {
        await refuseNonEmptyTargets(client);
      } else {
        await protectNonExportedDependencies(client);
        for (const name of [
          'authIdentities',
          'storageObjects',
          'storageBuckets',
          'authUsers',
          'kv',
        ] as const) {
          await deleteRowsMissingFromStage(client, specsByName.get(name)!);
        }
      }

      const triggerStates = await disableUpdatedAtTriggers(client);
      for (const name of [
        'authUsers',
        'storageBuckets',
        'authIdentities',
        'storageObjects',
        'kv',
      ] as const) {
        const dataset = validatedExport.datasets.get(name);
        if (dataset === undefined) throw new Error(`Missing validated dataset ${name}`);
        await upsertDataset(client, dataset);
      }
      await restoreUpdatedAtTriggers(client, triggerStates);

      await verifyImportedDatasets(client, validatedExport);
      await client.query('COMMIT');
    } catch (error) {
      await client.query('ROLLBACK');
      throw error;
    }
  } finally {
    await client.end();
  }
}

function redactedErrorMessage(error: unknown): string {
  let message = error instanceof Error ? error.message : String(error);
  const secrets = new Set<string>();
  for (const [name, value] of Object.entries(process.env)) {
    if (
      value !== undefined &&
      value.length >= 4 &&
      /(PASSWORD|PASSWD|PWD|SECRET|TOKEN|API_KEY|PRIVATE_KEY|DATABASE_URL)/i.test(name)
    ) {
      secrets.add(value);
    }
  }
  const databaseUrl = environmentValue('DATABASE_URL');
  if (databaseUrl !== undefined) {
    try {
      const password = decodeURIComponent(new URL(databaseUrl).password);
      if (password.length >= 4) secrets.add(password);
    } catch {
      // Invalid DATABASE_URL is reported by databaseConfig without echoing it.
    }
  }
  for (const secret of secrets) message = message.split(secret).join('[REDACTED]');
  return message;
}

async function main(): Promise<void> {
  const options = parseCli(process.argv.slice(2));
  if (options.help) {
    printHelp();
    return;
  }

  const exportDirectory = options.exportDirectory ?? await newestCompleteExportDirectory();
  console.log('Validating the local source export before opening a database transaction.');
  const validatedExport = await validateExport(exportDirectory);
  console.log(
    `Export validated: ${validatedExport.directory} ` +
      `(${validatedExport.manifest.export.rowCount} rows).`,
  );
  console.log(
    options.replace
      ? 'Starting atomic replacement of the exported datasets.'
      : 'Starting import into empty target datasets.',
  );
  await importExport(validatedExport, options.replace);
  console.log(
    `Import committed: ${validatedExport.manifest.export.rowCount} rows; ` +
      'all dataset counts and canonical contents verified.',
  );
  console.log('Storage metadata was imported; this source export contains no object binaries.');
}

main().catch((error: unknown) => {
  console.error(`Import failed: ${redactedErrorMessage(error)}`);
  process.exitCode = 1;
});