ylrz_saas_pharmacy/java/scripts/ApplyLobsterTenantSchemaSync.java

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<Long> 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<String, String[]> expectedColumns() {
        Map<String, String[]> 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<String, String[]> expected) throws SQLException {
        for (Map.Entry<String, String[]> e : expected.entrySet()) {
            if (!tableExists(conn, e.getKey())) {
                System.out.println("  MISSING TABLE: " + e.getKey());
                continue;
            }
            List<String> 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;
    }
}