diff --git a/.gitignore b/.gitignore index f51373c..e2a5f74 100644 --- a/.gitignore +++ b/.gitignore @@ -61,6 +61,9 @@ chrome-profile/ # Database files db/ +!src/main/java/db/ +!src/main/java/db/migration/ +!src/main/java/db/migration/*.java !src/main/resources/db/ !src/main/resources/db/migration/ !src/main/resources/db/migration/*.sql diff --git "a/doc/\345\274\200\345\217\221\346\214\207\345\215\227.md" "b/doc/\345\274\200\345\217\221\346\214\207\345\215\227.md" index 2c9be5f..c4abcc7 100644 --- "a/doc/\345\274\200\345\217\221\346\214\207\345\215\227.md" +++ "b/doc/\345\274\200\345\217\221\346\214\207\345\215\227.md" @@ -193,7 +193,7 @@ MyBatis-Plus 会根据 mapper 和 entity 访问数据库。数据库文件包含 - `db/` 和 `*.db` 已被 `.gitignore` 忽略。 - 修改表结构前请备份本地数据库。 -- 前端有脚本 `front/scripts/migrate-sort-order.mjs` 用于特定字段迁移。 +- 历史脚本 `front/scripts/migrate-sort-order.mjs` 已安全禁用；它曾用 `sql.js` 整文件覆盖 SQLite，无法保护 WAL 中的并发写入。所有 Schema 变更必须通过后端 Flyway migration，并先在隔离副本演练。 ## 自动化执行层 diff --git a/front/scripts/migrate-sort-order.mjs b/front/scripts/migrate-sort-order.mjs index ad9af4a..3638b96 100644 --- a/front/scripts/migrate-sort-order.mjs +++ b/front/scripts/migrate-sort-order.mjs @@ -1,92 +1,8 @@ -// Migrate boss_option: add sort_order column and set city display order -// Usage: pnpm exec node scripts/migrate-sort-order.mjs - -import fs from 'fs'; -import path from 'path'; -import { fileURLToPath } from 'url'; -import initSqlJs from 'sql.js'; -import { createRequire } from 'module'; - -const log = (...args) => console.log('[migrate-sort-order]', ...args); - -async function main() { - try { - const __filename = fileURLToPath(import.meta.url); - const __dirname = path.dirname(__filename); - const projectRoot = path.resolve(__dirname, '..', '..'); - const dbPath = path.resolve(projectRoot, 'db', 'getjobs.db'); - - if (!fs.existsSync(dbPath)) { - throw new Error(`Database file not found: ${dbPath}`); - } - - const require = createRequire(import.meta.url); - const wasmDir = path.dirname(require.resolve('sql.js/dist/sql-wasm.wasm')); - const SQL = await initSqlJs({ locateFile: (file) => path.join(wasmDir, file) }); - - const fileBuffer = fs.readFileSync(dbPath); - const u8 = new Uint8Array(fileBuffer); - const db = new SQL.Database(u8); - - const hasColumn = (table, column) => { - const res = db.exec(`PRAGMA table_info(${table});`); - if (!res || res.length === 0) return false; - const names = res[0].values.map((row) => String(row[1]).toLowerCase()); - return names.includes(String(column).toLowerCase()); - }; - - // Ensure boss_option exists - const tables = db.exec("SELECT name FROM sqlite_master WHERE type='table' AND name='boss_option';"); - if (!tables || tables.length === 0 || tables[0].values.length === 0) { - throw new Error("Table 'boss_option' not found in database."); - } - - // Add sort_order column if missing - if (!hasColumn('boss_option', 'sort_order')) { - log('Adding column sort_order to boss_option ...'); - db.exec('ALTER TABLE boss_option ADD COLUMN sort_order INTEGER;'); - } else { - log('Column sort_order already exists.'); - } - - // Reset sort_order for cities - db.exec("UPDATE boss_option SET sort_order = NULL WHERE type='city';"); - - // Define preferred city display order (as requested) - const cityOrder = [ - '全国', '北京', '上海', '广州', '深圳', - '杭州', '天津', '西安', '苏州', '武汉', - '厦门', '长沙', '成都', '郑州', '重庆' - ]; - - const stmt = db.prepare("UPDATE boss_option SET sort_order = ? WHERE type='city' AND name = ?;"); - let updated = 0; - cityOrder.forEach((name, idx) => { - stmt.run([idx + 1, name]); - updated += 1; - }); - stmt.free(); - - log(`Applied sort_order to ${updated} city rows.`); - - // Persist changes back to file - const out = db.export(); - fs.writeFileSync(dbPath, Buffer.from(out)); - log('Database updated:', dbPath); - - // Optional: verify a few rows - const verify = db.exec("SELECT id, name, sort_order FROM boss_option WHERE type='city' AND sort_order IS NOT NULL ORDER BY sort_order ASC LIMIT 10;"); - if (verify && verify.length > 0) { - const rows = verify[0].values.map((r) => ({ id: r[0], name: r[1], sort_order: r[2] })); - log('Top cities after migration:', rows); - } - - db.close(); - log('Migration completed successfully.'); - } catch (err) { - console.error('[migrate-sort-order] Migration failed:', err); - process.exitCode = 1; - } -} - -await main(); \ No newline at end of file +// Historical migration entrypoint retained only to give old instructions a safe failure. +// Schema changes are managed by backend Flyway migrations. Never overwrite a live SQLite file with sql.js. + +console.error( + '[migrate-sort-order] 已禁用：数据库结构现在只允许由后端 Flyway 管理。' + + '请停止服务、备份数据库，并通过正式 migration/rehearsal 流程升级。' +) +process.exitCode = 1 diff --git a/src/main/java/com/getjobs/application/controller/BossAnalyticsController.java b/src/main/java/com/getjobs/application/controller/BossAnalyticsController.java index 89545b7..52fffc0 100644 --- a/src/main/java/com/getjobs/application/controller/BossAnalyticsController.java +++ b/src/main/java/com/getjobs/application/controller/BossAnalyticsController.java @@ -108,7 +108,7 @@ public BossService.PagedResult list( } /** - * 刷新 boss_data（列顺序检查 + VACUUM） + * 刷新 Boss 数据视图；只重新读取统计，不执行 Schema 或数据库维护操作。 */ @GetMapping("/reload") public Map reload() { diff --git a/src/main/java/com/getjobs/application/init/ZhilianOptionInitializer.java b/src/main/java/com/getjobs/application/init/ZhilianOptionInitializer.java index b3a70ce..78a2083 100644 --- a/src/main/java/com/getjobs/application/init/ZhilianOptionInitializer.java +++ b/src/main/java/com/getjobs/application/init/ZhilianOptionInitializer.java @@ -10,14 +10,13 @@ import org.springframework.boot.CommandLineRunner; import org.springframework.context.annotation.DependsOn; import org.springframework.stereotype.Component; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.support.TransactionTemplate; -import javax.sql.DataSource; import java.net.URI; import java.net.http.HttpClient; import java.net.http.HttpRequest; import java.net.http.HttpResponse; -import java.sql.Connection; -import java.sql.Statement; import java.time.Duration; import java.time.LocalDateTime; import java.util.ArrayList; @@ -37,13 +36,11 @@ public class ZhilianOptionInitializer implements CommandLineRunner { private static final ObjectMapper MAPPER = new ObjectMapper(); private static final Duration HTTP_TIMEOUT = Duration.ofSeconds(12); - private final DataSource dataSource; private final ZhilianOptionMapper zhilianOptionMapper; + private final PlatformTransactionManager transactionManager; @Override public void run(String... args) { - ensureTableExists(); - List options; try { options = loadOfficialOptions(); @@ -53,24 +50,15 @@ public void run(String... args) { log.warn("智联官方筛选项同步失败，使用内置兜底选项：{}", e.getMessage()); } - replaceCityAndSalaryOptions(options); + replaceCityAndSalaryOptionsAtomically(options); } - private void ensureTableExists() { - String ddl = "CREATE TABLE IF NOT EXISTS zhilian_option (" + - " id INTEGER PRIMARY KEY AUTOINCREMENT," + - " type VARCHAR(50)," + - " name VARCHAR(100)," + - " code VARCHAR(100)," + - " sort_order INTEGER," + - " created_at DATETIME," + - " updated_at DATETIME" + - ")"; - try (Connection conn = dataSource.getConnection(); Statement stmt = conn.createStatement()) { - stmt.execute(ddl); - log.info("确保 zhilian_option 表已存在"); - } catch (Exception e) { - log.warn("创建 zhilian_option 表失败: {}", e.getMessage()); + void replaceCityAndSalaryOptionsAtomically(List options) { + try { + new TransactionTemplate(transactionManager) + .executeWithoutResult(status -> replaceCityAndSalaryOptions(options)); + } catch (RuntimeException e) { + throw new IllegalStateException("刷新智联城市/薪资筛选项失败，已回滚", e); } } @@ -104,25 +92,21 @@ private List loadOfficialOptions() throws Exception { private void replaceCityAndSalaryOptions(List options) { LocalDateTime now = LocalDateTime.now(); - try { - zhilianOptionMapper.delete( - new QueryWrapper() - .in("type", List.of("city", "salary")) - ); - for (OptionSeed option : options) { - ZhilianOptionEntity entity = new ZhilianOptionEntity(); - entity.setType(option.type()); - entity.setName(option.name()); - entity.setCode(option.code()); - entity.setSortOrder(option.sortOrder()); - entity.setCreatedAt(now); - entity.setUpdatedAt(now); - zhilianOptionMapper.insert(entity); - } - log.info("智联城市/薪资筛选项刷新完成：{} 条", options.size()); - } catch (Exception e) { - log.warn("刷新智联城市/薪资筛选项失败: {}", e.getMessage()); + zhilianOptionMapper.delete( + new QueryWrapper() + .in("type", List.of("city", "salary")) + ); + for (OptionSeed option : options) { + ZhilianOptionEntity entity = new ZhilianOptionEntity(); + entity.setType(option.type()); + entity.setName(option.name()); + entity.setCode(option.code()); + entity.setSortOrder(option.sortOrder()); + entity.setCreatedAt(now); + entity.setUpdatedAt(now); + zhilianOptionMapper.insert(entity); } + log.info("智联城市/薪资筛选项刷新完成：{} 条", options.size()); } static List buildOptionsFromOfficialBaseData(JsonNode data) { diff --git a/src/main/java/com/getjobs/application/service/BossService.java b/src/main/java/com/getjobs/application/service/BossService.java index c2f9219..aa0e464 100644 --- a/src/main/java/com/getjobs/application/service/BossService.java +++ b/src/main/java/com/getjobs/application/service/BossService.java @@ -527,119 +527,6 @@ public List getAllBlacklist() { // ==================== boss_data（岗位数据）相关方法 ==================== - /** - * 确保 boss_data 表的列顺序以 encrypt_id、encrypt_user_id 开头。 - * 若不满足，则进行一次在线迁移：创建新表、复制数据、替换旧表。 - * 该迁移会重建 boss_data，比普通补列风险更高，暂时保留在业务刷新入口中按需执行。 - */ - public void ensureBossDataColumnOrder() { - java.sql.Connection conn = null; - try { - conn = dataSource.getConnection(); - try (java.sql.Statement stmt = conn.createStatement()) { - java.util.List cols = new java.util.ArrayList<>(); - try (java.sql.ResultSet rs = stmt.executeQuery("PRAGMA table_info('boss_data')")) { - while (rs.next()) { - cols.add(rs.getString("name")); - } - } - if (cols.isEmpty()) return; // 表不存在或无列 - boolean needMigrate = true; - if (cols.size() >= 3) { - String c0 = cols.get(0) == null ? "" : cols.get(0).toLowerCase(); - String c1 = cols.get(1) == null ? "" : cols.get(1).toLowerCase(); - String c2 = cols.get(2) == null ? "" : cols.get(2).toLowerCase(); - String c3 = cols.size() > 3 && cols.get(3) != null ? cols.get(3).toLowerCase() : ""; - // 允许第一列是 id 或 encrypt_id；档案模式下 profile_id 可以紧跟 id。 - if ("id".equals(c0) && "encrypt_id".equals(c1) && "encrypt_user_id".equals(c2)) { - needMigrate = false; - } else if ("id".equals(c0) && "profile_id".equals(c1) && "encrypt_id".equals(c2) && "encrypt_user_id".equals(c3)) { - needMigrate = false; - } else if ("encrypt_id".equals(c0) && "encrypt_user_id".equals(c1)) { - needMigrate = false; - } - } - if (!needMigrate) return; - - stmt.execute("BEGIN TRANSACTION"); - // 新表：将 encrypt_id、encrypt_user_id 移到最前（紧随 id） - String createSql = "CREATE TABLE boss_data_new (" + - "id INTEGER PRIMARY KEY AUTOINCREMENT, " + - "profile_id INTEGER, " + - "encrypt_id TEXT, " + - "encrypt_user_id TEXT, " + - "company_name TEXT, " + - "job_name TEXT, " + - "salary TEXT, " + - "salary_min_k REAL, " + - "salary_max_k REAL, " + - "salary_median_k REAL, " + - "salary_months INTEGER, " + - "location TEXT, " + - "experience TEXT, " + - "degree TEXT, " + - "hr_name TEXT, " + - "hr_position TEXT, " + - "hr_active_status TEXT, " + - "delivery_status TEXT, " + - "failure_type TEXT, " + - "failure_reason TEXT, " + - "job_description TEXT, " + - "job_url TEXT, " + - "recruitment_status TEXT, " + - "company_address TEXT, " + - "industry TEXT, " + - "introduce TEXT, " + - "financing_stage TEXT, " + - "company_scale TEXT, " + - "source_keyword TEXT, " + - "scan_run_id TEXT, " + - "ai_score INTEGER, " + - "ai_decision TEXT, " + - "ai_reason TEXT, " + - "priority_company INTEGER DEFAULT 0, " + - "created_at TEXT, " + - "updated_at TEXT" + - ")"; - stmt.execute(createSql); - - String copySql = "INSERT INTO boss_data_new (" + - "id, profile_id, encrypt_id, encrypt_user_id, company_name, job_name, salary, salary_min_k, salary_max_k, salary_median_k, salary_months, location, experience, degree, " + - "hr_name, hr_position, hr_active_status, delivery_status, failure_type, failure_reason, job_description, job_url, recruitment_status, " + - "company_address, industry, introduce, financing_stage, company_scale, source_keyword, scan_run_id, ai_score, ai_decision, ai_reason, priority_company, created_at, updated_at" + - ") SELECT " + - "id, " + (cols.contains("profile_id") ? "profile_id" : "NULL") + ", encrypt_id, encrypt_user_id, company_name, job_name, salary, " + - (cols.contains("salary_min_k") ? "salary_min_k" : "NULL") + ", " + - (cols.contains("salary_max_k") ? "salary_max_k" : "NULL") + ", " + - (cols.contains("salary_median_k") ? "salary_median_k" : "NULL") + ", " + - (cols.contains("salary_months") ? "salary_months" : "NULL") + ", " + - "location, experience, degree, " + - "hr_name, hr_position, hr_active_status, delivery_status, " + - (cols.contains("failure_type") ? "failure_type" : "NULL") + ", " + - (cols.contains("failure_reason") ? "failure_reason" : "NULL") + ", job_description, job_url, recruitment_status, " + - "company_address, industry, introduce, financing_stage, company_scale, " + - (cols.contains("source_keyword") ? "source_keyword" : "NULL") + ", " + - (cols.contains("scan_run_id") ? "scan_run_id" : "NULL") + ", " + - (cols.contains("ai_score") ? "ai_score" : "NULL") + ", " + - (cols.contains("ai_decision") ? "ai_decision" : "NULL") + ", " + - (cols.contains("ai_reason") ? "ai_reason" : "NULL") + ", " + - (cols.contains("priority_company") ? "priority_company" : "0") + ", created_at, updated_at " + - "FROM boss_data"; - stmt.execute(copySql); - - stmt.execute("DROP TABLE boss_data"); - stmt.execute("ALTER TABLE boss_data_new RENAME TO boss_data"); - stmt.execute("COMMIT"); - log.info("已调整 boss_data 表列顺序：将 encrypt_id、encrypt_user_id 前置"); - } - } catch (Exception e) { - log.warn("调整 boss_data 列顺序失败：{}", e.getMessage()); - try { if (conn != null) conn.createStatement().execute("ROLLBACK"); } catch (Exception ignore) {} - } finally { - try { if (conn != null) conn.close(); } catch (Exception ignore) {} - } - } - /** * 判断岗位是否已存在（相同 encrypt_id AND encrypt_user_id） */ @@ -1750,18 +1637,13 @@ public String normalizeExplicitBossScanRunId(String scanRunId) { } /** - * 刷新数据：执行列顺序检查，并执行 VACUUM 以优化数据库；返回当前总数 + * 刷新数据视图并返回当前档案总数。该入口不再执行任何数据库维护或 DDL。 */ public Map reloadBossData() { Map resp = new HashMap<>(); Connection conn = null; try { - ensureBossDataColumnOrder(); conn = dataSource.getConnection(); - try (Statement st = conn.createStatement()) { - try { st.execute("PRAGMA wal_checkpoint(TRUNCATE)"); } catch (Exception ignore) {} - try { st.execute("VACUUM"); } catch (Exception ignore) {} - } Long profileId = profileService.getCurrentProfileIdOrNull(); long total = profileId == null ? 0 : scalarCount(conn, "SELECT COUNT(*) FROM boss_data WHERE profile_id=" + profileId); resp.put("success", true); diff --git a/src/main/java/com/getjobs/application/service/DatabaseSchemaService.java b/src/main/java/com/getjobs/application/service/DatabaseSchemaService.java index 1511df4..7a02c4f 100644 --- a/src/main/java/com/getjobs/application/service/DatabaseSchemaService.java +++ b/src/main/java/com/getjobs/application/service/DatabaseSchemaService.java @@ -4,16 +4,23 @@ import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import com.getjobs.application.config.RuntimeDirectoryInitializer; +import org.springframework.context.annotation.DependsOn; import org.springframework.stereotype.Service; import javax.sql.DataSource; import java.sql.Connection; import java.sql.ResultSet; +import java.sql.SQLException; import java.sql.Statement; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; @Slf4j @Service @RequiredArgsConstructor +@DependsOn("flywayInitializer") public class DatabaseSchemaService { private final DataSource dataSource; private final RuntimeDirectoryInitializer runtimeDirectoryInitializer; @@ -21,24 +28,35 @@ public class DatabaseSchemaService { @PostConstruct public void initializeSchema() { runtimeDirectoryInitializer.ensureRuntimeDirectories(); - try (Connection conn = dataSource.getConnection(); Statement stmt = conn.createStatement()) { + try (Connection conn = dataSource.getConnection()) { + validateSchema(conn); + log.info("数据库 schema 校验完成"); + } catch (Exception e) { + throw new IllegalStateException("数据库 schema 不完整，已阻止应用继续启动: " + e.getMessage(), e); + } + } + + /** + * 只供 Flyway V5 调用的一次性旧库兼容迁移。运行期不得调用。 + */ + public static void migrateLegacySchema(Connection conn) throws Exception { + try (Statement stmt = conn.createStatement()) { ensureCoreTables(stmt); ensureProfileColumns(stmt); ensureAiColumns(stmt); ensurePlatformConfigColumns(stmt); ensurePlatformDataColumns(stmt); + addColumn(stmt, "liepin_data", "delivered", "INTEGER DEFAULT 0"); + addColumn(stmt, "job51_data", "delivered", "INTEGER DEFAULT 0"); backfillProfileIds(stmt); ensurePriorityCompanySchema(stmt); backfillProfileIds(stmt); normalizeActiveProfile(stmt); ensureIndexes(stmt); - log.info("数据库 schema 初始化完成"); - } catch (Exception e) { - log.warn("数据库 schema 初始化失败: {}", e.getMessage()); } } - private void ensureCoreTables(Statement stmt) throws Exception { + private static void ensureCoreTables(Statement stmt) throws Exception { createTableIfNotExists(stmt, "profile", "id INTEGER PRIMARY KEY AUTOINCREMENT, " + "name TEXT NOT NULL, " + @@ -201,7 +219,7 @@ private void ensureCoreTables(Statement stmt) throws Exception { "update_time DATETIME"); } - private void ensureProfileColumns(Statement stmt) { + private static void ensureProfileColumns(Statement stmt) { addColumn(stmt, "profile", "is_active", "INTEGER DEFAULT 0"); addColumn(stmt, "profile", "created_at", "DATETIME"); addColumn(stmt, "profile", "updated_at", "DATETIME"); @@ -230,7 +248,7 @@ private void ensureProfileColumns(Statement stmt) { addProfileColumn(stmt, "zhilian_data"); } - private void ensureAiColumns(Statement stmt) { + private static void ensureAiColumns(Statement stmt) { addColumn(stmt, "ai", "profile_id", "INTEGER"); addColumn(stmt, "ai", "introduce", "TEXT"); addColumn(stmt, "ai", "prompt", "TEXT"); @@ -271,7 +289,7 @@ private void ensureAiColumns(Statement stmt) { addColumn(stmt, "priority_company", "updated_at", "DATETIME"); } - private void ensurePlatformConfigColumns(Statement stmt) { + private static void ensurePlatformConfigColumns(Statement stmt) { addColumn(stmt, "boss_config", "profile_id", "INTEGER"); addColumn(stmt, "boss_config", "debugger", "INTEGER DEFAULT 0"); addColumn(stmt, "boss_config", "wait_time", "INTEGER DEFAULT 10"); @@ -305,7 +323,7 @@ private void ensurePlatformConfigColumns(Statement stmt) { addColumn(stmt, "zhilian_config", "updated_at", "DATETIME"); } - private void ensurePlatformDataColumns(Statement stmt) { + private static void ensurePlatformDataColumns(Statement stmt) { addColumn(stmt, "boss_data", "profile_id", "INTEGER"); addColumn(stmt, "boss_data", "encrypt_id", "TEXT"); addColumn(stmt, "boss_data", "encrypt_user_id", "TEXT"); @@ -364,7 +382,7 @@ private void ensurePlatformDataColumns(Statement stmt) { addColumn(stmt, "zhilian_data", "update_time", "DATETIME"); } - private void backfillProfileIds(Statement stmt) { + private static void backfillProfileIds(Statement stmt) { Long profileId = findCurrentProfileId(stmt); if (profileId == null) { return; @@ -379,25 +397,25 @@ private void backfillProfileIds(Statement stmt) { backfillProfileId(stmt, "zhilian_data", profileId); } - public void createTableIfNotExists(Statement stmt, String table, String columnsSql) throws Exception { + private static void createTableIfNotExists(Statement stmt, String table, String columnsSql) throws Exception { stmt.execute("CREATE TABLE IF NOT EXISTS " + table + " (" + columnsSql + ")"); } - public void addProfileColumn(Statement stmt, String table) { + private static void addProfileColumn(Statement stmt, String table) { addColumn(stmt, table, "profile_id", "INTEGER"); } - public void addColumn(Statement stmt, String table, String column, String type) { + private static void addColumn(Statement stmt, String table, String column, String type) { try { if (tableExists(stmt, table) && !columnExists(stmt, table, column)) { stmt.execute("ALTER TABLE " + table + " ADD COLUMN " + column + " " + type); } } catch (Exception e) { - log.debug("补列失败 {}.{}: {}", table, column, e.getMessage()); + throw new IllegalStateException("补列失败 " + table + "." + column + ": " + e.getMessage(), e); } } - private void ensurePriorityCompanySchema(Statement stmt) throws Exception { + private static void ensurePriorityCompanySchema(Statement stmt) throws Exception { createTableIfNotExists(stmt, "priority_company", "id INTEGER PRIMARY KEY AUTOINCREMENT, " + "profile_id INTEGER, " + @@ -417,6 +435,7 @@ private void ensurePriorityCompanySchema(Statement stmt) throws Exception { return; } + stmt.execute("DROP TABLE IF EXISTS priority_company_profile_new"); createTableIfNotExists(stmt, "priority_company_profile_new", "id INTEGER PRIMARY KEY AUTOINCREMENT, " + "profile_id INTEGER, " + @@ -427,25 +446,30 @@ private void ensurePriorityCompanySchema(Statement stmt) throws Exception { "updated_at DATETIME, " + "UNIQUE(profile_id, company_name)"); String profileExpr = hasProfileId ? "profile_id" : "NULL"; - stmt.executeUpdate("INSERT OR IGNORE INTO priority_company_profile_new " + + long sourceCount = scalarCount(stmt, "priority_company"); + stmt.executeUpdate("INSERT INTO priority_company_profile_new " + "(id, profile_id, company_name, enabled, remark, created_at, updated_at) " + "SELECT id, " + profileExpr + ", company_name, enabled, remark, created_at, updated_at " + "FROM priority_company"); + long copiedCount = scalarCount(stmt, "priority_company_profile_new"); + if (sourceCount != copiedCount) { + throw new IllegalStateException("priority_company 重建行数不一致: " + sourceCount + " -> " + copiedCount); + } stmt.execute("DROP TABLE priority_company"); stmt.execute("ALTER TABLE priority_company_profile_new RENAME TO priority_company"); createPriorityCompanyUniqueIndex(stmt); } - private void createPriorityCompanyUniqueIndex(Statement stmt) { + private static void createPriorityCompanyUniqueIndex(Statement stmt) { try { stmt.execute("CREATE UNIQUE INDEX IF NOT EXISTS idx_priority_company_profile_name " + "ON priority_company(profile_id, company_name)"); } catch (Exception e) { - log.debug("创建重点公司唯一索引失败: {}", e.getMessage()); + throw new IllegalStateException("创建重点公司唯一索引失败: " + e.getMessage(), e); } } - private void ensureIndexes(Statement stmt) { + private static void ensureIndexes(Statement stmt) { createIndexIfNotExists(stmt, "idx_boss_data_profile_run_encrypt", "boss_data", @@ -468,29 +492,44 @@ private void ensureIndexes(Statement stmt) { "profile_id, platform, job_key, scan_run_id"); } - private void createIndexIfNotExists(Statement stmt, String indexName, String table, String columnsSql) { + private static void createIndexIfNotExists(Statement stmt, String indexName, String table, String columnsSql) { try { if (tableExists(stmt, table)) { stmt.execute("CREATE INDEX IF NOT EXISTS " + indexName + " ON " + table + "(" + columnsSql + ")"); } } catch (Exception e) { - log.debug("创建索引失败 {}.{}: {}", table, indexName, e.getMessage()); + throw new IllegalStateException("创建索引失败 " + table + "." + indexName + ": " + e.getMessage(), e); } } - private boolean hasGlobalCompanyUnique(Statement stmt) { - try (ResultSet rs = stmt.executeQuery("SELECT sql FROM sqlite_master WHERE type='table' AND name='priority_company'")) { - if (rs.next()) { - String sql = rs.getString("sql"); - return sql != null && sql.toUpperCase().contains("COMPANY_NAME TEXT NOT NULL UNIQUE"); + private static boolean hasGlobalCompanyUnique(Statement stmt) { + try (Statement indexStatement = stmt.getConnection().createStatement(); + ResultSet indexes = indexStatement.executeQuery("PRAGMA index_list('priority_company')")) { + while (indexes.next()) { + if (indexes.getInt("unique") != 1) { + continue; + } + String indexName = indexes.getString("name"); + try (Statement columnStatement = stmt.getConnection().createStatement(); + ResultSet columns = columnStatement.executeQuery("PRAGMA index_info('" + indexName.replace("'", "''") + "')")) { + int count = 0; + boolean companyNameOnly = true; + while (columns.next()) { + count++; + companyNameOnly &= "company_name".equalsIgnoreCase(columns.getString("name")); + } + if (count == 1 && companyNameOnly) { + return true; + } + } } } catch (Exception e) { - log.debug("检查 priority_company 唯一约束失败: {}", e.getMessage()); + throw new IllegalStateException("检查 priority_company 唯一约束失败: " + e.getMessage(), e); } return false; } - private void normalizeActiveProfile(Statement stmt) { + private static void normalizeActiveProfile(Statement stmt) { try { Long activeId = null; try (ResultSet rs = stmt.executeQuery("SELECT id FROM profile WHERE is_active = 1 ORDER BY id ASC LIMIT 1")) { @@ -509,45 +548,123 @@ private void normalizeActiveProfile(Statement stmt) { stmt.executeUpdate("UPDATE profile SET is_active = CASE WHEN id = " + activeId + " THEN 1 ELSE 0 END"); } } catch (Exception e) { - log.warn("规范化当前档案失败: {}", e.getMessage()); + throw new IllegalStateException("规范化当前档案失败: " + e.getMessage(), e); } } - private Long findCurrentProfileId(Statement stmt) { + private static Long findCurrentProfileId(Statement stmt) { try (ResultSet rs = stmt.executeQuery("SELECT id FROM profile WHERE is_active = 1 ORDER BY id ASC LIMIT 1")) { if (rs.next()) { return rs.getLong("id"); } } catch (Exception e) { - log.debug("查询当前档案失败: {}", e.getMessage()); + throw new IllegalStateException("查询当前档案失败: " + e.getMessage(), e); } try (ResultSet rs = stmt.executeQuery("SELECT id FROM profile ORDER BY id ASC LIMIT 1")) { if (rs.next()) { return rs.getLong("id"); } } catch (Exception e) { - log.debug("查询首个档案失败: {}", e.getMessage()); + throw new IllegalStateException("查询首个档案失败: " + e.getMessage(), e); } return null; } - private void backfillProfileId(Statement stmt, String table, Long profileId) { + private static void backfillProfileId(Statement stmt, String table, Long profileId) { try { if (tableExists(stmt, table) && columnExists(stmt, table, "profile_id")) { stmt.executeUpdate("UPDATE " + table + " SET profile_id = " + profileId + " WHERE profile_id IS NULL"); } } catch (Exception e) { - log.debug("回填 {}.profile_id 失败: {}", table, e.getMessage()); + throw new IllegalStateException("回填 " + table + ".profile_id 失败: " + e.getMessage(), e); + } + } + + private static long scalarCount(Statement stmt, String table) throws SQLException { + try (ResultSet rs = stmt.executeQuery("SELECT COUNT(*) FROM " + table)) { + return rs.next() ? rs.getLong(1) : 0L; + } + } + + public static void validateSchema(Connection conn) throws SQLException { + List requiredTables = List.of( + "profile", "config", "cookie", "ai", "resume_profile", "priority_company", + "job_ai_analysis", "job_analysis_task", "boss_config", "boss_data", + "boss_blacklist", "boss_option", "boss_industry", "zhilian_config", + "zhilian_data", "zhilian_option", "liepin_config", "liepin_data", + "liepin_option", "job51_config", "job51_data", "job51_option" + ); + Map> requiredColumns = new LinkedHashMap<>(); + requiredColumns.put("profile", Set.of("id", "is_active")); + requiredColumns.put("config", Set.of("config_key", "config_value")); + requiredColumns.put("cookie", Set.of("platform", "cookie_value")); + requiredColumns.put("ai", Set.of("profile_id", "apply_threshold", "priority_apply_threshold")); + requiredColumns.put("priority_company", Set.of("profile_id", "company_name")); + requiredColumns.put("boss_data", Set.of( + "profile_id", "encrypt_id", "encrypt_user_id", "delivery_status", "failure_type", + "failure_reason", "scan_run_id", "source_keyword", "salary_min_k", "salary_max_k", + "salary_median_k", "salary_months" + )); + requiredColumns.put("zhilian_data", Set.of("profile_id", "job_id", "delivery_status", "scan_run_id")); + requiredColumns.put("liepin_data", Set.of("job_id", "delivered")); + requiredColumns.put("job51_data", Set.of("job_id", "delivered")); + requiredColumns.put("job_analysis_task", Set.of("profile_id", "platform", "status", "scan_run_id")); + List requiredIndexes = List.of( + "idx_priority_company_profile_name", + "idx_boss_blacklist_type_value", + "idx_boss_option_type_code", + "idx_boss_industry_code", + "idx_boss_data_profile_run_encrypt", + "idx_boss_data_profile_delivery_status", + "idx_boss_data_profile_created_at", + "idx_boss_data_profile_company_job", + "idx_job_ai_analysis_profile_platform_job_run", + "idx_zhilian_data_profile_scan_run", + "idx_liepin_data_company_job", + "idx_job51_data_company_job", + "idx_job_analysis_task_profile_platform_status", + "idx_job_analysis_task_scan_run" + ); + + try (Statement stmt = conn.createStatement()) { + for (String table : requiredTables) { + if (!tableExists(stmt, table)) { + throw new SQLException("缺少必要数据表: " + table); + } + } + for (Map.Entry> entry : requiredColumns.entrySet()) { + for (String column : entry.getValue()) { + if (!columnExists(stmt, entry.getKey(), column)) { + throw new SQLException("缺少必要字段: " + entry.getKey() + "." + column); + } + } + } + for (String index : requiredIndexes) { + if (!indexExists(stmt, index)) { + throw new SQLException("缺少必要索引: " + index); + } + } + } catch (SQLException e) { + throw e; + } catch (Exception e) { + throw new SQLException("Schema 校验失败: " + e.getMessage(), e); } } - private boolean tableExists(Statement stmt, String table) throws Exception { + private static boolean tableExists(Statement stmt, String table) throws Exception { try (ResultSet rs = stmt.executeQuery("SELECT 1 FROM sqlite_master WHERE type='table' AND name='" + table + "' LIMIT 1")) { return rs.next(); } } - private boolean columnExists(Statement stmt, String table, String column) throws Exception { + private static boolean indexExists(Statement stmt, String index) throws Exception { + try (ResultSet rs = stmt.executeQuery( + "SELECT 1 FROM sqlite_master WHERE type='index' AND name='" + index + "' LIMIT 1")) { + return rs.next(); + } + } + + private static boolean columnExists(Statement stmt, String table, String column) throws Exception { try (ResultSet rs = stmt.executeQuery("PRAGMA table_info('" + table + "')")) { while (rs.next()) { if (column.equalsIgnoreCase(rs.getString("name"))) { diff --git a/src/main/java/com/getjobs/application/service/Job51Service.java b/src/main/java/com/getjobs/application/service/Job51Service.java index 7762429..7f207c8 100644 --- a/src/main/java/com/getjobs/application/service/Job51Service.java +++ b/src/main/java/com/getjobs/application/service/Job51Service.java @@ -8,7 +8,6 @@ import com.getjobs.application.mapper.Job51Mapper; import com.getjobs.application.mapper.Job51OptionMapper; import com.getjobs.worker.job51.Job51Config; -import jakarta.annotation.PostConstruct; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; @@ -160,48 +159,7 @@ public String normalizeOptionCode(String type, String input) { return v; } - // ==================== 表初始化与数据导入 ==================== - - @PostConstruct - public void ensureJob51OptionTableAndData() { - // 仅确保表存在；城市选项完全由数据库维护 - ensureJob51OptionTable(); - ensureJob51DataTable(); - } - - private void ensureJob51OptionTable() { - String createSql = "CREATE TABLE IF NOT EXISTS job51_option (" + - " id INTEGER PRIMARY KEY AUTOINCREMENT," + - " type VARCHAR(50)," + - " name VARCHAR(100)," + - " code VARCHAR(100)," + - " sort_order INTEGER," + - " created_at DATETIME," + - " updated_at DATETIME" + - ")"; - try (Connection conn = dataSource.getConnection(); Statement stmt = conn.createStatement()) { - stmt.execute(createSql); - } catch (Exception e) { - log.warn("创建 job51_option 表失败: {}", e.getMessage()); - } - } - - // 初始化逻辑移除：数据由外部迁移并在数据库维护，无需自动填充 - - private void insertOption(String type, String name, String code, int sortOrder, LocalDateTime now) { - try { - Job51OptionEntity e = new Job51OptionEntity(); - e.setType(type); - e.setName(name); - e.setCode(code); - e.setSortOrder(sortOrder); - e.setCreatedAt(now); - e.setUpdatedAt(now); - job51OptionMapper.insert(e); - } catch (Exception ex) { - log.warn("写入选项失败 type={} name={} code={}: {}", type, name, code, ex.getMessage()); - } - } + // 表结构只由 Flyway 管理；Service 运行时不再执行 DDL。 /** * 选择性更新：若传入 ID 则按 ID 更新；否则更新第一条记录（不存在则插入） @@ -248,40 +206,6 @@ public Job51ConfigEntity saveOrUpdateFirstSelective(Job51ConfigEntity incoming) // ==================== 51job 岗位数据表与持久化 ==================== - /** 创建 job51_data 表（如不存在） */ - private void ensureJob51DataTable() { - String createSql = "CREATE TABLE IF NOT EXISTS job51_data (" + - " job_id BIGINT PRIMARY KEY," + - " job_title VARCHAR(200)," + - " job_link VARCHAR(300)," + - " job_salary_text VARCHAR(100)," + - " job_area VARCHAR(100)," + - " job_edu_req VARCHAR(50)," + - " job_exp_req VARCHAR(50)," + - " job_publish_time VARCHAR(50)," + - " comp_id BIGINT," + - " comp_name VARCHAR(200)," + - " comp_industry VARCHAR(100)," + - " comp_scale VARCHAR(50)," + - " hr_id VARCHAR(64)," + - " hr_name VARCHAR(50)," + - " hr_title VARCHAR(100)," + - " delivered INTEGER DEFAULT 0," + - " create_time TEXT," + - " update_time TEXT" + - ")"; - try (Connection conn = dataSource.getConnection(); Statement stmt = conn.createStatement()) { - stmt.execute(createSql); - // 兼容旧库：添加 delivered 列（已存在则忽略） - try { stmt.execute("ALTER TABLE job51_data ADD COLUMN delivered INTEGER DEFAULT 0"); } catch (Exception ignored) {} - // 兼容旧库：尝试移除 account_id（已不存在或不支持则忽略） - try { stmt.execute("ALTER TABLE job51_data DROP COLUMN account_id"); } catch (Exception ignored) {} - log.info("确保 job51_data 表已存在"); - } catch (Exception e) { - log.warn("创建 job51_data 表失败: {}", e.getMessage()); - } - } - /** 批量插入（仅不存在时），默认 delivered=0 */ public void batchInsertIfNotExists(List entities) { if (entities == null || entities.isEmpty()) return; @@ -779,16 +703,12 @@ else if (s.contains("元/天")) { private String nullSafe(String s) { return (s == null || s.isEmpty()) ? "未知" : s; } - /** 刷新 51job 数据：执行 VACUUM 并返回当前总数 */ + /** 刷新 51job 数据视图并返回当前总数；不执行数据库维护或 DDL。 */ public java.util.Map reloadJob51Data() { java.util.Map resp = new java.util.HashMap<>(); Connection conn = null; try { conn = dataSource.getConnection(); - try (Statement st = conn.createStatement()) { - try { st.execute("PRAGMA wal_checkpoint(TRUNCATE)"); } catch (Exception ignore) {} - try { st.execute("VACUUM"); } catch (Exception ignore) {} - } long total = scalarCount(conn, "SELECT COUNT(*) FROM job51_data"); resp.put("success", true); resp.put("message", "刷新完成"); @@ -805,4 +725,4 @@ private long scalarCount(Connection conn, String sql) throws Exception { return rs.next() ? rs.getLong(1) : 0L; } } -} \ No newline at end of file +} diff --git a/src/main/java/com/getjobs/application/service/LiepinService.java b/src/main/java/com/getjobs/application/service/LiepinService.java index 1fbd595..14ad59a 100644 --- a/src/main/java/com/getjobs/application/service/LiepinService.java +++ b/src/main/java/com/getjobs/application/service/LiepinService.java @@ -13,10 +13,8 @@ import java.time.LocalDateTime; import java.util.*; -import jakarta.annotation.PostConstruct; import javax.sql.DataSource; import java.sql.Connection; -import java.sql.Statement; import java.sql.PreparedStatement; import java.sql.Timestamp; import java.sql.Types; @@ -37,47 +35,7 @@ public class LiepinService { private final LiepinMapper liepinMapper; private final DataSource dataSource; - // ==================== 记录表初始化与快照保存 ==================== - - @PostConstruct - public void ensureTableExists() { - String createSql = "CREATE TABLE IF NOT EXISTS liepin_data (" + - " job_id BIGINT PRIMARY KEY," + - " job_title VARCHAR(200)," + - " job_link VARCHAR(300)," + - " job_salary_text VARCHAR(100)," + - " job_area VARCHAR(100)," + - " job_edu_req VARCHAR(50)," + - " job_exp_req VARCHAR(50)," + - " job_publish_time VARCHAR(50)," + - " comp_id BIGINT," + - " comp_name VARCHAR(200)," + - " comp_industry VARCHAR(100)," + - " comp_scale VARCHAR(50)," + - " hr_id VARCHAR(64)," + - " hr_name VARCHAR(50)," + - " hr_title VARCHAR(100)," + - " hr_im_id VARCHAR(64)," + - " delivered INTEGER DEFAULT 0," + - " create_time DATETIME," + - " update_time DATETIME" + - ")"; - try (Connection conn = dataSource.getConnection(); Statement stmt = conn.createStatement()) { - stmt.execute(createSql); - // 兼容旧库：尝试添加 delivered 列（如已存在则忽略错误） - try { - stmt.execute("ALTER TABLE liepin_data ADD COLUMN delivered INTEGER DEFAULT 0"); - } catch (Exception ignored) {} - // 兼容旧库：尝试移除无数据列（SQLite 3.35+ 支持；不支持则忽略错误） - try { stmt.execute("ALTER TABLE liepin_data DROP COLUMN job_function"); } catch (Exception ignored) {} - try { stmt.execute("ALTER TABLE liepin_data DROP COLUMN job_city"); } catch (Exception ignored) {} - try { stmt.execute("ALTER TABLE liepin_data DROP COLUMN comp_full_name"); } catch (Exception ignored) {} - try { stmt.execute("ALTER TABLE liepin_data DROP COLUMN comp_kind"); } catch (Exception ignored) {} - log.info("确保 liepin_data 表已存在"); - } catch (Exception e) { - log.warn("创建 liepin_data 表失败: {}", e.getMessage()); - } - } + // 表结构只由 Flyway 管理；Service 运行时不再执行 DDL。 /** * 保存或更新一条岗位快照（以 job_id 作为主键） diff --git a/src/main/java/com/getjobs/worker/boss/Boss.java b/src/main/java/com/getjobs/worker/boss/Boss.java index 3ef2ff6..7d453f9 100644 --- a/src/main/java/com/getjobs/worker/boss/Boss.java +++ b/src/main/java/com/getjobs/worker/boss/Boss.java @@ -71,8 +71,6 @@ public interface ProgressCallback { // 通过 Lombok @RequiredArgsConstructor 使用构造器注入 bossService 与 aiService public void prepare() { - // 调整 boss_data 表结构：将 encrypt_id、encrypt_user_id 前置 - try { bossService.ensureBossDataColumnOrder(); } catch (Throwable ignore) {} // 从数据库加载黑名单 this.blackCompanies = bossService.getBlackCompanies(); this.blackRecruiters = bossService.getBlackRecruiters(); diff --git a/src/main/java/db/migration/V5__consolidate_legacy_schema.java b/src/main/java/db/migration/V5__consolidate_legacy_schema.java new file mode 100644 index 0000000..14419aa --- /dev/null +++ b/src/main/java/db/migration/V5__consolidate_legacy_schema.java @@ -0,0 +1,49 @@ +package db.migration; + +import com.getjobs.application.service.DatabaseSchemaService; +import org.flywaydb.core.api.migration.BaseJavaMigration; +import org.flywaydb.core.api.migration.Context; + +import java.io.IOException; +import java.io.InputStream; +import java.nio.charset.StandardCharsets; +import java.sql.Statement; + +/** + * 将历史运行时补表/补列逻辑收敛到一次性 Flyway 迁移。 + * + *

旧的非空数据库可能被 baseline 为 v4 而没有实际执行 V1-V4。本迁移会先用 + * CREATE IF NOT EXISTS 补齐历史表，再补运行时漂移列和全部索引。

+ */ +public class V5__consolidate_legacy_schema extends BaseJavaMigration { + @Override + public void migrate(Context context) throws Exception { + executeResource(context, "db/migration/V1__init_schema.sql"); + executeResource(context, "db/migration/V4__add_job_analysis_task.sql"); + DatabaseSchemaService.migrateLegacySchema(context.getConnection()); + executeResource(context, "db/migration/V2__add_indexes.sql"); + DatabaseSchemaService.validateSchema(context.getConnection()); + } + + private void executeResource(Context context, String resourcePath) throws Exception { + String script = readResource(resourcePath); + try (Statement statement = context.getConnection().createStatement()) { + for (String sql : script.split(";")) { + String trimmed = sql.trim(); + if (!trimmed.isEmpty()) { + statement.execute(trimmed); + } + } + } + } + + private String readResource(String resourcePath) throws IOException { + ClassLoader classLoader = Thread.currentThread().getContextClassLoader(); + try (InputStream input = classLoader.getResourceAsStream(resourcePath)) { + if (input == null) { + throw new IOException("找不到迁移资源: " + resourcePath); + } + return new String(input.readAllBytes(), StandardCharsets.UTF_8); + } + } +} diff --git a/src/test/java/com/getjobs/application/init/ZhilianOptionInitializerTest.java b/src/test/java/com/getjobs/application/init/ZhilianOptionInitializerTest.java index 633ee23..c6536d9 100644 --- a/src/test/java/com/getjobs/application/init/ZhilianOptionInitializerTest.java +++ b/src/test/java/com/getjobs/application/init/ZhilianOptionInitializerTest.java @@ -2,12 +2,22 @@ import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; +import com.getjobs.application.entity.ZhilianOptionEntity; +import com.getjobs.application.mapper.ZhilianOptionMapper; import org.junit.jupiter.api.Test; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.TransactionStatus; import java.util.List; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.assertj.core.api.Assertions.tuple; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; class ZhilianOptionInitializerTest { private static final ObjectMapper MAPPER = new ObjectMapper(); @@ -79,4 +89,23 @@ void fallbackOptionsContainOfficialSalaryCodes() { tuple("salary", "50K\u4ee5\u4e0a", "50001,9999999") ); } + + @Test + void failedOptionReplacementRollsBackInsteadOfLeavingPartialRows() { + ZhilianOptionMapper mapper = mock(ZhilianOptionMapper.class); + PlatformTransactionManager transactionManager = mock(PlatformTransactionManager.class); + TransactionStatus transactionStatus = mock(TransactionStatus.class); + when(transactionManager.getTransaction(any())).thenReturn(transactionStatus); + when(mapper.insert(any(ZhilianOptionEntity.class))).thenThrow(new IllegalStateException("insert failed")); + ZhilianOptionInitializer initializer = new ZhilianOptionInitializer(mapper, transactionManager); + + assertThatThrownBy(() -> initializer.replaceCityAndSalaryOptionsAtomically( + ZhilianOptionInitializer.fallbackOptions())) + .isInstanceOf(IllegalStateException.class) + .hasMessage("刷新智联城市/薪资筛选项失败，已回滚") + .hasRootCauseMessage("insert failed"); + + verify(transactionManager).rollback(transactionStatus); + verify(transactionManager, never()).commit(transactionStatus); + } } diff --git a/src/test/java/com/getjobs/application/service/DatabaseMigrationRehearsalTest.java b/src/test/java/com/getjobs/application/service/DatabaseMigrationRehearsalTest.java new file mode 100644 index 0000000..47166b9 --- /dev/null +++ b/src/test/java/com/getjobs/application/service/DatabaseMigrationRehearsalTest.java @@ -0,0 +1,113 @@ +package com.getjobs.application.service; + +import org.flywaydb.core.Flyway; +import org.junit.jupiter.api.Assumptions; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.attribute.FileTime; +import java.security.MessageDigest; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.ResultSet; +import java.sql.Statement; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +class DatabaseMigrationRehearsalTest { + private static final List COUNTED_TABLES = List.of( + "profile", "config", "cookie", "ai", "resume_profile", "priority_company", + "job_ai_analysis", "job_analysis_task", "boss_data", "zhilian_data", "liepin_data", "job51_data" + ); + + @TempDir + Path tempDir; + + @Test + void migratesIsolatedCopyWithoutChangingSourceDatabase() throws Exception { + String configuredPath = System.getenv("P0_REHEARSAL_DB"); + Assumptions.assumeTrue(configuredPath != null && !configuredPath.isBlank(), + "设置 P0_REHEARSAL_DB 后才执行真实数据库副本演练"); + + Path source = Path.of(configuredPath).toAbsolutePath().normalize(); + assertThat(source).isRegularFile(); + assertNoActiveSidecar(source); + byte[] sourceHashBefore = sha256(source); + FileTime sourceMtimeBefore = Files.getLastModifiedTime(source); + + Path rehearsal = tempDir.resolve("rehearsal.db"); + Files.copy(source, rehearsal); + String rehearsalUrl = "jdbc:sqlite:" + rehearsal; + Map countsBefore = tableCounts(rehearsalUrl); + + Flyway.configure() + .dataSource(rehearsalUrl, null, null) + .locations("classpath:db/migration") + .baselineOnMigrate(true) + .baselineVersion("4") + .validateOnMigrate(true) + .load() + .migrate(); + + try (Connection connection = DriverManager.getConnection(rehearsalUrl)) { + DatabaseSchemaService.validateSchema(connection); + assertThat(scalarText(connection, "PRAGMA integrity_check")).isEqualTo("ok"); + assertThat(scalarLong(connection, + "SELECT COUNT(*) FROM flyway_schema_history WHERE success=1 AND version='5'")) + .isEqualTo(1L); + } + assertThat(tableCounts(rehearsalUrl)).containsAllEntriesOf(countsBefore); + assertThat(sha256(source)).isEqualTo(sourceHashBefore); + assertThat(Files.getLastModifiedTime(source)).isEqualTo(sourceMtimeBefore); + } + + private void assertNoActiveSidecar(Path source) { + for (String suffix : List.of("-wal", "-shm", "-journal")) { + Path sidecar = Path.of(source.toString() + suffix); + assertThat(sidecar) + .as("检测到 SQLite 活跃或未收敛侧车文件，拒绝复制: %s", sidecar) + .doesNotExist(); + } + } + + private Map tableCounts(String url) throws Exception { + Map counts = new LinkedHashMap<>(); + try (Connection connection = DriverManager.getConnection(url); Statement statement = connection.createStatement()) { + for (String table : COUNTED_TABLES) { + try (ResultSet resultSet = statement.executeQuery("SELECT COUNT(*) FROM " + table)) { + counts.put(table, resultSet.next() ? resultSet.getLong(1) : 0L); + } + } + } + return counts; + } + + private long scalarLong(Connection connection, String sql) throws Exception { + try (Statement statement = connection.createStatement(); ResultSet resultSet = statement.executeQuery(sql)) { + return resultSet.next() ? resultSet.getLong(1) : 0L; + } + } + + private String scalarText(Connection connection, String sql) throws Exception { + try (Statement statement = connection.createStatement(); ResultSet resultSet = statement.executeQuery(sql)) { + return resultSet.next() ? resultSet.getString(1) : null; + } + } + + private byte[] sha256(Path path) throws Exception { + MessageDigest digest = MessageDigest.getInstance("SHA-256"); + try (var input = Files.newInputStream(path)) { + byte[] buffer = new byte[8192]; + int read; + while ((read = input.read(buffer)) != -1) { + digest.update(buffer, 0, read); + } + } + return digest.digest(); + } +} diff --git a/src/test/java/com/getjobs/application/service/DatabaseMigrationTest.java b/src/test/java/com/getjobs/application/service/DatabaseMigrationTest.java new file mode 100644 index 0000000..e9443db --- /dev/null +++ b/src/test/java/com/getjobs/application/service/DatabaseMigrationTest.java @@ -0,0 +1,175 @@ +package com.getjobs.application.service; + +import org.flywaydb.core.Flyway; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.nio.file.Path; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.ResultSet; +import java.sql.Statement; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class DatabaseMigrationTest { + @TempDir + Path tempDir; + + @Test + void freshDatabaseMigratesThroughV5AndMatchesSchemaContract() throws Exception { + String url = sqliteUrl(tempDir.resolve("fresh.db")); + + Flyway flyway = flyway(url); + flyway.migrate(); + + try (Connection connection = DriverManager.getConnection(url)) { + DatabaseSchemaService.validateSchema(connection); + assertThat(scalar(connection, + "SELECT COUNT(*) FROM flyway_schema_history WHERE success=1 AND version='5'")) + .isEqualTo(1L); + assertThat(columns(connection, "ai")).contains("apply_threshold", "priority_apply_threshold"); + assertThat(columns(connection, "boss_data")) + .contains("source_keyword", "salary_min_k", "salary_max_k", "salary_median_k", "salary_months"); + } + } + + @Test + void nonEmptyLegacyDatabaseIsBaselinedThenSafelyCompletedByV5() throws Exception { + String url = sqliteUrl(tempDir.resolve("legacy.db")); + try (Connection connection = DriverManager.getConnection(url); Statement statement = connection.createStatement()) { + statement.execute("CREATE TABLE profile (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL)"); + statement.execute("INSERT INTO profile(name) VALUES ('legacy-profile')"); + statement.execute("CREATE TABLE priority_company (" + + "id INTEGER PRIMARY KEY AUTOINCREMENT, company_name TEXT NOT NULL UNIQUE)"); + statement.execute("INSERT INTO priority_company(company_name) VALUES ('legacy-company')"); + } + + flyway(url).migrate(); + + try (Connection connection = DriverManager.getConnection(url); Statement statement = connection.createStatement()) { + DatabaseSchemaService.validateSchema(connection); + assertThat(scalar(connection, "SELECT COUNT(*) FROM priority_company")).isEqualTo(1L); + assertThat(scalar(connection, + "SELECT COUNT(*) FROM flyway_schema_history WHERE type='BASELINE' AND version='4'")) + .isEqualTo(1L); + assertThat(scalar(connection, + "SELECT COUNT(*) FROM flyway_schema_history WHERE success=1 AND version='5'")) + .isEqualTo(1L); + + statement.execute("INSERT INTO profile(name, is_active) VALUES ('second-profile', 0)"); + long secondProfileId = scalar(connection, "SELECT MAX(id) FROM profile"); + statement.execute("INSERT INTO priority_company(profile_id, company_name) VALUES (" + + secondProfileId + ", 'legacy-company')"); + assertThat(scalar(connection, + "SELECT COUNT(*) FROM priority_company WHERE company_name='legacy-company'")) + .isEqualTo(2L); + } + } + + @Test + void legacyStandaloneCompanyUniqueIndexIsDetectedAndRebuilt() throws Exception { + String url = sqliteUrl(tempDir.resolve("legacy-index.db")); + try (Connection connection = DriverManager.getConnection(url); Statement statement = connection.createStatement()) { + statement.execute("CREATE TABLE profile (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL)"); + statement.execute("INSERT INTO profile(name) VALUES ('legacy-profile')"); + statement.execute("CREATE TABLE priority_company (" + + "id INTEGER PRIMARY KEY AUTOINCREMENT, profile_id INTEGER, company_name TEXT NOT NULL)"); + statement.execute("CREATE UNIQUE INDEX legacy_company_name_unique ON priority_company(company_name)"); + statement.execute("INSERT INTO priority_company(profile_id, company_name) VALUES (1, 'legacy-company')"); + } + + flyway(url).migrate(); + + try (Connection connection = DriverManager.getConnection(url); Statement statement = connection.createStatement()) { + DatabaseSchemaService.validateSchema(connection); + statement.execute("INSERT INTO profile(name, is_active) VALUES ('second-profile', 0)"); + long secondProfileId = scalar(connection, "SELECT MAX(id) FROM profile"); + statement.execute("INSERT INTO priority_company(profile_id, company_name) VALUES (" + + secondProfileId + ", 'legacy-company')"); + assertThat(scalar(connection, + "SELECT COUNT(*) FROM priority_company WHERE company_name='legacy-company'")) + .isEqualTo(2L); + } + } + + @Test + void failedLegacyMigrationRollsBackSchemaAndPreservesRows() throws Exception { + String url = sqliteUrl(tempDir.resolve("legacy-duplicate.db")); + try (Connection connection = DriverManager.getConnection(url); Statement statement = connection.createStatement()) { + statement.execute("CREATE TABLE profile (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL)"); + statement.execute("INSERT INTO profile(name) VALUES ('legacy-profile')"); + statement.execute("CREATE TABLE priority_company (" + + "id INTEGER PRIMARY KEY AUTOINCREMENT, profile_id INTEGER, company_name TEXT NOT NULL)"); + statement.execute("INSERT INTO priority_company(profile_id, company_name) VALUES (1, 'duplicate-company')"); + statement.execute("INSERT INTO priority_company(profile_id, company_name) VALUES (1, 'duplicate-company')"); + } + + assertThatThrownBy(() -> flyway(url).migrate()) + .hasMessageContaining("Migration failed") + .hasStackTraceContaining( + "UNIQUE constraint failed: priority_company.profile_id, priority_company.company_name"); + + try (Connection connection = DriverManager.getConnection(url)) { + assertThat(scalar(connection, "SELECT COUNT(*) FROM priority_company")).isEqualTo(2L); + assertThat(columns(connection, "profile")).doesNotContain("is_active"); + assertThat(tableExists(connection, "boss_data")).isFalse(); + assertThat(tableExists(connection, "priority_company_profile_new")).isFalse(); + assertThat(scalar(connection, + "SELECT COUNT(*) FROM flyway_schema_history WHERE success=1 AND version='5'")) + .isZero(); + } + } + + @Test + void schemaValidationFailsWhenCriticalObjectIsMissing() throws Exception { + String url = sqliteUrl(tempDir.resolve("broken.db")); + try (Connection connection = DriverManager.getConnection(url); Statement statement = connection.createStatement()) { + statement.execute("CREATE TABLE profile (id INTEGER PRIMARY KEY, is_active INTEGER)"); + + org.assertj.core.api.Assertions.assertThatThrownBy(() -> DatabaseSchemaService.validateSchema(connection)) + .isInstanceOf(java.sql.SQLException.class) + .hasMessageContaining("缺少必要数据表"); + } + } + + private Flyway flyway(String url) { + return Flyway.configure() + .dataSource(url, null, null) + .locations("classpath:db/migration") + .baselineOnMigrate(true) + .baselineVersion("4") + .validateOnMigrate(true) + .load(); + } + + private String sqliteUrl(Path path) { + return "jdbc:sqlite:" + path.toAbsolutePath(); + } + + private long scalar(Connection connection, String sql) throws Exception { + try (Statement statement = connection.createStatement(); ResultSet resultSet = statement.executeQuery(sql)) { + return resultSet.next() ? resultSet.getLong(1) : 0L; + } + } + + private java.util.Set columns(Connection connection, String table) throws Exception { + java.util.Set columns = new java.util.HashSet<>(); + try (Statement statement = connection.createStatement(); + ResultSet resultSet = statement.executeQuery("PRAGMA table_info('" + table + "')")) { + while (resultSet.next()) { + columns.add(resultSet.getString("name")); + } + } + return columns; + } + + private boolean tableExists(Connection connection, String table) throws Exception { + try (Statement statement = connection.createStatement(); + ResultSet resultSet = statement.executeQuery( + "SELECT 1 FROM sqlite_master WHERE type='table' AND name='" + table + "'")) { + return resultSet.next(); + } + } +} diff --git a/src/test/java/com/getjobs/application/service/DatabaseSchemaServiceAiThresholdTest.java b/src/test/java/com/getjobs/application/service/DatabaseSchemaServiceAiThresholdTest.java deleted file mode 100644 index 733ebdd..0000000 --- a/src/test/java/com/getjobs/application/service/DatabaseSchemaServiceAiThresholdTest.java +++ /dev/null @@ -1,55 +0,0 @@ -package com.getjobs.application.service; - -import com.getjobs.application.config.RuntimeDirectoryInitializer; -import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.io.TempDir; -import org.sqlite.SQLiteDataSource; - -import java.nio.file.Path; -import java.sql.Connection; -import java.sql.ResultSet; -import java.sql.Statement; -import java.util.HashSet; -import java.util.Set; - -import static org.assertj.core.api.Assertions.assertThat; -import static org.mockito.Mockito.mock; - -class DatabaseSchemaServiceAiThresholdTest { - @TempDir - Path tempDir; - - @Test - void addsThresholdColumnsToExistingAiTable() throws Exception { - SQLiteDataSource dataSource = new SQLiteDataSource(); - dataSource.setUrl("jdbc:sqlite:" + tempDir.resolve("schema-test.db")); - - try (Connection connection = dataSource.getConnection(); Statement statement = connection.createStatement()) { - statement.execute(""" - CREATE TABLE ai ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - profile_id INTEGER, - introduce TEXT, - prompt TEXT - ) - """); - } - - DatabaseSchemaService service = new DatabaseSchemaService( - dataSource, - mock(RuntimeDirectoryInitializer.class) - ); - service.initializeSchema(); - - Set columns = new HashSet<>(); - try (Connection connection = dataSource.getConnection(); - Statement statement = connection.createStatement(); - ResultSet resultSet = statement.executeQuery("PRAGMA table_info(ai)")) { - while (resultSet.next()) { - columns.add(resultSet.getString("name")); - } - } - - assertThat(columns).contains("apply_threshold", "priority_apply_threshold"); - } -} diff --git a/tasks/2026-08-24-p0-2-data-migration-safety.md b/tasks/2026-08-24-p0-2-data-migration-safety.md new file mode 100644 index 0000000..bb842f5 --- /dev/null +++ b/tasks/2026-08-24-p0-2-data-migration-safety.md @@ -0,0 +1,71 @@ +# P0.2 数据一致性与迁移止血 + +## 背景 + +审计确认项目同时依赖 Flyway、`DatabaseSchemaService`、平台 Service 和手工 `sql.js` 脚本修改 Schema。部分异常只记录 warning 后继续，Boss 刷新与 Worker 还可能在线重建 `boss_data`。旧的非空数据库若没有 Flyway 历史，会被 baseline 到 v4，可能跳过 V1-V4 并遗留缺表、缺列或缺索引。 + +当前 `db/getjobs.db` 只读画像显示：V1-V4 均成功、`integrity_check=ok`、无 WAL/SHM、未发现已检查的重复/孤儿数据。该结果仅说明当前库尚未损坏，不代表现有迁移机制安全。 + +## 目标 + +1. 以一次性 Flyway V5 兼容迁移收敛旧库缺失的表、列和索引。 +2. Flyway 完成后，应用启动阶段只读校验 Schema；缺失时阻止启动，不再静默带病运行。 +3. 禁止普通 API、Worker prepare 和平台 Service 初始化执行 DROP/重建/补列。 +4. 禁用会整文件覆盖 SQLite 且不处理 WAL/SHM 的旧手工迁移脚本。 +5. 在临时数据库和当前数据库隔离副本上验证迁移前后行数与完整性。 + +## 允许修改范围 + +- `src/main/resources/db/migration/` 与对应 Java migration。 +- `DatabaseSchemaService` 的迁移/校验职责边界。 +- Boss、猎聘、51job、智联中与运行时 DDL 直接相关的调用。 +- 旧手工迁移脚本的安全禁用提示。 +- 隔离数据库迁移测试、任务文档与数据库说明。 + +## 禁止修改范围 + +- 不写入、迁移、覆盖或删除 `db/getjobs.db`。 +- 不新增业务 UNIQUE/FK、不清洗历史重复数据、不改写已有的非空 Profile 归属；仅沿用旧迁移对 NULL `profile_id` 的兼容回填。 +- 不修改投递、AI、Provider、Cookie 或平台采集业务逻辑。 +- 不进行真实招聘平台、Webhook 或计费 AI 调用。 +- 不删除旧数据库字段；本轮只保证兼容和停止在线 destructive DDL。 + +## 已确定实现要求 + +- V5 对 fresh DB、已有 V1-V4 DB、非空无历史旧库均可执行。 +- V5 必须在 Flyway 事务内工作；`priority_company` 重建前后校验行数，不使用 `INSERT OR IGNORE` 静默丢数据。 +- V5 补齐 V1/V4 缺表、运行时漂移列和 V2/V4 索引。 +- 启动 Schema 校验必须验证关键表、关键列、关键索引；失败抛异常阻止启动。 +- Boss reload 只做只读计数，不再重建表、checkpoint 或 VACUUM。 +- 平台 Service 不再在 `@PostConstruct` 中执行 DDL。 +- 原库 rehearsal 必须先复制到测试临时目录，且源库存在 WAL/SHM/journal 时拒绝复制。 + +## 验收标准 + +- fresh SQLite 从 V1-V5 初始化成功。 +- 无 Flyway 历史的旧 schema 被 baseline v4 后执行 V5，补齐缺失对象并保留原行数。 +- 当前 `db/getjobs.db` 的隔离副本执行 V5 后 `integrity_check=ok`，关键表行数不减少，原库时间戳/hash 不变。 +- 任一关键表、列或索引缺失时，Schema 校验明确失败。 +- 源码中不再存在从 Boss reload/Worker prepare、Liepin/51job `@PostConstruct` 触发的 DROP/ALTER/CREATE。 +- 完整后端测试、前端 lint/typecheck/build 与扩展测试通过。 + +## 测试命令 + +```powershell +.\gradlew.bat test +$env:P0_REHEARSAL_DB = (Resolve-Path db/getjobs.db).Path +.\gradlew.bat test --tests com.getjobs.application.service.DatabaseMigrationRehearsalTest +Remove-Item Env:P0_REHEARSAL_DB +pnpm --dir front lint +pnpm --dir front typecheck +pnpm --dir front build +$extensionTests = Get-ChildItem chrome-extension/tests/*.test.cjs | ForEach-Object { $_.FullName } +node --test $extensionTests +``` + +## 返回格式 + +- 迁移对象、职责变化和兼容边界。 +- fresh/legacy/真实副本的版本、完整性与行数对账结果。 +- 原库未修改的 hash/mtime 证据。 +- 测试、diff、Commit、Push 与 PR 状态。