import { synchronizeMysqlPartitions } from './partition-storage.mjs';

export function createMysqlAdapter(context) {
  const {
    mkdir,
    dirname,
    sqliteSchema,
    mysqlSchema,
    optional,
    buildSeedOperations,
    stateFromRows,
    readSqliteRows,
    readMysqlRows,
    createRepository
  } = context;

  async function createMysqlStore({ seed }) {
    const { default: mysql } = await import('mysql2/promise');
    const connectionUrl = process.env.DATABASE_URL;
    const database = process.env.MYSQL_DATABASE;
  
    if (!connectionUrl && (!process.env.MYSQL_HOST || !process.env.MYSQL_USER || !database)) {
      throw new Error('MySQL 配置不完整：请设置 DATABASE_URL，或 MYSQL_HOST、MYSQL_USER、MYSQL_DATABASE');
    }
  
    const pool = connectionUrl
      ? mysql.createPool(connectionUrl)
      : mysql.createPool({
          host: process.env.MYSQL_HOST,
          port: Number(process.env.MYSQL_PORT || 3306),
          user: process.env.MYSQL_USER,
          password: process.env.MYSQL_PASSWORD || '',
          database,
          waitForConnections: true,
          connectionLimit: Number(process.env.MYSQL_CONNECTION_LIMIT || 10),
          charset: 'utf8mb4',
          timezone: 'Z',
          enableKeepAlive: true
        });
  
    const mysqlTableNames = mysqlSchema.map(statement =>
      statement.match(/^CREATE TABLE IF NOT EXISTS\s+([a-z0-9_]+)/i)?.[1]
    ).filter(Boolean);
    const [databaseTables] = await pool.execute(`
      SELECT TABLE_NAME FROM information_schema.TABLES
      WHERE TABLE_SCHEMA = DATABASE() AND TABLE_TYPE = 'BASE TABLE'
    `);
    const existingTableNames = new Set(databaseTables.map(row => row.TABLE_NAME));
    const existingAppTables = mysqlTableNames.filter(table => existingTableNames.has(table));
    let hasSchemaMetadata = false;
    let existingSchemaVersion = null;
    if (existingTableNames.has('schema_metadata')) {
      const [metadataRows] = await pool.execute('SELECT id, schema_version FROM schema_metadata WHERE id = 1');
      hasSchemaMetadata = metadataRows.length > 0;
      existingSchemaVersion = hasSchemaMetadata ? Number(metadataRows[0].schema_version) : null;
    }
    if (existingAppTables.length && (!hasSchemaMetadata || ![15, 16, 17].includes(existingSchemaVersion))) {
      for (const table of [...mysqlTableNames].reverse()) {
        await pool.query(`DROP TABLE IF EXISTS \`${table}\``);
      }
    }

    for (const statement of mysqlSchema) await pool.query(statement);
    // Development schemas are created from the current DDL as a whole. MySQL 8.4 lacks
    // MariaDB-style conditional column addition; outdated schemas are rejected by the
    // version check below and should be rebuilt instead of migrated column by column.
    const [resultLockTriggers] = await pool.execute(`
      SELECT TRIGGER_NAME FROM information_schema.TRIGGERS
      WHERE TRIGGER_SCHEMA = DATABASE() AND TRIGGER_NAME LIKE 'trg_results_lock_archived_%'
    `);
    const existingResultLockTriggers = new Set(resultLockTriggers.map(item => item.TRIGGER_NAME));
    const mysqlResultLockTriggers = {
      trg_results_lock_archived_insert: `CREATE TRIGGER trg_results_lock_archived_insert BEFORE INSERT ON results FOR EACH ROW
        BEGIN
          IF EXISTS (SELECT 1 FROM registrations registration JOIN exams exam ON exam.id = registration.exam_id
            WHERE registration.id = NEW.registration_id AND exam.archived_at IS NOT NULL) THEN
            SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = '归档考试成绩已永久锁定';
          END IF;
        END`,
      trg_results_lock_archived_update: `CREATE TRIGGER trg_results_lock_archived_update BEFORE UPDATE ON results FOR EACH ROW
        BEGIN
          IF EXISTS (SELECT 1 FROM registrations registration JOIN exams exam ON exam.id = registration.exam_id
            WHERE registration.id IN (OLD.registration_id, NEW.registration_id) AND exam.archived_at IS NOT NULL) THEN
            SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = '归档考试成绩已永久锁定';
          END IF;
        END`,
      trg_results_lock_archived_delete: `CREATE TRIGGER trg_results_lock_archived_delete BEFORE DELETE ON results FOR EACH ROW
        BEGIN
          IF EXISTS (SELECT 1 FROM registrations registration JOIN exams exam ON exam.id = registration.exam_id
            WHERE registration.id = OLD.registration_id AND exam.archived_at IS NOT NULL) THEN
            SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = '归档考试成绩已永久锁定';
          END IF;
        END`
    };
    for (const [name, statement] of Object.entries(mysqlResultLockTriggers)) {
      if (!existingResultLockTriggers.has(name)) await pool.query(statement);
    }
    const [legacyRegistrationNumberIndexes] = await pool.query("SHOW INDEX FROM registrations WHERE Key_name = 'uq_registrations_number'");
    if (legacyRegistrationNumberIndexes.length) await pool.query('ALTER TABLE registrations DROP INDEX uq_registrations_number');
    const [existing] = await pool.execute('SELECT id FROM schema_metadata WHERE id = 1');
    if (existing.length) {
      const [metadataRows] = await pool.execute('SELECT app_version, schema_version FROM schema_metadata WHERE id = 1');
      if (Number(metadataRows[0]?.schema_version || 1) < 9) {
        await pool.query('DROP TABLE IF EXISTS admit_card_subjects');
        await pool.query('DROP TABLE IF EXISTS admit_cards');
        await pool.query('DROP TABLE IF EXISTS exam_arrangement_plans');
        await pool.query('DROP TABLE IF EXISTS admission_number_rules');
        const admissionTables = ['admission_number_rules', 'exam_arrangement_plans', 'admit_cards', 'admit_card_subjects'];
        for (const table of admissionTables) {
          const statement = mysqlSchema.find(item => item.includes(`CREATE TABLE IF NOT EXISTS ${table} (`));
          if (!statement) throw new Error(`缺少 ${table} 的 MySQL 表定义`);
          await pool.query(statement);
        }
        const extension = seed();
        for (const rule of extension.admissionNumberRules) await pool.execute(
          `INSERT INTO admission_number_rules (
            id, code, name, description, \`separator\`, segments_json, example, active, created_at
          ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
          [rule.id, rule.code, rule.name, rule.description, rule.separator || '', JSON.stringify(rule.segments || []),
            rule.example || '', rule.active === false ? 0 : 1, rule.createdAt]
        );
        await pool.execute('UPDATE schema_metadata SET schema_version = 9, app_version = 9 WHERE id = 1');
        metadataRows[0].schema_version = 9;
        metadataRows[0].app_version = 9;
      }
      if (Number(metadataRows[0]?.schema_version || 1) < 10) {
        await pool.execute(`UPDATE admit_cards card LEFT JOIN test_centers center ON center.id = card.center_id SET
          card.center_code = COALESCE(center.code, card.center_code),
          card.center_address = COALESCE(CONCAT_WS(' ', center.province_name, center.city_name, center.district_name, center.address), card.center_address)`);
        await pool.execute(`UPDATE admit_card_subjects assignment LEFT JOIN test_rooms room ON room.id = assignment.room_id SET
          assignment.building = COALESCE(room.building, assignment.building),
          assignment.floor = COALESCE(room.floor, assignment.floor)`);
        await pool.execute('UPDATE schema_metadata SET schema_version = 10, app_version = 10 WHERE id = 1');
        metadataRows[0].schema_version = 10;
        metadataRows[0].app_version = 10;
      }
      if (Number(metadataRows[0]?.schema_version || 1) < 11) {
        await pool.execute("UPDATE exam_subjects SET pass_rule = 'fixed_score', pass_value = pass_score");
        await pool.execute('UPDATE schema_metadata SET schema_version = 11, app_version = 11 WHERE id = 1');
        metadataRows[0].schema_version = 11;
        metadataRows[0].app_version = 11;
      }
      if (Number(metadataRows[0]?.schema_version || 1) < 12) {
        await pool.execute("UPDATE exams SET pass_policy = 'rank_percent' WHERE pass_policy = 'score_ratio'");
        await pool.execute('UPDATE schema_metadata SET schema_version = 12, app_version = 12 WHERE id = 1');
        metadataRows[0].schema_version = 12;
        metadataRows[0].app_version = 12;
      }
      if (Number(metadataRows[0]?.schema_version || 1) < 13) {
        await pool.execute('UPDATE schema_metadata SET schema_version = 13, app_version = 13 WHERE id = 1');
        metadataRows[0].schema_version = 13;
        metadataRows[0].app_version = 13;
      }
      if (Number(metadataRows[0]?.schema_version || 1) < 15) {
        throw new Error('数据库结构已升级到 v15，请重建开发数据库后重新启动');
      }
      if (Number(metadataRows[0]?.schema_version || 1) < 16) {
        await pool.execute('UPDATE schema_metadata SET schema_version = 16 WHERE id = 1');
        metadataRows[0].schema_version = 16;
      }
      if (Number(metadataRows[0]?.schema_version || 1) < 17) {
        await pool.query(`ALTER TABLE users
          ADD COLUMN totp_enabled BOOLEAN NOT NULL DEFAULT FALSE AFTER must_change_password,
          ADD COLUMN totp_secret_encrypted VARCHAR(512) NULL AFTER totp_enabled,
          ADD COLUMN totp_recovery_codes VARCHAR(2048) NOT NULL DEFAULT '[]' AFTER totp_secret_encrypted,
          ADD COLUMN totp_last_used_step BIGINT NULL AFTER totp_recovery_codes`);
        await pool.execute('UPDATE schema_metadata SET schema_version = 17 WHERE id = 1');
        metadataRows[0].schema_version = 17;
      }
      if (Number(metadataRows[0]?.schema_version || 1) < 18) {
        await pool.query("ALTER TABLE users MODIFY COLUMN role ENUM('admin', 'candidate', 'admission_school') NOT NULL");
        const [profileColumns] = await pool.query("SHOW COLUMNS FROM candidate_profiles WHERE Field IN ('specialty_types', 'specialty_certificate', 'policy_eligibility')");
        const existingProfileColumns = new Set(profileColumns.map(item => item.Field));
        if (!existingProfileColumns.has('specialty_types')) await pool.query('ALTER TABLE candidate_profiles ADD COLUMN specialty_types JSON NOT NULL DEFAULT (JSON_ARRAY()) AFTER guardian_phone');
        if (!existingProfileColumns.has('specialty_certificate')) await pool.query('ALTER TABLE candidate_profiles ADD COLUMN specialty_certificate VARCHAR(255) NULL AFTER specialty_types');
        if (!existingProfileColumns.has('policy_eligibility')) await pool.query('ALTER TABLE candidate_profiles ADD COLUMN policy_eligibility VARCHAR(255) NULL AFTER specialty_certificate');
        await pool.execute('UPDATE schema_metadata SET schema_version = 18, app_version = 18 WHERE id = 1');
        metadataRows[0].schema_version = 18;
      }
      if (Number(metadataRows[0]?.app_version || 1) < 2) {
        const extension = seed();
        const connection = await pool.getConnection();
        try {
          await connection.beginTransaction();
          for (const school of extension.schools) await connection.execute(
            'INSERT IGNORE INTO schools (id, name, code, address, active) VALUES (?, ?, ?, ?, ?)',
            [school.id, school.name, school.code, optional(school.address), school.active === false ? 0 : 1]
          );
          for (const schoolClass of extension.classes) await connection.execute(
            'INSERT IGNORE INTO school_classes (id, school_id, name, grade, active) VALUES (?, ?, ?, ?, ?)',
            [schoolClass.id, schoolClass.schoolId, schoolClass.name, schoolClass.grade, schoolClass.active === false ? 0 : 1]
          );
          await connection.execute("UPDATE users SET admin_level = COALESCE(admin_level, 'super'), active = COALESCE(active, TRUE) WHERE role = 'admin'");
          for (const user of extension.users.filter(item => item.role === 'admin')) await connection.execute(
            `INSERT IGNORE INTO users (
              id, username, password_hash, role, admin_level, school_id, class_id, active, display_name, created_at
            ) VALUES (?, ?, ?, 'admin', ?, ?, ?, ?, ?, ?)`,
            [user.id, user.username, user.passwordHash, user.adminLevel, optional(user.schoolId), optional(user.classId), user.active === false ? 0 : 1, user.displayName, user.createdAt]
          );
          for (const profile of extension.candidateProfiles) await connection.execute(
            `UPDATE candidate_profiles SET school_id = COALESCE(school_id, ?), class_id = COALESCE(class_id, ?)
             WHERE school = ? AND grade = ?`,
            [optional(profile.schoolId), optional(profile.classId), profile.school, profile.grade]
          );
          const [centerRows] = await connection.execute('SELECT id FROM test_centers LIMIT 1');
          if (!centerRows.length) for (const center of extension.testCenters) await connection.execute(
            'INSERT INTO test_centers (id, school_id, name, address, contact, rooms, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?)',
            [center.id, center.schoolId, center.name, center.address, optional(center.contact), center.rooms || '', center.updatedAt]
          );
          const [ruleRows] = await connection.execute('SELECT id FROM number_rules LIMIT 1');
          if (!ruleRows.length) for (const rule of extension.numberRules) {
            await connection.execute(
              'INSERT INTO number_rules (id, name, `separator`, active, created_by, updated_at) VALUES (?, ?, ?, ?, ?, ?)',
              [rule.id, rule.name, rule.separator || '', rule.active ? 1 : 0, optional(rule.createdBy), rule.updatedAt]
            );
            for (const [index, segment] of rule.segments.entries()) await connection.execute(
              'INSERT INTO number_rule_segments (id, rule_id, position, type, value, width) VALUES (?, ?, ?, ?, ?, ?)',
              [segment.id, rule.id, Number(segment.position || index + 1), segment.type, optional(segment.value), Number(segment.width || 0)]
            );
          }
          const [workflowRows] = await connection.execute('SELECT id FROM workflow_definitions LIMIT 1');
          if (!workflowRows.length) for (const workflow of extension.workflows) {
            await connection.execute(
              'INSERT INTO workflow_definitions (id, business_type, name, active, updated_by, updated_at) VALUES (?, ?, ?, ?, ?, ?)',
              [workflow.id, workflow.businessType, workflow.name, workflow.active === false ? 0 : 1, optional(workflow.updatedBy), workflow.updatedAt]
            );
            for (const [index, step] of workflow.steps.entries()) await connection.execute(
              'INSERT INTO workflow_steps (id, workflow_id, position, name, admin_level) VALUES (?, ?, ?, ?, ?)',
              [step.id, workflow.id, Number(step.position || index + 1), step.name, step.adminLevel]
            );
          }
          await connection.execute('UPDATE schema_metadata SET schema_version = 2, app_version = 2 WHERE id = 1');
          await connection.commit();
        } catch (error) {
          await connection.rollback();
          throw error;
        } finally {
          connection.release();
        }
      }
      if (Number(metadataRows[0]?.app_version || 1) < 3) {
        const extension = seed();
        const connection = await pool.getConnection();
        try {
          await connection.beginTransaction();
          for (const center of extension.testCenters) await connection.execute(
            `UPDATE test_centers SET
              code = COALESCE(NULLIF(code, ''), ?), manager_name = COALESCE(manager_name, ?),
              manager_phone = COALESCE(manager_phone, ?), emergency_phone = COALESCE(emergency_phone, ?),
              gate_open_time = COALESCE(gate_open_time, ?), transport = COALESCE(transport, ?),
              status = COALESCE(status, 'active'), notes = COALESCE(notes, ?)
            WHERE id = ?`,
            [center.code, optional(center.managerName), optional(center.managerPhone), optional(center.emergencyPhone),
              optional(center.gateOpenTime), optional(center.transport), optional(center.notes), center.id]
          );
          await connection.execute("UPDATE test_centers SET code = CONCAT('CENTER-', RIGHT(id, 8)) WHERE code IS NULL OR code = ''");
          const [roomRows] = await connection.execute('SELECT id FROM test_rooms LIMIT 1');
          if (!roomRows.length) for (const room of extension.testRooms) await connection.execute(
            `INSERT INTO test_rooms (
              id, center_id, code, name, building, floor, capacity, seat_start, seat_end, room_type, status, notes
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
            [room.id, room.centerId, room.code, room.name, room.building, optional(room.floor), Number(room.capacity),
              Number(room.seatStart), Number(room.seatEnd), room.roomType, room.status, optional(room.notes)]
          );
          const centerWorkflow = extension.workflows.find(item => item.businessType === 'center_change');
          const [centerWorkflowRows] = await connection.execute("SELECT id FROM workflow_definitions WHERE business_type = 'center_change' AND active = 1");
          if (centerWorkflow && !centerWorkflowRows.length) {
            await connection.execute(
              'INSERT INTO workflow_definitions (id, business_type, name, active, updated_by, updated_at) VALUES (?, ?, ?, ?, ?, ?)',
              [centerWorkflow.id, centerWorkflow.businessType, centerWorkflow.name, 1, optional(centerWorkflow.updatedBy), centerWorkflow.updatedAt]
            );
            for (const [index, step] of centerWorkflow.steps.entries()) await connection.execute(
              'INSERT INTO workflow_steps (id, workflow_id, position, name, admin_level) VALUES (?, ?, ?, ?, ?)',
              [step.id, centerWorkflow.id, Number(step.position || index + 1), step.name, step.adminLevel]
            );
          }
          await connection.execute('UPDATE schema_metadata SET schema_version = 3, app_version = 3 WHERE id = 1');
          await connection.commit();
        } catch (error) {
          await connection.rollback();
          throw error;
        } finally {
          connection.release();
        }
      }
      if (Number(metadataRows[0]?.app_version || 1) < 4) {
        const extension = seed();
        const connection = await pool.getConnection();
        try {
          await connection.beginTransaction();
          for (const user of extension.users.filter(item => item.role === 'candidate')) await connection.execute(
            `UPDATE users SET candidate_number = COALESCE(NULLIF(candidate_number, ''), ?),
              must_change_password = COALESCE(must_change_password, ?) WHERE id = ?`,
            [optional(user.candidateNumber), user.mustChangePassword ? 1 : 0, user.id]
          );
          await connection.execute(`UPDATE users SET candidate_number = COALESCE(
            (SELECT registration_number FROM registrations WHERE registrations.user_id = users.id AND registration_number IS NOT NULL AND registration_number <> '' ORDER BY created_at LIMIT 1),
            CONCAT('CAND-', RIGHT(id, 10))
          ) WHERE role = 'candidate' AND (candidate_number IS NULL OR candidate_number = '')`);
          for (const profile of extension.candidateProfiles) await connection.execute(
            `UPDATE candidate_profiles SET native_place = COALESCE(native_place, ?), birth_date = COALESCE(birth_date, ?),
              ethnicity = COALESCE(ethnicity, ?), postal_code = COALESCE(postal_code, ?), guardian_name = COALESCE(guardian_name, ?),
              guardian_phone = COALESCE(guardian_phone, ?), profile_completed = ? WHERE id = ?`,
            [optional(profile.nativePlace), optional(profile.birthDate), optional(profile.ethnicity), optional(profile.postalCode),
              optional(profile.guardianName), optional(profile.guardianPhone), profile.profileCompleted ? 1 : 0, profile.id]
          );
          await connection.execute(`UPDATE registrations JOIN users ON users.id = registrations.user_id
            SET registrations.registration_number = users.candidate_number
            WHERE registrations.registration_number IS NULL OR registrations.registration_number = ''`);
          await connection.execute('UPDATE schema_metadata SET schema_version = 4, app_version = 4 WHERE id = 1');
          await connection.commit();
        } catch (error) {
          await connection.rollback();
          throw error;
        } finally {
          connection.release();
        }
      }
      if (Number(metadataRows[0]?.app_version || 1) < 5) {
        const extension = seed();
        const batchWorkflow = extension.workflows.find(item => item.businessType === 'candidate_account_batch');
        const connection = await pool.getConnection();
        try {
          await connection.beginTransaction();
          const [batchWorkflowRows] = await connection.execute("SELECT id FROM workflow_definitions WHERE business_type = 'candidate_account_batch' AND active = 1");
          if (batchWorkflow && !batchWorkflowRows.length) {
            await connection.execute(
              'INSERT INTO workflow_definitions (id, business_type, name, active, updated_by, updated_at) VALUES (?, ?, ?, ?, ?, ?)',
              [batchWorkflow.id, batchWorkflow.businessType, batchWorkflow.name, 1, optional(batchWorkflow.updatedBy), batchWorkflow.updatedAt]
            );
            for (const [index, step] of batchWorkflow.steps.entries()) await connection.execute(
              'INSERT INTO workflow_steps (id, workflow_id, position, name, admin_level) VALUES (?, ?, ?, ?, ?)',
              [step.id, batchWorkflow.id, Number(step.position || index + 1), step.name, step.adminLevel]
            );
          }
          await connection.execute('UPDATE schema_metadata SET schema_version = 5, app_version = 5 WHERE id = 1');
          await connection.commit();
        } catch (error) {
          await connection.rollback();
          throw error;
        } finally {
          connection.release();
        }
      }
      if (Number(metadataRows[0]?.app_version || 1) < 6) {
        await pool.execute('UPDATE schema_metadata SET schema_version = 6, app_version = 6 WHERE id = 1');
      }
      if (Number(metadataRows[0]?.app_version || 1) < 7) {
        await pool.execute('UPDATE schema_metadata SET schema_version = 7, app_version = 7 WHERE id = 1');
      }
      if (Number(metadataRows[0]?.schema_version || 1) < 15) {
        throw new Error('数据库结构已升级到 v15，请重建开发数据库后重新启动');
      }
    }
    if (!existing.length) {
      const initialState = seed();
      const connection = await pool.getConnection();
      try {
        await connection.beginTransaction();
        const [insert] = await connection.execute(`
          INSERT IGNORE INTO schema_metadata (id, schema_version, app_version, self_registration_enabled, created_at)
          VALUES (1, 18, ?, ?, ?)
        `, [Number(initialState.meta?.version || 1), initialState.settings?.selfRegistrationEnabled ? 1 : 0, initialState.meta?.createdAt || new Date().toISOString()]);
        if (insert.affectedRows === 1) {
          for (const item of buildSeedOperations(initialState)) await connection.execute(item.sql, item.params);
        }
        await connection.commit();
      } catch (error) {
        await connection.rollback();
        throw error;
      } finally {
        connection.release();
      }
    }
  
    const [centerCodeIndexes] = await pool.query("SHOW INDEX FROM test_centers WHERE Key_name = 'uq_centers_code'");
    if (!centerCodeIndexes.length) {
      await pool.query('ALTER TABLE test_centers MODIFY COLUMN code VARCHAR(40) NOT NULL, ADD UNIQUE KEY uq_centers_code (code)');
    }
    const [candidateNumberIndexes] = await pool.query("SHOW INDEX FROM users WHERE Key_name = 'uq_users_candidate_number'");
    if (!candidateNumberIndexes.length) {
      await pool.query('ALTER TABLE users ADD UNIQUE KEY uq_users_candidate_number (candidate_number)');
    }

    const partitionConnection = await pool.getConnection();
    try {
      await synchronizeMysqlPartitions(partitionConnection);
    } finally {
      partitionConnection.release();
    }
  
    const transaction = async operations => {
      const connection = await pool.getConnection();
      try {
        await connection.beginTransaction();
        for (const item of operations) await connection.execute(item.sql, item.params);
        await connection.commit();
        await synchronizeMysqlPartitions(connection);
      } catch (error) {
        await connection.rollback();
        throw error;
      } finally {
        connection.release();
      }
    };
    const read = async () => {
      const connection = await pool.getConnection();
      try {
        await connection.beginTransaction();
        const state = stateFromRows(await readMysqlRows(connection));
        await connection.commit();
        return state;
      } catch (error) {
        await connection.rollback();
        throw error;
      } finally {
        connection.release();
      }
    };
    return createRepository({
      client: 'mysql',
      location: connectionUrl ? 'DATABASE_URL' : `${process.env.MYSQL_HOST}:${process.env.MYSQL_PORT || 3306}/${database}`,
      read,
      transaction,
      close: async () => pool.end()
    });
  }

  return createMysqlStore;
}