|
|
@@ -0,0 +1,366 @@
|
|
|
+package com.etotem.cfc.service;
|
|
|
+
|
|
|
+import lombok.extern.slf4j.Slf4j;
|
|
|
+import org.springframework.jdbc.core.JdbcTemplate;
|
|
|
+import org.springframework.stereotype.Service;
|
|
|
+import org.springframework.transaction.annotation.Transactional;
|
|
|
+
|
|
|
+import javax.annotation.Resource;
|
|
|
+import java.sql.ResultSet;
|
|
|
+import java.sql.SQLException;
|
|
|
+import java.util.List;
|
|
|
+
|
|
|
+/**
|
|
|
+ * F-6:三套旧 CF 账本一次性迁入个人 {@code cf_balance}。
|
|
|
+ *
|
|
|
+ * <p>迁移编号 417。旧表 {@code family_platform_balance}、
|
|
|
+ * {@code family_members.system_points}、{@code user_platform_balance} 只读保留,
|
|
|
+ * 本服务不会回写或清零旧值。</p>
|
|
|
+ *
|
|
|
+ * <p>幂等策略:每一笔旧账本记录都先按
|
|
|
+ * {@code (member_id, ref_type, ref_id)} 检查 {@code cf_balance_log};已有流水即跳过。
|
|
|
+ * 余额变更与流水写入处于同一事务,失败会整体回滚,下次启动可安全重试。</p>
|
|
|
+ */
|
|
|
+@Slf4j
|
|
|
+@Service
|
|
|
+public class CfLegacyLedgerMigrationService {
|
|
|
+
|
|
|
+ public static final String REF_TYPE_FAMILY_POOL = "migration_family_pool";
|
|
|
+ public static final String REF_TYPE_SYSTEM_POINTS = "migration_system_points";
|
|
|
+ public static final String REF_TYPE_USER_WALLET = "migration_user_wallet";
|
|
|
+
|
|
|
+ private static final String LOG_TYPE_ADJUST = "adjust";
|
|
|
+
|
|
|
+ private static final String FAMILY_POOL_SQL =
|
|
|
+ "SELECT fpb.id AS source_id, fpb.family_id AS family_id, f.creator_id AS user_id, "
|
|
|
+ + "fm.id AS member_id, COALESCE(fpb.total_earned, 0) AS total_earned, "
|
|
|
+ + "COALESCE(fpb.available, 0) AS available, COALESCE(fpb.frozen, 0) AS frozen, "
|
|
|
+ + "COALESCE(fpb.withdrawn, 0) AS withdrawn, COALESCE(fpb.exchanged, 0) AS exchanged "
|
|
|
+ + "FROM family_platform_balance fpb "
|
|
|
+ + "LEFT JOIN families f ON f.id = fpb.family_id "
|
|
|
+ + "LEFT JOIN family_members fm ON fm.id = ("
|
|
|
+ + " SELECT fm2.id FROM family_members fm2 "
|
|
|
+ + " WHERE fm2.family_id = fpb.family_id AND fm2.user_id = f.creator_id "
|
|
|
+ + " ORDER BY CASE WHEN fm2.is_admin = 1 THEN 0 ELSE 1 END, fm2.id ASC LIMIT 1"
|
|
|
+ + ")";
|
|
|
+
|
|
|
+ private static final String SYSTEM_POINTS_SQL =
|
|
|
+ "SELECT fm.id AS source_id, fm.id AS member_id, fm.family_id AS family_id, "
|
|
|
+ + "fm.user_id AS user_id, COALESCE(fm.system_points, 0) AS amount "
|
|
|
+ + "FROM family_members fm WHERE COALESCE(fm.system_points, 0) > 0";
|
|
|
+
|
|
|
+ private static final String USER_WALLET_SQL =
|
|
|
+ "SELECT upb.id AS source_id, fm.id AS member_id, fm.family_id AS family_id, "
|
|
|
+ + "upb.user_id AS user_id, COALESCE(upb.total_earned, 0) AS total_earned, "
|
|
|
+ + "COALESCE(upb.available, 0) AS available, COALESCE(upb.frozen, 0) AS frozen, "
|
|
|
+ + "COALESCE(upb.withdrawn, 0) AS withdrawn, COALESCE(upb.exchanged, 0) AS exchanged "
|
|
|
+ + "FROM user_platform_balance upb "
|
|
|
+ + "LEFT JOIN users u ON u.id = upb.user_id "
|
|
|
+ + "LEFT JOIN family_members fm ON fm.id = ("
|
|
|
+ + " SELECT fm2.id FROM family_members fm2 WHERE fm2.user_id = upb.user_id "
|
|
|
+ + " ORDER BY CASE WHEN upb.family_id IS NOT NULL AND fm2.family_id = upb.family_id THEN 0 "
|
|
|
+ + " WHEN u.family_id IS NOT NULL AND fm2.family_id = u.family_id THEN 1 ELSE 2 END, "
|
|
|
+ + " CASE WHEN fm2.is_admin = 1 THEN 0 ELSE 1 END, fm2.id ASC LIMIT 1"
|
|
|
+ + ")";
|
|
|
+
|
|
|
+ @Resource
|
|
|
+ private JdbcTemplate jdbcTemplate;
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 执行三类存量迁移。整个方法是一个事务:任何一笔失败都会回滚本批全部变更。
|
|
|
+ *
|
|
|
+ * @return 本次实际迁入的旧账本记录数(已迁移过的记录不计入)
|
|
|
+ */
|
|
|
+ @Transactional
|
|
|
+ public int migrate() {
|
|
|
+ if (!tableExists("cf_balance") || !tableExists("cf_balance_log")) {
|
|
|
+ log.warn("迁移417: cf_balance / cf_balance_log 不存在,跳过存量迁移");
|
|
|
+ return 0;
|
|
|
+ }
|
|
|
+
|
|
|
+ int migrated = 0;
|
|
|
+ migrated += migrateFamilyPlatformPools();
|
|
|
+ migrated += migrateMemberSystemPoints();
|
|
|
+ migrated += migrateUserWallets();
|
|
|
+ log.info("迁移417: 三套旧 CF 账本迁移完成,本次迁入 {} 笔", migrated);
|
|
|
+ return migrated;
|
|
|
+ }
|
|
|
+
|
|
|
+ int migrateFamilyPlatformPools() {
|
|
|
+ if (!tableExists("family_platform_balance") || !tableExists("families")
|
|
|
+ || !tableExists("family_members")) {
|
|
|
+ log.warn("迁移417: 家庭池或家庭成员表不存在,跳过 migration_family_pool");
|
|
|
+ return 0;
|
|
|
+ }
|
|
|
+
|
|
|
+ List<FamilyPoolRow> rows = jdbcTemplate.query(FAMILY_POOL_SQL,
|
|
|
+ (rs, rowNum) -> readFamilyPoolRow(rs));
|
|
|
+ int migrated = 0;
|
|
|
+ int unresolved = 0;
|
|
|
+ for (FamilyPoolRow row : rows) {
|
|
|
+ if (row.memberId == null) {
|
|
|
+ if (hasMigratableValue(row.totalEarned, row.available, row.frozen,
|
|
|
+ row.withdrawn, row.exchanged)) {
|
|
|
+ unresolved++;
|
|
|
+ log.error("迁移417: 家庭池 familyId={} 找不到创建者对应的 family_member",
|
|
|
+ row.familyId);
|
|
|
+ }
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ int availableDelta = mergeAvailable(row.available, row.frozen);
|
|
|
+ boolean applied = applyLedger(row.memberId, row.userId, row.familyId,
|
|
|
+ 0L, REF_TYPE_FAMILY_POOL, availableDelta, row.totalEarned,
|
|
|
+ row.withdrawn, row.exchanged,
|
|
|
+ "F-6 存量迁移:家庭池 available+frozen 合入 available");
|
|
|
+ if (applied) {
|
|
|
+ migrated++;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ assertNoUnresolvedSource("migration_family_pool", unresolved);
|
|
|
+ return migrated;
|
|
|
+ }
|
|
|
+
|
|
|
+ int migrateMemberSystemPoints() {
|
|
|
+ if (!tableExists("family_members")) {
|
|
|
+ log.warn("迁移417: family_members 表不存在,跳过 migration_system_points");
|
|
|
+ return 0;
|
|
|
+ }
|
|
|
+
|
|
|
+ List<SystemPointsRow> rows = jdbcTemplate.query(SYSTEM_POINTS_SQL,
|
|
|
+ (rs, rowNum) -> readSystemPointsRow(rs));
|
|
|
+ int migrated = 0;
|
|
|
+ for (SystemPointsRow row : rows) {
|
|
|
+ int availableDelta = mergeAvailable(row.amount, 0);
|
|
|
+ boolean applied = applyLedger(row.memberId, row.userId, row.familyId,
|
|
|
+ row.memberId, REF_TYPE_SYSTEM_POINTS, availableDelta, row.amount,
|
|
|
+ 0, 0,
|
|
|
+ "F-6 存量迁移:family_members.system_points 迁入个人 CF 值");
|
|
|
+ if (applied) {
|
|
|
+ migrated++;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return migrated;
|
|
|
+ }
|
|
|
+
|
|
|
+ int migrateUserWallets() {
|
|
|
+ if (!tableExists("user_platform_balance") || !tableExists("users")
|
|
|
+ || !tableExists("family_members")) {
|
|
|
+ log.warn("迁移417: 用户钱包或家庭成员表不存在,跳过 migration_user_wallet");
|
|
|
+ return 0;
|
|
|
+ }
|
|
|
+
|
|
|
+ List<UserWalletRow> rows = jdbcTemplate.query(USER_WALLET_SQL,
|
|
|
+ (rs, rowNum) -> readUserWalletRow(rs));
|
|
|
+ int migrated = 0;
|
|
|
+ int unresolved = 0;
|
|
|
+ for (UserWalletRow row : rows) {
|
|
|
+ if (row.memberId == null) {
|
|
|
+ if (hasMigratableValue(row.totalEarned, row.available, row.frozen,
|
|
|
+ row.withdrawn, row.exchanged)) {
|
|
|
+ unresolved++;
|
|
|
+ log.error("迁移417: 用户钱包 userId={} 找不到本人对应的 family_member",
|
|
|
+ row.userId);
|
|
|
+ }
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ int availableDelta = mergeAvailable(row.available, row.frozen);
|
|
|
+ boolean applied = applyLedger(row.memberId, row.userId, row.familyId,
|
|
|
+ row.sourceId, REF_TYPE_USER_WALLET, availableDelta, row.totalEarned,
|
|
|
+ row.withdrawn, row.exchanged,
|
|
|
+ "F-6 存量迁移:用户钱包 available+frozen 合入 available");
|
|
|
+ if (applied) {
|
|
|
+ migrated++;
|
|
|
+ }
|
|
|
+ }
|
|
|
+ assertNoUnresolvedSource("migration_user_wallet", unresolved);
|
|
|
+ return migrated;
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 将一笔旧账本增量写入个人余额并留迁移流水。
|
|
|
+ *
|
|
|
+ * @return true 表示本次实际迁入;false 表示已有迁移流水,已幂等跳过
|
|
|
+ */
|
|
|
+ private boolean applyLedger(Long memberId, Long userId, Long familyId, Long refId,
|
|
|
+ String refType, int availableDelta, int totalEarnedDelta,
|
|
|
+ int withdrawnDelta, int exchangedDelta, String remark) {
|
|
|
+ // 幂等键对齐 F-2 升级后的唯一键 uk_ref_type (member_id, type, ref_type, ref_id)
|
|
|
+ Integer existingLogs = jdbcTemplate.queryForObject(
|
|
|
+ "SELECT COUNT(*) FROM cf_balance_log "
|
|
|
+ + "WHERE member_id = ? AND type = ? AND ref_type = ? AND ref_id = ?",
|
|
|
+ Integer.class, memberId, LOG_TYPE_ADJUST, refType, refId);
|
|
|
+ if (!shouldMigrateRow(existingLogs)) {
|
|
|
+ log.debug("迁移417: {} 已存在迁移流水 memberId={} refId={},跳过",
|
|
|
+ refType, memberId, refId);
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+
|
|
|
+ // frozen 按存量口径合入 available;目标表 frozen 保留 F-2 后续业务产生的冻结值。
|
|
|
+ jdbcTemplate.update(
|
|
|
+ "INSERT INTO cf_balance "
|
|
|
+ + "(member_id, user_id, family_id, total_earned, available, frozen, withdrawn, exchanged, created_at, updated_at) "
|
|
|
+ + "VALUES (?, ?, ?, ?, ?, 0, ?, ?, NOW(), NOW()) "
|
|
|
+ + "ON DUPLICATE KEY UPDATE "
|
|
|
+ + "user_id = COALESCE(cf_balance.user_id, VALUES(user_id)), "
|
|
|
+ + "family_id = COALESCE(cf_balance.family_id, VALUES(family_id)), "
|
|
|
+ + "total_earned = cf_balance.total_earned + VALUES(total_earned), "
|
|
|
+ + "available = cf_balance.available + VALUES(available), "
|
|
|
+ + "withdrawn = cf_balance.withdrawn + VALUES(withdrawn), "
|
|
|
+ + "exchanged = cf_balance.exchanged + VALUES(exchanged), "
|
|
|
+ + "updated_at = NOW()",
|
|
|
+ memberId, userId, familyId, totalEarnedDelta, availableDelta,
|
|
|
+ withdrawnDelta, exchangedDelta);
|
|
|
+
|
|
|
+ Integer balanceAfter = jdbcTemplate.queryForObject(
|
|
|
+ "SELECT available FROM cf_balance WHERE member_id = ?",
|
|
|
+ Integer.class, memberId);
|
|
|
+ if (balanceAfter == null) {
|
|
|
+ throw new IllegalStateException("迁移417: cf_balance 写入后未找到 memberId=" + memberId);
|
|
|
+ }
|
|
|
+
|
|
|
+ jdbcTemplate.update(
|
|
|
+ "INSERT INTO cf_balance_log "
|
|
|
+ + "(member_id, user_id, family_id, type, ref_type, ref_id, amount, balance_after, remark, created_at) "
|
|
|
+ + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NOW())",
|
|
|
+ memberId, userId, familyId, LOG_TYPE_ADJUST, refType, refId,
|
|
|
+ availableDelta, balanceAfter, remark);
|
|
|
+ return true;
|
|
|
+ }
|
|
|
+
|
|
|
+ /** 旧账本可用值口径:available + frozen,合入新账本 available。 */
|
|
|
+ public static int mergeAvailable(Integer available, Integer frozen) {
|
|
|
+ return safeInt(available) + safeInt(frozen);
|
|
|
+ }
|
|
|
+
|
|
|
+ /** 幂等守卫:仅当该来源尚无迁移流水时才允许入账。 */
|
|
|
+ public static boolean shouldMigrateRow(Integer existingLogCount) {
|
|
|
+ return existingLogCount == null || existingLogCount <= 0;
|
|
|
+ }
|
|
|
+
|
|
|
+ /** 流动 CF 值守恒口径:迁移前后均按 available + frozen 计算。 */
|
|
|
+ public static long liquidValue(Integer available, Integer frozen) {
|
|
|
+ return (long) safeInt(available) + safeInt(frozen);
|
|
|
+ }
|
|
|
+
|
|
|
+ /** 历史消耗口径:已提现 + 已兑换。 */
|
|
|
+ public static long historicalValue(Integer withdrawn, Integer exchanged) {
|
|
|
+ return (long) safeInt(withdrawn) + safeInt(exchanged);
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 校验单笔旧账本迁入后的守恒:
|
|
|
+ * 流动值 available+frozen 必须等于迁入 available;历史 withdrawn/exchanged 必须原样带入。
|
|
|
+ */
|
|
|
+ public static boolean isConserved(Integer oldAvailable, Integer oldFrozen,
|
|
|
+ Integer oldWithdrawn, Integer oldExchanged,
|
|
|
+ int migratedAvailable, int migratedWithdrawn,
|
|
|
+ int migratedExchanged) {
|
|
|
+ return liquidValue(oldAvailable, oldFrozen) == migratedAvailable
|
|
|
+ && historicalValue(oldWithdrawn, oldExchanged)
|
|
|
+ == (long) migratedWithdrawn + migratedExchanged;
|
|
|
+ }
|
|
|
+
|
|
|
+ private static int safeInt(Integer value) {
|
|
|
+ return value == null ? 0 : value;
|
|
|
+ }
|
|
|
+
|
|
|
+ private static boolean hasMigratableValue(int totalEarned, int available, int frozen,
|
|
|
+ int withdrawn, int exchanged) {
|
|
|
+ return totalEarned != 0 || available != 0 || frozen != 0
|
|
|
+ || withdrawn != 0 || exchanged != 0;
|
|
|
+ }
|
|
|
+
|
|
|
+ /** 有价值但找不到目标成员的旧账本必须让整批回滚,禁止静默丢失 CF 值。 */
|
|
|
+ private static void assertNoUnresolvedSource(String refType, int unresolved) {
|
|
|
+ if (unresolved > 0) {
|
|
|
+ throw new IllegalStateException(
|
|
|
+ "迁移417: " + refType + " 有 " + unresolved + " 笔非零旧账本找不到目标成员");
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ private FamilyPoolRow readFamilyPoolRow(ResultSet rs) throws SQLException {
|
|
|
+ FamilyPoolRow row = new FamilyPoolRow();
|
|
|
+ row.sourceId = rs.getLong("source_id");
|
|
|
+ row.familyId = rs.getLong("family_id");
|
|
|
+ long userId = rs.getLong("user_id");
|
|
|
+ row.userId = rs.wasNull() ? null : userId;
|
|
|
+ long memberId = rs.getLong("member_id");
|
|
|
+ row.memberId = rs.wasNull() ? null : memberId;
|
|
|
+ row.totalEarned = rs.getInt("total_earned");
|
|
|
+ row.available = rs.getInt("available");
|
|
|
+ row.frozen = rs.getInt("frozen");
|
|
|
+ row.withdrawn = rs.getInt("withdrawn");
|
|
|
+ row.exchanged = rs.getInt("exchanged");
|
|
|
+ return row;
|
|
|
+ }
|
|
|
+
|
|
|
+ private SystemPointsRow readSystemPointsRow(ResultSet rs) throws SQLException {
|
|
|
+ SystemPointsRow row = new SystemPointsRow();
|
|
|
+ row.sourceId = rs.getLong("source_id");
|
|
|
+ row.memberId = rs.getLong("member_id");
|
|
|
+ row.familyId = rs.getLong("family_id");
|
|
|
+ long userId = rs.getLong("user_id");
|
|
|
+ row.userId = rs.wasNull() ? null : userId;
|
|
|
+ row.amount = rs.getInt("amount");
|
|
|
+ return row;
|
|
|
+ }
|
|
|
+
|
|
|
+ private UserWalletRow readUserWalletRow(ResultSet rs) throws SQLException {
|
|
|
+ UserWalletRow row = new UserWalletRow();
|
|
|
+ row.sourceId = rs.getLong("source_id");
|
|
|
+ row.memberId = rs.getLong("member_id");
|
|
|
+ row.familyId = rs.getLong("family_id");
|
|
|
+ long userId = rs.getLong("user_id");
|
|
|
+ row.userId = rs.wasNull() ? null : userId;
|
|
|
+ row.totalEarned = rs.getInt("total_earned");
|
|
|
+ row.available = rs.getInt("available");
|
|
|
+ row.frozen = rs.getInt("frozen");
|
|
|
+ row.withdrawn = rs.getInt("withdrawn");
|
|
|
+ row.exchanged = rs.getInt("exchanged");
|
|
|
+ return row;
|
|
|
+ }
|
|
|
+
|
|
|
+ private boolean tableExists(String tableName) {
|
|
|
+ try {
|
|
|
+ Integer count = jdbcTemplate.queryForObject(
|
|
|
+ "SELECT COUNT(*) FROM information_schema.TABLES "
|
|
|
+ + "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ?",
|
|
|
+ Integer.class, tableName);
|
|
|
+ return count != null && count > 0;
|
|
|
+ } catch (Exception e) {
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ static final class FamilyPoolRow {
|
|
|
+ Long sourceId;
|
|
|
+ Long familyId;
|
|
|
+ Long userId;
|
|
|
+ Long memberId;
|
|
|
+ int totalEarned;
|
|
|
+ int available;
|
|
|
+ int frozen;
|
|
|
+ int withdrawn;
|
|
|
+ int exchanged;
|
|
|
+ }
|
|
|
+
|
|
|
+ static final class SystemPointsRow {
|
|
|
+ Long sourceId;
|
|
|
+ Long memberId;
|
|
|
+ Long familyId;
|
|
|
+ Long userId;
|
|
|
+ int amount;
|
|
|
+ }
|
|
|
+
|
|
|
+ static final class UserWalletRow {
|
|
|
+ Long sourceId;
|
|
|
+ Long memberId;
|
|
|
+ Long familyId;
|
|
|
+ Long userId;
|
|
|
+ int totalEarned;
|
|
|
+ int available;
|
|
|
+ int frozen;
|
|
|
+ int withdrawn;
|
|
|
+ int exchanged;
|
|
|
+ }
|
|
|
+}
|