const DEFAULT_EXCLUDED_TABLES = new Set([
  'infra_file', 'infra_file_content', 'infra_file_config',
  'infra_api_access_log', 'infra_api_error_log', 'infra_job_log',
  'system_operate_log', 'system_login_log',
]);
const MIGRATION_TABLE_PREFIX = /^(?:_+)?(?:migration(?:_|$)|migrate(?:_|$)|(?:infra_file_oss|infra_file|file|oss)_migration(?:_|$))/i;
const URL_RUN = /[a-z][a-z0-9+.-]*:\/\/[^\s"'`<>]+/gi;
const NEXT_URL = /,(?=[a-z][a-z0-9+.-]*:\/\/)/gi;
const TRAILING_DELIMITERS = /[,\])}]+$/;

function lookup(value, replacements) {
  const replacement = replacements.get(value);
  return typeof replacement === 'string' && replacement !== value
    ? { value: replacement, changed: true, replacements: 1 }
    : { value, changed: false, replacements: 0 };
}

function rewriteUrlPart(part, replacements) {
  const exact = lookup(part, replacements);
  if (exact.changed || replacements.has(part)) return exact;
  // Closing prose/Markdown punctuation is separate from a URL. A dictionary
  // entry that includes that punctuation always takes precedence.
  const delimiters = part.match(TRAILING_DELIMITERS)?.[0] ?? '';
  for (let count = 1; count <= delimiters.length; count += 1) {
    const prefix = part.slice(0, -count);
    const result = lookup(prefix, replacements);
    if (result.changed) return { ...result, value: result.value + part.slice(-count) };
  }
  return exact;
}

function rewriteUrlRun(run, replacements) {
  const exact = rewriteUrlPart(run, replacements);
  if (exact.changed || replacements.has(run)) return exact;
  const parts = run.split(NEXT_URL);
  if (parts.length === 1) return exact;
  let count = 0;
  const rewritten = [];
  for (let index = 0; index < parts.length; index += 1) {
    const result = rewriteUrlPart(parts[index], replacements);
    rewritten.push(result.value);
    count += result.replacements;
    // An unknown query/fragment can contain a literal URL. Do not treat its
    // comma-delimited contents as independent business references.
    if (!result.changed && !replacements.has(parts[index]) && /[?#]/.test(parts[index])) {
      rewritten.push(...parts.slice(index + 1));
      break;
    }
  }
  return { value: rewritten.join(','), changed: count > 0, replacements: count };
}

function rewriteText(value, replacements) {
  const exact = lookup(value, replacements);
  if (exact.changed || replacements.has(value)) return exact;

  // Validate JSON first, then edit only string-value tokens. This preserves
  // whitespace, key spelling and untouched escapes in serialized JSON.
  if (/^[\s]*[\[{\"]/.test(value)) {
    let isJson = false;
    try {
      const parsed = JSON.parse(value);
      isJson = parsed !== null && (typeof parsed === 'object' || typeof parsed === 'string');
    } catch {
      // A regular text or HTML field may also begin with a JSON-like character.
    }
    if (isJson) {
      let count = 0;
      const rewritten = value.replace(/"(?:\\[\s\S]|[^"\\])*"/g, (token, offset) => {
        if (/^\s*:/.test(value.slice(offset + token.length))) return token;
        const result = rewriteText(JSON.parse(token), replacements);
        count += result.replacements;
        return result.changed ? JSON.stringify(result.value) : token;
      });
      return { value: rewritten, changed: count > 0, replacements: count };
    }
  }

  let count = 0;
  const rewritten = value.replace(URL_RUN, (run) => {
    const result = rewriteUrlRun(run, replacements);
    count += result.replacements;
    return result.value;
  });
  return { value: rewritten, changed: count > 0, replacements: count };
}

/**
 * Replace complete absolute URL references, without URL decoding, prefix
 * substitution, or query/fragment normalization. JSON values returned as
 * objects by mysql2 are supported; object keys are never rewritten.
 * The input is not mutated and replacement values are not processed again.
 */
export function rewriteReferences(value, replacements) {
  if (!(replacements instanceof Map)) throw new TypeError('replacements must be a Map');
  const active = new WeakSet();
  function visit(input) {
    if (typeof input === 'string') return rewriteText(input, replacements);
    if (input === null || typeof input !== 'object' || input instanceof Date || ArrayBuffer.isView(input)) {
      return { value: input, changed: false, replacements: 0 };
    }
    if (active.has(input)) throw new TypeError('Reference values must not contain cycles');
    active.add(input);
    const output = Array.isArray(input) ? [] : {};
    let count = 0;
    for (const [key, item] of Object.entries(input)) {
      const result = visit(item);
      // defineProperty also handles JSON keys such as "__proto__" safely.
      Object.defineProperty(output, key, { value: result.value, enumerable: true, writable: true, configurable: true });
      count += result.replacements;
    }
    active.delete(input);
    return { value: count > 0 ? output : input, changed: count > 0, replacements: count };
  }
  return visit(value);
}

/** Detect remaining legacy URLs in text/JSON values, including nested JSON. */
export function containsLegacyReference(value, sourceConfigIds) {
  const markers = [...sourceConfigIds].map((id) => `/infra/file/${String(id)}/get/`);
  if (!markers.length) return false;
  const visited = new WeakSet();
  function visit(input) {
    if (typeof input === 'string') {
      if (/^[\s]*[\[{\"]/.test(input)) {
        try {
          const parsed = JSON.parse(input);
          // Parse JSON before examining its text so keys do not count as
          // references, and \\u002f / nested JSON escapes are decoded properly.
          return visit(parsed);
        } catch {
          // Rich text or malformed JSON still receives the literal text check.
        }
      }
      const normalized = input.replaceAll('\\/', '/');
      return markers.some((marker) => normalized.includes(marker));
    }
    if (input === null || typeof input !== 'object' || input instanceof Date || ArrayBuffer.isView(input)) return false;
    if (visited.has(input)) return false;
    visited.add(input);
    return Object.values(input).some(visit);
  }
  return visit(value);
}

/**
 * Discover writable text/JSON columns with one primary-key column. All schema
 * names are bound values, never SQL identifiers. excludedTables accepts table
 * names or schema.table names in an Array/Set (case-insensitive exclusion).
 */
export async function discoverReferenceColumns(connection, schemas, excludedTables = []) {
  const schemaNames = [...new Set(schemas)];
  if (schemaNames.some((name) => typeof name !== 'string' || !name.trim())) {
    throw new TypeError('schemas must contain non-empty schema names');
  }
  if (schemaNames.length === 0) return { columns: [], skipped: [] };
  const excluded = new Set([...excludedTables].map((name) => String(name).toLowerCase()));
  const placeholders = schemaNames.map(() => '?').join(', ');
  const [rows] = await connection.execute(`
    SELECT c.TABLE_SCHEMA AS schemaName, c.TABLE_NAME AS tableName,
           c.COLUMN_NAME AS columnName, c.DATA_TYPE AS dataType,
           c.EXTRA AS extra, c.GENERATION_EXPRESSION AS generationExpression,
           pk.primaryKey, pk.primaryKeyCount
      FROM information_schema.COLUMNS c
      JOIN information_schema.TABLES t
        ON t.TABLE_SCHEMA = c.TABLE_SCHEMA AND t.TABLE_NAME = c.TABLE_NAME
      LEFT JOIN (
        SELECT TABLE_SCHEMA, TABLE_NAME, MIN(COLUMN_NAME) AS primaryKey,
               COUNT(*) AS primaryKeyCount
          FROM information_schema.KEY_COLUMN_USAGE
         WHERE CONSTRAINT_NAME = 'PRIMARY' AND TABLE_SCHEMA IN (${placeholders})
         GROUP BY TABLE_SCHEMA, TABLE_NAME
      ) pk ON pk.TABLE_SCHEMA = c.TABLE_SCHEMA AND pk.TABLE_NAME = c.TABLE_NAME
     WHERE c.TABLE_SCHEMA IN (${placeholders}) AND t.TABLE_TYPE = 'BASE TABLE'
       AND c.DATA_TYPE IN ('char', 'varchar', 'tinytext', 'text', 'mediumtext', 'longtext', 'json')
     ORDER BY c.TABLE_SCHEMA, c.TABLE_NAME, c.ORDINAL_POSITION
  `, [...schemaNames, ...schemaNames]);

  const columns = [];
  const skipped = [];
  const skippedTables = new Set();
  for (const row of rows) {
    const schema = row.schemaName;
    const table = row.tableName;
    const lowerTable = table.toLowerCase();
    const fullName = `${schema}.${table}`;
    let reason;
    if (DEFAULT_EXCLUDED_TABLES.has(lowerTable) || MIGRATION_TABLE_PREFIX.test(table)
        || excluded.has(lowerTable) || excluded.has(fullName.toLowerCase())) {
      reason = 'excluded table';
    } else if (Number(row.primaryKeyCount) !== 1) {
      reason = Number(row.primaryKeyCount) > 1 ? 'composite primary key' : 'no primary key';
    }
    if (reason) {
      if (!skippedTables.has(fullName)) {
        skipped.push({ schema, table, reason });
        skippedTables.add(fullName);
      }
      continue;
    }
    const extra = row.extra ?? '';
    if (row.columnName === row.primaryKey || /\bauto_increment\b/i.test(extra)
        || row.generationExpression || /\b(?:VIRTUAL|STORED)\s+GENERATED\b/i.test(extra)) continue;
    columns.push({ schema, table, column: row.columnName, primaryKey: row.primaryKey, dataType: row.dataType.toLowerCase() });
  }
  return { columns, skipped };
}