import { execFileSync } from 'node:child_process';
import { createHash, randomUUID } from 'node:crypto';
import fs from 'node:fs';
import os from 'node:os';
import path from 'node:path';

export const CONTROL_SCHEMA = 'mindspace_control';

export function assertUserId(value) {
  const userId = String(value ?? '').trim().toLowerCase();
  if (!/^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/.test(userId)) {
    throw new Error('user id 必须是标准 UUID');
  }
  return userId;
}

export function quotePgIdentifier(value) {
  return `"${String(value).replaceAll('"', '""')}"`;
}

export function deriveUserSpaceNames(value) {
  const userId = assertUserId(value);
  const compact = userId.replaceAll('-', '');
  const short = compact.slice(0, 12);
  return {
    userId,
    schemaName: `u_${compact}`,
    ownerRole: `ms_u_${short}_owner`,
    agentRole: `ms_u_${short}_agent`,
  };
}

export function buildControlSchemaSql() {
  return `
CREATE SCHEMA IF NOT EXISTS ${quotePgIdentifier(CONTROL_SCHEMA)};
REVOKE ALL ON SCHEMA ${quotePgIdentifier(CONTROL_SCHEMA)} FROM PUBLIC;

CREATE TABLE IF NOT EXISTS ${quotePgIdentifier(CONTROL_SCHEMA)}.user_spaces (
  user_id uuid PRIMARY KEY,
  schema_name name NOT NULL UNIQUE,
  owner_role name NOT NULL UNIQUE,
  agent_role name NOT NULL UNIQUE,
  backend_type text NOT NULL DEFAULT 'postgres' CHECK (backend_type IN ('sqlite', 'postgres')),
  source_sqlite_path text,
  migration_state text NOT NULL DEFAULT 'pending' CHECK (
    migration_state IN ('pending', 'snapshot_created', 'schema_converted', 'data_imported', 'verified', 'shadow', 'cutover', 'rolled_back', 'failed')
  ),
  schema_version integer NOT NULL DEFAULT 1,
  quota_bytes bigint NOT NULL DEFAULT 104857600 CHECK (quota_bytes > 0),
  created_at timestamptz NOT NULL DEFAULT CURRENT_TIMESTAMP,
  updated_at timestamptz NOT NULL DEFAULT CURRENT_TIMESTAMP,
  cutover_at timestamptz,
  rollback_until timestamptz
);

CREATE TABLE IF NOT EXISTS ${quotePgIdentifier(CONTROL_SCHEMA)}.migration_runs (
  id uuid PRIMARY KEY,
  user_id uuid NOT NULL REFERENCES ${quotePgIdentifier(CONTROL_SCHEMA)}.user_spaces(user_id) ON DELETE CASCADE,
  source_path text NOT NULL,
  source_size_bytes bigint NOT NULL DEFAULT 0,
  source_sha256 text,
  status text NOT NULL,
  table_count integer NOT NULL DEFAULT 0,
  source_row_count bigint NOT NULL DEFAULT 0,
  target_row_count bigint NOT NULL DEFAULT 0,
  detail_json jsonb NOT NULL DEFAULT '{}'::jsonb,
  started_at timestamptz NOT NULL DEFAULT CURRENT_TIMESTAMP,
  finished_at timestamptz
);

CREATE TABLE IF NOT EXISTS ${quotePgIdentifier(CONTROL_SCHEMA)}.dataset_catalog (
  user_id uuid NOT NULL REFERENCES ${quotePgIdentifier(CONTROL_SCHEMA)}.user_spaces(user_id) ON DELETE CASCADE,
  dataset_name text NOT NULL,
  physical_table name NOT NULL,
  config_json jsonb NOT NULL,
  sensitivity text NOT NULL DEFAULT 'normal' CHECK (sensitivity IN ('normal', 'personal', 'credential', 'health')),
  created_at timestamptz NOT NULL DEFAULT CURRENT_TIMESTAMP,
  updated_at timestamptz NOT NULL DEFAULT CURRENT_TIMESTAMP,
  PRIMARY KEY (user_id, dataset_name)
);

CREATE TABLE IF NOT EXISTS ${quotePgIdentifier(CONTROL_SCHEMA)}.audit_events (
  id bigserial PRIMARY KEY,
  user_id uuid,
  actor_type text NOT NULL,
  actor_id text,
  action text NOT NULL,
  object_type text,
  object_name text,
  statement_hash text,
  affected_rows bigint,
  duration_ms integer,
  result text NOT NULL,
  detail_json jsonb NOT NULL DEFAULT '{}'::jsonb,
  created_at timestamptz NOT NULL DEFAULT CURRENT_TIMESTAMP
);
REVOKE ALL ON ALL TABLES IN SCHEMA ${quotePgIdentifier(CONTROL_SCHEMA)} FROM PUBLIC;
`.trim();
}

export function buildProvisionUserSql(userId, { sourceSqlitePath = null, quotaBytes = 104857600 } = {}) {
  const names = deriveUserSpaceNames(userId);
  const schema = quotePgIdentifier(names.schemaName);
  const owner = quotePgIdentifier(names.ownerRole);
  const agent = quotePgIdentifier(names.agentRole);
  return {
    ...names,
    statements: [
      `DO $$ BEGIN CREATE ROLE ${owner} NOLOGIN; EXCEPTION WHEN duplicate_object THEN NULL; END $$`,
      `DO $$ BEGIN CREATE ROLE ${agent} NOLOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE NOREPLICATION; EXCEPTION WHEN duplicate_object THEN NULL; END $$`,
      `DO $$ BEGIN
         IF NOT EXISTS (
           SELECT 1 FROM pg_auth_members am
           JOIN pg_roles r ON r.oid = am.roleid
           JOIN pg_roles m ON m.oid = am.member
           WHERE r.rolname = '${names.ownerRole}' AND m.rolname = CURRENT_USER AND am.set_option
         ) THEN
           EXECUTE format('GRANT %I TO %I', '${names.ownerRole}', CURRENT_USER);
         END IF;
       END $$`,
      `CREATE SCHEMA IF NOT EXISTS ${schema} AUTHORIZATION ${owner}`,
      `REVOKE ALL ON SCHEMA ${schema} FROM PUBLIC`,
      `GRANT USAGE, CREATE ON SCHEMA ${schema} TO ${agent}`,
      `ALTER ROLE ${agent} SET search_path = ${schema}, pg_catalog`,
      `ALTER ROLE ${agent} SET statement_timeout = '30s'`,
      `ALTER ROLE ${agent} SET lock_timeout = '5s'`,
      `ALTER ROLE ${agent} SET idle_in_transaction_session_timeout = '60s'`,
      `ALTER ROLE ${agent} SET work_mem = '4MB'`,
      `ALTER DEFAULT PRIVILEGES FOR ROLE ${owner} IN SCHEMA ${schema} GRANT SELECT, INSERT, UPDATE, DELETE ON TABLES TO ${agent}`,
      `ALTER DEFAULT PRIVILEGES FOR ROLE ${owner} IN SCHEMA ${schema} GRANT USAGE, SELECT, UPDATE ON SEQUENCES TO ${agent}`,
      `INSERT INTO ${quotePgIdentifier(CONTROL_SCHEMA)}.user_spaces
        (user_id, schema_name, owner_role, agent_role, source_sqlite_path, quota_bytes, updated_at)
       VALUES ($1::uuid, $2::name, $3::name, $4::name, $5, $6, CURRENT_TIMESTAMP)
       ON CONFLICT (user_id) DO UPDATE SET
         source_sqlite_path = EXCLUDED.source_sqlite_path,
         quota_bytes = EXCLUDED.quota_bytes,
         updated_at = CURRENT_TIMESTAMP`,
    ],
    params: [names.userId, names.schemaName, names.ownerRole, names.agentRole, sourceSqlitePath, quotaBytes],
  };
}

export function mapSqliteTypeToPostgres(column) {
  const declared = String(column.type ?? '').trim().toUpperCase();
  if (Number(column.pk) === 1 && declared.includes('INT')) return 'bigint GENERATED BY DEFAULT AS IDENTITY';
  if (declared.includes('TEXT') && isSqliteTimestampDefault(column.dflt_value)) return 'timestamptz';
  if (declared.includes('INT')) return 'bigint';
  if (declared.includes('REAL') || declared.includes('FLOA') || declared.includes('DOUB')) return 'double precision';
  if (declared.includes('BLOB')) return 'bytea';
  if (declared.includes('NUM') || declared.includes('DEC')) return 'numeric';
  if (declared.includes('BOOL')) return 'boolean';
  return 'text';
}

export function isSqliteTimestampDefault(value) {
  return /^\(?datetime\(\s*'now'\s*(?:,\s*'(?:\+8 hours|localtime)'\s*)?\)\)?$/i.test(String(value ?? '').trim());
}

export function translateSqliteDefault(value) {
  if (value == null) return null;
  const input = String(value).trim();
  if (/^NULL$/i.test(input)) return 'NULL';
  if (/^[+-]?\d+(?:\.\d+)?$/.test(input)) return input;
  if (/^CURRENT_TIMESTAMP$/i.test(input)) return 'CURRENT_TIMESTAMP';
  if (/^\(?datetime\(\s*'now'\s*(?:,\s*'(?:\+8 hours|localtime)'\s*)?\)\)?$/i.test(input)) return 'CURRENT_TIMESTAMP';
  if (/^".*"$/.test(input)) return `'${input.slice(1, -1).replaceAll("'", "''")}'`;
  if (/^'.*'$/.test(input)) return input;
  return null;
}

export function buildPostgresTableSql(schemaName, tableName, columns) {
  if (!Array.isArray(columns) || columns.length === 0) throw new Error(`表 ${tableName} 没有字段`);
  const primaryKeys = columns.filter((column) => Number(column.pk) > 0).sort((a, b) => Number(a.pk) - Number(b.pk));
  const definitions = columns.map((column) => {
    const parts = [quotePgIdentifier(column.name), mapSqliteTypeToPostgres(column)];
    if (Number(column.notnull) === 1) parts.push('NOT NULL');
    const translatedDefault = translateSqliteDefault(column.dflt_value);
    if (translatedDefault != null) parts.push(`DEFAULT ${translatedDefault}`);
    if (primaryKeys.length === 1 && primaryKeys[0].name === column.name) parts.push('PRIMARY KEY');
    return parts.join(' ');
  });
  if (primaryKeys.length > 1) {
    definitions.push(`PRIMARY KEY (${primaryKeys.map((column) => quotePgIdentifier(column.name)).join(', ')})`);
  }
  return `CREATE TABLE ${quotePgIdentifier(schemaName)}.${quotePgIdentifier(tableName)} (\n  ${definitions.join(',\n  ')}\n)`;
}

function stableObjectName(prefix, ...parts) {
  const readable = `${prefix}_${parts.join('_')}`.replace(/[^a-zA-Z0-9_]+/g, '_').toLowerCase();
  const hash = createHash('sha256').update(readable).digest('hex').slice(0, 8);
  return `${readable.slice(0, 50)}_${hash}`;
}

export function extractSqliteChecks(createSql) {
  const sql = String(createSql ?? '');
  const checks = [];
  const pattern = /\bCHECK\s*\(/ig;
  let match;
  while ((match = pattern.exec(sql))) {
    const start = pattern.lastIndex;
    let depth = 1;
    let quote = '';
    let index = start;
    for (; index < sql.length && depth > 0; index += 1) {
      const char = sql[index];
      if (quote) {
        if (char === quote && sql[index + 1] === quote) index += 1;
        else if (char === quote) quote = '';
        continue;
      }
      if (char === "'" || char === '"') quote = char;
      else if (char === '(') depth += 1;
      else if (char === ')') depth -= 1;
    }
    if (depth === 0) checks.push(sql.slice(start, index - 1).trim());
    pattern.lastIndex = index;
  }
  return checks;
}

export function buildPostgresIndexSql(schemaName, tableName, index) {
  if (!index.columns?.length) return null;
  if (index.partial) return null;
  const prefix = index.unique ? 'ux' : 'ix';
  const name = stableObjectName(prefix, tableName, ...index.columns);
  return `CREATE ${index.unique ? 'UNIQUE ' : ''}INDEX ${quotePgIdentifier(name)} ON ${quotePgIdentifier(schemaName)}.${quotePgIdentifier(tableName)} (${index.columns.map(quotePgIdentifier).join(', ')})`;
}

export function buildPostgresForeignKeySql(schemaName, tableName, foreignKey) {
  const name = stableObjectName('fk', tableName, foreignKey.id, foreignKey.table);
  const onUpdate = String(foreignKey.onUpdate ?? 'NO ACTION').toUpperCase();
  const onDelete = String(foreignKey.onDelete ?? 'NO ACTION').toUpperCase();
  const allowedActions = new Set(['NO ACTION', 'RESTRICT', 'CASCADE', 'SET NULL', 'SET DEFAULT']);
  if (!allowedActions.has(onUpdate) || !allowedActions.has(onDelete)) throw new Error('SQLite 外键动作不受支持');
  return `ALTER TABLE ${quotePgIdentifier(schemaName)}.${quotePgIdentifier(tableName)} ADD CONSTRAINT ${quotePgIdentifier(name)} FOREIGN KEY (${foreignKey.from.map(quotePgIdentifier).join(', ')}) REFERENCES ${quotePgIdentifier(schemaName)}.${quotePgIdentifier(foreignKey.table)} (${foreignKey.to.map(quotePgIdentifier).join(', ')}) ON UPDATE ${onUpdate} ON DELETE ${onDelete}`;
}

export function buildPostgresCheckSql(schemaName, tableName, expression, index) {
  if (/;|--|\/\*/.test(expression)) throw new Error(`CHECK 表达式包含不安全内容：${tableName}`);
  const name = stableObjectName('ck', tableName, index);
  return `ALTER TABLE ${quotePgIdentifier(schemaName)}.${quotePgIdentifier(tableName)} ADD CONSTRAINT ${quotePgIdentifier(name)} CHECK (${expression})`;
}

export function convertSqliteValueForPostgres(value, column) {
  if (value == null) return null;
  if (isSqliteTimestampDefault(column.dflt_value) && typeof value === 'string') {
    const trimmed = value.trim();
    if (/^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}(?:\.\d+)?$/.test(trimmed)) {
      return `${trimmed.replace(' ', 'T')}+08:00`;
    }
  }
  return value;
}

function sqliteJson(sqliteBin, databasePath, sql) {
  const output = execFileSync(sqliteBin, ['-readonly', '-json', databasePath, sql], {
    encoding: 'utf8',
    maxBuffer: 32 * 1024 * 1024,
  }).trim();
  return output ? JSON.parse(output) : [];
}

function quoteSqliteIdentifier(value) {
  return `"${String(value).replaceAll('"', '""')}"`;
}

export function inspectSqliteDatabase(databasePath, { sqliteBin = 'sqlite3' } = {}) {
  const resolved = path.resolve(databasePath);
  if (!fs.statSync(resolved).isFile()) throw new Error(`SQLite 文件不存在：${resolved}`);
  const integrity = sqliteJson(sqliteBin, resolved, 'PRAGMA integrity_check;');
  if (integrity[0]?.integrity_check !== 'ok') throw new Error(`SQLite integrity_check 未通过：${resolved}`);
  const tables = sqliteJson(
    sqliteBin,
    resolved,
    "SELECT name, sql FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name;",
  ).map((table) => {
    const columns = sqliteJson(sqliteBin, resolved, `PRAGMA table_info(${quoteSqliteIdentifier(table.name)});`);
    const rows = sqliteJson(sqliteBin, resolved, `SELECT * FROM ${quoteSqliteIdentifier(table.name)};`);
    const indexes = sqliteJson(sqliteBin, resolved, `PRAGMA index_list(${quoteSqliteIdentifier(table.name)});`).map((index) => ({
      name: index.name,
      unique: Boolean(index.unique),
      origin: index.origin,
      partial: Boolean(index.partial),
      columns: sqliteJson(sqliteBin, resolved, `PRAGMA index_info(${quoteSqliteIdentifier(index.name)});`).map((item) => item.name),
    })).filter((index) => index.origin !== 'pk');
    const foreignKeyRows = sqliteJson(sqliteBin, resolved, `PRAGMA foreign_key_list(${quoteSqliteIdentifier(table.name)});`);
    const foreignKeys = [...new Set(foreignKeyRows.map((item) => item.id))].map((id) => {
      const items = foreignKeyRows.filter((item) => item.id === id).sort((a, b) => a.seq - b.seq);
      return {
        id,
        table: items[0].table,
        from: items.map((item) => item.from),
        to: items.map((item) => item.to),
        onUpdate: items[0].on_update,
        onDelete: items[0].on_delete,
      };
    });
    return { ...table, columns, rows, indexes, foreignKeys, checks: extractSqliteChecks(table.sql) };
  });
  return { path: resolved, sizeBytes: fs.statSync(resolved).size, tables };
}

export function createSqliteSnapshot(sourcePath, { sqliteBin = 'sqlite3' } = {}) {
  const target = path.join(os.tmpdir(), `mindspace-userdata-${Date.now()}-${process.pid}.sqlite`);
  if (/\s/.test(target)) throw new Error('SQLite snapshot 临时路径不能包含空格');
  execFileSync(sqliteBin, [path.resolve(sourcePath), `.backup ${target}`], { stdio: 'pipe' });
  // A standalone snapshot must not depend on the source WAL/SHM files.
  execFileSync(sqliteBin, [target, 'PRAGMA journal_mode=DELETE;'], { stdio: 'pipe' });
  return target;
}

export async function provisionUserSpace(client, userId, options = {}) {
  const provision = buildProvisionUserSql(userId, options);
  await client.query('BEGIN');
  try {
    for (let index = 0; index < provision.statements.length; index += 1) {
      const params = index === provision.statements.length - 1 ? provision.params : [];
      await client.query(provision.statements[index], params);
    }
    await client.query('COMMIT');
    return provision;
  } catch (error) {
    await client.query('ROLLBACK');
    throw error;
  }
}

export async function migrateSqliteSnapshot(client, snapshot, userId, { replaceShadow = false, sourcePath = snapshot.path } = {}) {
  const names = deriveUserSpaceNames(userId);
  const owner = quotePgIdentifier(names.ownerRole);
  const agent = quotePgIdentifier(names.agentRole);
  const migrationId = randomUUID();
  let sourceRows = 0;
  let targetRows = 0;
  await client.query('BEGIN');
  try {
    const stateResult = await client.query(
      `SELECT migration_state FROM ${quotePgIdentifier(CONTROL_SCHEMA)}.user_spaces WHERE user_id = $1::uuid FOR UPDATE`,
      [names.userId],
    );
    const migrationState = stateResult.rows[0]?.migration_state;
    if (migrationState === 'cutover') throw new Error('用户空间已经 cutover，禁止迁移器覆盖');
    const existingResult = await client.query(
      `SELECT count(*)::integer AS count FROM pg_catalog.pg_tables WHERE schemaname = $1`,
      [names.schemaName],
    );
    if (Number(existingResult.rows[0]?.count ?? 0) > 0 && !replaceShadow) {
      throw new Error('目标影子空间已有表；如确认重建，显式使用 --replace-shadow');
    }
    await client.query(
      `INSERT INTO ${quotePgIdentifier(CONTROL_SCHEMA)}.migration_runs
        (id, user_id, source_path, source_size_bytes, status, started_at)
       VALUES ($1::uuid, $2::uuid, $3, $4, 'importing', CURRENT_TIMESTAMP)`,
      [migrationId, names.userId, sourcePath, snapshot.sizeBytes],
    );
    for (const table of snapshot.tables) {
      await client.query(`DROP TABLE IF EXISTS ${quotePgIdentifier(names.schemaName)}.${quotePgIdentifier(table.name)} CASCADE`);
      await client.query(buildPostgresTableSql(names.schemaName, table.name, table.columns));
      await client.query(`ALTER TABLE ${quotePgIdentifier(names.schemaName)}.${quotePgIdentifier(table.name)} OWNER TO ${owner}`);
      await client.query(
        `GRANT SELECT, INSERT, UPDATE, DELETE ON TABLE ${quotePgIdentifier(names.schemaName)}.${quotePgIdentifier(table.name)} TO ${agent}`,
      );
      for (const row of table.rows) {
        const columns = Object.keys(row);
        const params = columns.map((_, index) => `$${index + 1}`);
        await client.query(
          `INSERT INTO ${quotePgIdentifier(names.schemaName)}.${quotePgIdentifier(table.name)} (${columns.map(quotePgIdentifier).join(', ')}) VALUES (${params.join(', ')})`,
          columns.map((column) => convertSqliteValueForPostgres(
            row[column],
            table.columns.find((item) => item.name === column) ?? {},
          )),
        );
      }
      for (const column of table.columns.filter(
        (item) => Number(item.pk) === 1 && String(item.type ?? '').toUpperCase().includes('INT'),
      )) {
        const qualifiedTable = `${names.schemaName}.${table.name}`;
        await client.query(
          `SELECT setval(
             pg_get_serial_sequence($1, $2),
             GREATEST(COALESCE(MAX(${quotePgIdentifier(column.name)}), 0), 1),
             COALESCE(MAX(${quotePgIdentifier(column.name)}), 0) > 0
           )
           FROM ${quotePgIdentifier(names.schemaName)}.${quotePgIdentifier(table.name)}`,
          [qualifiedTable, column.name],
        );
      }
      sourceRows += table.rows.length;
      const count = await client.query(
        `SELECT count(*)::bigint AS count FROM ${quotePgIdentifier(names.schemaName)}.${quotePgIdentifier(table.name)}`,
      );
      targetRows += Number(count.rows[0].count);
    }
    for (const table of snapshot.tables) {
      for (const index of table.indexes ?? []) {
        const sql = buildPostgresIndexSql(names.schemaName, table.name, index);
        if (sql) await client.query(sql);
      }
      for (const foreignKey of table.foreignKeys ?? []) {
        await client.query(buildPostgresForeignKeySql(names.schemaName, table.name, foreignKey));
      }
      for (let index = 0; index < (table.checks ?? []).length; index += 1) {
        await client.query(buildPostgresCheckSql(names.schemaName, table.name, table.checks[index], index));
      }
    }
    await client.query(`GRANT USAGE, SELECT, UPDATE ON ALL SEQUENCES IN SCHEMA ${quotePgIdentifier(names.schemaName)} TO ${agent}`);
    const registry = snapshot.tables.find((table) => table.name === '__page_data_datasets');
    for (const row of registry?.rows ?? []) {
      await client.query(
        `INSERT INTO ${quotePgIdentifier(CONTROL_SCHEMA)}.dataset_catalog
          (user_id, dataset_name, physical_table, config_json, updated_at)
         VALUES ($1::uuid, $2, $3::name, $4::jsonb, CURRENT_TIMESTAMP)
         ON CONFLICT (user_id, dataset_name) DO UPDATE SET
           physical_table = EXCLUDED.physical_table,
           config_json = EXCLUDED.config_json,
           updated_at = CURRENT_TIMESTAMP`,
        [names.userId, row.name, row.table_name, row.config_json],
      );
    }
    await client.query(
      `UPDATE ${quotePgIdentifier(CONTROL_SCHEMA)}.user_spaces
       SET migration_state = 'shadow', updated_at = CURRENT_TIMESTAMP
       WHERE user_id = $1::uuid`,
      [names.userId],
    );
    await client.query(
      `UPDATE ${quotePgIdentifier(CONTROL_SCHEMA)}.migration_runs
       SET status = 'verified', table_count = $2, source_row_count = $3,
           target_row_count = $4, finished_at = CURRENT_TIMESTAMP
       WHERE id = $1::uuid`,
      [migrationId, snapshot.tables.length, sourceRows, targetRows],
    );
    await client.query('COMMIT');
    return { migrationId, ...names, tableCount: snapshot.tables.length, sourceRows, targetRows };
  } catch (error) {
    await client.query('ROLLBACK');
    throw error;
  }
}