450 lines
27 KiB
JavaScript
450 lines
27 KiB
JavaScript
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, 18, 19, 20].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]?.schema_version || 1) < 19) {
|
|
const [schoolColumns] = await pool.query("SHOW COLUMNS FROM schools WHERE Field IN ('is_source_school', 'is_admission_school')");
|
|
const existingSchoolColumns = new Set(schoolColumns.map(item => item.Field));
|
|
if (!existingSchoolColumns.has('is_source_school')) await pool.query('ALTER TABLE schools ADD COLUMN is_source_school BOOLEAN NOT NULL DEFAULT TRUE AFTER address');
|
|
if (!existingSchoolColumns.has('is_admission_school')) await pool.query('ALTER TABLE schools ADD COLUMN is_admission_school BOOLEAN NOT NULL DEFAULT TRUE AFTER is_source_school');
|
|
const [specialtyColumns] = await pool.query("SHOW COLUMNS FROM candidate_profiles WHERE Field IN ('specialty_category', 'specialty_type')");
|
|
const existingSpecialtyColumns = new Set(specialtyColumns.map(item => item.Field));
|
|
if (!existingSpecialtyColumns.has('specialty_category')) await pool.query('ALTER TABLE candidate_profiles ADD COLUMN specialty_category VARCHAR(30) NULL AFTER guardian_phone');
|
|
if (!existingSpecialtyColumns.has('specialty_type')) await pool.query('ALTER TABLE candidate_profiles ADD COLUMN specialty_type VARCHAR(40) NULL AFTER specialty_category');
|
|
const [registrationColumns] = await pool.query("SHOW COLUMNS FROM registrations WHERE Field = 'feature_score'");
|
|
if (!registrationColumns.length) await pool.query('ALTER TABLE registrations ADD COLUMN feature_score DECIMAL(8,2) NOT NULL DEFAULT 0 AFTER number_rule_id');
|
|
await pool.execute('UPDATE schema_metadata SET schema_version = 19, app_version = 19 WHERE id = 1');
|
|
metadataRows[0].schema_version = 19;
|
|
}
|
|
if (Number(metadataRows[0]?.schema_version || 1) < 20) {
|
|
await pool.query("ALTER TABLE admission_records MODIFY COLUMN kind ENUM('setting', 'plan', 'preference', 'placement', 'notification', 'indicator_qualification', 'qualification_publication', 'cutoff_publication') NOT NULL");
|
|
await pool.execute('UPDATE schema_metadata SET schema_version = 20, app_version = 20 WHERE id = 1');
|
|
metadataRows[0].schema_version = 20;
|
|
}
|
|
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;
|
|
}
|