默认topic

nacos_config/rc-service-other/application.yml

@@ -128,6 +128,13 @@ task:
         topic: ${TASK_CENTER_EVENT_MQ_TOPIC:RC_DEFAULT_APP_ORDINARY}
         tag: ${TASK_CENTER_EVENT_MQ_TAG:task_center_event}
 
+cp:
+  relation-broadcast:
+    mq:
+      enabled: ${CP_RELATION_BROADCAST_MQ_ENABLED:true}
+      topic: ${CP_RELATION_BROADCAST_MQ_TOPIC:RC_DEFAULT_APP_ORDINARY}
+      tag: ${CP_RELATION_BROADCAST_MQ_TAG:cp_relation_broadcast}
+
 voice-room:
   region-broadcast:
     go:

rc-common-business/common-business-mq/business-mq-model/src/main/java/com/red/circle/mq/business/model/event/pay/RechargeSuccessEvent.java

@@ -0,0 +1,36 @@
+package com.red.circle.mq.business.model.event.pay;
+
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import com.fasterxml.jackson.databind.ser.std.ToStringSerializer;
+import java.io.Serializable;
+import java.util.Map;
+import lombok.Data;
+import lombok.experimental.Accessors;
+
+/**
+ * 通用充值成功事件。
+ */
+@Data
+@Accessors(chain = true)
+public class RechargeSuccessEvent implements Serializable {
+
+  private static final long serialVersionUID = 1L;
+
+  private String eventId;
+  private String sysOrigin;
+
+  @JsonSerialize(using = ToStringSerializer.class)
+  private Long userId;
+
+  @JsonSerialize(using = ToStringSerializer.class)
+  private Long amountCents;
+
+  private String amountUsd;
+  private String currency;
+  private String payPlatform;
+  private String paymentMethod;
+  private String sourceOrderId;
+  private String orderId;
+  private String occurredAt;
+  private Map<String, Object> payload;
+}

rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/constant/QueueNameConstant.java

@@ -17,13 +17,18 @@ public interface QueueNameConstant {
    */
   String LOGIN_REGISTER = "LOGIN_REGISTER";
 
-  /**
-   * 购买成功
-   */
-  String BUY_SUCCESS = "BUY_SUCCESS";
-
-  /**
-   * 视频语音房间会话
+  /**
+   * 购买成功
+   */
+  String BUY_SUCCESS = "BUY_SUCCESS";
+
+  /**
+   * 通用充值成功。
+   */
+  String RECHARGE_SUCCESS = "RECHARGE_SUCCESS";
+
+  /**
+   * 视频语音房间会话
    */
   String LIVE_VOICE_SESSION = "LIVE_VOICE_SESSION";
 

rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/streams/RechargeSuccessSink.java

@@ -0,0 +1,12 @@
+package com.red.circle.mq.rocket.business.streams;
+
+import com.red.circle.mq.rocket.business.constant.QueueNameConstant;
+
+/**
+ * 通用充值成功事件。
+ */
+public interface RechargeSuccessSink {
+
+  String INPUT = QueueNameConstant.RECHARGE_SUCCESS;
+  String TAG = "recharge_success_v1";
+}

rc-service/rc-service-console/console-adapter/src/main/java/com/red/circle/console/adapter/app/activity/ResidentCpActivityRestController.java

@@ -0,0 +1,36 @@
+package com.red.circle.console.adapter.app.activity;
+
+import com.red.circle.console.app.dto.clienobject.app.activity.cp.ResidentCpConfigCO;
+import com.red.circle.console.app.dto.cmd.app.activity.cp.ResidentCpConfigSaveCmd;
+import com.red.circle.console.app.service.app.activity.ResidentCpActivityService;
+import com.red.circle.console.infra.annotations.OpsOperationReqLog;
+import lombok.RequiredArgsConstructor;
+import org.springframework.validation.annotation.Validated;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+/**
+ * 常驻 CP 后台接口.
+ */
+@Validated
+@RestController
+@RequiredArgsConstructor
+@RequestMapping("/resident-activity/cp")
+public class ResidentCpActivityRestController {
+
+  private final ResidentCpActivityService residentCpActivityService;
+
+  @GetMapping("/config")
+  public ResidentCpConfigCO getConfig(String sysOrigin) {
+    return residentCpActivityService.getConfig(sysOrigin);
+  }
+
+  @OpsOperationReqLog(value = "保存 CP 等级积分配置")
+  @PostMapping("/config/save")
+  public void saveConfig(@RequestBody @Validated ResidentCpConfigSaveCmd cmd) {
+    residentCpActivityService.saveConfig(cmd);
+  }
+}

rc-service/rc-service-console/console-application/src/main/java/com/red/circle/console/app/service/app/activity/ResidentCpActivityServiceImpl.java

@@ -0,0 +1,167 @@
+package com.red.circle.console.app.service.app.activity;
+
+import com.red.circle.console.app.dto.clienobject.app.activity.cp.ResidentCpConfigCO;
+import com.red.circle.console.app.dto.clienobject.app.activity.cp.ResidentCpLevelConfigCO;
+import com.red.circle.console.app.dto.cmd.app.activity.cp.ResidentCpConfigSaveCmd;
+import com.red.circle.console.app.dto.cmd.app.activity.cp.ResidentCpLevelConfigSaveCmd;
+import com.red.circle.console.app.service.app.sys.CpCabinConfigService;
+import com.red.circle.framework.core.asserts.ResponseAssert;
+import com.red.circle.framework.core.response.ResponseErrorCode;
+import com.red.circle.framework.dto.PageQuery;
+import com.red.circle.framework.dto.PageResult;
+import com.red.circle.other.inner.model.cmd.sys.SysCpCabinConfigQryCmd;
+import com.red.circle.other.inner.model.dto.sys.SysCpCabinConfigDTO;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import lombok.RequiredArgsConstructor;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.stereotype.Service;
+
+/**
+ * 常驻 CP 配置服务实现.
+ */
+@Service
+@RequiredArgsConstructor
+public class ResidentCpActivityServiceImpl implements ResidentCpActivityService {
+
+  private static final int CP_LEVEL_COUNT = 5;
+  private static final int QUERY_LIMIT = 50;
+
+  private final CpCabinConfigService cpCabinConfigService;
+
+  @Override
+  public ResidentCpConfigCO getConfig(String sysOrigin) {
+    String normalizedSysOrigin = normalizeSysOrigin(sysOrigin);
+    List<SysCpCabinConfigDTO> configs = listCpCabinConfigs(normalizedSysOrigin);
+    List<ResidentCpLevelConfigCO> levels = new ArrayList<>(CP_LEVEL_COUNT);
+    for (int level = 1; level <= CP_LEVEL_COUNT; level++) {
+      SysCpCabinConfigDTO config = level <= configs.size() ? configs.get(level - 1) : null;
+      levels.add(toLevelCO(level, config));
+    }
+    return new ResidentCpConfigCO()
+        .setConfigured(!configs.isEmpty())
+        .setSysOrigin(normalizedSysOrigin)
+        .setLevels(levels);
+  }
+
+  @Override
+  public void saveConfig(ResidentCpConfigSaveCmd cmd) {
+    validateConfig(cmd);
+    String normalizedSysOrigin = normalizeSysOrigin(cmd.getSysOrigin());
+    List<SysCpCabinConfigDTO> configs = listCpCabinConfigs(normalizedSysOrigin);
+    Map<Long, SysCpCabinConfigDTO> configById = configs.stream()
+        .filter(item -> Objects.nonNull(item.getId()))
+        .collect(Collectors.toMap(SysCpCabinConfigDTO::getId, Function.identity(), (left, right) -> left));
+
+    for (ResidentCpLevelConfigSaveCmd levelCmd : sortedLevels(cmd.getLevels())) {
+      SysCpCabinConfigDTO config = Optional.ofNullable(levelCmd.getCabinConfigId())
+          .map(configById::get)
+          .orElseGet(() -> levelCmd.getLevel() <= configs.size()
+              ? configs.get(levelCmd.getLevel() - 1)
+              : null);
+      boolean exists = Objects.nonNull(config) && Objects.nonNull(config.getId());
+      if (!exists) {
+        config = new SysCpCabinConfigDTO();
+      }
+      config.setSysOrigin(normalizedSysOrigin)
+          .setName(defaultLevelName(levelCmd.getLevel(), levelCmd.getLevelName()))
+          .setCabinUnlockValue(defaultRequiredValue(levelCmd.getRequiredIntimacyValue()))
+          .setSort(levelCmd.getLevel().longValue());
+
+      if (exists) {
+        cpCabinConfigService.updateCpCabin(config);
+      } else {
+        cpCabinConfigService.addCpCabin(config);
+      }
+    }
+  }
+
+  private List<SysCpCabinConfigDTO> listCpCabinConfigs(String sysOrigin) {
+    SysCpCabinConfigQryCmd query = new SysCpCabinConfigQryCmd().setSysOrigin(sysOrigin);
+    PageQuery pageQuery = new PageQuery();
+    pageQuery.setCursor(1);
+    pageQuery.setLimit(QUERY_LIMIT);
+    pageQuery.setSearchCount(false);
+    query.setPageQuery(pageQuery);
+    PageResult<SysCpCabinConfigDTO> result = cpCabinConfigService.getCpCabin(query);
+    return Optional.ofNullable(result.getRecords())
+        .stream()
+        .flatMap(java.util.Collection::stream)
+        .sorted(Comparator
+            .comparing((SysCpCabinConfigDTO item) -> Optional.ofNullable(item.getSort())
+                .orElse(Long.MAX_VALUE))
+            .thenComparing(item -> Optional.ofNullable(item.getCabinUnlockValue())
+                .orElse(Long.MAX_VALUE))
+            .thenComparing(item -> Optional.ofNullable(item.getId()).orElse(Long.MAX_VALUE)))
+        .toList();
+  }
+
+  private ResidentCpLevelConfigCO toLevelCO(int level, SysCpCabinConfigDTO config) {
+    return new ResidentCpLevelConfigCO()
+        .setLevel(level)
+        .setLevelName(Objects.isNull(config)
+            ? defaultLevelName(level, null)
+            : defaultLevelName(level, config.getName()))
+        .setRequiredIntimacyValue(Objects.isNull(config)
+            ? 0L
+            : defaultRequiredValue(config.getCabinUnlockValue()))
+        .setCabinConfigId(Objects.isNull(config) ? null : config.getId());
+  }
+
+  private void validateConfig(ResidentCpConfigSaveCmd cmd) {
+    ResponseAssert.notBlank(ResponseErrorCode.REQUEST_PARAMETER_ERROR, cmd.getSysOrigin());
+    List<ResidentCpLevelConfigSaveCmd> levels = sortedLevels(cmd.getLevels());
+    ResponseAssert.isTrue(ResponseErrorCode.REQUEST_PARAMETER_ERROR,
+        "CP levels must contain exactly 5 items.", levels.size() == CP_LEVEL_COUNT);
+    Set<Integer> levelSet = new HashSet<>();
+    long previousRequiredValue = -1L;
+    for (ResidentCpLevelConfigSaveCmd level : levels) {
+      ResponseAssert.isTrue(ResponseErrorCode.REQUEST_PARAMETER_ERROR,
+          "CP level must be between 1 and 5.",
+          Objects.nonNull(level.getLevel()) && level.getLevel() >= 1
+              && level.getLevel() <= CP_LEVEL_COUNT && levelSet.add(level.getLevel()));
+      long requiredValue = defaultRequiredValue(level.getRequiredIntimacyValue());
+      ResponseAssert.isTrue(ResponseErrorCode.REQUEST_PARAMETER_ERROR,
+          "requiredIntimacyValue must be greater than or equal to zero.", requiredValue >= 0);
+      ResponseAssert.isTrue(ResponseErrorCode.REQUEST_PARAMETER_ERROR,
+          "requiredIntimacyValue must be ascending by level.",
+          requiredValue >= previousRequiredValue);
+      previousRequiredValue = requiredValue;
+    }
+  }
+
+  private List<ResidentCpLevelConfigSaveCmd> sortedLevels(
+      List<ResidentCpLevelConfigSaveCmd> levels) {
+    if (Objects.isNull(levels)) {
+      return List.of();
+    }
+    return levels.stream()
+        .filter(Objects::nonNull)
+        .sorted(Comparator.comparing(ResidentCpLevelConfigSaveCmd::getLevel,
+            Comparator.nullsLast(Integer::compareTo)))
+        .toList();
+  }
+
+  private String normalizeSysOrigin(String sysOrigin) {
+    return Objects.toString(sysOrigin, "").trim().toUpperCase();
+  }
+
+  private Long defaultRequiredValue(Long requiredValue) {
+    return requiredValue == null ? 0L : requiredValue;
+  }
+
+  private String defaultLevelName(Integer level, String levelName) {
+    if (StringUtils.isNotBlank(levelName)) {
+      return levelName.trim();
+    }
+    return "CP " + level + "级";
+  }
+}

rc-service/rc-service-console/console-application/src/main/java/com/red/circle/console/app/service/app/datav/CountryDashboardServiceImpl.java

@@ -78,18 +78,9 @@ public class CountryDashboardServiceImpl implements CountryDashboardService {
     QueryCondition condition = QueryCondition.of(cmd);
     Map<String, CountryDashboardMetricCO> rows = new LinkedHashMap<>();
 
-    boolean usePrecomputedMetrics = shouldUsePrecomputedMetrics(condition);
-    if (usePrecomputedMetrics) {
-      mergeSql(rows, countryDashboardDAO.listPrecomputedPeriodMetrics(
-          condition.periodType, condition.startTime, condition.endTime,
-          condition.countryKeyword, condition.statTimezone, condition.sysOrigin));
-    } else {
-      mergeLiveSql(rows, condition);
-      mergeBankSalaryExchange(rows, condition);
-    }
-    if (usePrecomputedMetrics && condition.shouldOverlayLiveSalaryExchange()) {
-      overlayLiveSalaryExchange(rows, condition);
-    }
+    mergeSql(rows, countryDashboardDAO.listPrecomputedPeriodMetrics(
+        condition.periodType, condition.startTime, condition.endTime,
+        condition.countryKeyword, condition.statTimezone, condition.sysOrigin));
 
     mergeMongo(rows, condition, listGiftConsume(condition),
         (row, amount) -> row.setGiftConsume(add(row.getGiftConsume(), amount)));
@@ -303,29 +294,15 @@ public class CountryDashboardServiceImpl implements CountryDashboardService {
   @Override
   public PageResult<CountryDashboardRechargeDetailCO> pageRechargeDetails(
       CountryDashboardRechargeDetailQryCmd cmd) {
-    QueryCondition condition = QueryCondition.of(cmd);
     int cursor = Math.max(1, cmd == null || cmd.getCursor() == null ? 1 : cmd.getCursor());
     int limit = Math.max(1, cmd == null || cmd.getLimit() == null ? 100 : cmd.getLimit());
     limit = Math.min(100, limit);
-    int offset = (cursor - 1) * limit;
-    String countryCodes = normalizeCountryCodes(cmd == null ? null : cmd.getCountryCodes());
-
-    Long total = countryDashboardDAO.countRechargeDetails(
-        condition.startTime, condition.endTime, condition.countryKeyword, countryCodes,
-        condition.sysOrigin, condition.storageTimezoneOffset, condition.statTimezoneOffset);
-    List<CountryDashboardRechargeDetailCO> records = countryDashboardDAO.listRechargeDetails(
-            condition.startTime, condition.endTime, condition.countryKeyword, countryCodes,
-            condition.sysOrigin, condition.storageTimezoneOffset, condition.statTimezoneOffset,
-            offset, limit)
-        .stream()
-        .map(this::toRechargeDetailCO)
-        .toList();
 
     PageResult<CountryDashboardRechargeDetailCO> result = new PageResult<>();
     result.setCurrent(cursor);
     result.setSize(limit);
-    result.setTotal(total == null ? 0L : total);
-    result.setRecords(records);
+    result.setTotal(0L);
+    result.setRecords(List.of());
     return result;
   }
 

rc-service/rc-service-console/console-client/src/main/java/com/red/circle/console/app/dto/clienobject/app/activity/cp/ResidentCpConfigCO.java

@@ -0,0 +1,27 @@
+package com.red.circle.console.app.dto.clienobject.app.activity.cp;
+
+import com.red.circle.framework.dto.ClientObject;
+import java.io.Serial;
+import java.util.ArrayList;
+import java.util.List;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.experimental.Accessors;
+
+/**
+ * 常驻 CP 配置.
+ */
+@Data
+@Accessors(chain = true)
+@EqualsAndHashCode(callSuper = true)
+public class ResidentCpConfigCO extends ClientObject {
+
+  @Serial
+  private static final long serialVersionUID = 1L;
+
+  private Boolean configured = Boolean.FALSE;
+
+  private String sysOrigin;
+
+  private List<ResidentCpLevelConfigCO> levels = new ArrayList<>();
+}

rc-service/rc-service-console/console-client/src/main/java/com/red/circle/console/app/dto/clienobject/app/activity/cp/ResidentCpLevelConfigCO.java

@@ -0,0 +1,31 @@
+package com.red.circle.console.app.dto.clienobject.app.activity.cp;
+
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import com.fasterxml.jackson.databind.ser.std.ToStringSerializer;
+import com.red.circle.framework.dto.ClientObject;
+import java.io.Serial;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.experimental.Accessors;
+
+/**
+ * 常驻 CP 等级配置.
+ */
+@Data
+@Accessors(chain = true)
+@EqualsAndHashCode(callSuper = true)
+public class ResidentCpLevelConfigCO extends ClientObject {
+
+  @Serial
+  private static final long serialVersionUID = 1L;
+
+  private Integer level;
+
+  private String levelName;
+
+  @JsonSerialize(using = ToStringSerializer.class)
+  private Long requiredIntimacyValue;
+
+  @JsonSerialize(using = ToStringSerializer.class)
+  private Long cabinConfigId;
+}

rc-service/rc-service-console/console-client/src/main/java/com/red/circle/console/app/dto/cmd/app/activity/cp/ResidentCpConfigSaveCmd.java

@@ -0,0 +1,28 @@
+package com.red.circle.console.app.dto.cmd.app.activity.cp;
+
+import jakarta.validation.Valid;
+import jakarta.validation.constraints.NotBlank;
+import jakarta.validation.constraints.NotEmpty;
+import java.io.Serial;
+import java.io.Serializable;
+import java.util.List;
+import lombok.Data;
+import lombok.experimental.Accessors;
+
+/**
+ * 常驻 CP 配置保存命令.
+ */
+@Data
+@Accessors(chain = true)
+public class ResidentCpConfigSaveCmd implements Serializable {
+
+  @Serial
+  private static final long serialVersionUID = 1L;
+
+  @NotBlank(message = "sysOrigin required.")
+  private String sysOrigin;
+
+  @Valid
+  @NotEmpty(message = "levels required.")
+  private List<ResidentCpLevelConfigSaveCmd> levels;
+}

rc-service/rc-service-console/console-client/src/main/java/com/red/circle/console/app/dto/cmd/app/activity/cp/ResidentCpLevelConfigSaveCmd.java

@@ -0,0 +1,32 @@
+package com.red.circle.console.app.dto.cmd.app.activity.cp;
+
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import com.fasterxml.jackson.databind.ser.std.ToStringSerializer;
+import jakarta.validation.constraints.NotNull;
+import java.io.Serial;
+import java.io.Serializable;
+import lombok.Data;
+import lombok.experimental.Accessors;
+
+/**
+ * 常驻 CP 等级配置保存命令.
+ */
+@Data
+@Accessors(chain = true)
+public class ResidentCpLevelConfigSaveCmd implements Serializable {
+
+  @Serial
+  private static final long serialVersionUID = 1L;
+
+  @NotNull(message = "level required.")
+  private Integer level;
+
+  private String levelName;
+
+  @NotNull(message = "requiredIntimacyValue required.")
+  @JsonSerialize(using = ToStringSerializer.class)
+  private Long requiredIntimacyValue;
+
+  @JsonSerialize(using = ToStringSerializer.class)
+  private Long cabinConfigId;
+}

rc-service/rc-service-console/console-client/src/main/java/com/red/circle/console/app/service/app/activity/ResidentCpActivityService.java

@@ -0,0 +1,14 @@
+package com.red.circle.console.app.service.app.activity;
+
+import com.red.circle.console.app.dto.clienobject.app.activity.cp.ResidentCpConfigCO;
+import com.red.circle.console.app.dto.cmd.app.activity.cp.ResidentCpConfigSaveCmd;
+
+/**
+ * 常驻 CP 配置管理.
+ */
+public interface ResidentCpActivityService {
+
+  ResidentCpConfigCO getConfig(String sysOrigin);
+
+  void saveConfig(ResidentCpConfigSaveCmd cmd);
+}

rc-service/rc-service-order/order-application/src/main/java/com/red/circle/order/app/common/InAppPurchaseCommon.java

@@ -8,8 +8,9 @@ import com.red.circle.framework.core.asserts.ResponseAssert;
 import com.red.circle.framework.core.response.CommonErrorCode;
 import com.red.circle.framework.dto.ResultResponse;
 import com.red.circle.framework.web.props.EnvProperties;
-import com.red.circle.mq.business.model.event.pay.BuySuccessEvent;
-import com.red.circle.order.app.command.pay.web.strategy.PayEnv;
+import com.red.circle.mq.business.model.event.pay.BuySuccessEvent;
+import com.red.circle.order.app.common.task.RechargeSuccessEventPublisher;
+import com.red.circle.order.app.command.pay.web.strategy.PayEnv;
 import com.red.circle.order.domain.gateway.UserRechargeCountGateway;
 import com.red.circle.order.infra.database.mongo.order.InAppPurchaseCollectionReceiptService;
 import com.red.circle.order.infra.database.mongo.order.InAppPurchaseDetailsService;
@@ -76,9 +77,10 @@ public class InAppPurchaseCommon {
   private final OrderPurchaseHistoryService orderPurchaseHistoryService;
   private final InAppPurchaseCollectionReceiptService inAppPurchaseCollectionReceiptService;
   private final UserLevelClient userLevelClient;
-  private final PropsActivityClient propsActivityClient;
+  private final PropsActivityClient propsActivityClient;
   private final UserOneTimeTaskClient userOneTimeTaskClient;
   private final UserProfileClient userProfileClient;
+  private final RechargeSuccessEventPublisher rechargeSuccessEventPublisher;
 
   public void inAppPurchaseReceipt(InAppPurchaseDetails order) {
     inAppPurchaseReceipt(order,
@@ -192,13 +194,15 @@ public class InAppPurchaseCommon {
     userSvipClient.incrIntegral(order.getAcceptUserId(),
         amount.longValue());
 
-    // 被邀请人充值邀请人获得佣金
-    inviteUserClient.processRechargeCommission(amount,
-        order.getSysOrigin(), order.getAcceptUserId());
-
-      // 发送首充奖励
-      completedFirstCharge(order.getAcceptUserId(), order.getSysOrigin());
-  }
+    // 被邀请人充值邀请人获得佣金
+    inviteUserClient.processRechargeCommission(amount,
+        order.getSysOrigin(), order.getAcceptUserId());
+
+    rechargeSuccessEventPublisher.reportPaymentRecharge(order);
+
+      // 发送首充奖励
+      completedFirstCharge(order.getAcceptUserId(), order.getSysOrigin());
+  }
 
     private void completedFirstCharge(Long userId, String sysOrigin) {
         UserProfileDTO userProfileDTO = userProfileClient.getByUserId(userId).getBody();

rc-service/rc-service-order/order-application/src/main/java/com/red/circle/order/app/common/task/RechargeSuccessEventPublisher.java

@@ -0,0 +1,214 @@
+package com.red.circle.order.app.common.task;
+
+import com.red.circle.component.mq.MessageEvent;
+import com.red.circle.mq.business.model.event.pay.BuySuccessEvent;
+import com.red.circle.mq.business.model.event.pay.RechargeSuccessEvent;
+import com.red.circle.mq.rocket.business.service.MessageSenderService;
+import com.red.circle.mq.rocket.business.streams.RechargeSuccessSink;
+import com.red.circle.order.infra.database.mongo.order.entity.InAppPurchaseDetails;
+import com.red.circle.order.infra.database.mongo.order.entity.assist.InAppPurchaseProduct;
+import com.red.circle.order.inner.model.enums.PayApplicationCommodityType;
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.sql.Timestamp;
+import java.time.Instant;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+public class RechargeSuccessEventPublisher {
+
+  private static final String GOOGLE = "GOOGLE";
+  private static final String APPLE = "APPLE";
+  private static final String THIRD_PARTY = "THIRD_PARTY";
+  private static final String UNKNOWN = "UNKNOWN";
+
+  private final ObjectProvider<MessageSenderService> messageSenderServiceProvider;
+
+  @Value("${recharge.success.mq.enabled:true}")
+  private boolean mqEnabled;
+
+  @Value("${recharge.success.mq.topic:RC_DEFAULT_APP_ORDINARY}")
+  private String mqTopic;
+
+  @Value("${recharge.success.mq.tag:" + RechargeSuccessSink.TAG + "}")
+  private String mqTag;
+
+  public RechargeSuccessEventPublisher(ObjectProvider<MessageSenderService> messageSenderServiceProvider) {
+    this.messageSenderServiceProvider = messageSenderServiceProvider;
+  }
+
+  public void reportAppPurchaseRecharge(BuySuccessEvent event) {
+    if (Objects.isNull(event)) {
+      return;
+    }
+    Long userId = event.getAcceptUserId();
+    Long purchaseHistoryId = event.getPurchaseHistoryId();
+    BigDecimal amountUsd = Objects.nonNull(event.getProductConfig())
+        ? event.getProductConfig().getUnitPrice()
+        : null;
+    Long amountCents = toCents(amountUsd);
+    if (Objects.isNull(userId) || Objects.isNull(purchaseHistoryId) || !positive(amountCents)) {
+      return;
+    }
+
+    Map<String, Object> payload = new HashMap<>();
+    payload.put("payerUserId", event.getUserId());
+    payload.put("acceptUserId", userId);
+    payload.put("purchaseHistoryId", purchaseHistoryId);
+    payload.put("candyQuantity", event.getCandyQuantity());
+    payload.put("platform", event.getPlatform());
+    payload.put("payPlatform", event.getPayPlatform());
+
+    String sourceOrderId = String.valueOf(purchaseHistoryId);
+    String payPlatform = normalizePayPlatform(defaultIfBlank(event.getPayPlatform(), event.getPlatform()));
+    String paymentMethod = classifyPaymentMethod(payPlatform);
+    publish(new RechargeSuccessEvent()
+        .setEventId("RECHARGE_SUCCESS:" + paymentMethod + ":" + sourceOrderId)
+        .setSysOrigin(event.getSysOrigin())
+        .setUserId(userId)
+        .setAmountCents(amountCents)
+        .setAmountUsd(formatAmount(amountUsd))
+        .setCurrency("USD")
+        .setPayPlatform(payPlatform)
+        .setPaymentMethod(paymentMethod)
+        .setSourceOrderId(sourceOrderId)
+        .setOrderId(sourceOrderId)
+        .setOccurredAt(toInstantString(event.getBuyDateTime()))
+        .setPayload(payload));
+  }
+
+  public void reportPaymentRecharge(InAppPurchaseDetails order) {
+    if (Objects.isNull(order) || !order.receiptTypeEqPayment()) {
+      return;
+    }
+    InAppPurchaseProduct product = order.firstProduct();
+    if (Objects.isNull(product) || !PayApplicationCommodityType.GOLD.eq(product.getCode())) {
+      return;
+    }
+    String payPlatform = normalizePayPlatform(Objects.nonNull(order.getFactory()) ? order.getFactory().getFactoryCode() : "");
+    String paymentMethod = classifyPaymentMethod(payPlatform);
+
+    BigDecimal amountUsd = firstPositive(order.getAmountUsd(), product.getAmountUsd());
+    Long amountCents = toCents(amountUsd);
+    if (Objects.isNull(order.getAcceptUserId()) || isBlank(order.getId()) || !positive(amountCents)) {
+      return;
+    }
+
+    Map<String, Object> payload = new HashMap<>();
+    payload.put("trackId", order.getTrackId());
+    payload.put("sourceOrderId", order.getId());
+    payload.put("orderId", order.getOrderId());
+    payload.put("factoryCode", payPlatform);
+    payload.put("factoryChannelCode", Objects.nonNull(order.getFactory()) ? order.getFactory().getFactoryChannelCode() : null);
+    payload.put("platform", Objects.nonNull(order.getFactory()) ? order.getFactory().getPlatform() : null);
+    payload.put("productCode", product.getCode());
+    payload.put("productContent", product.getContent());
+
+    publish(new RechargeSuccessEvent()
+        .setEventId("RECHARGE_SUCCESS:" + paymentMethod + ":" + payPlatform + ":" + order.getId())
+        .setSysOrigin(order.getSysOrigin())
+        .setUserId(order.getAcceptUserId())
+        .setAmountCents(amountCents)
+        .setAmountUsd(formatAmount(amountUsd))
+        .setCurrency(defaultIfBlank(order.getCurrency(), "USD"))
+        .setPayPlatform(payPlatform)
+        .setPaymentMethod(paymentMethod)
+        .setSourceOrderId(order.getId())
+        .setOrderId(defaultIfBlank(order.getOrderId(), order.getId()))
+        .setOccurredAt(firstTime(order.getPurchaseDateMs(), order.getUpdateTime(), order.getCreateTime()))
+        .setPayload(payload));
+  }
+
+  private void publish(RechargeSuccessEvent event) {
+    if (!mqEnabled) {
+      return;
+    }
+    try {
+      MessageSenderService senderService = messageSenderServiceProvider.getIfAvailable();
+      if (Objects.isNull(senderService)) {
+        log.warn("Recharge success mq sender missing, skip event report. eventId={}", event.getEventId());
+        return;
+      }
+      senderService.sendEventMessage(MessageEvent.builder()
+          .consumeId(event.getEventId())
+          .topicString(defaultIfBlank(mqTopic, "RC_DEFAULT_APP_ORDINARY"))
+          .tag(defaultIfBlank(mqTag, RechargeSuccessSink.TAG))
+          .body(event)
+          .build());
+    } catch (Exception e) {
+      log.warn("Recharge success mq report error. eventId={}", event.getEventId(), e);
+    }
+  }
+
+  private String normalizePayPlatform(String value) {
+    String normalized = Objects.toString(value, "").trim().toUpperCase();
+    return normalized.isEmpty() ? UNKNOWN : normalized;
+  }
+
+  private String classifyPaymentMethod(String payPlatform) {
+    if (Objects.equals(GOOGLE, payPlatform)) {
+      return GOOGLE;
+    }
+    if (Objects.equals(APPLE, payPlatform)) {
+      return APPLE;
+    }
+    if (Objects.equals(UNKNOWN, payPlatform)) {
+      return UNKNOWN;
+    }
+    return THIRD_PARTY;
+  }
+
+  private Long toCents(BigDecimal amount) {
+    if (Objects.isNull(amount) || amount.compareTo(BigDecimal.ZERO) <= 0) {
+      return null;
+    }
+    return amount.multiply(BigDecimal.valueOf(100)).setScale(0, RoundingMode.HALF_UP).longValue();
+  }
+
+  private BigDecimal firstPositive(BigDecimal first, BigDecimal second) {
+    if (Objects.nonNull(first) && first.compareTo(BigDecimal.ZERO) > 0) {
+      return first;
+    }
+    return second;
+  }
+
+  private boolean positive(Long value) {
+    return Objects.nonNull(value) && value > 0;
+  }
+
+  private String toInstantString(Timestamp timestamp) {
+    return Objects.nonNull(timestamp) ? timestamp.toInstant().toString() : Instant.now().toString();
+  }
+
+  private String firstTime(Timestamp first, Timestamp second, Timestamp third) {
+    if (Objects.nonNull(first)) {
+      return first.toInstant().toString();
+    }
+    if (Objects.nonNull(second)) {
+      return second.toInstant().toString();
+    }
+    if (Objects.nonNull(third)) {
+      return third.toInstant().toString();
+    }
+    return Instant.now().toString();
+  }
+
+  private String formatAmount(BigDecimal amount) {
+    return Objects.isNull(amount) ? null : amount.stripTrailingZeros().toPlainString();
+  }
+
+  private String defaultIfBlank(String value, String fallback) {
+    return isBlank(value) ? fallback : value.trim();
+  }
+
+  private boolean isBlank(String value) {
+    return Objects.toString(value, "").trim().isEmpty();
+  }
+}

rc-service/rc-service-order/order-application/src/main/java/com/red/circle/order/app/listener/BuySuccessListener.java

@@ -17,6 +17,7 @@ import com.red.circle.framework.dto.ResultResponse;
 import com.red.circle.mq.business.model.event.pay.BuySuccessEvent;
 import com.red.circle.mq.business.model.event.pay.PurchaseProductEvent;
 import com.red.circle.mq.rocket.business.streams.BuySuccessSink;
+import com.red.circle.order.app.common.task.RechargeSuccessEventPublisher;
 import com.red.circle.order.app.common.task.TaskCenterGoClient;
 import com.red.circle.other.inner.endpoint.activity.CumulativeRechargeClient;
 import com.red.circle.other.inner.endpoint.activity.PropsActivityClient;
@@ -71,6 +72,7 @@ public class BuySuccessListener implements MessageListener {
   private final CumulativeRechargeClient cumulativeRechargeClient;
   private final RedisService redisService;
   private final TaskCenterGoClient taskCenterGoClient;
+  private final RechargeSuccessEventPublisher rechargeSuccessEventPublisher;
 
   @Override
   public Action consume(ConsumerMessage message) {
@@ -133,6 +135,7 @@ public class BuySuccessListener implements MessageListener {
         eventBody.getPlatform(),
         eventBody.getPayPlatform(),
         Objects.nonNull(eventBody.getBuyDateTime()) ? eventBody.getBuyDateTime().toInstant() : null);
+    rechargeSuccessEventPublisher.reportAppPurchaseRecharge(eventBody);
 
     userExpandClient.updatePurchasingToTrue(eventBody.getUserId());
 

rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/command/user/ProcessCpApplyCmd.java

@@ -4,9 +4,10 @@ import com.red.circle.common.business.core.enums.SysOriginPlatformEnum;
 import com.red.circle.external.inner.endpoint.message.ImMessageClient;
 import com.red.circle.framework.core.asserts.ResponseAssert;
 import com.red.circle.framework.core.response.CommonErrorCode;
-import com.red.circle.mq.business.model.event.user.CpApplyEvent;
-import com.red.circle.mq.rocket.business.producer.UserMqMessageService;
-import com.red.circle.other.app.dto.cmd.user.relation.cp.ProcessApplyCmd;
+import com.red.circle.mq.business.model.event.user.CpApplyEvent;
+import com.red.circle.mq.rocket.business.producer.UserMqMessageService;
+import com.red.circle.other.app.common.cp.CpRelationBroadcastMqClient;
+import com.red.circle.other.app.dto.cmd.user.relation.cp.ProcessApplyCmd;
 import com.red.circle.other.domain.gateway.user.UserProfileGateway;
 import com.red.circle.other.infra.database.cache.service.cp.CpRelationshipCacheService;
 import com.red.circle.other.infra.database.rds.entity.user.user.CpApply;
@@ -47,9 +48,10 @@ public class ProcessCpApplyCmd {
   private final ImMessageClient imMessageClient;
   private final WalletGoldClient walletGoldClient;
   private final UserMqMessageService userMqMessageService;
-  private final CpRelationshipService cpRelationshipService;
-  private final UserProfileGateway userProfileGateway;
-  private final CpRelationshipCacheService cpRelationshipCacheService;
+  private final CpRelationshipService cpRelationshipService;
+  private final UserProfileGateway userProfileGateway;
+  private final CpRelationshipCacheService cpRelationshipCacheService;
+  private final CpRelationBroadcastMqClient cpRelationBroadcastMqClient;
 
   public void execute(ProcessApplyCmd cmd) {
     process(cmd);
@@ -106,6 +108,8 @@ public class ProcessCpApplyCmd {
     refuseStatusWaitIfLimitReached(cmd, cpApply, relationType);
 
     cpRelationshipCacheService.remove(Arrays.asList(cpApply.getSendApplyUserId(), cpApply.getAcceptApplyUserId()));
+    cpRelationBroadcastMqClient.publish(cmd.requireReqSysOrigin(), cpApply.getId(),
+        cpApply.getSendApplyUserId(), cpApply.getAcceptApplyUserId(), relationType);
   }
 
   private void refuseStatusWaitIfLimitReached(ProcessApplyCmd cmd, CpApply cpApply,

rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/command/user/query/UserCpRightsConfigQueryExe.java

@@ -27,6 +27,7 @@ import org.springframework.stereotype.Component;
 public class UserCpRightsConfigQueryExe {
 
   private static final String RELATIONSHIP_STATUS_NONE = "NONE";
+  private static final int CP_RIGHTS_LEVEL_COUNT = 5;
 
   private final CpCabinConfigService cpCabinConfigService;
   private final CpRelationshipService cpRelationshipService;
@@ -37,7 +38,7 @@ public class UserCpRightsConfigQueryExe {
     CpRelationship relationship = findRelationship(cmd.getReqUserId(), cmd.getCpUserId(),
         relationType);
     Long intimacyValue = getIntimacyValue(relationship);
-    List<CpCabinConfig> configs = cpCabinConfigService.getCpCabinConfig(
+    List<CpCabinConfig> configs = cpCabinConfigService.getCpRightsLevelConfig(
         cmd.getReqSysOrigin().getOrigin());
 
     return CpRightsConfigCO.builder()
@@ -91,15 +92,23 @@ public class UserCpRightsConfigQueryExe {
 
   private List<CpRightsConfigCO.LevelCO> toLevels(String relationType, List<CpCabinConfig> configs,
       boolean hasRelationship, Long intimacyValue) {
-    if (Objects.isNull(configs) || configs.isEmpty()) {
-      return Collections.emptyList();
-    }
-    return IntStream.range(0, configs.size())
-        .mapToObj(index -> toLevel(relationType, configs.get(index), index + 1, hasRelationship,
+    List<CpCabinConfig> resolvedConfigs = Objects.isNull(configs)
+        ? Collections.emptyList()
+        : configs;
+    return IntStream.range(0, CP_RIGHTS_LEVEL_COUNT)
+        .mapToObj(index -> toLevel(relationType, getLevelConfig(resolvedConfigs, index),
+            index + 1, hasRelationship,
             intimacyValue))
         .toList();
   }
 
+  private CpCabinConfig getLevelConfig(List<CpCabinConfig> configs, int index) {
+    if (index < configs.size()) {
+      return configs.get(index);
+    }
+    return new CpCabinConfig().setCabinUnlockValue(0L);
+  }
+
   private CpRightsConfigCO.LevelCO toLevel(String relationType, CpCabinConfig config, int level,
       boolean hasRelationship, Long intimacyValue) {
     Long requiredIntimacyValue = Optional.ofNullable(config.getCabinUnlockValue()).orElse(0L);

rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/common/cp/CpRelationBroadcastMqClient.java

@@ -0,0 +1,137 @@
+package com.red.circle.other.app.common.cp;
+
+import com.red.circle.component.mq.MessageEvent;
+import com.red.circle.mq.rocket.business.service.MessageSenderService;
+import com.red.circle.other.domain.gateway.user.UserProfileGateway;
+import com.red.circle.other.domain.model.user.UserProfile;
+import com.red.circle.other.infra.enums.user.user.CpRelationshipType;
+import java.time.Instant;
+import java.util.Objects;
+import lombok.Data;
+import lombok.RequiredArgsConstructor;
+import lombok.experimental.Accessors;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class CpRelationBroadcastMqClient {
+
+  private static final String MQ_TOPIC = "RC_DEFAULT_APP_ORDINARY";
+  private static final String MQ_TAG = "cp_relation_broadcast";
+
+  private final ObjectProvider<MessageSenderService> messageSenderServiceProvider;
+  private final UserProfileGateway userProfileGateway;
+
+  public void publish(String sysOrigin, Long applyId, Long userId, Long cpUserId, String relationType) {
+    if (Objects.isNull(userId) || Objects.isNull(cpUserId)) {
+      log.warn("CP relation broadcast mq skip: user id missing. applyId={}, userId={}, cpUserId={}",
+          applyId, userId, cpUserId);
+      return;
+    }
+
+    String normalizedRelationType = CpRelationshipType.normalize(relationType);
+    String eventId = buildEventId(applyId, userId, cpUserId, normalizedRelationType);
+    CpRelationBroadcastEvent event = new CpRelationBroadcastEvent()
+        .setSysOrigin(defaultIfBlank(sysOrigin, "LIKEI"))
+        .setEventId(eventId)
+        .setApplyId(applyId)
+        .setRelationType(normalizedRelationType)
+        .setUserId(userId)
+        .setCpUserId(cpUserId)
+        .setSendUserId(userId)
+        .setAcceptUserId(cpUserId)
+        .setOccurredAt(Instant.now().toString())
+        .setUser(loadUser(userId))
+        .setCpUser(loadUser(cpUserId));
+
+    try {
+      MessageSenderService senderService = messageSenderServiceProvider.getIfAvailable();
+      if (Objects.isNull(senderService)) {
+        log.warn("CP relation broadcast mq sender missing, skip. eventId={}", eventId);
+        return;
+      }
+      senderService.sendEventMessage(MessageEvent.builder()
+          .consumeId(eventId)
+          .topicString(MQ_TOPIC)
+          .tag(MQ_TAG)
+          .body(event)
+          .build());
+    } catch (Exception e) {
+      log.warn("CP relation broadcast mq publish error. eventId={}", eventId, e);
+    }
+  }
+
+  private CpRelationUser loadUser(Long userId) {
+    CpRelationUser fallback = new CpRelationUser()
+        .setId(userId)
+        .setUserId(userId)
+        .setAccount(Objects.toString(userId, ""))
+        .setUserNickname(Objects.toString(userId, ""));
+    try {
+      UserProfile profile = userProfileGateway.getByUserId(userId);
+      if (Objects.isNull(profile)) {
+        return fallback;
+      }
+      return new CpRelationUser()
+          .setId(profile.getId())
+          .setUserId(profile.getId())
+          .setAccount(defaultIfBlank(profile.getAccount(), fallback.getAccount()))
+          .setUserNickname(defaultIfBlank(profile.getUserNickname(), fallback.getUserNickname()))
+          .setNickname(defaultIfBlank(profile.getUserNickname(), fallback.getUserNickname()))
+          .setUserAvatar(profile.getUserAvatar())
+          .setAvatar(profile.getUserAvatar())
+          .setCountryCode(profile.getCountryCode())
+          .setCountryName(profile.getCountryName())
+          .setOriginSys(profile.getOriginSys());
+    } catch (Exception e) {
+      log.warn("CP relation broadcast load user profile failed. userId={}", userId, e);
+      return fallback;
+    }
+  }
+
+  private String buildEventId(Long applyId, Long userId, Long cpUserId, String relationType) {
+    if (Objects.nonNull(applyId)) {
+      return "CP_RELATION:" + relationType + ":" + applyId;
+    }
+    return "CP_RELATION:" + relationType + ":" + userId + ":" + cpUserId;
+  }
+
+  private String defaultIfBlank(String value, String fallback) {
+    String normalized = Objects.toString(value, "").trim();
+    return normalized.isEmpty() ? fallback : normalized;
+  }
+
+  @Data
+  @Accessors(chain = true)
+  private static class CpRelationBroadcastEvent {
+    private String sysOrigin;
+    private String eventId;
+    private Long applyId;
+    private String relationType;
+    private Long userId;
+    private Long cpUserId;
+    private Long sendUserId;
+    private Long acceptUserId;
+    private String occurredAt;
+    private CpRelationUser user;
+    private CpRelationUser cpUser;
+  }
+
+  @Data
+  @Accessors(chain = true)
+  private static class CpRelationUser {
+    private Long id;
+    private Long userId;
+    private String account;
+    private String nickname;
+    private String userNickname;
+    private String avatar;
+    private String userAvatar;
+    private String countryCode;
+    private String countryName;
+    private String originSys;
+  }
+}

rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/rds/service/user/user/CpCabinConfigService.java

@@ -12,7 +12,9 @@ import java.util.List;
  *
  * @author zongpubin on 2023-11-20 10:18
  */
-public interface CpCabinConfigService extends BaseService<CpCabinConfig> {
-
-  List<CpCabinConfig> getCpCabinConfig(String sysOriginName);
-}
+public interface CpCabinConfigService extends BaseService<CpCabinConfig> {
+
+  List<CpCabinConfig> getCpCabinConfig(String sysOriginName);
+
+  List<CpCabinConfig> getCpRightsLevelConfig(String sysOriginName);
+}

rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/rds/service/user/user/impl/CpCabinConfigServiceImpl.java

@@ -20,16 +20,27 @@ import org.springframework.stereotype.Service;
 
 @Service
 @AllArgsConstructor
-public class CpCabinConfigServiceImpl extends
-    BaseServiceImpl<CpCabinConfigDAO, CpCabinConfig> implements
-    CpCabinConfigService {
-
-  @Override
-  public List<CpCabinConfig> getCpCabinConfig(String sysOriginName) {
-    return query()
-        .eq(CpCabinConfig::getSysOrigin, sysOriginName)
+public class CpCabinConfigServiceImpl extends
+    BaseServiceImpl<CpCabinConfigDAO, CpCabinConfig> implements
+    CpCabinConfigService {
+
+  private static final int CP_RIGHTS_LEVEL_COUNT = 5;
+
+  @Override
+  public List<CpCabinConfig> getCpCabinConfig(String sysOriginName) {
+    return query()
+        .eq(CpCabinConfig::getSysOrigin, sysOriginName)
         .orderByAsc(CpCabinConfig::getCabinUnlockValue)
-        .last(PageConstant.formatLimit(20))
-        .list();
-  }
-}
+        .last(PageConstant.formatLimit(20))
+        .list();
+  }
+
+  @Override
+  public List<CpCabinConfig> getCpRightsLevelConfig(String sysOriginName) {
+    return query()
+        .eq(CpCabinConfig::getSysOrigin, sysOriginName)
+        .last("ORDER BY sort IS NULL, sort ASC, cabin_unlock_value ASC "
+            + PageConstant.formatLimit(CP_RIGHTS_LEVEL_COUNT))
+        .list();
+  }
+}