import test from 'node:test';
import assert from 'node:assert/strict';
import { createServer } from 'node:http';
import { once } from 'node:events';
import { encodePath, normalizeConfig } from '../lib/core.mjs';
import { applyChange } from '../lib/journal.mjs';
import { discoverStorageConfig } from '../lib/auto-storage.mjs';
import { loadSourceIndex } from '../lib/source-index.mjs';
import { loadSettings, runApply, runPlan, runRollback } from '../migrate.mjs';

// Only point this at the disposable test MySQL container: this suite recreates
// its dedicated codex_oss_migration_test schema, never the application database.
test('MySQL migration, resume, exact reference replacement and rollback', {
  skip: !process.env.MIGRATION_TEST_DB_PORT, timeout: 60000
}, async t => {
  const { default: mysql } = await import('mysql2/promise');
  const db = await mysql.createConnection({ host: '127.0.0.1', port: Number(process.env.MIGRATION_TEST_DB_PORT), user: 'root', password: '', supportBigNumbers: true, bigNumberStrings: true, jsonStrings: true, dateStrings: true });
  const objects = new Map();
  let putCount = 0, failPut = false, sourceDownloads = 0;
  const oss = {
    async head(key) { if (!objects.has(key)) throw Object.assign(new Error('missing'), { code: 'NoSuchKey', status: 404 }); return { res: { headers: { 'content-length': String(objects.get(key).length) } } }; },
    async get(key) { return { content: Buffer.from(objects.get(key)) }; },
    async put(key, bytes, options) {
      assert.equal(options.headers['x-oss-forbid-overwrite'], 'true');
      if (failPut) throw Object.assign(new Error('test failure'), { code: 'TestOssUnavailable' });
      assert.ok(!objects.has(key)); objects.set(key, Buffer.from(bytes)); putCount++;
    }
  };
  const server = createServer(async (request, response) => {
    try {
      const url = new URL(request.url, 'http://localhost');
      if (url.pathname.startsWith('/objects/')) {
        const bytes = objects.get(decodeURIComponent(url.pathname.slice('/objects/'.length)));
        if (!bytes) { response.writeHead(404).end(); return; }
        response.end(bytes); return;
      }
      const prefix = '/admin-api/infra/file/5/get/';
      if (url.pathname.startsWith(prefix)) {
        sourceDownloads++;
        const path = decodeURIComponent(url.pathname.slice(prefix.length).replaceAll('+', ' '));
        const [[row]] = await db.execute('SELECT content FROM infra_file_content WHERE config_id=5 AND path=? AND deleted=0 ORDER BY id DESC LIMIT 1', [path]);
        if (row) response.end(row.content); else response.writeHead(404).end();
        return;
      }
      response.writeHead(404).end();
    } catch { response.writeHead(500).end(); }
  });
  server.listen(0, '127.0.0.1'); await once(server, 'listening');
  const origin = `http://127.0.0.1:${server.address().port}`;
  const schema = 'codex_oss_migration_test';
  let config;
  const firstPath = '头像/张 三+100%.png';
  const firstUrl = `${origin}/admin-api/infra/file/5/get/${encodePath(firstPath)}`;
  const orphanUrl = `${origin}/admin-api/infra/file/5/get/orphan.png`;
  const caseUrl = `${origin}/admin-api/infra/file/5/get/Case.png`;
  const originals = [Buffer.from('first current image'), Buffer.from('second orphan image'), Buffer.from('case latest image')];
  try {
    await db.query(`DROP DATABASE IF EXISTS ${schema}`);
    await db.query(`CREATE DATABASE ${schema} CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci`);
    await db.changeUser({ database: schema });
    await db.query('CREATE TABLE infra_file_config (id BIGINT PRIMARY KEY,storage INT,master BIT(1),config TEXT,deleted BIT(1) DEFAULT 0) ENGINE=InnoDB');
    await db.query('CREATE TABLE infra_file_content (id BIGINT PRIMARY KEY,config_id BIGINT,path VARCHAR(512),content MEDIUMBLOB,deleted BIT(1) DEFAULT 0) ENGINE=InnoDB');
    await db.query('CREATE TABLE infra_file (id BIGINT PRIMARY KEY,config_id BIGINT,path VARCHAR(512),url VARCHAR(1024),type VARCHAR(128),size INT,deleted BIT(1) DEFAULT 0,update_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP) ENGINE=InnoDB');
    await db.query('CREATE TABLE member_user (id BIGINT PRIMARY KEY,avatar VARCHAR(1024),wechat_qr_code VARCHAR(255),extra_json JSON,update_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP) ENGINE=InnoDB');
    await db.query('CREATE TABLE notes (id INT PRIMARY KEY,body TEXT) ENGINE=InnoDB');
    await db.execute('INSERT INTO infra_file_config VALUES (5,1,0,?,0),(30,20,1,?,0)', [JSON.stringify({ domain: origin }), JSON.stringify({ endpoint: 'oss-cn-hangzhou.aliyuncs.com', bucket: 'test', accessKey: 'test-key', accessSecret: 'test-secret', domain: `${origin}/objects` })]);
    const records = [
      ['9007199254740992', 'Case.png', Buffer.from('case old image'), 0],
      ['9007199254740993', firstPath, Buffer.from('first old image'), 0],
      ['9007199254740994', firstPath, originals[0], 0],
      ['9007199254740995', firstPath, Buffer.from('deleted newest'), 1],
      ['9007199254740996', 'orphan.png', originals[1], 0],
      ['9007199254740997', 'case.PNG', originals[2], 0]
    ];
    for (const [contentId, path, content, deleted] of records) await db.execute('INSERT INTO infra_file_content VALUES (?,5,?,?,?)', [contentId, path, content, deleted]);
    for (const [fileId, path, url] of [['9007199254741001', firstPath, firstUrl], ['9007199254741002', 'Case.png', caseUrl], ['9007199254741003', 'case.PNG', caseUrl.replace('Case.png', 'case.PNG')]]) {
      await db.execute("INSERT INTO infra_file (id,config_id,path,url,type,size,update_time) VALUES (?,5,?,?,'image/png',1,'2020-01-01 00:00:00')", [fileId, path, url]);
    }
    const payload = JSON.stringify({ a: [firstUrl, orphanUrl], nested: { image: caseUrl }, outside: 'https://outside.example/image.png' });
    await db.execute("INSERT INTO member_user VALUES (9007199254741101,?,?,?,'2020-01-01 00:00:00')", [firstUrl, orphanUrl, payload]);
    await db.execute('INSERT INTO notes VALUES (1,?)', [`<img src="${caseUrl}">${orphanUrl},${caseUrl}`]);
    config = normalizeConfig({ runId: 'integration', mysql: { database: schema, user: 'root' }, sourceConfigIds: ['5'], targetConfigId: '30', downloadBaseUrl: `${origin}/admin-api/infra/file`, batchSize: 1, retries: 1 });
    const settings = await loadSettings(db, config);

    await t.test('automatic discovery uses the existing database and unique OSS without configuration writes', async () => {
      const automatic = await discoverStorageConfig(db, { mysql: config.mysql, _autoDiscovery: true, _projectDownloadBaseUrl: config.downloadBaseUrl });
      assert.deepEqual(automatic.sourceConfigIds, ['5']); assert.equal(automatic.targetConfigId, '30');
      assert.equal(automatic.sourceMode, 'database'); assert.ok(automatic.runId.startsWith('db-images-'));
      assert.ok(automatic.oldDomains.includes(origin)); assert.equal(objects.size, 0);
      const [tables] = await db.query("SHOW TABLES LIKE 'infra_file_oss_migration_%'"); assert.equal(tables.length, 0);
    });

    await t.test('plan is read-only and uses latest active content with bigint/collation semantics', async () => {
      const statements = [];
      const readOnlyDb = {
        execute(sql, params) { statements.push(sql); return db.execute(sql, params); },
        query(sql, params) { statements.push(sql); return db.query(sql, params); }
      };
      const plan = await runPlan(readOnlyDb, config, settings);
      assert.equal(plan.status, 'planned'); assert.equal(plan.candidates, 3); assert.equal(plan.withoutFileRecord, 1);
      assert.equal(plan.bytes, null); assert.equal(plan.estimatedBytes, 2); assert.equal(plan.sizePending, 1); assert.equal(plan.contentVerified, false);
      assert.ok(statements.every(sql => /^\s*SELECT/i.test(sql)));
      assert.ok(statements.every(sql => !/\b(?:SHA2|OCTET_LENGTH|content)\b|NOT EXISTS/i.test(sql)), 'plan must never scan image bytes or use per-file correlated subqueries');
      assert.equal(plan.references.changedRows, 2); assert.equal(objects.size, 0);
      const [rows] = await db.query("SHOW TABLES LIKE 'infra_file_oss_migration_%'"); assert.equal(rows.length, 0);
    });

    await t.test('metadata upper bound excludes newer versions without misclassifying their files as missing', async () => {
      const index = await loadSourceIndex(db, config, { maxId: '9007199254740996' });
      assert.equal(index.groups.length, 2); assert.equal(index.missingFiles.length, 0);
      assert.equal(index.groups[0].path, firstPath); assert.equal(index.groups[1].path, 'orphan.png');
      const full = await loadSourceIndex(db, config);
      assert.deepEqual(full.groups[2].spellings.slice().sort(), ['Case.png', 'case.PNG'].sort());
      assert.equal(full.groups[2].files.length, 2);
    });

    await t.test('limited trial, resume and repeated apply preserve originals and journals', async () => {
      const trial = await runApply(db, config, settings, 1, () => oss);
      assert.equal(trial.status, 'partial'); assert.equal(trial.items.switched, 1); assert.equal(objects.size, 1);
      const resumedConfig = await discoverStorageConfig(db, { mysql: config.mysql, _autoDiscovery: true }, { command: 'apply' });
      assert.equal(resumedConfig.runId, 'integration'); assert.deepEqual(resumedConfig.sourceConfigIds, ['5']);
      assert.equal(resumedConfig.sourceMode, 'http'); assert.equal(resumedConfig.downloadBaseUrl, config.downloadBaseUrl);
      assert.deepEqual((await loadSettings(db, resumedConfig)).safe, settings.safe);
      // Simulate interruption after all remaining uploads reached verified but
      // before their first file transaction. Resume must reuse the checkpoint.
      const interruptedDb = {
        execute: (...args) => db.execute(...args), query: (...args) => db.query(...args),
        beginTransaction: async () => { throw new Error('simulated interruption before transition'); }
      };
      await assert.rejects(runApply(interruptedDb, config, settings, Infinity, () => oss), /simulated interruption/);
      const [[checkpoints]] = await db.query("SELECT COUNT(*) AS n FROM infra_file_oss_migration_items WHERE run_id='integration' AND status='verified'");
      assert.equal(Number(checkpoints.n), 2);
      const downloadsBeforeResume = sourceDownloads;
      const putsBeforeResume = putCount;
      const done = await runApply(db, config, settings, Infinity, () => oss);
      assert.equal(done.status, 'complete'); assert.equal(done.items.switched, 3); assert.equal(objects.size, 3); assert.equal(putCount, 3);
      assert.equal(done.revalidated, 2); assert.equal(done.uploaded, 0);
      assert.equal(sourceDownloads, downloadsBeforeResume); assert.equal(putCount, putsBeforeResume);
      const [[member]] = await db.query('SELECT * FROM member_user');
      assert.ok(member.avatar.startsWith(`${origin}/objects/`)); assert.ok(member.wechat_qr_code.startsWith(`${origin}/objects/`));
      assert.deepEqual(Object.values(JSON.parse(member.extra_json).nested).map(url => url.startsWith(`${origin}/objects/`)), [true]);
      assert.equal(JSON.parse(member.extra_json).outside, 'https://outside.example/image.png');
      const [[counts]] = await db.query('SELECT COUNT(*) AS n FROM infra_file_content'); assert.equal(Number(counts.n), 6);
      const [files] = await db.query('SELECT * FROM infra_file ORDER BY id');
      assert.ok(files.every(file => file.config_id === '30'));
      assert.equal(files[1].path, files[2].path); // case/accent-insensitive source lookup remains equivalent
      assert.equal(files[0].size, originals[0].length); // stale old metadata is repaired
      assert.ok(files.every(file => file.update_time === '2020-01-01 00:00:00'));
      const [[before]] = await db.query('SELECT COUNT(*) AS n FROM infra_file_oss_migration_changes');
      const repeated = await runApply(db, config, settings, Infinity, () => oss); assert.equal(repeated.status, 'complete');
      const [[after]] = await db.query('SELECT COUNT(*) AS n FROM infra_file_oss_migration_changes'); assert.equal(before.n, after.n); assert.equal(putCount, 3);
      const [[entry]] = await db.query("SELECT data FROM infra_file_oss_migration_items WHERE JSON_UNQUOTE(JSON_EXTRACT(data,'$.path')) LIKE '%100%.png'");
      assert.ok(JSON.parse(entry.data).aliases.includes(firstUrl)); // logs must never rewrite themselves
    });

    await t.test('resume does not rewrite new references using a deleted OSS object', async () => {
      const [[entry]] = await db.query("SELECT data FROM infra_file_oss_migration_items WHERE JSON_UNQUOTE(JSON_EXTRACT(data,'$.path')) LIKE '%100%.png'");
      const item = JSON.parse(entry.data), bytes = objects.get(item.key);
      objects.delete(item.key);
      await db.execute('INSERT INTO notes VALUES (2,?)', [firstUrl]);
      const result = await runApply(db, config, settings, Infinity, () => oss);
      assert.equal(result.status, 'partial'); assert.equal(result.failed, 1);
      const [[note]] = await db.query('SELECT body FROM notes WHERE id=2'); assert.equal(note.body, firstUrl);
      objects.set(item.key, bytes); await db.query('DELETE FROM notes WHERE id=2');
    });

    await t.test('rollback preserves later user edits and can resume after conflict resolution', async () => {
      const [[migrated]] = await db.query('SELECT avatar FROM member_user');
      await db.execute('UPDATE member_user SET avatar=?', ['https://user-new.example/avatar.png']);
      const rollback = await runRollback(db, config); assert.equal(rollback.status, 'rollback_partial'); assert.ok(rollback.conflicts.length > 0);
      const [[current]] = await db.query('SELECT avatar FROM member_user'); assert.equal(current.avatar, 'https://user-new.example/avatar.png');
      await db.execute('UPDATE member_user SET avatar=?', [migrated.avatar]);
      const resumed = await runRollback(db, config); assert.equal(resumed.status, 'rolled_back');
      const [[member]] = await db.query('SELECT * FROM member_user'); assert.equal(member.avatar, firstUrl); assert.equal(member.wechat_qr_code, orphanUrl); assert.deepEqual(JSON.parse(member.extra_json), JSON.parse(payload));
      const [files] = await db.query('SELECT * FROM infra_file'); assert.ok(files.every(file => file.config_id === '5')); assert.equal(objects.size, 3);
      await assert.rejects(runApply(db, config, settings, Infinity, () => oss), /回滚/);
    });

    await t.test('failed uploads persist a partial state and never switch database addresses', async () => {
      const failureConfig = { ...config, runId: 'upload-failure', objectPrefix: 'migration/failure' };
      const failureSettings = await loadSettings(db, failureConfig); failPut = true;
      const result = await runApply(db, failureConfig, failureSettings, Infinity, () => oss); failPut = false;
      assert.equal(result.status, 'partial'); assert.equal(result.items.failed, 3);
      const [[run]] = await db.query("SELECT status FROM infra_file_oss_migration_runs WHERE run_id='upload-failure'"); assert.equal(run.status, 'partial');
      const [[member]] = await db.query('SELECT avatar FROM member_user'); assert.equal(member.avatar, firstUrl);
    });

    await t.test('non-strict MySQL truncation warnings roll back both data and journal', async () => {
      await db.query('CREATE TABLE short_url (id INT PRIMARY KEY,url VARCHAR(8)) ENGINE=InnoDB');
      await db.query("INSERT INTO short_url VALUES (1,'old')");
      const [[mode]] = await db.query('SELECT @@SESSION.sql_mode AS value');
      await db.query("SET SESSION sql_mode=''"); await db.beginTransaction();
      await assert.rejects(applyChange(db, schema, 'warning', { schema, table: 'short_url', primaryKey: 'id', pkValue: 1, before: { url: 'old' }, after: { url: 'this-would-be-truncated' } }), /警告/);
      await db.rollback(); await db.execute('SET SESSION sql_mode=?', [mode.value]);
      const [[row]] = await db.query('SELECT url FROM short_url'); assert.equal(row.url, 'old');
      const [[count]] = await db.query("SELECT COUNT(*) AS n FROM infra_file_oss_migration_changes WHERE run_id='warning'"); assert.equal(Number(count.n), 0);
    });

    await t.test('batch CAS conflict isolates the edited file and commits unrelated verified items', async () => {
      const conflictConfig = { ...config, runId: 'batch-conflict', objectPrefix: 'migration/conflict' };
      const conflictSettings = await loadSettings(db, conflictConfig);
      let changed = false, rollbacks = 0;
      const concurrentEditDb = {
        query: (...args) => db.query(...args), execute: (...args) => db.execute(...args),
        async beginTransaction() {
          if (!changed) {
            changed = true;
            await db.execute('UPDATE infra_file SET url=? WHERE id=?', ['https://user-change.example/new.png', '9007199254741001']);
          }
          return db.beginTransaction();
        },
        commit: () => db.commit(),
        rollback() { rollbacks++; return db.rollback(); }
      };
      const result = await runApply(concurrentEditDb, conflictConfig, conflictSettings, Infinity, () => oss);
      assert.equal(result.status, 'partial'); assert.equal(result.failed, 1);
      assert.equal(result.items.switched, 2); assert.equal(result.items.verified, 1);
      assert.ok(rollbacks >= 2, 'the batch and the conflicted single item are rolled back');
      const [[edited]] = await db.query('SELECT config_id,url FROM infra_file WHERE id=9007199254741001');
      assert.equal(edited.config_id, '5'); assert.equal(edited.url, 'https://user-change.example/new.png');
      const [[member]] = await db.query('SELECT avatar FROM member_user'); assert.equal(member.avatar, firstUrl);
      const rollback = await runRollback(db, conflictConfig); assert.equal(rollback.status, 'rolled_back');
      await db.execute('UPDATE infra_file SET url=? WHERE id=9007199254741001', [firstUrl]);
    });

    await t.test('database-source mode works and ambiguous raw-plus links require review', async () => {
      const rawUrl = `${origin}/admin-api/infra/file/5/get/${firstPath}`;
      await db.execute('INSERT INTO notes VALUES (3,?)', [rawUrl]);
      const directConfig = { ...config, runId: 'database-mode', sourceMode: 'database' };
      const directSettings = await loadSettings(db, directConfig);
      const result = await runApply(db, directConfig, directSettings, Infinity, () => oss);
      assert.equal(result.status, 'needs-review'); assert.equal(result.items.switched, 3);
      const [[note]] = await db.query('SELECT body FROM notes WHERE id=3'); assert.equal(note.body, rawUrl);
      assert.ok(result.references.unresolvedRows > 0);
    });
  } finally {
    server.close(); server.closeAllConnections();
    await db.query(`DROP DATABASE IF EXISTS ${schema}`);
    await db.end();
  }
});