import java.io.IOException; import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; import java.sql.*; import java.util.ArrayList; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; /** * Audit + apply lobster/profile DDL gaps on tenant DB(s). * Usage: java -cp mysql-connector-j.jar ApplyLobsterTenantSchemaSync [tenantId...] */ public class ApplyLobsterTenantSchemaSync { private static final String MASTER_HOST = getenv("DB_HOST", "cq-cdb-8fjmemkb.sql.tencentcdb.com"); private static final int MASTER_PORT = Integer.parseInt(getenv("DB_PORT", "27220")); private static final String MASTER_USER = getenv("DB_USER", "root"); private static final String MASTER_PASSWORD = getenv("DB_PASSWORD", "Ylrz_1q2w3e4r5t6y"); private static final String MASTER_DB = getenv("DB_NAME", "ylrz_saas"); public static void main(String[] args) throws Exception { List tenantIds = new ArrayList<>(); if (args.length == 0) { tenantIds.add(33L); } else { for (String arg : args) { tenantIds.add(Long.parseLong(arg.trim())); } } try (Connection master = open(MASTER_HOST, MASTER_PORT, MASTER_DB, MASTER_USER, MASTER_PASSWORD)) { for (Long tenantId : tenantIds) { TenantDb tenant = loadTenant(master, tenantId); if (tenant == null) { System.err.println("[SKIP] tenant " + tenantId + " not found"); continue; } System.out.println("\n========== tenant " + tenantId + " -> " + tenant.database + " =========="); try (Connection conn = open(tenant.host, tenant.port, tenant.database, tenant.user, tenant.password)) { auditBefore(conn); applyCustomerProfileDomain(conn); applyLobsterUserProfile(conn); applySqlFile(conn, "../fs-service/src/main/resources/db/tenant-initTable-migration-lobster-schema-sync.sql"); auditAfter(conn); } } } } private static void auditBefore(Connection conn) throws SQLException { System.out.println("[AUDIT before]"); printCount(conn, "customer_profile_* tables", "SELECT COUNT(*) FROM information_schema.TABLES WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME LIKE 'customer_profile_%'"); printMissingColumns(conn, expectedColumns()); printNullVarCodes(conn); } private static void auditAfter(Connection conn) throws SQLException { System.out.println("[AUDIT after]"); printCount(conn, "customer_profile_* tables", "SELECT COUNT(*) FROM information_schema.TABLES WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME LIKE 'customer_profile_%'"); printMissingColumns(conn, expectedColumns()); printNullVarCodes(conn); } private static Map expectedColumns() { Map m = new LinkedHashMap<>(); m.put("company_workflow_lobster", new String[]{ "execution_mode", "enable_content_personalization", "enable_flow_personalization", "strict_fixed_workflow", "auto_start_instance_on_tag", "template_kind", "source_template_id" }); m.put("lobster_user_profile", new String[]{ "channel_type", "lifecycle_stage", "value_score", "total_purchase", "interaction_count", "last_active_time", "variable_snapshot", "internal_tags" }); m.put("lobster_workflow_instance", new String[]{ "binding_id", "dept_id", "company_user_id", "qw_user_id", "wx_account_id", "channel_type" }); m.put("company_tag_template_binding", new String[]{ "exclude_tags", "target_channels", "source_template_id", "task_template_id", "task_name" }); m.put("company_lobster_profile_config", new String[]{"field_type"}); m.put("lobster_conversation_summary", new String[]{"external_user_id", "channel_type"}); m.put("lobster_user_preference", new String[]{"channel_type"}); m.put("lobster_chat_record", new String[]{"channel_type"}); m.put("lobster_vector_store", new String[]{"category", "vec_key", "text_content", "vector"}); m.put("lobster_sales_corpus", new String[]{"corpus_kind", "industry_type"}); return m; } private static void printMissingColumns(Connection conn, Map expected) throws SQLException { for (Map.Entry e : expected.entrySet()) { if (!tableExists(conn, e.getKey())) { System.out.println(" MISSING TABLE: " + e.getKey()); continue; } List missing = new ArrayList<>(); for (String col : e.getValue()) { if (!columnExists(conn, e.getKey(), col)) { missing.add(col); } } if (missing.isEmpty()) { System.out.println(" OK columns: " + e.getKey()); } else { System.out.println(" MISSING columns on " + e.getKey() + ": " + String.join(", ", missing)); } } } private static void printNullVarCodes(Connection conn) throws SQLException { if (!tableExists(conn, "company_workflow_lobster_variable")) { return; } try (Statement st = conn.createStatement(); ResultSet rs = st.executeQuery( "SELECT COUNT(*) FROM company_workflow_lobster_variable WHERE del_flag=0 AND (var_code IS NULL OR var_code='')")) { rs.next(); System.out.println(" var_code empty rows (active): " + rs.getInt(1)); } } private static void applyCustomerProfileDomain(Connection conn) throws SQLException { System.out.println("[APPLY] customer profile domain"); try (Statement st = conn.createStatement()) { st.execute( "CREATE TABLE IF NOT EXISTS `customer_profile_identity` (" + "`id` BIGINT NOT NULL AUTO_INCREMENT," + "`company_id` BIGINT NOT NULL," + "`channel_type` VARCHAR(8) NOT NULL," + "`profile_key` VARCHAR(128) NOT NULL," + "`customer_id` BIGINT DEFAULT NULL," + "`shard_id` INT NOT NULL," + "`create_time` DATETIME DEFAULT CURRENT_TIMESTAMP," + "`update_time` DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP," + "PRIMARY KEY (`id`)," + "UNIQUE KEY `uk_identity` (`company_id`, `channel_type`, `profile_key`)," + "KEY `idx_customer` (`company_id`, `customer_id`)," + "KEY `idx_shard` (`company_id`, `shard_id`)" + ") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4"); for (int i = 0; i < 32; i++) { st.execute(projectionSql(i)); st.execute(eventSql(i)); } } } private static void applyLobsterUserProfile(Connection conn) throws SQLException { System.out.println("[APPLY] lobster_user_profile"); if (!tableExists(conn, "lobster_user_profile")) { try (Statement st = conn.createStatement()) { st.execute( "CREATE TABLE IF NOT EXISTS `lobster_user_profile` (" + "`id` bigint NOT NULL AUTO_INCREMENT," + "`company_id` bigint NOT NULL," + "`external_user_id` varchar(128) NOT NULL," + "`channel_type` varchar(20) NOT NULL DEFAULT 'QW'," + "`nickname` varchar(128) DEFAULT NULL," + "`lifecycle_stage` varchar(32) DEFAULT 'NEW'," + "`value_score` int DEFAULT 0," + "`total_purchase` decimal(14,2) DEFAULT 0," + "`interaction_count` int DEFAULT 0," + "`last_active_time` datetime DEFAULT NULL," + "`internal_tags` varchar(1024) DEFAULT NULL," + "`variable_snapshot` text," + "`current_state` varchar(64) DEFAULT NULL," + "`create_time` datetime DEFAULT CURRENT_TIMESTAMP," + "`update_time` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP," + "PRIMARY KEY (`id`)," + "UNIQUE KEY `uk_user` (`company_id`,`channel_type`,`external_user_id`)," + "KEY `idx_lifecycle` (`company_id`,`lifecycle_stage`)" + ") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='Lobster legacy user profile'"); System.out.println(" created table lobster_user_profile"); } return; } addColumnIfMissing(conn, "lobster_user_profile", "channel_type", "ALTER TABLE lobster_user_profile ADD COLUMN channel_type varchar(20) NOT NULL DEFAULT 'QW' AFTER external_user_id"); addColumnIfMissing(conn, "lobster_user_profile", "lifecycle_stage", "ALTER TABLE lobster_user_profile ADD COLUMN lifecycle_stage varchar(32) DEFAULT 'NEW' AFTER nickname"); addColumnIfMissing(conn, "lobster_user_profile", "value_score", "ALTER TABLE lobster_user_profile ADD COLUMN value_score int DEFAULT 0 AFTER lifecycle_stage"); addColumnIfMissing(conn, "lobster_user_profile", "total_purchase", "ALTER TABLE lobster_user_profile ADD COLUMN total_purchase decimal(14,2) DEFAULT 0 AFTER value_score"); addColumnIfMissing(conn, "lobster_user_profile", "interaction_count", "ALTER TABLE lobster_user_profile ADD COLUMN interaction_count int DEFAULT 0 AFTER total_purchase"); addColumnIfMissing(conn, "lobster_user_profile", "last_active_time", "ALTER TABLE lobster_user_profile ADD COLUMN last_active_time datetime DEFAULT NULL AFTER interaction_count"); addColumnIfMissing(conn, "lobster_user_profile", "internal_tags", "ALTER TABLE lobster_user_profile ADD COLUMN internal_tags varchar(1024) DEFAULT NULL AFTER last_active_time"); addColumnIfMissing(conn, "lobster_user_profile", "variable_snapshot", "ALTER TABLE lobster_user_profile ADD COLUMN variable_snapshot text AFTER internal_tags"); addColumnIfMissing(conn, "lobster_user_profile", "current_state", "ALTER TABLE lobster_user_profile ADD COLUMN current_state varchar(64) DEFAULT NULL AFTER variable_snapshot"); fixUkUserIndex(conn); fixNullVarCodes(conn); } private static void fixUkUserIndex(Connection conn) throws SQLException { if (!indexExists(conn, "lobster_user_profile", "uk_user")) { try (Statement st = conn.createStatement()) { st.execute("ALTER TABLE lobster_user_profile ADD UNIQUE KEY uk_user (company_id, channel_type, external_user_id)"); System.out.println(" added index lobster_user_profile.uk_user"); } } } private static void fixNullVarCodes(Connection conn) throws SQLException { if (!tableExists(conn, "company_workflow_lobster_variable")) { return; } try (Statement st = conn.createStatement()) { int updated = st.executeUpdate( "UPDATE company_workflow_lobster_variable " + "SET var_code = CONCAT('var_', id) " + "WHERE del_flag = 0 AND (var_code IS NULL OR TRIM(var_code) = '')"); if (updated > 0) { System.out.println(" backfilled empty var_code rows: " + updated); } } } private static void applySqlFile(Connection conn, String relativePath) throws SQLException, IOException { Path path = Paths.get(relativePath).normalize(); if (!Files.exists(path)) { path = Paths.get("java", "fs-service", "src", "main", "resources", "db", "tenant-initTable-migration-lobster-schema-sync.sql"); } if (!Files.exists(path)) { System.out.println("[SKIP] schema sync file not found: " + relativePath); return; } System.out.println("[APPLY] " + path.toAbsolutePath()); String sql = Files.readString(path, StandardCharsets.UTF_8); // Strip line comments; execute as one batch (allowMultiQueries=true on connection) StringBuilder batch = new StringBuilder(); for (String line : sql.split("\\R")) { String trimmed = line.trim(); if (trimmed.startsWith("--")) { continue; } batch.append(line).append('\n'); } try (Statement st = conn.createStatement()) { st.execute(batch.toString()); } catch (SQLException e) { String msg = e.getMessage() != null ? e.getMessage() : ""; if (msg.contains("Duplicate column") || msg.contains("already exists") || msg.contains("Duplicate key name")) { System.out.println(" [WARN] idempotent skip: " + msg); return; } throw e; } } private static String projectionSql(int shard) { return "CREATE TABLE IF NOT EXISTS `customer_profile_projection_" + shard + "` (" + "`id` BIGINT NOT NULL AUTO_INCREMENT," + "`company_id` BIGINT NOT NULL," + "`channel_type` VARCHAR(8) NOT NULL," + "`profile_key` VARCHAR(128) NOT NULL," + "`customer_id` BIGINT DEFAULT NULL," + "`shard_id` INT NOT NULL," + "`nickname` VARCHAR(128) DEFAULT NULL," + "`lifecycle_stage` VARCHAR(32) DEFAULT 'NEW'," + "`value_score` INT DEFAULT 0," + "`total_purchase` DECIMAL(14,2) DEFAULT 0," + "`interaction_count` INT DEFAULT 0," + "`last_active_time` DATETIME DEFAULT NULL," + "`internal_tags` VARCHAR(1024) DEFAULT NULL," + "`current_state` VARCHAR(64) DEFAULT NULL," + "`profile_ext` JSON DEFAULT NULL," + "`create_time` DATETIME DEFAULT CURRENT_TIMESTAMP," + "`update_time` DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP," + "PRIMARY KEY (`id`)," + "UNIQUE KEY `uk_profile` (`company_id`, `channel_type`, `profile_key`)," + "KEY `idx_customer` (`company_id`, `customer_id`)," + "KEY `idx_active` (`company_id`, `last_active_time`)" + ") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4"; } private static String eventSql(int shard) { return "CREATE TABLE IF NOT EXISTS `customer_profile_event_" + shard + "` (" + "`id` BIGINT NOT NULL AUTO_INCREMENT," + "`company_id` BIGINT NOT NULL," + "`channel_type` VARCHAR(8) NOT NULL," + "`profile_key` VARCHAR(128) NOT NULL," + "`source` VARCHAR(64) NOT NULL," + "`source_version` VARCHAR(128) NOT NULL," + "`event_payload` JSON DEFAULT NULL," + "`create_time` DATETIME DEFAULT CURRENT_TIMESTAMP," + "PRIMARY KEY (`id`)," + "UNIQUE KEY `uk_event` (`company_id`, `channel_type`, `profile_key`, `source`, `source_version`)," + "KEY `idx_profile_time` (`company_id`, `channel_type`, `profile_key`, `create_time`)" + ") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4"; } private static void addColumnIfMissing(Connection conn, String table, String column, String ddl) throws SQLException { if (columnExists(conn, table, column)) { return; } try (Statement st = conn.createStatement()) { st.execute(ddl); System.out.println(" added column: " + table + "." + column); } } private static void printCount(Connection conn, String label, String sql) throws SQLException { try (Statement st = conn.createStatement(); ResultSet rs = st.executeQuery(sql)) { rs.next(); System.out.println(" " + label + ": " + rs.getInt(1)); } } private static boolean tableExists(Connection conn, String table) throws SQLException { try (PreparedStatement ps = conn.prepareStatement( "SELECT COUNT(*) FROM information_schema.TABLES WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=?")) { ps.setString(1, table); try (ResultSet rs = ps.executeQuery()) { return rs.next() && rs.getInt(1) > 0; } } } private static boolean columnExists(Connection conn, String table, String column) throws SQLException { try (PreparedStatement ps = conn.prepareStatement( "SELECT COUNT(*) FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=? AND COLUMN_NAME=?")) { ps.setString(1, table); ps.setString(2, column); try (ResultSet rs = ps.executeQuery()) { return rs.next() && rs.getInt(1) > 0; } } } private static boolean indexExists(Connection conn, String table, String index) throws SQLException { try (PreparedStatement ps = conn.prepareStatement( "SELECT COUNT(*) FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=? AND INDEX_NAME=?")) { ps.setString(1, table); ps.setString(2, index); try (ResultSet rs = ps.executeQuery()) { return rs.next() && rs.getInt(1) > 0; } } } private static TenantDb loadTenant(Connection master, long tenantId) throws SQLException { try (PreparedStatement ps = master.prepareStatement( "SELECT db_url, db_account, db_pwd FROM tenant_info WHERE id=? AND status=1 LIMIT 1")) { ps.setLong(1, tenantId); try (ResultSet rs = ps.executeQuery()) { if (!rs.next()) { return null; } return parseJdbcUrl(rs.getString("db_url"), rs.getString("db_account"), rs.getString("db_pwd")); } } } private static TenantDb parseJdbcUrl(String jdbcUrl, String user, String password) { String body = jdbcUrl.substring("jdbc:mysql://".length()); int slash = body.indexOf('/'); String hostPort = body.substring(0, slash); String rest = body.substring(slash + 1); int q = rest.indexOf('?'); String database = q >= 0 ? rest.substring(0, q) : rest; int colon = hostPort.indexOf(':'); TenantDb t = new TenantDb(); t.host = colon >= 0 ? hostPort.substring(0, colon) : hostPort; t.port = colon >= 0 ? Integer.parseInt(hostPort.substring(colon + 1)) : 3306; t.database = database; t.user = user != null ? user : MASTER_USER; t.password = password != null ? password : MASTER_PASSWORD; return t; } private static Connection open(String host, int port, String database, String user, String password) throws SQLException { String url = "jdbc:mysql://" + host + ":" + port + "/" + database + "?useUnicode=true&characterEncoding=utf8&serverTimezone=GMT%2B8&allowMultiQueries=true"; return DriverManager.getConnection(url, user, password); } private static String getenv(String key, String defaultValue) { String v = System.getenv(key); return v != null && !v.isEmpty() ? v : defaultValue; } private static final class TenantDb { String host; int port; String database; String user; String password; } }