hy/yumi-java
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(other): 国家资料级联同步用户与团队冗余快照

Browse Source 
- 新增国家级/用户级 Redis 写锁（token+Lua 安全释放），统一 country -> user 锁序
- 后台改国家码/名称时按 countryId 游标分批同步 RDS 与 mongo 全部国家冗余字段，
  含注册国家码同事务重命名、团队区域重算（CAS + 投影回滚）与两次缓存失效
- 建团（BD 邀请/后台）、改用户资料、注册、编辑团队等写国家快照路径接入锁防竞态，
  团队国家码一律取国家主表标准值
- 违规调整/账号状态改为仅写变更字段，防止旧快照覆盖同步结果
- 团队账单结算改用账单自身 region，配合团队迁区
- 补充 user_base_info (country_id,id) 索引与 mongo 复合索引迁移脚本

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
hy001 2026-07-16 20:11:57 +08:00
parent e956250483
commit c884827072
50 changed files with 1969 additions and 359 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

177
rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/command/team/BdInviteMessageProcessExe.java
View File

@ -14,19 +14,23 @@ import com.red.circle.mq.rocket.business.producer.UserMqMessageService;
import com.red.circle.other.app.convertor.user.UserProfileAppConvertor;
import com.red.circle.other.app.dto.cmd.team.DbInviteMessageProcessCmd;
import com.red.circle.other.app.util.OfficialNoticeUtils;
import com.red.circle.other.domain.gateway.user.UserProfileGateway;
import com.red.circle.other.domain.gateway.user.ability.RegisterDeviceGateway;
import com.red.circle.other.infra.database.mongo.entity.team.bd.BdInviteAgentMessage;
import com.red.circle.other.domain.gateway.user.UserProfileGateway;
import com.red.circle.other.domain.gateway.user.ability.RegisterDeviceGateway;
import com.red.circle.other.infra.common.sys.CountryDataSyncLockService;
import com.red.circle.other.infra.common.user.UserCountryWriteLockService;
import com.red.circle.other.infra.database.mongo.entity.team.bd.BdInviteAgentMessage;
import com.red.circle.other.infra.database.mongo.entity.team.bd.MessageInviteGroup;
import com.red.circle.other.infra.database.mongo.entity.team.team.TeamMember;
import com.red.circle.other.infra.database.mongo.entity.team.team.TeamProfile;
import com.red.circle.other.infra.database.mongo.service.team.bd.BdInviteAgentMessageService;
import com.red.circle.other.infra.database.mongo.service.team.team.TeamBillCycleService;
import com.red.circle.other.infra.database.mongo.service.team.team.TeamMemberService;
import com.red.circle.other.infra.database.mongo.service.team.team.TeamProfileService;
import com.red.circle.other.infra.database.rds.entity.team.BusinessDevelopmentBaseInfo;
import com.red.circle.other.infra.database.rds.entity.team.BusinessDevelopmentTeam;
import com.red.circle.other.infra.database.rds.service.sys.AdministratorService;
import com.red.circle.other.infra.database.mongo.service.team.team.TeamProfileService;
import com.red.circle.other.infra.database.rds.entity.sys.SysCountryCode;
import com.red.circle.other.infra.database.rds.entity.team.BusinessDevelopmentBaseInfo;
import com.red.circle.other.infra.database.rds.entity.team.BusinessDevelopmentTeam;
import com.red.circle.other.infra.database.rds.service.sys.AdministratorService;
import com.red.circle.other.infra.database.rds.service.sys.SysCountryCodeService;
import com.red.circle.other.infra.database.rds.service.team.BusinessDevelopmentBaseInfoService;
import com.red.circle.other.infra.database.rds.service.team.BusinessDevelopmentTeamService;
import com.red.circle.other.infra.database.rds.service.team.UserHistoryIdentityService;
@ -49,8 +53,9 @@ import com.red.circle.tool.core.json.JacksonUtils;
import com.red.circle.tool.core.sequence.IdWorkerUtils;
import com.red.circle.tool.core.text.StringUtils;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.Objects;
import java.util.Set;
import java.util.function.Supplier;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
@ -76,8 +81,11 @@ public class BdInviteMessageProcessExe {
private final BusinessDevelopmentTeamService businessDevelopmentTeamService;
private final BusinessDevelopmentBaseInfoService businessDevelopmentBaseInfoService;
private final UserMqMessageService userMqMessageService;
private final AdministratorService administratorService;
private final TaskMqMessage taskMqMessage;
private final AdministratorService administratorService;
private final TaskMqMessage taskMqMessage;
private final SysCountryCodeService sysCountryCodeService;
private final CountryDataSyncLockService countryDataSyncLockService;
private final UserCountryWriteLockService userCountryWriteLockService;
public void execute(DbInviteMessageProcessCmd cmd) {
@ -100,12 +108,8 @@ public class BdInviteMessageProcessExe {
// 6014 用户正在申请加入其他代理
ResponseAssert.isFalse(TeamErrorCode.USER_APPLY_JOIN_OTHER_AGENT,
teamMemberService.existsMember(cmd.requiredReqUserId()));
ResponseAssert.isTrue(CommonErrorCode.OPERATING_FAILURE,
bdInviteAgentMessageService.updateStatus(cmd.getId(), message.getStatus(),
cmd.getStatus(), message.getGroup()));
if (Objects.equals(cmd.getStatus(), 1)) {
Long teamId = createTeam(cmd, message);
if (Objects.equals(cmd.getStatus(), 1)) {
Long teamId = createTeam(cmd, message);
// 主播设备指纹校验
Boolean hasAnchorWithSameDevice = registerDeviceGateway.checkDeviceFingerprintHasAnchor(
@ -128,10 +132,13 @@ public class BdInviteMessageProcessExe {
userMqMessageService.sendBadgeOperateEvent(message.getInviteUserId(), BadgeKeyEnum.HOST.name(), BadgeOperateEvent.OperationType.WEAR.name());
sendMessage(message, OfficialNoticeTypeEnum.AGREE_BD_INVITE_AGENT);
return;
}
sendMessage(message, OfficialNoticeTypeEnum.REFUSE_BD_INVITE_AGENT);
return;
}
ResponseAssert.isTrue(CommonErrorCode.OPERATING_FAILURE,
bdInviteAgentMessageService.updateStatus(cmd.getId(), message.getStatus(),
cmd.getStatus(), message.getGroup()));
sendMessage(message, OfficialNoticeTypeEnum.REFUSE_BD_INVITE_AGENT);
}
private void sendAdminTaskMessage(Long userId) {
@ -156,9 +163,62 @@ public class BdInviteMessageProcessExe {
message.getInviteUserId());
ResponseAssert.isFalse(TeamErrorCode.DEVICE_FINGERPRINT_USED_BY_ANCHOR, hasAnchorWithSameDevice);
UserProfileDTO userProfile = userProfileAppConvertor.toUserProfileDTO(
userProfileGateway.getByUserId(message.getInviteUserId()));
TeamProfile teamProfile = new TeamProfile()
UserProfileDTO preflightUser = userProfileAppConvertor.toUserProfileDTO(
userProfileGateway.getByUserId(message.getInviteUserId()));
ResponseAssert.notNull(UserErrorCode.USER_INFO_NOT_FOUND, preflightUser);
Long expectedCountryId = preflightUser.getCountryId();
// 锁顺序固定为 country -> user，邀请状态与团队全部初始化完成后才释放.
return executeTeamCreationLocks(expectedCountryId, message.getInviteUserId(), () ->
createTeamLocked(cmd, message, bdBaseInfo, expectedCountryId));
}
private Long createTeamLocked(DbInviteMessageProcessCmd cmd, BdInviteAgentMessage message,
BusinessDevelopmentBaseInfo bdBaseInfo, Long expectedCountryId) {
TeamCreationContext creationContext = buildTeamProfile(cmd, message, bdBaseInfo);
ResponseAssert.isTrue(CommonErrorCode.OPERATION_CONFLICT,
Objects.equals(expectedCountryId, creationContext.userProfile().getCountryId()));
// 接受状态必须在取得国家写锁后更新，锁冲突时保留待处理状态供安全重试.
ResponseAssert.isTrue(CommonErrorCode.OPERATING_FAILURE,
bdInviteAgentMessageService.updateStatus(cmd.getId(), message.getStatus(),
cmd.getStatus(), message.getGroup()));
teamProfileService.create(creationContext.teamProfile());
UserProfileDTO userProfile = creationContext.userProfile();
TeamProfile teamProfile = creationContext.teamProfile();
// 初始化账单
teamBillCycleService.createIfAbsent(
new TeamBillCmd()
.setSysOrigin(teamProfile.getSysOrigin())
.setTeamId(teamProfile.getId())
.setRegion(teamProfile.getRegion())
.setOperationTime(TimestampUtils.now())
.setOperationBackUser(teamProfile.getCreateUser())
);
teamMemberService.addMember(new TeamMember()
.setSortId(IdWorkerUtils.getId())
.setSysOrigin(teamProfile.getSysOrigin())
.setMemberId(userProfile.getId())
.setTeamId(teamProfile.getId())
.setRole(TeamMemberRole.OWN)
.setRemarks("")
.setCreateTime(TimestampUtils.now())
.setCreateUserOrigin(1)
.setUpdateTime(TimestampUtils.now())
.setUpdateUserOrigin(1)
.setActiveTime(TimestampUtils.now())
);
userHistoryIdentityService.saveHost(userProfile.getId());
return teamProfile.getId();
}
private TeamCreationContext buildTeamProfile(DbInviteMessageProcessCmd cmd,
BdInviteAgentMessage message, BusinessDevelopmentBaseInfo bdBaseInfo) {
UserProfileDTO userProfile = userProfileAppConvertor.toUserProfileDTO(
userProfileGateway.getByUserId(message.getInviteUserId()));
TeamProfile teamProfile = new TeamProfile()
.setId(IdWorkerUtils.getId())
.setRegion(bdBaseInfo.getRegion())
.setSysOrigin(cmd.requireReqSysOrigin())
@ -168,11 +228,7 @@ public class BdInviteMessageProcessExe {
.setSetting(new TeamSetting()
.setMaxMember(1000)
)
.setCountry(new TeamCountry()
.setCountryId(userProfile.getCountryId())
.setCountryCode(userProfile.getCountryCode())
.setCountryName(userProfile.getCountryName())
)
.setCountry(resolveTeamCountry(userProfile))
.setCounter(new TeamCounter()
.setMemberQuantity(1L)
.setAdminQuantity(0L)
@ -192,39 +248,36 @@ public class BdInviteMessageProcessExe {
.setCreateTime(TimestampUtils.now())
.setUpdateTime(TimestampUtils.now())
.setCreateUser(cmd.requiredReqUserId())
.setCreateUserOrigin(OperationUserOrigin.APP.getValue())
.setUpdateUser(cmd.requiredReqUserId())
.setUpdateUserOrigin(OperationUserOrigin.APP.getValue());
teamProfileService.create(teamProfile);
// 初始化账单
teamBillCycleService.createIfAbsent(
new TeamBillCmd()
.setSysOrigin(teamProfile.getSysOrigin())
.setTeamId(teamProfile.getId())
.setRegion(teamProfile.getRegion())
.setOperationTime(TimestampUtils.now())
.setOperationBackUser(teamProfile.getCreateUser())
);
teamMemberService.addMember(new TeamMember()
.setSortId(IdWorkerUtils.getId())
.setSysOrigin(teamProfile.getSysOrigin())
.setMemberId(userProfile.getId())
.setTeamId(teamProfile.getId())
.setRole(TeamMemberRole.OWN)
.setRemarks("")
.setCreateTime(TimestampUtils.now())
.setCreateUserOrigin(1)
.setUpdateTime(TimestampUtils.now())
.setUpdateUserOrigin(1)
.setActiveTime(TimestampUtils.now())
);
userHistoryIdentityService.saveHost(userProfile.getId());
return teamProfile.getId();
}
.setCreateUserOrigin(OperationUserOrigin.APP.getValue())
.setUpdateUser(cmd.requiredReqUserId())
.setUpdateUserOrigin(OperationUserOrigin.APP.getValue());
return new TeamCreationContext(userProfile, teamProfile);
}
private TeamCountry resolveTeamCountry(UserProfileDTO userProfile) {
if (userProfile.getCountryId() == null) {
return new TeamCountry()
.setCountryCode(userProfile.getCountryCode())
.setCountryName(userProfile.getCountryName());
}
// countryId 是稳定归属，团队冗余码与名称始终取国家主表的最新标准值.
SysCountryCode country = sysCountryCodeService.getById(userProfile.getCountryId());
ResponseAssert.notNull(CommonErrorCode.NOT_FOUND_RECORD_INFO, country);
return new TeamCountry()
.setCountryId(country.getId())
.setCountryCode(country.getAlphaTwo())
.setCountryName(country.getCountryName());
}
private <T> T executeTeamCreationLocks(Long countryId, Long ownUserId, Supplier<T> action) {
Supplier<T> userLocked = () -> userCountryWriteLockService.execute(
ownUserId == null ? List.of() : List.of(ownUserId), action);
return countryId == null ? userLocked.get()
: countryDataSyncLockService.execute(countryId, userLocked);
}
private record TeamCreationContext(UserProfileDTO userProfile, TeamProfile teamProfile) {
}
private void sendMessage(BdInviteAgentMessage message, OfficialNoticeTypeEnum noticeTypeEnum) {
UserProfileDTO userProfile = userProfileAppConvertor.toUserProfileDTO(

40
rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/command/user/UpdateUserProfileCmdExe.java
View File

@ -13,9 +13,11 @@ import com.red.circle.other.app.convertor.user.UserProfileAppConvertor;
import com.red.circle.other.app.dto.cmd.user.user.UserProfileModifyCmd;
import com.red.circle.other.domain.gateway.approval.ProfileApprovalGateway;
import com.red.circle.other.domain.gateway.user.UserProfileGateway;
import com.red.circle.other.domain.model.approval.ProfileApprovalContent;
import com.red.circle.other.domain.model.user.UserProfile;
import com.red.circle.other.infra.database.cache.service.cp.CpRelationshipCacheService;
import com.red.circle.other.domain.model.approval.ProfileApprovalContent;
import com.red.circle.other.domain.model.user.UserProfile;
import com.red.circle.other.infra.common.sys.CountryDataSyncLockService;
import com.red.circle.other.infra.common.user.UserCountryWriteLockService;
import com.red.circle.other.infra.database.cache.service.cp.CpRelationshipCacheService;
import com.red.circle.other.infra.database.mongo.service.live.RoomProfileManagerService;
import com.red.circle.other.infra.database.rds.entity.sys.SysCountryCode;
import com.red.circle.other.infra.database.rds.entity.user.user.BaseInfo;
@ -57,11 +59,33 @@ public class UpdateUserProfileCmdExe {
private final SysCountryCodeService sysCountryCodeService;
private final ProfileApprovalGateway profileApprovalGateway;
private final UserProfileAppConvertor userProfileAppConvertor;
private final RoomProfileManagerService roomProfileManagerService;
private final CpRelationshipCacheService cpRelationshipCacheService;
public UserProfileDTO execute(UserProfileModifyCmd cmd) {
//敏感词
private final RoomProfileManagerService roomProfileManagerService;
private final CpRelationshipCacheService cpRelationshipCacheService;
private final CountryDataSyncLockService countryDataSyncLockService;
private final UserCountryWriteLockService userCountryWriteLockService;
public UserProfileDTO execute(UserProfileModifyCmd cmd) {
if (cmd.getCountryId() == null) {
return executeInternal(cmd);
}
BaseInfo preflight = baseInfoService.getById(cmd.requiredReqUserId());
ResponseAssert.notNull(UserErrorCode.USER_INFO_NOT_FOUND, preflight);
Long sourceCountryId = preflight.getCountryId();
// 跨国修改按 country -> user 持锁，避免用户在目标国家游标扫过后迁入而漏掉运行投影.
return countryDataSyncLockService.execute(
Arrays.asList(sourceCountryId, cmd.getCountryId()),
() -> userCountryWriteLockService.execute(
Collections.singletonList(cmd.requiredReqUserId()), () -> {
BaseInfo lockedUser = baseInfoService.getById(cmd.requiredReqUserId());
ResponseAssert.notNull(UserErrorCode.USER_INFO_NOT_FOUND, lockedUser);
ResponseAssert.isTrue(CommonErrorCode.OPERATION_CONFLICT,
Objects.equals(sourceCountryId, lockedUser.getCountryId()));
return executeInternal(cmd);
}));
}
private UserProfileDTO executeInternal(UserProfileModifyCmd cmd) {
//敏感词
if(ObjectUtil.isNotEmpty(cmd.getUserNickname())){
ResponseAssert.isFalse(DynamicErrorCode.SENSITIVE_WORD_ERROR, SensitiveWordFilter.checkSensitiveWord(cmd.getUserNickname()));
}

29
rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/command/violation/user/UserViolationAdjustCmdExe.java
View File

@ -117,19 +117,22 @@ public class UserViolationAdjustCmdExe {
default:
return null;
}
}
private void updateUserProfile(UserProfile userProfile, ViolationTypeEnum violationType, String adjustedContent) {
switch (violationType) {
case USER_NICKNAME:
userProfile.setUserNickname(adjustedContent);
break;
case USER_AVATAR:
userProfile.setUserAvatar(adjustedContent);
break;
}
userProfileGateway.updateSelectiveById(userProfile);
}
}
private void updateUserProfile(UserProfile userProfile, ViolationTypeEnum violationType, String adjustedContent) {
// 违规处理只提交实际变更字段，避免把此前读取的完整用户国家快照重新写回主表.
UserProfile updateProfile = new UserProfile();
updateProfile.setId(userProfile.getId());
switch (violationType) {
case USER_NICKNAME:
updateProfile.setUserNickname(adjustedContent);
break;
case USER_AVATAR:
updateProfile.setUserAvatar(adjustedContent);
break;
}
userProfileGateway.updateSelectiveById(updateProfile);
}
private void sendAdjustMessage(Long userId, ViolationTypeEnum violationType) {
String content = buildAdjustContent(userId, violationType);

4
rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/listener/team/TeamBillSettleListener.java
View File

@ -154,8 +154,8 @@ public class TeamBillSettleListener implements MessageListener {
List<TeamMemberTarget> teamMemberTargets = getTeamMemberTargets(teamBillCycle);
log.warn("团队账单结算teamMemberTargets."+teamMemberTargets);
// 获得区域配置细信息
SysRegionConfig regionConfig = sysRegionConfigService.getById(teamProfile.getRegion());
// 历史待结算账单必须全程使用账单所属区域，避免团队迁区后混用新区域汇率与旧政策.
SysRegionConfig regionConfig = sysRegionConfigService.getById(teamBillCycle.getRegion());
if (Objects.isNull(regionConfig)) {
log.warn("没有获得团队的区域数据,团队ID:{}", teamProfile.getId());
teamBillCycleService.updateStatus(teamBillCycle.getId(), TeamBillCycleStatus.HANG_UP,

13
rc-service/rc-service-other/other-application/src/test/java/com/red/circle/other/app/command/team/BdInviteMessageProcessExeTest.java
View File

@ -14,12 +14,15 @@ import com.red.circle.other.app.dto.cmd.team.DbInviteMessageProcessCmd;
import com.red.circle.other.domain.gateway.user.UserProfileGateway;
import com.red.circle.other.domain.gateway.user.ability.RegisterDeviceGateway;
import com.red.circle.other.domain.model.user.UserProfile;
import com.red.circle.other.infra.common.sys.CountryDataSyncLockService;
import com.red.circle.other.infra.common.user.UserCountryWriteLockService;
import com.red.circle.other.infra.database.mongo.entity.team.bd.BdInviteAgentMessage;
import com.red.circle.other.infra.database.mongo.service.team.bd.BdInviteAgentMessageService;
import com.red.circle.other.infra.database.mongo.service.team.team.TeamBillCycleService;
import com.red.circle.other.infra.database.mongo.service.team.team.TeamMemberService;
import com.red.circle.other.infra.database.mongo.service.team.team.TeamProfileService;
import com.red.circle.other.infra.database.rds.service.sys.AdministratorService;
import com.red.circle.other.infra.database.rds.service.sys.SysCountryCodeService;
import com.red.circle.other.infra.database.rds.service.team.BusinessDevelopmentBaseInfoService;
import com.red.circle.other.infra.database.rds.service.team.BusinessDevelopmentTeamService;
import com.red.circle.other.infra.database.rds.service.team.UserHistoryIdentityService;
@ -45,6 +48,11 @@ class BdInviteMessageProcessExeTest {
UserMqMessageService userMqMessageService = mock(UserMqMessageService.class);
AdministratorService administratorService = mock(AdministratorService.class);
TaskMqMessage taskMqMessage = mock(TaskMqMessage.class);
SysCountryCodeService sysCountryCodeService = mock(SysCountryCodeService.class);
CountryDataSyncLockService countryDataSyncLockService =
mock(CountryDataSyncLockService.class);
UserCountryWriteLockService userCountryWriteLockService =
mock(UserCountryWriteLockService.class);
BdInviteMessageProcessExe exe = new BdInviteMessageProcessExe(
teamMemberService,
@ -59,7 +67,10 @@ class BdInviteMessageProcessExeTest {
businessDevelopmentBaseInfoService,
userMqMessageService,
administratorService,
taskMqMessage
taskMqMessage,
sysCountryCodeService,
countryDataSyncLockService,
userCountryWriteLockService
);
DbInviteMessageProcessCmd cmd = new DbInviteMessageProcessCmd();

26
rc-service/rc-service-other/other-application/src/test/java/com/red/circle/other/app/command/user/UpdateUserProfileCmdExeTest.java
View File

@ -1,7 +1,10 @@
package com.red.circle.other.app.command.user;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyList;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@ -13,9 +16,12 @@ import com.red.circle.other.app.dto.cmd.user.user.UserProfileModifyCmd;
import com.red.circle.other.domain.gateway.approval.ProfileApprovalGateway;
import com.red.circle.other.domain.gateway.user.UserProfileGateway;
import com.red.circle.other.domain.model.user.UserProfile;
import com.red.circle.other.infra.common.sys.CountryDataSyncLockService;
import com.red.circle.other.infra.common.user.UserCountryWriteLockService;
import com.red.circle.other.infra.database.cache.service.cp.CpRelationshipCacheService;
import com.red.circle.other.infra.database.mongo.service.live.RoomProfileManagerService;
import com.red.circle.other.infra.database.rds.entity.sys.SysCountryCode;
import com.red.circle.other.infra.database.rds.entity.user.user.BaseInfo;
import com.red.circle.other.infra.database.rds.service.sys.SysCountryCodeService;
import com.red.circle.other.infra.database.rds.service.user.user.BaseInfoService;
import com.red.circle.other.inner.model.dto.user.PhotoItem;
@ -23,6 +29,7 @@ import com.red.circle.other.inner.model.dto.user.UserProfileDTO;
import com.red.circle.component.redis.service.RedisService;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
@ -39,6 +46,12 @@ class UpdateUserProfileCmdExeTest {
UserProfileAppConvertor userProfileAppConvertor = mock(UserProfileAppConvertor.class);
RoomProfileManagerService roomProfileManagerService = mock(RoomProfileManagerService.class);
CpRelationshipCacheService cpRelationshipCacheService = mock(CpRelationshipCacheService.class);
CountryDataSyncLockService countryDataSyncLockService =
new CountryDataSyncLockService(redisService);
UserCountryWriteLockService userCountryWriteLockService =
new UserCountryWriteLockService(redisService);
when(redisService.setIfAbsent(anyString(), any(), anyLong(), any(TimeUnit.class)))
.thenReturn(true);
UpdateUserProfileCmdExe exe = new UpdateUserProfileCmdExe(
ossServiceClient,
userProfileGateway,
@ -48,7 +61,9 @@ class UpdateUserProfileCmdExeTest {
profileApprovalGateway,
userProfileAppConvertor,
roomProfileManagerService,
cpRelationshipCacheService
cpRelationshipCacheService,
countryDataSyncLockService,
userCountryWriteLockService
);
UserProfileModifyCmd cmd = new UserProfileModifyCmd();
@ -68,6 +83,7 @@ class UpdateUserProfileCmdExeTest {
.setCountryName("Japan");
when(userProfileGateway.getByUserId(1L)).thenReturn(currentUserProfile);
when(baseInfoService.getById(1L)).thenReturn(new BaseInfo().setId(1L).setCountryId(1L));
when(userProfileAppConvertor.toUserProfile(cmd)).thenReturn(updateUserProfile);
when(sysCountryCodeService.getById(2L)).thenReturn(newCountry);
when(userProfileAppConvertor.toUserProfileDTO(currentUserProfile)).thenReturn(new UserProfileDTO());
@ -94,6 +110,10 @@ class UpdateUserProfileCmdExeTest {
UserProfileAppConvertor userProfileAppConvertor = mock(UserProfileAppConvertor.class);
RoomProfileManagerService roomProfileManagerService = mock(RoomProfileManagerService.class);
CpRelationshipCacheService cpRelationshipCacheService = mock(CpRelationshipCacheService.class);
CountryDataSyncLockService countryDataSyncLockService =
new CountryDataSyncLockService(redisService);
UserCountryWriteLockService userCountryWriteLockService =
new UserCountryWriteLockService(redisService);
UpdateUserProfileCmdExe exe = new UpdateUserProfileCmdExe(
ossServiceClient,
userProfileGateway,
@ -103,7 +123,9 @@ class UpdateUserProfileCmdExeTest {
profileApprovalGateway,
userProfileAppConvertor,
roomProfileManagerService,
cpRelationshipCacheService
cpRelationshipCacheService,
countryDataSyncLockService,
userCountryWriteLockService
);
String backgroundUrl = "https://example.com/profile-bg.jpg";

78
rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/common/sys/CountryDataSyncLockService.java Normal file
View File

@ -0,0 +1,78 @@
package com.red.circle.other.infra.common.sys;
import com.red.circle.component.redis.service.RedisService;
import com.red.circle.framework.core.asserts.ResponseAssert;
import com.red.circle.framework.core.response.CommonErrorCode;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Comparator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
/**
* 国家资料级联全局写锁，防止同一国家同时更新或创建新团队.
*/
@Service
@Slf4j
@RequiredArgsConstructor
public class CountryDataSyncLockService {
private static final int LOCK_SECONDS = 600;
private static final String LOCK_PREFIX = "sys_country_code_sync_lock:v2:";
private static final String RELEASE_LOCK_SCRIPT =
"if redis.call('GET', KEYS[1]) == ARGV[1] then "
+ "return redis.call('DEL', KEYS[1]) "
+ "end "
+ "return 0";
private final RedisService redisService;
/**
* 以唯一 token 持有国家锁，超时后不会误删新持有者的锁.
*/
public <T> T execute(Long countryId, Supplier<T> action) {
return execute(List.of(countryId), action);
}
/**
* 按国家ID排序持有多把锁，用于用户跨国迁移同时保护原国家与目标国家.
*/
public <T> T execute(Collection<Long> countryIds, Supplier<T> action) {
Map<String, String> lockedKeys = new LinkedHashMap<>();
try {
countryIds.stream().filter(Objects::nonNull).distinct().sorted(Comparator.naturalOrder())
.forEach(countryId -> {
String key = key(countryId);
String token = UUID.randomUUID().toString();
ResponseAssert.isTrue(CommonErrorCode.OPERATION_CONFLICT,
redisService.setIfAbsent(key, token, LOCK_SECONDS, TimeUnit.SECONDS));
lockedKeys.put(key, token);
});
return action.get();
} finally {
List<Map.Entry<String, String>> entries = new ArrayList<>(lockedKeys.entrySet());
for (int index = entries.size() - 1; index >= 0; index--) {
try {
Map.Entry<String, String> entry = entries.get(index);
redisService.execute(RELEASE_LOCK_SCRIPT, Long.class, List.of(entry.getKey()),
entry.getValue());
} catch (RuntimeException ex) {
// 解锁故障不应覆盖已成功的业务结果，token锁会在TTL后自动释放.
log.warn("释放国家资料级联写锁失败", ex);
}
}
}
}
private String key(Long countryId) {
return LOCK_PREFIX + countryId;
}
}

61
rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/common/user/UserAccountCommon.java
View File

@ -15,10 +15,12 @@ import com.red.circle.framework.web.util.ValidatorUtils;
import com.red.circle.mq.business.model.event.task.TaskApprovalEvent;
import com.red.circle.mq.rocket.business.producer.TaskMqMessage;
import com.red.circle.other.domain.gateway.user.UserAccountShortProductionGateway;
import com.red.circle.other.domain.gateway.user.UserProfileGateway;
import com.red.circle.other.domain.model.user.UserProfile;
import com.red.circle.other.infra.convertor.user.UserProfileInfraConvertor;
import com.red.circle.other.infra.database.mongo.service.user.profile.UserRunProfileService;
import com.red.circle.other.domain.gateway.user.UserProfileGateway;
import com.red.circle.other.domain.model.user.UserProfile;
import com.red.circle.other.infra.common.sys.CountryDataSyncLockService;
import com.red.circle.other.infra.convertor.user.UserProfileInfraConvertor;
import com.red.circle.other.infra.database.mongo.service.user.profile.UserRunProfileService;
import com.red.circle.other.infra.database.rds.entity.sys.SysCountryCode;
import com.red.circle.other.infra.database.rds.entity.user.user.ApprovalUserAccountStatusLog;
import com.red.circle.other.infra.database.rds.entity.user.user.AuthType;
import com.red.circle.other.infra.database.rds.entity.user.user.BaseInfo;
@ -26,7 +28,8 @@ import com.red.circle.other.infra.database.rds.entity.user.user.MobileAuth;
import com.red.circle.other.infra.database.rds.entity.user.user.RegisterInfo;
import com.red.circle.other.infra.database.rds.entity.user.user.UserInviteUser;
import com.red.circle.other.infra.database.rds.service.user.device.ArchiveDeviceService;
import com.red.circle.other.infra.database.rds.service.user.device.LatestMobileDeviceService;
import com.red.circle.other.infra.database.rds.service.user.device.LatestMobileDeviceService;
import com.red.circle.other.infra.database.rds.service.sys.SysCountryCodeService;
import com.red.circle.other.infra.database.rds.service.user.user.ApprovalUserAccountStatusLogService;
import com.red.circle.other.infra.database.rds.service.user.user.AuthTypeService;
import com.red.circle.other.infra.database.rds.service.user.user.BaseInfoService;
@ -67,8 +70,10 @@ import org.springframework.transaction.interceptor.TransactionAspectSupport;
@RequiredArgsConstructor
public class UserAccountCommon {
private final AuthClient authClient;
private final BaseInfoService baseInfoService;
private final AuthClient authClient;
private final BaseInfoService baseInfoService;
private final SysCountryCodeService sysCountryCodeService;
private final CountryDataSyncLockService countryDataSyncLockService;
private final AuthTypeService authTypeService;
private final ImAccountClient imAccountClient;
private final NewsletterClient newsletterClient;
@ -93,14 +98,11 @@ public class UserAccountCommon {
BaseInfo baseInfo = toBaseInfo(cmd);
String userSig = requestUserSign(baseInfo.getId());
try {
// 客户端时区判断问题出现了，新用户注册就被冻结现象
baseInfo.setFreezingTime(TimestampUtils.convert(LocalDateTime.now().minusDays(1)));
mobileRegister(cmd, baseInfo);
UserProfile inviteUser = inviteRegister(cmd, baseInfo);
baseInfoService.save(baseInfo);
registerInfoService.save(toRegisterInfo(baseInfo.getId(), cmd));
authTypeService.save(toAuthType(baseInfo.getId(), cmd));
// saveUserInviteData(baseInfo.getId(), cmd, inviteUser);
// 客户端时区判断问题出现了，新用户注册就被冻结现象
baseInfo.setFreezingTime(TimestampUtils.convert(LocalDateTime.now().minusDays(1)));
UserProfile inviteUser = inviteRegister(cmd, baseInfo);
saveAccountRecordsWithCanonicalCountry(cmd, baseInfo);
// saveUserInviteData(baseInfo.getId(), cmd, inviteUser);
} catch (DuplicateKeyException ex) {
log.error("RegisterAddCmdExe:{},{}", ex, JacksonUtils.toJson(cmd));
TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();
@ -109,8 +111,33 @@ public class UserAccountCommon {
return new UserAccountDTO()
.setUserProfile(userProfileInfraConvertor.toUserProfileDTO(baseInfo))
.setUserSig(userSig);
}
.setUserSig(userSig);
}
private void saveAccountRecordsWithCanonicalCountry(CreateAccountCmd cmd, BaseInfo baseInfo) {
if (baseInfo.getCountryId() == null) {
saveAccountRecords(cmd, baseInfo);
return;
}
// 邀请注册可能在前置处理后改变最终国家；落库前按最终 countryId 持锁并重读主表，
// 避免国家级联完成后才写入旧国家码的注册竞态.
countryDataSyncLockService.execute(baseInfo.getCountryId(), () -> {
SysCountryCode country = sysCountryCodeService.getById(baseInfo.getCountryId());
ResponseAssert.notNull(ResponseErrorCode.REQUEST_PARAMETER_ERROR, country);
baseInfo.setCountryCode(country.getAlphaTwo());
baseInfo.setCountryName(country.getCountryName());
saveAccountRecords(cmd, baseInfo);
return null;
});
}
private void saveAccountRecords(CreateAccountCmd cmd, BaseInfo baseInfo) {
// 账号核心记录在国家锁内连续写入，锁冲突时不会遗留先写入的手机号认证记录.
mobileRegister(cmd, baseInfo);
baseInfoService.save(baseInfo);
registerInfoService.save(toRegisterInfo(baseInfo.getId(), cmd));
authTypeService.save(toAuthType(baseInfo.getId(), cmd));
}
/**
* 处理邀请注册信息

67
rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/common/user/UserCountryWriteLockService.java Normal file
View File

@ -0,0 +1,67 @@
package com.red.circle.other.infra.common.user;
import com.red.circle.component.redis.service.RedisService;
import com.red.circle.framework.core.asserts.ResponseAssert;
import com.red.circle.framework.core.response.CommonErrorCode;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Comparator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
/**
* 用户国家资料写锁，统一串行化单用户改国家与国家配置级联同步.
*/
@Slf4j
@Service
@RequiredArgsConstructor
public class UserCountryWriteLockService {
private static final int LOCK_SECONDS = 120;
private static final String LOCK_PREFIX = "user_country_write_lock:v2:";
private static final String RELEASE_LOCK_SCRIPT =
"if redis.call('GET', KEYS[1]) == ARGV[1] then "
+ "return redis.call('DEL', KEYS[1]) "
+ "end "
+ "return 0";
private final RedisService redisService;
/**
* 按用户ID排序加锁，避免多个批处理交叉申请时形成相反的锁顺序.
*/
public <T> T execute(Collection<Long> userIds, Supplier<T> action) {
Map<String, String> lockedKeys = new LinkedHashMap<>();
try {
userIds.stream().filter(Objects::nonNull).distinct().sorted(Comparator.naturalOrder())
.forEach(userId -> {
String key = LOCK_PREFIX + userId;
String token = UUID.randomUUID().toString();
ResponseAssert.isTrue(CommonErrorCode.OPERATION_CONFLICT,
redisService.setIfAbsent(key, token, LOCK_SECONDS, TimeUnit.SECONDS));
lockedKeys.put(key, token);
});
return action.get();
} finally {
List<Map.Entry<String, String>> entries = new ArrayList<>(lockedKeys.entrySet());
for (int index = entries.size() - 1; index >= 0; index--) {
try {
Map.Entry<String, String> entry = entries.get(index);
// token 校验防止旧持有者超时后删掉新持有者的锁.
redisService.execute(RELEASE_LOCK_SCRIPT, Long.class, List.of(entry.getKey()),
entry.getValue());
} catch (RuntimeException ex) {
log.warn("释放用户国家资料写锁失败", ex);
}
}
}
}
}

11
rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/cache/service/user/UserRunProfileCacheService.java vendored
View File

@ -1,4 +1,6 @@
package com.red.circle.other.infra.database.cache.service.user;
package com.red.circle.other.infra.database.cache.service.user;
import java.util.Collection;
/**
@ -13,7 +15,12 @@ public interface UserRunProfileCacheService {
*
* @param userId 用户id
*/
void remove(Long userId);
void remove(Long userId);
/**
* 批量移除用户运行资料缓存.
*/
void remove(Collection<Long> userIds);
/**
* 添加缓存.

14
rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/cache/service/user/impl/UserRunProfileCacheServiceImpl.java vendored
View File

@ -30,9 +30,17 @@ public class UserRunProfileCacheServiceImpl implements UserRunProfileCacheServic
private static final long CACHE_TIME = 60;
@Override
public void remove(Long userId) {
redisService.delete(getUserRunProfileKey(userId));
}
public void remove(Long userId) {
redisService.delete(getUserRunProfileKey(userId));
}
@Override
public void remove(Collection<Long> userIds) {
if (CollectionUtils.isEmpty(userIds)) {
return;
}
redisService.delete(userIds.stream().map(this::getUserRunProfileKey).toList());
}
@Override
public void save(String json, Long userId) {

12
rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/mongo/entity/activity/AgentActivityCount.java
View File

@ -9,8 +9,9 @@ import lombok.AccessLevel;
import lombok.Data;
import lombok.experimental.Accessors;
import lombok.experimental.FieldDefaults;
import org.springframework.data.annotation.Id;
import org.springframework.data.mongodb.core.mapping.Document;
import org.springframework.data.annotation.Id;
import org.springframework.data.mongodb.core.index.CompoundIndex;
import org.springframework.data.mongodb.core.mapping.Document;
/**
* 代理活动.
@ -18,9 +19,10 @@ import org.springframework.data.mongodb.core.mapping.Document;
* @author pengshigang on 2023/5/25
*/
@Data
@Accessors(chain = true)
@Document("agent_activity_count")
@FieldDefaults(level = AccessLevel.PRIVATE)
@Accessors(chain = true)
@Document("agent_activity_count")
@CompoundIndex(name = "idx_team_history", def = "{'teamId': 1, 'history': 1}")
@FieldDefaults(level = AccessLevel.PRIVATE)
public class AgentActivityCount implements Serializable {
@Serial

12
rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/mongo/entity/live/ActiveVoiceRoom.java
View File

@ -11,8 +11,9 @@ import lombok.AccessLevel;
import lombok.Data;
import lombok.experimental.Accessors;
import lombok.experimental.FieldDefaults;
import org.springframework.data.annotation.Id;
import org.springframework.data.mongodb.core.mapping.Document;
import org.springframework.data.annotation.Id;
import org.springframework.data.mongodb.core.index.CompoundIndex;
import org.springframework.data.mongodb.core.mapping.Document;
/**
* 活跃语音房间.
@ -20,9 +21,10 @@ import org.springframework.data.mongodb.core.mapping.Document;
* @author pengliang on 2020/12/11
*/
@Data
@Accessors(chain = true)
@Document("active_voice_room")
@FieldDefaults(level = AccessLevel.PRIVATE)
@Accessors(chain = true)
@Document("active_voice_room")
@CompoundIndex(name = "idx_user_id", def = "{'userId': 1}")
@FieldDefaults(level = AccessLevel.PRIVATE)
public class ActiveVoiceRoom implements Serializable {
@Serial

12
rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/mongo/entity/live/RoomProfileManager.java
View File

@ -10,8 +10,9 @@ import java.util.Optional;
import java.util.function.Predicate;
import java.util.stream.Collectors;
import lombok.Data;
import lombok.EqualsAndHashCode;
import org.springframework.data.mongodb.core.mapping.Document;
import lombok.EqualsAndHashCode;
import org.springframework.data.mongodb.core.index.CompoundIndex;
import org.springframework.data.mongodb.core.mapping.Document;
/**
* <p>
@ -22,9 +23,10 @@ import org.springframework.data.mongodb.core.mapping.Document;
* @since 2022/6/24
*/
@Data
@EqualsAndHashCode(callSuper = false)
@Document("room_profile_manager")
public class RoomProfileManager extends RoomProfile implements Serializable {
@EqualsAndHashCode(callSuper = false)
@Document("room_profile_manager")
@CompoundIndex(name = "idx_user_id", def = "{'userId': 1}")
public class RoomProfileManager extends RoomProfile implements Serializable {
@Serial
private static final long serialVersionUID = 1L;

8
rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/mongo/entity/team/team/TeamBillCycle.java
View File

@ -14,7 +14,8 @@ import java.util.List;
import lombok.Data;
import lombok.experimental.Accessors;
import org.springframework.data.annotation.Id;
import org.springframework.data.mongodb.core.index.CompoundIndex;
import org.springframework.data.mongodb.core.index.CompoundIndex;
import org.springframework.data.mongodb.core.index.CompoundIndexes;
import org.springframework.data.mongodb.core.mapping.Document;
/**
@ -25,7 +26,10 @@ import org.springframework.data.mongodb.core.mapping.Document;
@Data
@Accessors(chain = true)
@Document("team_bill_cycle")
@CompoundIndex(name = "idx_billBelong_status", def = "{'billBelong': 1, 'status': 1}")
@CompoundIndexes({
@CompoundIndex(name = "idx_billBelong_status", def = "{'billBelong': 1, 'status': 1}"),
@CompoundIndex(name = "idx_team_status", def = "{'teamId': 1, 'status': 1}")
})
public class TeamBillCycle implements Serializable {
@Serial

16
rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/mongo/entity/team/team/TeamMember.java
View File

@ -7,8 +7,10 @@ import java.io.Serializable;
import java.sql.Timestamp;
import lombok.Data;
import lombok.experimental.Accessors;
import org.springframework.data.annotation.Id;
import org.springframework.data.mongodb.core.mapping.Document;
import org.springframework.data.annotation.Id;
import org.springframework.data.mongodb.core.index.CompoundIndex;
import org.springframework.data.mongodb.core.index.CompoundIndexes;
import org.springframework.data.mongodb.core.mapping.Document;
/**
* 团队成员.
@ -16,9 +18,13 @@ import org.springframework.data.mongodb.core.mapping.Document;
* @author pengliang on 2022/12/1
*/
@Data
@Accessors(chain = true)
@Document("team_member")
public class TeamMember implements Serializable {
@Accessors(chain = true)
@Document("team_member")
@CompoundIndexes({
@CompoundIndex(name = "idx_team_member", def = "{'teamId': 1, 'memberId': 1}"),
@CompoundIndex(name = "idx_member_team", def = "{'memberId': 1, 'teamId': 1}")
})
public class TeamMember implements Serializable {
private static final long serialVersionUID = 1L;

16
rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/mongo/entity/team/team/TeamMemberTarget.java
View File

@ -12,8 +12,10 @@ import java.util.List;
import java.util.Objects;
import lombok.Data;
import lombok.experimental.Accessors;
import org.springframework.data.annotation.Id;
import org.springframework.data.mongodb.core.mapping.Document;
import org.springframework.data.annotation.Id;
import org.springframework.data.mongodb.core.index.CompoundIndex;
import org.springframework.data.mongodb.core.index.CompoundIndexes;
import org.springframework.data.mongodb.core.mapping.Document;
/**
* 团队成员目标.
@ -21,9 +23,13 @@ import org.springframework.data.mongodb.core.mapping.Document;
* @author pengliang on 2022/12/1
*/
@Data
@Accessors(chain = true)
@Document("team_member_target")
public class TeamMemberTarget implements Serializable {
@Accessors(chain = true)
@Document("team_member_target")
@CompoundIndexes({
@CompoundIndex(name = "idx_team_history", def = "{'teamId': 1, 'history': 1}"),
@CompoundIndex(name = "idx_user_history", def = "{'userId': 1, 'history': 1}")
})
public class TeamMemberTarget implements Serializable {
@Serial
private static final long serialVersionUID = 1L;

16
rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/mongo/entity/team/team/TeamProfile.java
View File

@ -15,8 +15,10 @@ import java.util.List;
import java.util.Objects;
import lombok.Data;
import lombok.experimental.Accessors;
import org.springframework.data.annotation.Id;
import org.springframework.data.mongodb.core.mapping.Document;
import org.springframework.data.annotation.Id;
import org.springframework.data.mongodb.core.index.CompoundIndex;
import org.springframework.data.mongodb.core.index.CompoundIndexes;
import org.springframework.data.mongodb.core.mapping.Document;
/**
* 团队资料.
@ -24,9 +26,13 @@ import org.springframework.data.mongodb.core.mapping.Document;
* @author pengliang on 2022/12/1
*/
@Data
@Accessors(chain = true)
@Document("team_profile")
public class TeamProfile implements Serializable {
@Accessors(chain = true)
@Document("team_profile")
@CompoundIndexes({
@CompoundIndex(name = "idx_country_id_cursor",
def = "{'country.countryId': 1, '_id': 1}")
})
public class TeamProfile implements Serializable {
@Serial
private static final long serialVersionUID = 1L;

7
rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/mongo/service/activity/AgentActivityCountService.java
View File

@ -31,7 +31,12 @@ public interface AgentActivityCountService {
/**
* 删除代理积分.
*/
void removeByTeamIds(Collection<Long> teamIds);
void removeByTeamIds(Collection<Long> teamIds);
/**
* 团队迁区时同步当前活动统计的区域，保留已累计积分.
*/
void updateCurrentRegion(Long teamId, String regionId);
/**
* 修改状态为历史top用户.

14
rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/mongo/service/activity/impl/AgentActivityCountServiceImpl.java
View File

@ -117,7 +117,7 @@ public class AgentActivityCountServiceImpl implements AgentActivityCountService
}
@Override
public void reduceTarget(Collection<Long> teamIds, Long target) {
public void reduceTarget(Collection<Long> teamIds, Long target) {
if (CollectionUtils.isEmpty(teamIds) || target <= 0) {
return;
@ -136,7 +136,17 @@ public class AgentActivityCountServiceImpl implements AgentActivityCountService
new Update().set("target", count.getTarget() > target ? count.getTarget() - target : 0),
AgentActivityCount.class));
}
}
@Override
public void updateCurrentRegion(Long teamId, String regionId) {
if (teamId == null || StringUtils.isBlank(regionId)) {
return;
}
mongoTemplate.updateMulti(Query.query(Criteria.where("teamId").is(teamId)
.and("history").ne(Boolean.TRUE)),
new Update().set("regionId", regionId), AgentActivityCount.class);
}
@Override
public void removeByTeamIds(Collection<Long> teamIds) {

20
rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/mongo/service/live/ActiveVoiceRoomService.java
View File

@ -3,7 +3,8 @@ package com.red.circle.other.infra.database.mongo.service.live;
import com.red.circle.common.business.core.enums.SysOriginPlatformEnum;
import com.red.circle.other.infra.database.mongo.entity.live.ActiveVoiceRoom;
import com.red.circle.other.inner.model.cmd.live.AppActiveVoiceRoomQryCmd;
import java.util.List;
import java.util.List;
import java.util.Collection;
/**
* 活跃语音房间.
@ -41,7 +42,22 @@ public interface ActiveVoiceRoomService {
* @param roomAccount 房间账号
* @param quantity 数量
*/
void updateQuantityByRoomAccount(String roomAccount, Long quantity);
void updateQuantityByRoomAccount(String roomAccount, Long quantity);
/**
* 批量同步当前活跃房间中的用户国家快照.
*/
void updateUserCountry(Collection<Long> userIds, String countryCode, String countryName);
/**
* 查询一批房主当前的活跃房间，供国家同步精确失效发现缓存.
*/
List<ActiveVoiceRoom> listByUserIds(Collection<Long> userIds);
/**
* 更新房主的默认区域；存在生效中定向置顶配置的房间保持其固定区域不变.
*/
void updateUserDefaultRegion(Collection<Long> userIds, String regionCode);
/**
* 移除首页房间.

10
rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/mongo/service/live/RoomProfileManagerService.java
View File

@ -6,7 +6,8 @@ import com.red.circle.other.infra.database.mongo.entity.live.RoomProfileManager;
import com.red.circle.other.infra.database.mongo.entity.live.RoomSetting;
import com.red.circle.other.inner.model.cmd.room.RoomProfileOpsQryCmd;
import com.red.circle.other.inner.model.dto.material.UseBadgeDTO;
import java.util.List;
import java.util.List;
import java.util.Collection;
import java.util.Map;
import java.util.Set;
@ -220,7 +221,12 @@ public interface RoomProfileManagerService {
/**
* 修改房间归属国家.
*/
void updateUserCountry(Long userId, String countryCode, String countryName);
void updateUserCountry(Long userId, String countryCode, String countryName);
/**
* 批量修改用户当前房间的归属国家.
*/
void updateUserCountry(Collection<Long> userIds, String countryCode, String countryName);
/**
* 使用语言.

73
rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/mongo/service/live/impl/ActiveVoiceRoomServiceImpl.java
View File

@ -6,10 +6,12 @@ import com.red.circle.framework.mybatis.constant.PageConstant;
import com.red.circle.other.infra.common.sys.CountryCodeAliasSupport;
import com.red.circle.other.infra.database.mongo.dto.query.live.ActiveVoiceRoomQuery;
import com.red.circle.other.infra.database.mongo.dto.query.live.FamilyOnlineRoomQuery;
import com.red.circle.other.infra.database.mongo.entity.live.ActiveVoiceRoom;
import com.red.circle.other.infra.database.mongo.service.live.ActiveVoiceRoomService;
import com.red.circle.other.infra.database.mongo.entity.live.ActiveVoiceRoom;
import com.red.circle.other.infra.database.mongo.entity.live.RoomTopFixed;
import com.red.circle.other.infra.database.mongo.service.live.ActiveVoiceRoomService;
import com.red.circle.other.inner.model.cmd.live.AppActiveVoiceRoomQryCmd;
import com.red.circle.tool.core.collection.CollectionUtils;
import com.red.circle.tool.core.collection.CollectionUtils;
import com.red.circle.tool.core.date.TimestampUtils;
import com.red.circle.tool.core.text.StringUtils;
import java.util.*;
@ -77,10 +79,69 @@ public class ActiveVoiceRoomServiceImpl implements ActiveVoiceRoomService {
}
@Override
public void updateQuantityByRoomAccount(String roomAccount, Long quantity) {
public void updateQuantityByRoomAccount(String roomAccount, Long quantity) {
mongoTemplate.updateFirst(Query.query(Criteria.where("roomAccount").is(roomAccount)),
new Update().set("onlineQuantity", quantity), ActiveVoiceRoom.class);
}
new Update().set("onlineQuantity", quantity), ActiveVoiceRoom.class);
}
@Override
public void updateUserCountry(Collection<Long> userIds, String countryCode, String countryName) {
if (CollectionUtils.isEmpty(userIds)) {
return;
}
mongoTemplate.updateMulti(Query.query(Criteria.where("userId").in(userIds)),
new Update()
.set("countryCode", countryCode)
.set("countryName", countryName),
ActiveVoiceRoom.class);
}
@Override
public List<ActiveVoiceRoom> listByUserIds(Collection<Long> userIds) {
if (CollectionUtils.isEmpty(userIds)) {
return CollectionUtils.newArrayList();
}
return mongoTemplate.find(Query.query(Criteria.where("userId").in(userIds)),
ActiveVoiceRoom.class);
}
@Override
public void updateUserDefaultRegion(Collection<Long> userIds, String regionCode) {
if (CollectionUtils.isEmpty(userIds) || StringUtils.isBlank(regionCode)) {
return;
}
Query activeRoomQuery = Query.query(Criteria.where("userId").in(userIds));
activeRoomQuery.fields().include("id");
Set<Long> activeRoomIds = mongoTemplate.find(activeRoomQuery, ActiveVoiceRoom.class).stream()
.map(ActiveVoiceRoom::getId)
.filter(Objects::nonNull)
.collect(java.util.stream.Collectors.toSet());
if (CollectionUtils.isEmpty(activeRoomIds)) {
return;
}
// 定向置顶房间的区域由 room_top_fixed 控制，国家或团队变化不能覆盖该运营配置.
java.sql.Timestamp now = TimestampUtils.now();
Criteria effectiveFixed = new Criteria().andOperator(
Criteria.where("id").in(activeRoomIds),
Criteria.where("startTime").lte(now),
new Criteria().orOperator(
Criteria.where("expiredTime").is(null),
Criteria.where("expiredTime").gt(now)));
Set<Long> fixedRegionRoomIds = mongoTemplate.find(Query.query(effectiveFixed),
RoomTopFixed.class).stream()
.filter(fixed -> StringUtils.isNotBlank(fixed.getRegion()))
.map(RoomTopFixed::getId)
.collect(java.util.stream.Collectors.toSet());
Criteria updateCriteria = Criteria.where("userId").in(userIds);
if (CollectionUtils.isNotEmpty(fixedRegionRoomIds)) {
updateCriteria.and("id").nin(fixedRegionRoomIds);
}
mongoTemplate.updateMulti(Query.query(updateCriteria),
new Update().set("region", regionCode), ActiveVoiceRoom.class);
}
@Override
public void removeByIds(List<Long> ids) {

14
rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/mongo/service/live/impl/RoomProfileManagerServiceImpl.java
View File

@ -24,6 +24,7 @@ import com.red.circle.tool.core.obj.ObjectUtils;
import com.red.circle.tool.core.text.StringUtils;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.Collection;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@ -651,6 +652,19 @@ public class RoomProfileManagerServiceImpl implements RoomProfileManagerService
RoomProfileManager.class);
}
@Override
public void updateUserCountry(Collection<Long> userIds, String countryCode, String countryName) {
if (CollectionUtils.isEmpty(userIds)) {
return;
}
// updateMulti 不携带查询排序，避免 MongoDB updateMany 拒绝 sort 参数.
mongoTemplate.updateMulti(Query.query(currentRoomCriteria().and("userId").in(userIds)),
new Update().set("updateTime", TimestampUtils.now())
.set("countryCode", countryCode)
.set("countryName", countryName),
RoomProfileManager.class);
}
private Query currentRoomQueryByUserId(Long userId) {
return Query.query(currentRoomCriteria().and("userId").is(userId))
.with(Sort.by(

12
rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/mongo/service/team/team/TeamMemberTargetService.java
View File

@ -130,7 +130,17 @@ public interface TeamMemberTargetService {
/**
* 修改本月，将成员账单归类于历史账单.
*/
void updateThisMonthUserTargetHistoryStatusByTeamId(Long teamId);
void updateThisMonthUserTargetHistoryStatusByTeamId(Long teamId);
/**
* 团队跨区时只迁移当前有效目标的区域，保留已累计流水和每日目标.
*/
void updateCurrentRegionByTeamId(Long teamId, String region);
/**
* 国家码修改时按用户同步当前有效目标的国家码，不覆盖其他国家团队成员.
*/
void updateCurrentCountryCodeByUserIds(Collection<Long> userIds, String countryCode);
/**
* 获取指定目标记录.

41
rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/mongo/service/team/team/TeamProfileService.java
View File

@ -3,7 +3,8 @@ package com.red.circle.other.infra.database.mongo.service.team.team;
import com.red.circle.other.infra.database.mongo.entity.team.team.TeamProfile;
import com.red.circle.other.inner.model.cmd.team.StatusTeamBatchCmd;
import com.red.circle.other.inner.model.cmd.team.TeamTableQryCmd;
import com.red.circle.other.inner.model.dto.agency.agency.TeamContact;
import com.red.circle.other.inner.model.dto.agency.agency.TeamContact;
import com.red.circle.other.inner.model.dto.agency.agency.TeamCountry;
import com.red.circle.other.inner.model.dto.agency.agency.TeamCounter;
import com.red.circle.other.inner.model.dto.agency.agency.TeamRemark;
import java.util.List;
@ -121,7 +122,43 @@ public interface TeamProfileService {
/**
* 获取团队资料列表.
*/
List<TeamProfile> listByIds(Set<Long> ids);
List<TeamProfile> listByIds(Set<Long> ids);
/**
* 按稳定国家ID获取全部团队.
*/
List<TeamProfile> listByCountryId(Long countryId);
/**
* 按国家ID和团队ID游标分页读取，避免一次加载大国家下全部团队.
*/
List<TeamProfile> listCountryTeamsAfterId(Long countryId, Long lastId, Integer limit);
/**
* 查询国家下存在团队的应用来源.
*/
Set<String> listOriginSysByCountryId(Long countryId);
/**
* 判断团队国家编码快照是否仍有未同步数据.
*/
boolean existsCountryCodeMismatch(Long countryId, String countryCode);
/**
* 判断团队国家名称快照是否仍有未同步数据.
*/
boolean existsCountryNameMismatch(Long countryId, String countryName);
/**
* 判断指定来源的团队是否仍未迁到目标区域.
*/
boolean existsCountryRegionMismatch(Long countryId, String sysOrigin, String region);
/**
* 仅当团队仍属于指定国家ID时更新国家与区域，避免覆盖并发修改.
*/
boolean updateCountryAndRegion(Long teamId, Long expectedCountryId, String expectedRegion,
String region, TeamCountry country, Long updateUser);
/**
* 获取团队资料映射表.

7
rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/mongo/service/team/team/impl/TeamBillCycleServiceImpl.java
View File

@ -721,9 +721,10 @@ public class TeamBillCycleServiceImpl implements TeamBillCycleService {
}
@Override
public void updateUnpaidTeamBillRegion(Long teamId, String region, Long updateUser) {
mongoTemplate.updateFirst(Query.query(Criteria.where("teamId").is(teamId)
.and("status").is(TeamBillCycleStatus.UNPAID)),
public void updateUnpaidTeamBillRegion(Long teamId, String region, Long updateUser) {
// 理论上每队只有一条未出账账单；updateMulti 同时修复可能存在的历史重复脏数据.
mongoTemplate.updateMulti(Query.query(Criteria.where("teamId").is(teamId)
.and("status").is(TeamBillCycleStatus.UNPAID)),
new Update()
.set("region", region)
.set("updateUser", updateUser)

24
rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/mongo/service/team/team/impl/TeamMemberTargetServiceImpl.java
View File

@ -82,7 +82,7 @@ public class TeamMemberTargetServiceImpl implements TeamMemberTargetService {
}
@Override
public void updateTargetCover(Long userId, Integer billBelong, TeamMemberTargetIndex target) {
public void updateTargetCover(Long userId, Integer billBelong, TeamMemberTargetIndex target) {
mongoTemplate.updateFirst(getUserBillQuery(userId, billBelong),
new Update().set("dailyTargets", List.of(target))
.set("updateTime", TimestampUtils.now())
@ -822,6 +822,28 @@ public class TeamMemberTargetServiceImpl implements TeamMemberTargetService {
return aggregated;
}
@Override
public void updateCurrentRegionByTeamId(Long teamId, String region) {
if (teamId == null || StringUtils.isBlank(region)) {
return;
}
mongoTemplate.updateMulti(Query.query(Criteria.where("teamId").is(teamId)
.and("history").ne(Boolean.TRUE)),
new Update().set("region", region).set("updateTime", TimestampUtils.now()),
TeamMemberTarget.class);
}
@Override
public void updateCurrentCountryCodeByUserIds(Collection<Long> userIds, String countryCode) {
if (CollectionUtils.isEmpty(userIds) || StringUtils.isBlank(countryCode)) {
return;
}
mongoTemplate.updateMulti(Query.query(Criteria.where("userId").in(userIds)
.and("history").ne(Boolean.TRUE)),
new Update().set("countryCode", countryCode).set("updateTime", TimestampUtils.now()),
TeamMemberTarget.class);
}
@Override
public List<TeamMemberTarget> listManualSalaryTargets(String sysOrigin, String countryCode,
Collection<Long> userIds, Integer billBelong) {

101
rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/mongo/service/team/team/impl/TeamProfileServiceImpl.java
View File

@ -12,15 +12,17 @@ import com.red.circle.other.infra.database.mongo.service.team.team.TeamProfileSe
import com.red.circle.other.inner.enums.team.TeamMemberRole;
import com.red.circle.other.inner.model.cmd.team.StatusTeamBatchCmd;
import com.red.circle.other.inner.model.cmd.team.TeamTableQryCmd;
import com.red.circle.other.inner.model.dto.agency.agency.TeamContact;
import com.red.circle.other.inner.model.dto.agency.agency.TeamCounter;
import com.red.circle.other.inner.model.dto.agency.agency.TeamContact;
import com.red.circle.other.inner.model.dto.agency.agency.TeamCounter;
import com.red.circle.other.inner.model.dto.agency.agency.TeamCountry;
import com.red.circle.other.inner.model.dto.agency.agency.TeamRemark;
import com.red.circle.other.inner.model.dto.agency.agency.TeamSetting;
import com.red.circle.other.inner.model.dto.agency.agency.TeamStatus;
import com.red.circle.tool.core.collection.CollectionUtils;
import com.red.circle.tool.core.date.TimestampUtils;
import com.red.circle.tool.core.text.StringUtils;
import java.util.List;
import com.red.circle.tool.core.date.TimestampUtils;
import com.red.circle.tool.core.text.StringUtils;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
@ -41,19 +43,19 @@ import org.springframework.stereotype.Service;
*/
@Service
@RequiredArgsConstructor
public class TeamProfileServiceImpl implements TeamProfileService {
private final MongoTemplate mongoTemplate;
@Override
public void createBatch(List<TeamProfile> teamProfiles) {
mongoTemplate.insert(teamProfiles, TeamProfile.class);
}
@Override
public void create(TeamProfile teamProfile) {
mongoTemplate.save(teamProfile);
}
public class TeamProfileServiceImpl implements TeamProfileService {
private final MongoTemplate mongoTemplate;
@Override
public void createBatch(List<TeamProfile> teamProfiles) {
mongoTemplate.insert(teamProfiles, TeamProfile.class);
}
@Override
public void create(TeamProfile teamProfile) {
mongoTemplate.save(teamProfile);
}
@Override
public boolean existsOwnUserId(Long ownUserId) {
@ -348,9 +350,66 @@ public class TeamProfileServiceImpl implements TeamProfileService {
}
@Override
public List<TeamProfile> listByIds(Set<Long> ids) {
return mongoTemplate.find(Query.query(Criteria.where("id").in(ids)), TeamProfile.class);
}
public List<TeamProfile> listByIds(Set<Long> ids) {
return mongoTemplate.find(Query.query(Criteria.where("id").in(ids)), TeamProfile.class);
}
@Override
public List<TeamProfile> listByCountryId(Long countryId) {
return mongoTemplate.find(
Query.query(Criteria.where("country.countryId").is(countryId))
.with(Sort.by(Sort.Order.asc("id"))),
TeamProfile.class);
}
@Override
public List<TeamProfile> listCountryTeamsAfterId(Long countryId, Long lastId, Integer limit) {
return mongoTemplate.find(Query.query(Criteria.where("country.countryId").is(countryId)
.and("id").gt(lastId))
.with(Sort.by(Sort.Order.asc("id"))).limit(limit), TeamProfile.class);
}
@Override
public Set<String> listOriginSysByCountryId(Long countryId) {
return new HashSet<>(mongoTemplate.findDistinct(
Query.query(Criteria.where("country.countryId").is(countryId)), "sysOrigin",
TeamProfile.class, String.class));
}
@Override
public boolean existsCountryCodeMismatch(Long countryId, String countryCode) {
return mongoTemplate.exists(Query.query(Criteria.where("country.countryId").is(countryId)
.and("country.countryCode").ne(countryCode)), TeamProfile.class);
}
@Override
public boolean existsCountryNameMismatch(Long countryId, String countryName) {
return mongoTemplate.exists(Query.query(Criteria.where("country.countryId").is(countryId)
.and("country.countryName").ne(countryName)), TeamProfile.class);
}
@Override
public boolean existsCountryRegionMismatch(Long countryId, String sysOrigin, String region) {
return mongoTemplate.exists(Query.query(Criteria.where("country.countryId").is(countryId)
.and("sysOrigin").is(sysOrigin).and("region").ne(region)), TeamProfile.class);
}
@Override
public boolean updateCountryAndRegion(Long teamId, Long expectedCountryId,
String expectedRegion, String region, TeamCountry country, Long updateUser) {
Update update = new Update()
.set("country", country)
.set("updateTime", TimestampUtils.now())
.set("updateUser", updateUser)
.set("originUpdateUser", 1);
if (StringUtils.isNotBlank(region)) {
update.set("region", region);
}
return mongoTemplate.updateFirst(Query.query(Criteria.where("id").is(teamId)
.and("country.countryId").is(expectedCountryId)
.and("region").is(expectedRegion)), update, TeamProfile.class)
.getMatchedCount() == 1;
}
@Override
public Map<Long, TeamProfile> mapProfileByIds(Set<Long> ids) {

15
rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/mongo/service/user/active/UserDailyActiveService.java
View File

@ -1,4 +1,6 @@
package com.red.circle.other.infra.database.mongo.service.user.active;
package com.red.circle.other.infra.database.mongo.service.user.active;
import java.util.Collection;
/**
* 用户每日活跃日志服务.
@ -13,6 +15,11 @@ public interface UserDailyActiveService {
* @param registerCountryCode 注册国家code
* @param language 用户语言
*/
void record(Long userId, String sysOrigin, String registerCountryCode, String language);
}
void record(Long userId, String sysOrigin, String registerCountryCode, String language);
/**
* 同步用户当天已生成的活跃国家快照，避免国家码修改后大屏当天仍显示旧码.
*/
void updateTodayCountryCode(Collection<Long> userIds, String countryCode);
}

32
rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/mongo/service/user/active/impl/UserDailyActiveServiceImpl.java
View File

@ -6,9 +6,10 @@ import com.red.circle.other.infra.database.mongo.service.user.active.UserDailyAc
import com.red.circle.other.infra.utils.ZonedDateTimeUtils;
import com.red.circle.tool.core.sequence.IdWorkerUtils;
import java.sql.Timestamp;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.util.concurrent.TimeUnit;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Collection;
import java.util.concurrent.TimeUnit;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.mongodb.core.MongoTemplate;
@ -30,7 +31,7 @@ public class UserDailyActiveServiceImpl implements UserDailyActiveService {
private final StringRedisTemplate stringRedisTemplate;
@Override
public void record(Long userId, String sysOrigin, String registerCountryCode, String language) {
public void record(Long userId, String sysOrigin, String registerCountryCode, String language) {
try {
ZonedDateTime now = ZonedDateTimeUtils.nowAsiaShanghai();
String activeDate = now.format(DateTimeFormatter.ofPattern("yyyy-MM-dd"));
@ -67,7 +68,22 @@ public class UserDailyActiveServiceImpl implements UserDailyActiveService {
}
} catch (Exception e) {
log.warn("[用户每日活跃] 记录失败 userId={}", userId, e);
}
}
}
}
}
@Override
public void updateTodayCountryCode(Collection<Long> userIds, String countryCode) {
if (userIds == null || userIds.isEmpty()) {
return;
}
ZonedDateTime now = ZonedDateTimeUtils.nowAsiaShanghai();
String activeDate = now.format(DateTimeFormatter.ofPattern("yyyy-MM-dd"));
mongoTemplate.updateMulti(
Query.query(Criteria.where("userId").in(userIds).and("activeDate").is(activeDate)),
new Update()
.set("registerCountryCode", countryCode)
.set("updateTime", Timestamp.from(now.toInstant())),
UserDailyActiveLog.class);
}
}

13
rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/mongo/service/user/profile/UserRunProfileService.java
View File

@ -24,7 +24,18 @@ public interface UserRunProfileService {
/**
* 修改资料信息-为空的不做处理.
*/
void updateSelectiveById(UserRunProfile userRunProfile);
void updateSelectiveById(UserRunProfile userRunProfile);
/**
* 仅修改账号状态，避免携带运行资料快照覆盖其他并发变更.
*/
void updateAccountStatus(Long id, String accountStatus);
/**
* 批量同步用户运行资料中的国家冗余字段.
*/
void updateCountry(Collection<Long> userIds, Long countryId, String countryCode,
String countryName);
/**
* 修改个性签名.

38
rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/mongo/service/user/profile/impl/UserRunProfileServiceImpl.java
View File

@ -56,14 +56,42 @@ public class UserRunProfileServiceImpl implements UserRunProfileService {
}
@Override
public void updateSelectiveById(UserRunProfile userRunProfile) {
if (Objects.isNull(userRunProfile) || Objects.isNull(userRunProfile.getId())) {
return;
public void updateSelectiveById(UserRunProfile userRunProfile) {
if (Objects.isNull(userRunProfile) || Objects.isNull(userRunProfile.getId())) {
return;
}
mongoTemplate.updateFirst(Query.query(Criteria.where("id").is(userRunProfile.getId())),
getUpdateSelective(userRunProfile),
UserRunProfile.class);
}
UserRunProfile.class);
}
@Override
public void updateAccountStatus(Long id, String accountStatus) {
if (Objects.isNull(id) || StringUtils.isBlank(accountStatus)) {
return;
}
// 账号状态更新只能写目标字段，防止旧的完整快照把国家迁移结果覆盖回去。
mongoTemplate.updateFirst(Query.query(Criteria.where("id").is(id)),
new Update()
.set("accountStatus", accountStatus)
.set("updateTime", TimestampUtils.now()),
UserRunProfile.class);
}
@Override
public void updateCountry(Collection<Long> userIds, Long countryId, String countryCode,
String countryName) {
if (CollectionUtils.isEmpty(userIds)) {
return;
}
mongoTemplate.updateMulti(Query.query(Criteria.where("id").in(userIds)),
new Update()
.set("countryId", countryId)
.set("countryCode", countryCode)
.set("countryName", countryName)
.set("updateTime", TimestampUtils.now()),
UserRunProfile.class);
}
@Override
public void updateUserAutograph(Long id, String autograph) {

45
rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/rds/dao/user/user/BaseInfoDAO.java
View File

@ -2,9 +2,10 @@ package com.red.circle.other.infra.database.rds.dao.user.user;
import com.red.circle.framework.mybatis.dao.BaseDAO;
import com.red.circle.other.infra.database.rds.entity.user.user.BaseInfo;
import com.red.circle.other.inner.model.cmd.user.UserBaseInfoPageQryCmd;
import java.util.List;
import com.red.circle.other.infra.database.rds.entity.user.user.BaseInfo;
import com.red.circle.other.inner.model.cmd.user.UserBaseInfoPageQryCmd;
import java.util.Collection;
import java.util.List;
import org.apache.ibatis.annotations.Param;
/**
@ -15,7 +16,43 @@ import org.apache.ibatis.annotations.Param;
* @author pengliang
* @since 2019-08-26
*/
public interface BaseInfoDAO extends BaseDAO<BaseInfo> {
public interface BaseInfoDAO extends BaseDAO<BaseInfo> {
/**
* 按稳定国家ID游标查询用户，供国家资料批量同步使用.
*/
List<BaseInfo> listCountryUsersAfterId(@Param("countryId") Long countryId,
@Param("lastId") Long lastId, @Param("limit") Integer limit);
/**
* 查询国家下涉及的应用来源.
*/
List<String> listDistinctOriginSysByCountryId(@Param("countryId") Long countryId);
/**
* 查询国家下仍保留的用户国家码，供同值保存时识别历史残留编码.
*/
List<String> listDistinctCountryCodesByCountryId(@Param("countryId") Long countryId);
/**
* 判断国家编码冗余字段是否仍有未同步数据.
*/
Boolean existsCountryCodeMismatch(@Param("countryId") Long countryId,
@Param("countryCode") String countryCode);
/**
* 判断国家名称冗余字段是否仍有未同步数据.
*/
Boolean existsCountryNameMismatch(@Param("countryId") Long countryId,
@Param("countryName") String countryName);
/**
* 按国家ID与用户ID双重约束批量更新，避免覆盖并发切换国家的用户.
*/
int updateCountryByIds(@Param("countryId") Long countryId,
@Param("userIds") Collection<Long> userIds,
@Param("countryCode") String countryCode,
@Param("countryName") String countryName);
/**
* 获取最大的account

19
rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/rds/dao/user/user/ExpandDAO.java
View File

@ -3,9 +3,10 @@ package com.red.circle.other.infra.database.rds.dao.user.user;
import com.red.circle.framework.mybatis.dao.BaseDAO;
import com.red.circle.other.infra.database.rds.entity.user.user.UserExpand;
import com.red.circle.other.inner.model.dto.user.ActiveUserCountryCodeDTO;
import java.time.LocalDateTime;
import java.util.List;
import com.red.circle.other.inner.model.dto.user.ActiveUserCountryCodeDTO;
import java.time.LocalDateTime;
import java.util.Collection;
import java.util.List;
import org.apache.ibatis.annotations.Param;
/**
@ -16,7 +17,15 @@ import org.apache.ibatis.annotations.Param;
* @author pengliang
* @since 2019-10-30
*/
public interface ExpandDAO extends BaseDAO<UserExpand> {
public interface ExpandDAO extends BaseDAO<UserExpand> {
/**
* 国家码重命名时，仅改同一国家ID下命中明确旧码的注册国家快照.
* 未命中旧码的跨国注册历史保持不变.
*/
int updateRegisterCountryCodeByCountry(@Param("countryId") Long countryId,
@Param("sourceCountryCodes") Collection<String> sourceCountryCodes,
@Param("newCountryCode") String newCountryCode);
/**
* 最新两小时在线用户分布国家
@ -24,7 +33,7 @@ public interface ExpandDAO extends BaseDAO<UserExpand> {
* @param lastActiveTime 活跃时间
* @return list
*/
List<ActiveUserCountryCodeDTO> findLatestActiveUserCountryCode(
List<ActiveUserCountryCodeDTO> findLatestActiveUserCountryCode(
@Param("lastActiveTime") LocalDateTime lastActiveTime);
}

36
rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/rds/service/user/user/BaseInfoService.java
View File

@ -13,7 +13,8 @@ import com.red.circle.other.inner.model.cmd.user.UserBaseInfoPageQryCmd;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;
/**
@ -34,7 +35,38 @@ public interface BaseInfoService extends BaseService<BaseInfo> {
/**
* 获取用户信息.
*/
List<BaseInfo> listUser(SysOriginUserQryCmd qryCmd);
List<BaseInfo> listUser(SysOriginUserQryCmd qryCmd);
/**
* 按国家ID游标读取用户，避免批量同步时一次加载全部用户.
*/
List<BaseInfo> listCountryUsersAfterId(Long countryId, Long lastId, Integer limit);
/**
* 查询国家下涉及的应用来源.
*/
Set<String> listOriginSysByCountryId(Long countryId);
/**
* 查询国家下当前存在的用户国家码.
*/
Set<String> listCountryCodesByCountryId(Long countryId);
/**
* 国家编码冗余字段是否存在未同步数据.
*/
boolean existsCountryCodeMismatch(Long countryId, String countryCode);
/**
* 国家名称冗余字段是否存在未同步数据.
*/
boolean existsCountryNameMismatch(Long countryId, String countryName);
/**
* 按国家ID约束批量同步用户国家资料.
*/
int updateCountryByIds(Long countryId, Collection<Long> userIds, String countryCode,
String countryName);
/**
* 归属系统.

15
rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/rds/service/user/user/ExpandService.java
View File

@ -2,8 +2,9 @@ package com.red.circle.other.infra.database.rds.service.user.user;
import com.red.circle.framework.mybatis.service.BaseService;
import com.red.circle.other.infra.database.rds.entity.user.user.UserExpand;
import com.red.circle.other.inner.model.dto.user.ActiveUserCountryCodeDTO;
import java.time.LocalDateTime;
import com.red.circle.other.inner.model.dto.user.ActiveUserCountryCodeDTO;
import java.time.LocalDateTime;
import java.util.Collection;
import java.util.List;
import java.util.Locale;
import java.util.Map;
@ -17,14 +18,20 @@ import java.util.Set;
* @author pengliang
* @since 2019-10-30
*/
public interface ExpandService extends BaseService<UserExpand> {
public interface ExpandService extends BaseService<UserExpand> {
/**
* 国家码重命名时同步仍指向旧码的注册国家快照.
*/
int updateRegisterCountryCodeByCountry(Long countryId, Collection<String> sourceCountryCodes,
String newCountryCode);
/**
* 不存在创建.
*
* @param userExpand ignore
*/
void notExistsCreate(UserExpand userExpand);
void notExistsCreate(UserExpand userExpand);
/**
* 初始化扩展信息.

40
rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/rds/service/user/user/impl/BaseInfoServiceImpl.java
View File

@ -55,7 +55,7 @@ public class BaseInfoServiceImpl extends BaseServiceImpl<BaseInfoDAO, BaseInfo>
}
@Override
public List<BaseInfo> listUser(SysOriginUserQryCmd qryCmd) {
public List<BaseInfo> listUser(SysOriginUserQryCmd qryCmd) {
//DAY,MONTH,YEAR
return query()
.eq(StringUtils.isNotBlank(qryCmd.getCountryCode()), BaseInfo::getCountryCode, qryCmd.getCountryCode())
@ -63,8 +63,42 @@ public class BaseInfoServiceImpl extends BaseServiceImpl<BaseInfoDAO, BaseInfo>
TimestampUtils.convert(getStartDate(qryCmd.getDate(), qryCmd.getDateType())))
.le(StringUtils.isNotBlank(qryCmd.getDate()), BaseInfo::getCreateTime,
TimestampUtils.convert(getEndDate(qryCmd.getDate(), qryCmd.getDateType())))
.list();
}
.list();
}
@Override
public List<BaseInfo> listCountryUsersAfterId(Long countryId, Long lastId, Integer limit) {
return baseInfoDAO.listCountryUsersAfterId(countryId, lastId, limit);
}
@Override
public Set<String> listOriginSysByCountryId(Long countryId) {
return new HashSet<>(baseInfoDAO.listDistinctOriginSysByCountryId(countryId));
}
@Override
public Set<String> listCountryCodesByCountryId(Long countryId) {
return new HashSet<>(baseInfoDAO.listDistinctCountryCodesByCountryId(countryId));
}
@Override
public boolean existsCountryCodeMismatch(Long countryId, String countryCode) {
return Boolean.TRUE.equals(baseInfoDAO.existsCountryCodeMismatch(countryId, countryCode));
}
@Override
public boolean existsCountryNameMismatch(Long countryId, String countryName) {
return Boolean.TRUE.equals(baseInfoDAO.existsCountryNameMismatch(countryId, countryName));
}
@Override
public int updateCountryByIds(Long countryId, Collection<Long> userIds, String countryCode,
String countryName) {
if (CollectionUtils.isEmpty(userIds)) {
return 0;
}
return baseInfoDAO.updateCountryByIds(countryId, userIds, countryCode, countryName);
}
/**
* 获取开始时间

24
rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/rds/service/user/user/impl/ExpandServiceImpl.java
View File

@ -7,8 +7,9 @@ import com.red.circle.tool.core.collection.CollectionUtils;
import com.red.circle.other.infra.database.rds.dao.user.user.ExpandDAO;
import com.red.circle.other.infra.database.rds.entity.user.user.UserExpand;
import com.red.circle.other.infra.database.rds.service.user.user.ExpandService;
import com.red.circle.other.inner.model.dto.user.ActiveUserCountryCodeDTO;
import java.time.LocalDateTime;
import com.red.circle.other.inner.model.dto.user.ActiveUserCountryCodeDTO;
import java.time.LocalDateTime;
import java.util.Collection;
import java.util.List;
import java.util.Locale;
import java.util.Map;
@ -30,9 +31,22 @@ import org.springframework.stereotype.Service;
*/
@RequiredArgsConstructor
@Service
public class ExpandServiceImpl extends BaseServiceImpl<ExpandDAO, UserExpand> implements ExpandService {
private Boolean exists(Long userId) {
public class ExpandServiceImpl extends BaseServiceImpl<ExpandDAO, UserExpand> implements ExpandService {
private final ExpandDAO expandDAO;
@Override
public int updateRegisterCountryCodeByCountry(Long countryId,
Collection<String> sourceCountryCodes,
String newCountryCode) {
if (countryId == null || CollectionUtils.isEmpty(sourceCountryCodes)) {
return 0;
}
return expandDAO.updateRegisterCountryCodeByCountry(countryId, sourceCountryCodes,
newCountryCode);
}
private Boolean exists(Long userId) {
return Optional.ofNullable(
query().select(UserExpand::getUserId).eq(UserExpand::getUserId, userId).last(PageConstant.LIMIT_ONE)
.getOne()

15
rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/gateway/user/UserProfileGatewayImpl.java
View File

@ -428,12 +428,15 @@ public class UserProfileGatewayImpl implements UserProfileGateway {
}
@Override
public void removeCacheAll(Collection<Long> ids) {
if (CollectionUtils.isNotEmpty(ids)) {
userRunProfileService.remove(ids);
userCacheService.remove(ids);
}
}
public void removeCacheAll(Collection<Long> ids) {
if (CollectionUtils.isNotEmpty(ids)) {
userRunProfileService.remove(ids);
userCacheService.remove(ids);
// 批量失效必须与单用户路径一致，否则国家或区域修改后仍会读到旧运行缓存.
userRunProfileCacheService.remove(ids);
userRegionCacheService.remove(ids);
}
}
@Override
public void removeCache(Long id) {

16
rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/gateway/user/UserRunProfileTransportGatewayImpl.java
View File

@ -215,14 +215,14 @@ public class UserRunProfileTransportGatewayImpl implements UserRunProfileTranspo
}
@Override
public UserProfile updateUserRunProfile(Long id, String accountStatus) {
UserRunProfile userRunProfiles = userRunProfileService.getById(id);
if (Objects.nonNull(userRunProfiles)) {
userRunProfiles.setAccountStatus(accountStatus);
userRunProfiles.setUpdateTime(TimestampUtils.now());
userRunProfileService.updateSelectiveById(userRunProfiles);
return userProfileInfraConvertor.toUserProfile(userRunProfiles);
}
public UserProfile updateUserRunProfile(Long id, String accountStatus) {
UserRunProfile userRunProfiles = userRunProfileService.getById(id);
if (Objects.nonNull(userRunProfiles)) {
userRunProfileService.updateAccountStatus(id, accountStatus);
userRunProfiles.setAccountStatus(accountStatus);
userRunProfiles.setUpdateTime(TimestampUtils.now());
return userProfileInfraConvertor.toUserProfile(userRunProfiles);
}
return null;
}

65
rc-service/rc-service-other/other-infrastructure/src/main/resources/dao/user/user/BaseInfoDAO.xml
View File

@ -1,8 +1,69 @@
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
"http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.red.circle.other.infra.database.rds.dao.user.user.BaseInfoDAO">
<select id="maxUserAccount" resultType="java.lang.Long">
<mapper namespace="com.red.circle.other.infra.database.rds.dao.user.user.BaseInfoDAO">
<select id="listCountryUsersAfterId"
resultType="com.red.circle.other.infra.database.rds.entity.user.user.BaseInfo">
SELECT id,
country_code AS countryCode,
country_name AS countryName,
origin_sys AS originSys
FROM user_base_info
WHERE country_id = #{countryId}
AND id &gt; #{lastId}
ORDER BY id ASC
LIMIT #{limit}
</select>
<select id="listDistinctOriginSysByCountryId" resultType="java.lang.String">
SELECT DISTINCT origin_sys
FROM user_base_info
WHERE country_id = #{countryId}
AND origin_sys IS NOT NULL
AND origin_sys != ''
</select>
<select id="listDistinctCountryCodesByCountryId" resultType="java.lang.String">
SELECT DISTINCT country_code
FROM user_base_info
WHERE country_id = #{countryId}
AND country_code IS NOT NULL
AND country_code != ''
</select>
<select id="existsCountryCodeMismatch" resultType="java.lang.Boolean">
SELECT EXISTS(
SELECT 1
FROM user_base_info
WHERE country_id = #{countryId}
AND (country_code IS NULL OR country_code != #{countryCode})
LIMIT 1
)
</select>
<select id="existsCountryNameMismatch" resultType="java.lang.Boolean">
SELECT EXISTS(
SELECT 1
FROM user_base_info
WHERE country_id = #{countryId}
AND (country_name IS NULL OR country_name != #{countryName})
LIMIT 1
)
</select>
<update id="updateCountryByIds">
UPDATE user_base_info
SET country_code = #{countryCode},
country_name = #{countryName},
update_time = CURRENT_TIMESTAMP
WHERE country_id = #{countryId}
AND id IN
<foreach collection="userIds" item="userId" open="(" separator="," close=")">
#{userId}
</foreach>
</update>
<select id="maxUserAccount" resultType="java.lang.Long">
SELECT MAX(account + 0)
FROM user_base_info
</select>

18
rc-service/rc-service-other/other-infrastructure/src/main/resources/dao/user/user/ExpandDAO.xml
View File

@ -1,9 +1,21 @@
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
"http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.red.circle.other.infra.database.rds.dao.user.user.ExpandDAO">
<select id="findLatestActiveUserCountryCode"
<mapper namespace="com.red.circle.other.infra.database.rds.dao.user.user.ExpandDAO">
<update id="updateRegisterCountryCodeByCountry">
UPDATE user_expand ue
INNER JOIN user_base_info ubi ON ubi.id = ue.user_id
SET ue.register_country_code = #{newCountryCode},
ue.update_time = CURRENT_TIMESTAMP
WHERE ubi.country_id = #{countryId}
AND ue.register_country_code IN
<foreach collection="sourceCountryCodes" item="sourceCountryCode" open="(" separator="," close=")">
#{sourceCountryCode}
</foreach>
</update>
<select id="findLatestActiveUserCountryCode"
resultType="com.red.circle.other.inner.model.dto.user.ActiveUserCountryCodeDTO">
SELECT alpha_three AS name,
COUNT(1) AS `value`

32
rc-service/rc-service-other/other-inner-endpoint/src/main/java/com/red/circle/other/app/inner/service/sys/CountryCodeRdsUpdateService.java Normal file
View File

@ -0,0 +1,32 @@
package com.red.circle.other.app.inner.service.sys;
import com.red.circle.other.infra.database.rds.entity.sys.SysCountryCode;
import com.red.circle.other.infra.database.rds.service.sys.SysCountryCodeService;
import com.red.circle.other.infra.database.rds.service.user.user.ExpandService;
import java.util.Set;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
/**
* 国家主记录与注册国家快照的同库事务更新服务.
*/
@Service
@RequiredArgsConstructor
public class CountryCodeRdsUpdateService {
private final ExpandService expandService;
private final SysCountryCodeService sysCountryCodeService;
/**
* 先更新国家主记录以触发唯一键校验，再在同一事务内重命名旧注册国家码.
* 任一步失败都会回滚，避免国家主记录已变更但注册快照仍残留旧码.
*/
@Transactional(rollbackFor = Exception.class)
public int update(SysCountryCode countryUpdate, Set<String> sourceCountryCodes,
String newCountryCode) {
sysCountryCodeService.updateCountryCode(countryUpdate);
return expandService.updateRegisterCountryCodeByCountry(countryUpdate.getId(),
sourceCountryCodes, newCountryCode);
}
}

474
rc-service/rc-service-other/other-inner-endpoint/src/main/java/com/red/circle/other/app/inner/service/sys/CountryDataSyncService.java Normal file
View File

@ -0,0 +1,474 @@
package com.red.circle.other.app.inner.service.sys;
import com.red.circle.common.business.core.util.CountryCodeAliasUtils;
import com.red.circle.component.redis.service.RedisService;
import com.red.circle.framework.core.asserts.ResponseAssert;
import com.red.circle.framework.core.response.CommonErrorCode;
import com.red.circle.other.infra.database.cache.service.user.UserCacheService;
import com.red.circle.other.infra.database.cache.service.user.RegionRoomCacheService;
import com.red.circle.other.infra.database.cache.service.user.UserRegionCacheService;
import com.red.circle.other.infra.database.cache.service.user.UserRunProfileCacheService;
import com.red.circle.other.infra.database.mongo.entity.live.ActiveVoiceRoom;
import com.red.circle.other.infra.database.mongo.entity.live.RoomProfile;
import com.red.circle.other.infra.database.mongo.entity.team.team.TeamMember;
import com.red.circle.other.infra.database.mongo.entity.team.team.TeamProfile;
import com.red.circle.other.infra.database.mongo.entity.user.region.SysRegionConfig;
import com.red.circle.other.infra.database.mongo.service.live.ActiveVoiceRoomService;
import com.red.circle.other.infra.database.mongo.service.live.RoomProfileManagerService;
import com.red.circle.other.infra.database.mongo.service.activity.AgentActivityCountService;
import com.red.circle.other.infra.database.mongo.service.team.team.TeamBillCycleService;
import com.red.circle.other.infra.database.mongo.service.team.team.TeamMemberService;
import com.red.circle.other.infra.database.mongo.service.team.team.TeamMemberTargetService;
import com.red.circle.other.infra.database.mongo.service.team.team.TeamProfileService;
import com.red.circle.other.infra.database.mongo.service.user.active.UserDailyActiveService;
import com.red.circle.other.infra.database.mongo.service.user.profile.UserRunProfileService;
import com.red.circle.other.infra.database.mongo.service.user.region.SysRegionConfigService;
import com.red.circle.other.infra.database.rds.entity.sys.SysCountryCode;
import com.red.circle.other.infra.database.rds.entity.user.user.BaseInfo;
import com.red.circle.other.infra.database.rds.service.user.user.BaseInfoService;
import com.red.circle.other.infra.common.user.UserCountryWriteLockService;
import com.red.circle.other.inner.model.dto.agency.agency.TeamCountry;
import com.red.circle.other.inner.model.cache.EmptyRoomCacheKeys;
import com.red.circle.tool.core.collection.CollectionUtils;
import com.red.circle.tool.core.text.StringUtils;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.stream.Collectors;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
/**
* 国家资料冗余字段同步服务.
*
* <p>用户和团队均以不可变 countryId 为归属依据；国家码只作为需要同步的属性，
* 避免新旧国家曾共用同一编码时误迁数据。</p>
*/
@Slf4j
@Service
@RequiredArgsConstructor
public class CountryDataSyncService {
private static final int USER_BATCH_SIZE = 300;
private static final int TEAM_BATCH_SIZE = 200;
private final BaseInfoService baseInfoService;
private final RedisService redisService;
private final RegionRoomCacheService regionRoomCacheService;
private final UserCountryWriteLockService userCountryWriteLockService;
private final AgentActivityCountService agentActivityCountService;
private final TeamBillCycleService teamBillCycleService;
private final TeamMemberService teamMemberService;
private final TeamMemberTargetService teamMemberTargetService;
private final TeamProfileService teamProfileService;
private final UserCacheService userCacheService;
private final UserRegionCacheService userRegionCacheService;
private final UserRunProfileCacheService userRunProfileCacheService;
private final UserRunProfileService userRunProfileService;
private final RoomProfileManagerService roomProfileManagerService;
private final ActiveVoiceRoomService activeVoiceRoomService;
private final UserDailyActiveService userDailyActiveService;
private final SysRegionConfigService sysRegionConfigService;
/**
* 在写国家主记录前生成同步计划并校验每个应用来源的目标区域唯一性.
*/
public CountrySyncPlan prepare(Long countryId, String oldCountryCode, String targetCountryCode,
String targetCountryName, boolean countryCodeChanged, boolean countryNameChanged) {
boolean userCodeMismatch = baseInfoService.existsCountryCodeMismatch(countryId,
targetCountryCode);
boolean userNameMismatch = baseInfoService.existsCountryNameMismatch(countryId,
targetCountryName);
boolean teamCodeMismatch = teamProfileService.existsCountryCodeMismatch(countryId,
targetCountryCode);
boolean teamNameMismatch = teamProfileService.existsCountryNameMismatch(countryId,
targetCountryName);
// 后台表单始终会提交 alphaTwo，因此以真实变更或编码残留判定是否重算区域.
boolean syncRegion = countryCodeChanged || userCodeMismatch || teamCodeMismatch;
Set<String> registerSourceCountryCodes = new HashSet<>();
if (syncRegion) {
registerSourceCountryCodes.addAll(
baseInfoService.listCountryCodesByCountryId(countryId).stream()
.filter(StringUtils::isNotBlank)
.filter(code -> !sameCountryCode(code, targetCountryCode))
.collect(Collectors.toSet()));
if (countryCodeChanged && StringUtils.isNotBlank(oldCountryCode)
&& !sameCountryCode(oldCountryCode, targetCountryCode)) {
registerSourceCountryCodes.add(oldCountryCode);
}
}
Map<String, TargetRegion> targetRegions;
boolean teamRegionMismatch;
if (syncRegion) {
Set<String> origins = new HashSet<>(baseInfoService.listOriginSysByCountryId(countryId));
origins.addAll(teamProfileService.listOriginSysByCountryId(countryId));
origins.removeIf(StringUtils::isBlank);
targetRegions = resolveTargetRegions(countryId, targetCountryCode, origins);
teamRegionMismatch = targetRegions.entrySet().stream()
.anyMatch(entry -> teamProfileService.existsCountryRegionMismatch(countryId,
entry.getKey(), entry.getValue().regionId()));
} else {
targetRegions = Map.of();
teamRegionMismatch = false;
}
boolean syncRequired = countryCodeChanged || countryNameChanged || userCodeMismatch
|| userNameMismatch || teamCodeMismatch || teamNameMismatch || teamRegionMismatch;
return new CountrySyncPlan(syncRequired, syncRegion, targetRegions,
registerSourceCountryCodes);
}
/**
* 执行用户与团队同步；所有步骤均可按 countryId 重复执行.
*/
public CountrySyncResult synchronize(SysCountryCode oldCountry, SysCountryCode targetCountry,
Long updateUser, CountrySyncPlan plan) {
if (!plan.syncRequired()) {
return new CountrySyncResult(0, 0, 0);
}
// 先改用户再扫团队：建团流程会从用户取国家快照，可避免批次末尾新团队落入旧编码.
int userCount = synchronizeUsers(targetCountry, plan);
TeamSyncResult teamResult = synchronizeTeams(targetCountry, updateUser, plan);
validateSynchronized(targetCountry, plan);
log.info("国家资料级联同步完成 countryId={}, code={} -> {}, users={}, teams={}, movedTeams={}",
targetCountry.getId(), oldCountry.getAlphaTwo(), targetCountry.getAlphaTwo(), userCount,
teamResult.updatedTeams(), teamResult.movedTeams());
return new CountrySyncResult(userCount, teamResult.updatedTeams(), teamResult.movedTeams());
}
private void validateSynchronized(SysCountryCode targetCountry, CountrySyncPlan plan) {
boolean mismatch = baseInfoService.existsCountryCodeMismatch(targetCountry.getId(),
targetCountry.getAlphaTwo())
|| baseInfoService.existsCountryNameMismatch(targetCountry.getId(),
targetCountry.getCountryName())
|| teamProfileService.existsCountryCodeMismatch(targetCountry.getId(),
targetCountry.getAlphaTwo())
|| teamProfileService.existsCountryNameMismatch(targetCountry.getId(),
targetCountry.getCountryName());
if (plan.syncRegion()) {
mismatch = mismatch || plan.targetRegionByOrigin().entrySet().stream()
.anyMatch(entry -> teamProfileService.existsCountryRegionMismatch(targetCountry.getId(),
entry.getKey(), entry.getValue().regionId()));
}
// 结束前二次校验游标后方新数据；不一致则明确失败，同值保存可幂等重试.
ResponseAssert.isFalse(CommonErrorCode.OPERATION_CONFLICT, mismatch);
}
private int synchronizeUsers(SysCountryCode targetCountry, CountrySyncPlan plan) {
long lastId = 0L;
int updated = 0;
while (true) {
List<BaseInfo> users = baseInfoService.listCountryUsersAfterId(targetCountry.getId(), lastId,
USER_BATCH_SIZE);
if (CollectionUtils.isEmpty(users)) {
break;
}
List<Long> userIds = users.stream().map(BaseInfo::getId).filter(Objects::nonNull).toList();
if (CollectionUtils.isEmpty(userIds)) {
break;
}
Integer batchUpdated = userCountryWriteLockService.execute(userIds, () -> {
// 加锁后重查稳定countryId，剔除刚被后台切到其他国家的用户.
List<BaseInfo> confirmedUsers = baseInfoService.listByUserId(userIds).stream()
.filter(user -> Objects.equals(user.getCountryId(), targetCountry.getId()))
.toList();
List<Long> confirmedUserIds = confirmedUsers.stream().map(BaseInfo::getId).toList();
if (CollectionUtils.isEmpty(confirmedUserIds)) {
return 0;
}
List<ActiveVoiceRoom> activeRooms = activeVoiceRoomService.listByUserIds(
confirmedUserIds);
List<RoomProfile> roomProfiles = roomProfileManagerService.listProfileByUserIds(
new HashSet<>(confirmedUserIds));
// 先同步跨库投影，user_base_info 最后落盘作为本批次完成标记；失败后同值保存可重试.
userRunProfileService.updateCountry(confirmedUserIds, targetCountry.getId(),
targetCountry.getAlphaTwo(), targetCountry.getCountryName());
roomProfileManagerService.updateUserCountry(confirmedUserIds, targetCountry.getAlphaTwo(),
targetCountry.getCountryName());
activeVoiceRoomService.updateUserCountry(confirmedUserIds, targetCountry.getAlphaTwo(),
targetCountry.getCountryName());
if (plan.syncRegion()) {
updateActiveRoomRegions(confirmedUsers, plan);
}
userDailyActiveService.updateTodayCountryCode(confirmedUserIds,
targetCountry.getAlphaTwo());
teamMemberTargetService.updateCurrentCountryCodeByUserIds(confirmedUserIds,
targetCountry.getAlphaTwo());
clearUserCaches(confirmedUserIds);
// 发现页缓存与空房快照都包含国家/区域冗余字段，主表落盘前先清理以保留重试标记.
clearRoomCaches(activeRooms, roomProfiles, plan.targetRegionByOrigin());
int affected = baseInfoService.updateCountryByIds(targetCountry.getId(), confirmedUserIds,
targetCountry.getAlphaTwo(), targetCountry.getCountryName());
clearUserCachesBestEffort(confirmedUserIds);
clearRoomCachesBestEffort(activeRooms, roomProfiles, plan.targetRegionByOrigin());
return affected;
});
updated += batchUpdated;
lastId = userIds.get(userIds.size() - 1);
}
return updated;
}
private TeamSyncResult synchronizeTeams(SysCountryCode targetCountry, Long updateUser,
CountrySyncPlan plan) {
int updatedTeams = 0;
int movedTeams = 0;
TeamCountry targetTeamCountry = new TeamCountry()
.setCountryId(targetCountry.getId())
.setCountryCode(targetCountry.getAlphaTwo())
.setCountryName(targetCountry.getCountryName());
long lastTeamId = 0L;
while (true) {
List<TeamProfile> teams = teamProfileService.listCountryTeamsAfterId(targetCountry.getId(),
lastTeamId, TEAM_BATCH_SIZE);
if (CollectionUtils.isEmpty(teams)) {
break;
}
for (TeamProfile currentTeam : teams) {
lastTeamId = currentTeam.getId();
List<Long> ownerLockIds = currentTeam.getOwnUserId() == null
? List.of() : List.of(currentTeam.getOwnUserId());
TeamItemSyncResult itemResult = userCountryWriteLockService.execute(ownerLockIds,
() -> synchronizeTeam(targetCountry, updateUser, plan, targetTeamCountry,
currentTeam.getId()));
if (itemResult.updated()) {
updatedTeams++;
if (itemResult.moved()) {
movedTeams++;
}
}
}
}
return new TeamSyncResult(updatedTeams, movedTeams);
}
private TeamItemSyncResult synchronizeTeam(SysCountryCode targetCountry, Long updateUser,
CountrySyncPlan plan, TeamCountry targetTeamCountry, Long teamId) {
// 取得团长国家锁后重读，避免扫描到正在初始化的半成品团队.
TeamProfile currentTeam = teamProfileService.getById(teamId);
if (currentTeam == null || currentTeam.getCountry() == null
|| !Objects.equals(currentTeam.getCountry().getCountryId(), targetCountry.getId())) {
return new TeamItemSyncResult(false, false);
}
TargetRegion targetRegion = plan.syncRegion()
? plan.targetRegionByOrigin().get(currentTeam.getSysOrigin())
: new TargetRegion(currentTeam.getRegion(), null);
ResponseAssert.isTrue(CommonErrorCode.CONFIGURATION_ERROR, targetRegion != null
&& StringUtils.isNotBlank(targetRegion.regionId()));
boolean regionChanged = !Objects.equals(currentTeam.getRegion(), targetRegion.regionId());
Set<Long> memberIds = regionChanged
? teamMemberService.getMemberIdsByTeamId(currentTeam.getId()) : Set.of();
if (regionChanged) {
synchronizeTeamRegionProjections(currentTeam, targetRegion, memberIds, updateUser);
}
// team_profile 最后以当前区域做CAS写入，避免覆盖并发发生的手工团队迁区.
boolean teamUpdated = teamProfileService.updateCountryAndRegion(currentTeam.getId(),
targetCountry.getId(), currentTeam.getRegion(), targetRegion.regionId(),
targetTeamCountry, updateUser);
if (!teamUpdated && regionChanged) {
// CAS失败说明团队被并发修改，立即按最新团队区域恢复先行投影，避免覆盖人工迁区结果.
restoreTeamRegionProjections(currentTeam.getId(), memberIds, updateUser);
}
ResponseAssert.isTrue(CommonErrorCode.OPERATION_CONFLICT, teamUpdated);
if (regionChanged) {
clearUserRegionCachesBestEffort(memberIds);
}
return new TeamItemSyncResult(true, regionChanged);
}
private void synchronizeTeamRegionProjections(TeamProfile team, TargetRegion targetRegion,
Set<Long> memberIds, Long updateUser) {
List<ActiveVoiceRoom> activeRooms = CollectionUtils.isEmpty(memberIds)
? CollectionUtils.newArrayList() : activeVoiceRoomService.listByUserIds(memberIds);
List<RoomProfile> roomProfiles = CollectionUtils.isEmpty(memberIds)
? CollectionUtils.newArrayList()
: roomProfileManagerService.listProfileByUserIds(memberIds);
if (CollectionUtils.isNotEmpty(memberIds)) {
// 活跃房间保存的是区域编码；定向置顶房间由服务层保留其固定区域.
activeVoiceRoomService.updateUserDefaultRegion(memberIds, targetRegion.regionCode());
userRegionCacheService.remove(memberIds);
}
// 仅迁移当前有效数据，保留历史账单、累计流水、目标和活动积分.
teamBillCycleService.updateUnpaidTeamBillRegion(team.getId(), targetRegion.regionId(),
updateUser);
teamMemberTargetService.updateCurrentRegionByTeamId(team.getId(), targetRegion.regionId());
agentActivityCountService.updateCurrentRegion(team.getId(), targetRegion.regionId());
clearRoomCaches(activeRooms, roomProfiles, Map.of(team.getSysOrigin(), targetRegion));
}
private void restoreTeamRegionProjections(Long teamId, Set<Long> memberIds, Long updateUser) {
TeamProfile latestTeam = teamProfileService.getById(teamId);
if (latestTeam == null || StringUtils.isBlank(latestTeam.getRegion())) {
return;
}
SysRegionConfig latestRegion = sysRegionConfigService.getById(latestTeam.getRegion());
if (latestRegion == null || StringUtils.isBlank(latestRegion.getRegionCode())) {
return;
}
synchronizeTeamRegionProjections(latestTeam,
new TargetRegion(latestRegion.getId(), latestRegion.getRegionCode()), memberIds, updateUser);
}
private void clearRoomCaches(List<ActiveVoiceRoom> activeRooms, List<RoomProfile> roomProfiles,
Map<String, TargetRegion> targetRegions) {
Set<String> discoveryKeys = new HashSet<>();
for (ActiveVoiceRoom room : activeRooms) {
if (StringUtils.isNotBlank(room.getSysOrigin()) && StringUtils.isNotBlank(room.getRegion())) {
discoveryKeys.add(room.getSysOrigin() + room.getRegion());
discoveryKeys.add(room.getSysOrigin() + room.getRegion() + ":ALL_REGION");
}
}
targetRegions.forEach((origin, region) -> {
if (StringUtils.isNotBlank(origin) && region != null
&& StringUtils.isNotBlank(region.regionCode())) {
discoveryKeys.add(origin + region.regionCode());
discoveryKeys.add(origin + region.regionCode() + ":ALL_REGION");
}
});
discoveryKeys.forEach(regionRoomCacheService::remove);
List<String> emptyRoomPayloadKeys = roomProfiles.stream().map(RoomProfile::getId)
.filter(Objects::nonNull).map(EmptyRoomCacheKeys::payload).toList();
if (CollectionUtils.isNotEmpty(emptyRoomPayloadKeys)) {
// payload 删除后，小型区域ZSet中的旧成员会被 reader 有界清理，无需全库扫描.
redisService.delete(emptyRoomPayloadKeys);
}
}
private void clearRoomCachesBestEffort(List<ActiveVoiceRoom> activeRooms,
List<RoomProfile> roomProfiles, Map<String, TargetRegion> targetRegions) {
try {
clearRoomCaches(activeRooms, roomProfiles, targetRegions);
} catch (RuntimeException ex) {
log.warn("国家资料同步后二次房间缓存失效失败", ex);
}
}
private Map<String, TargetRegion> resolveTargetRegions(Long countryId, String countryCode,
Set<String> origins) {
Map<String, TargetRegion> result = new HashMap<>();
for (String origin : origins) {
List<SysRegionConfig> matched = sysRegionConfigService.listAvailable(origin).stream()
.filter(region -> containsExactCountryCode(region.getCountryCodes(), countryCode))
.toList();
if (matched.size() != 1) {
log.error("国家码目标区域配置不唯一 countryId={}, countryCode={}, sysOrigin={}, matches={}",
countryId, countryCode, origin,
matched.stream().map(SysRegionConfig::getId).collect(Collectors.toList()));
}
ResponseAssert.isTrue(CommonErrorCode.CONFIGURATION_ERROR, matched.size() == 1
&& StringUtils.isNotBlank(matched.get(0).getRegionCode()));
result.put(origin, new TargetRegion(matched.get(0).getId(),
matched.get(0).getRegionCode()));
}
return result;
}
private boolean containsExactCountryCode(String csvCodes, String countryCode) {
if (StringUtils.isBlank(csvCodes) || StringUtils.isBlank(countryCode)) {
return false;
}
String normalized = CountryCodeAliasUtils.normalize(countryCode);
for (String code : csvCodes.split(",")) {
if (Objects.equals(normalized, CountryCodeAliasUtils.normalize(code))) {
return true;
}
}
return false;
}
private boolean sameCountryCode(String left, String right) {
if (StringUtils.isBlank(left) || StringUtils.isBlank(right)) {
return Objects.equals(left, right);
}
return Objects.equals(CountryCodeAliasUtils.normalize(left),
CountryCodeAliasUtils.normalize(right));
}
private void updateActiveRoomRegions(List<BaseInfo> users, CountrySyncPlan plan) {
Set<Long> userIds = users.stream().map(BaseInfo::getId).filter(Objects::nonNull)
.collect(Collectors.toSet());
Map<Long, TeamMember> memberByUser = teamMemberService.mapByMemberIds(userIds);
Map<Long, TeamProfile> teamById = teamProfileService.mapProfileByIds(memberByUser.values()
.stream().map(TeamMember::getTeamId).filter(Objects::nonNull).collect(Collectors.toSet()));
Set<String> teamRegionIds = teamById.values().stream().map(TeamProfile::getRegion)
.filter(StringUtils::isNotBlank).collect(Collectors.toSet());
Map<String, SysRegionConfig> teamRegions = sysRegionConfigService.mapAvailable(teamRegionIds);
Map<String, List<Long>> usersByRegionCode = new HashMap<>();
for (BaseInfo user : users) {
TeamMember member = memberByUser.get(user.getId());
TeamProfile team = member == null ? null : teamById.get(member.getTeamId());
SysRegionConfig teamRegion = team == null ? null : teamRegions.get(team.getRegion());
String regionCode = teamRegion == null ? null : teamRegion.getRegionCode();
if (StringUtils.isBlank(regionCode)) {
TargetRegion countryRegion = plan.targetRegionByOrigin().get(user.getOriginSys());
regionCode = countryRegion == null ? null : countryRegion.regionCode();
}
ResponseAssert.isTrue(CommonErrorCode.CONFIGURATION_ERROR,
StringUtils.isNotBlank(regionCode));
usersByRegionCode.computeIfAbsent(regionCode, ignored -> CollectionUtils.newArrayList())
.add(user.getId());
}
usersByRegionCode.forEach(
(regionCode, regionUserIds) -> activeVoiceRoomService.updateUserDefaultRegion(
regionUserIds, regionCode));
}
private void clearUserCaches(List<Long> userIds) {
userRunProfileCacheService.remove(userIds);
userRegionCacheService.remove(userIds);
userCacheService.remove(userIds);
}
private void clearUserCachesBestEffort(List<Long> userIds) {
try {
// 第二次失效封住“第一次删缓存到主表落盘”之间极短窗口内被回填的旧值.
clearUserCaches(userIds);
} catch (RuntimeException ex) {
log.warn("国家资料同步后二次缓存失效失败 userCount={}", userIds.size(), ex);
}
}
private void clearUserRegionCachesBestEffort(Collection<Long> userIds) {
try {
userRegionCacheService.remove(userIds);
} catch (RuntimeException ex) {
log.warn("团队区域同步后二次区域缓存失效失败 userCount={}", userIds.size(), ex);
}
}
public record CountrySyncPlan(boolean syncRequired, boolean syncRegion,
Map<String, TargetRegion> targetRegionByOrigin,
Set<String> registerSourceCountryCodes) {
}
public record TargetRegion(String regionId, String regionCode) {
}
public record CountrySyncResult(int updatedUsers, int updatedTeams, int movedTeams) {
}
private record TeamSyncResult(int updatedTeams, int movedTeams) {
}
private record TeamItemSyncResult(boolean updated, boolean moved) {
}
}

113
rc-service/rc-service-other/other-inner-endpoint/src/main/java/com/red/circle/other/app/inner/service/sys/impl/SysCountryCodeClientServiceImpl.java
View File

@ -1,29 +1,44 @@
package com.red.circle.other.app.inner.service.sys.impl;
import com.red.circle.framework.dto.PageResult;
import com.red.circle.other.app.inner.service.sys.SysCountryCodeClientService;
import com.red.circle.other.infra.convertor.sys.CountryCodeInnerConvertor;
import com.red.circle.other.infra.database.rds.service.sys.SysCountryCodeService;
import com.red.circle.other.inner.model.cmd.sys.SysCountryCodeCmd;
import com.red.circle.other.inner.model.cmd.sys.SysCountryCodePageQryCmd;
import com.red.circle.other.inner.model.dto.sys.SysCountryCodeDTO;
import java.util.List;
import java.util.Map;
import java.util.Set;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;
package com.red.circle.other.app.inner.service.sys.impl;
import com.red.circle.common.business.core.util.CountryCodeAliasUtils;
import com.red.circle.framework.core.asserts.ResponseAssert;
import com.red.circle.framework.core.response.CommonErrorCode;
import com.red.circle.framework.core.response.ResponseErrorCode;
import com.red.circle.framework.dto.PageResult;
import com.red.circle.other.app.inner.service.sys.CountryCodeRdsUpdateService;
import com.red.circle.other.app.inner.service.sys.CountryDataSyncService;
import com.red.circle.other.app.inner.service.sys.SysCountryCodeClientService;
import com.red.circle.other.infra.common.sys.CountryDataSyncLockService;
import com.red.circle.other.infra.convertor.sys.CountryCodeInnerConvertor;
import com.red.circle.other.infra.database.rds.entity.sys.SysCountryCode;
import com.red.circle.other.infra.database.rds.service.sys.SysCountryCodeService;
import com.red.circle.other.inner.model.cmd.sys.SysCountryCodeCmd;
import com.red.circle.other.inner.model.cmd.sys.SysCountryCodePageQryCmd;
import com.red.circle.other.inner.model.dto.sys.SysCountryCodeDTO;
import com.red.circle.tool.core.text.StringUtils;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
/**
* 系统国家服务 实现.
*
* @author pengliang on 2023/5/24
*/
@Service
@RequiredArgsConstructor
public class SysCountryCodeClientServiceImpl implements SysCountryCodeClientService {
private final SysCountryCodeService sysCountryCodeService;
private final CountryCodeInnerConvertor countryCodeInnerConvertor;
@Service
@RequiredArgsConstructor
@Slf4j
public class SysCountryCodeClientServiceImpl implements SysCountryCodeClientService {
private final SysCountryCodeService sysCountryCodeService;
private final CountryCodeInnerConvertor countryCodeInnerConvertor;
private final CountryDataSyncService countryDataSyncService;
private final CountryDataSyncLockService countryDataSyncLockService;
private final CountryCodeRdsUpdateService countryCodeRdsUpdateService;
@Override
public SysCountryCodeDTO getById(Long id) {
@ -71,9 +86,63 @@ public class SysCountryCodeClientServiceImpl implements SysCountryCodeClientServ
.convert(countryCodeInnerConvertor::toSysCountryCodeDTO);
}
@Override
@Override
public void updateCountryCode(SysCountryCodeCmd dto) {
sysCountryCodeService.updateCountryCode(countryCodeInnerConvertor.toSysCountryCode(dto));
SysCountryCode countryUpdate = countryCodeInnerConvertor.toSysCountryCode(dto);
if (dto.getId() == null) {
sysCountryCodeService.updateCountryCode(countryUpdate);
return;
}
// 开放、置顶等轻量操作也复用此接口；只有提交了国家编码或名称才检查级联同步.
boolean canonicalFieldsSubmitted = dto.getAlphaTwo() != null || dto.getCountryName() != null;
if (!canonicalFieldsSubmitted) {
sysCountryCodeService.updateCountryCode(countryUpdate);
return;
}
countryDataSyncLockService.execute(dto.getId(), () -> {
SysCountryCode oldCountry = sysCountryCodeService.getById(dto.getId());
ResponseAssert.notNull(CommonErrorCode.NOT_FOUND_RECORD_INFO, oldCountry);
String targetCode = dto.getAlphaTwo() == null
? oldCountry.getAlphaTwo()
: CountryCodeAliasUtils.normalize(dto.getAlphaTwo());
String targetName = dto.getCountryName() == null
? oldCountry.getCountryName()
: dto.getCountryName().trim();
ResponseAssert.isTrue(ResponseErrorCode.REQUEST_PARAMETER_ERROR,
StringUtils.isNotBlank(targetCode) && StringUtils.isNotBlank(targetName));
if (dto.getAlphaTwo() != null) {
countryUpdate.setAlphaTwo(targetCode);
}
if (dto.getCountryName() != null) {
countryUpdate.setCountryName(targetName);
}
boolean countryCodeChanged = !Objects.equals(
CountryCodeAliasUtils.normalize(oldCountry.getAlphaTwo()), targetCode);
boolean countryNameChanged = !Objects.equals(oldCountry.getCountryName(), targetName);
CountryDataSyncService.CountrySyncPlan syncPlan = countryDataSyncService.prepare(
oldCountry.getId(), oldCountry.getAlphaTwo(), targetCode, targetName,
countryCodeChanged, countryNameChanged);
// 国家主记录与注册国家快照在一个RDS事务内完成；跨库用户/团队投影随后按countryId幂等同步.
int registerSnapshots = countryCodeRdsUpdateService.update(countryUpdate,
syncPlan.registerSourceCountryCodes(), targetCode);
if (syncPlan.syncRequired()) {
SysCountryCode targetCountry = new SysCountryCode()
.setId(oldCountry.getId())
.setAlphaTwo(targetCode)
.setCountryName(targetName);
CountryDataSyncService.CountrySyncResult result = countryDataSyncService.synchronize(
oldCountry, targetCountry, dto.getUpdateUser(), syncPlan);
log.info("国家资料更新同步完成 countryId={}, registerSnapshots={}, result={}",
oldCountry.getId(), registerSnapshots, result);
}
return null;
});
}
@Override

249
rc-service/rc-service-other/other-inner-endpoint/src/main/java/com/red/circle/other/app/inner/service/team/impl/TeamProfileClientServiceImpl.java
View File

@ -17,10 +17,12 @@ import com.red.circle.mq.rocket.business.service.MessageSenderService;
import com.red.circle.mq.rocket.business.streams.TeamSalaryCountSink;
import com.red.circle.other.app.inner.service.team.TeamManualSalaryPaymentClientService;
import com.red.circle.other.app.inner.service.team.TeamProfileClientService;
import com.red.circle.other.domain.gateway.user.UserProfileGateway;
import com.red.circle.other.domain.gateway.user.ability.RegisterDeviceGateway;
import com.red.circle.other.domain.model.user.UserProfile;
import com.red.circle.other.infra.convertor.team.TeamInfraConvertor;
import com.red.circle.other.domain.gateway.user.UserProfileGateway;
import com.red.circle.other.domain.gateway.user.ability.RegisterDeviceGateway;
import com.red.circle.other.domain.model.user.UserProfile;
import com.red.circle.other.infra.common.sys.CountryDataSyncLockService;
import com.red.circle.other.infra.common.user.UserCountryWriteLockService;
import com.red.circle.other.infra.convertor.team.TeamInfraConvertor;
import com.red.circle.other.infra.database.mongo.dto.team.TeamBillSettleResult;
import com.red.circle.other.infra.database.mongo.entity.team.team.*;
import com.red.circle.other.infra.database.mongo.entity.user.region.SysRegionConfig;
@ -37,9 +39,11 @@ import com.red.circle.other.infra.database.mongo.service.team.team.TeamNoticeSer
import com.red.circle.other.infra.database.mongo.service.team.team.TeamPolicyManagerService;
import com.red.circle.other.infra.database.mongo.service.team.team.TeamProfileService;
import com.red.circle.other.infra.database.mongo.service.team.team.TeamSalaryPaymentService;
import com.red.circle.other.infra.database.mongo.service.user.region.SysRegionConfigService;
import com.red.circle.other.infra.database.rds.entity.team.BusinessDevelopmentBaseInfo;
import com.red.circle.other.infra.database.rds.entity.team.BusinessDevelopmentTeam;
import com.red.circle.other.infra.database.mongo.service.user.region.SysRegionConfigService;
import com.red.circle.other.infra.database.rds.entity.sys.SysCountryCode;
import com.red.circle.other.infra.database.rds.entity.team.BusinessDevelopmentBaseInfo;
import com.red.circle.other.infra.database.rds.entity.team.BusinessDevelopmentTeam;
import com.red.circle.other.infra.database.rds.service.sys.SysCountryCodeService;
import com.red.circle.other.infra.database.rds.service.team.*;
import com.red.circle.other.inner.asserts.team.TeamErrorCode;
import com.red.circle.other.inner.asserts.user.UserErrorCode;
@ -83,12 +87,14 @@ import com.red.circle.tool.core.sequence.IdWorkerUtils;
import com.red.circle.tool.core.text.StringUtils;
import com.red.circle.wallet.inner.endpoint.bank.UserBankCardClient;
import java.math.BigDecimal;
import java.util.List;
import java.math.BigDecimal;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.Optional;
import java.util.Set;
import java.util.function.Supplier;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;
@ -129,6 +135,9 @@ public class TeamProfileClientServiceImpl implements TeamProfileClientService {
private final RegisterDeviceGateway registerDeviceGateway;
private final UserMqMessageService userMqMessageService;
private final TeamManualSalaryPaymentClientService teamManualSalaryPaymentClientService;
private final SysCountryCodeService sysCountryCodeService;
private final CountryDataSyncLockService countryDataSyncLockService;
private final UserCountryWriteLockService userCountryWriteLockService;
@Override
public List<TeamProfileDTO> listByIds(Set<Long> ids) {
@ -199,9 +208,19 @@ public class TeamProfileClientServiceImpl implements TeamProfileClientService {
}
@Override
public void create(TeamProfileCmd cmd) {
teamProfileService.create(teamInfraConvertor.toTeamProfile(cmd));
@Override
public void create(TeamProfileCmd cmd) {
TeamProfile teamProfile = teamInfraConvertor.toTeamProfile(cmd);
Long countryId = teamProfile.getCountry() == null
? null : teamProfile.getCountry().getCountryId();
executeTeamCreationLocks(countryId, teamProfile.getOwnUserId(), () -> {
if (countryId != null) {
// 内部创建接口也可能携带旧国家快照，稳定 countryId 对应的主表值才可落库.
teamProfile.setCountry(canonicalTeamCountry(countryId));
}
teamProfileService.create(teamProfile);
return null;
});
}
@ -210,11 +229,68 @@ public class TeamProfileClientServiceImpl implements TeamProfileClientService {
teamApplicationProcessService.changeUserTeamWaitReject(userId);
}
@Override
public void createTeam(TeamCreateCmd cmd) {
UserProfile userProfile = userProfileGateway.getByUserId(cmd.getOwnUserId());
ResponseAssert.notNull(UserErrorCode.USER_INFO_NOT_FOUND, userProfile);
@Override
public void createTeam(TeamCreateCmd cmd) {
UserProfile preflightUser = userProfileGateway.getByUserId(cmd.getOwnUserId());
ResponseAssert.notNull(UserErrorCode.USER_INFO_NOT_FOUND, preflightUser);
Long expectedCountryId = preflightUser.getCountryId();
// 锁顺序固定为 country -> user，与国家批量同步一致，避免交叉死锁.
executeTeamCreationLocks(expectedCountryId, cmd.getOwnUserId(), () -> {
createTeamLocked(cmd, expectedCountryId);
return null;
});
}
private void createTeamLocked(TeamCreateCmd cmd, Long expectedCountryId) {
TeamCreationContext creationContext = createTeamProfile(cmd, expectedCountryId);
UserProfile userProfile = creationContext.userProfile();
TeamProfileDTO teamProfile = creationContext.teamProfile();
// 初始化账单
teamBillCycleService.createIfAbsent(
new TeamBillCmd()
.setSysOrigin(teamProfile.getSysOrigin())
.setTeamId(teamProfile.getId())
.setRegion(teamProfile.getRegion())
.setOperationTime(TimestampUtils.now())
.setOperationBackUser(teamProfile.getCreateUser())
);
teamMemberService.addMember(new TeamMember()
.setSortId(IdWorkerUtils.getId())
.setSysOrigin(teamProfile.getSysOrigin())
.setMemberId(userProfile.getId())
.setTeamId(teamProfile.getId())
.setRole(TeamMemberRole.OWN)
.setRemarks("")
.setCreateTime(TimestampUtils.now())
.setCreateUserOrigin(OperationUserOrigin.BACK.getValue())
.setCreateUser(cmd.getCreateUser())
.setUpdateTime(TimestampUtils.now())
.setUpdateUserOrigin(OperationUserOrigin.BACK.getValue())
.setUpdateUser(cmd.getCreateUser())
.setActiveTime(TimestampUtils.now())
);
// 处理等待消息, 全部标记拒绝
teamApplicationProcessService.changeUserTeamWaitReject(userProfile.getId());
// 标记身份
userHistoryIdentityService.saveAgentAndHost(userProfile.getId());
// 佩戴主播和代理徽章
userMqMessageService.sendBadgeOperateEvent(userProfile.getId(), BadgeKeyEnum.AGENCY.name(), BadgeOperateEvent.OperationType.WEAR.name());
userMqMessageService.sendBadgeOperateEvent(userProfile.getId(), BadgeKeyEnum.HOST.name(), BadgeOperateEvent.OperationType.WEAR.name());
redisService.unlock("CreateTeam:" + userProfile.getId());
}
private TeamCreationContext createTeamProfile(TeamCreateCmd cmd, Long expectedCountryId) {
UserProfile userProfile = userProfileGateway.getByUserId(cmd.getOwnUserId());
ResponseAssert.notNull(UserErrorCode.USER_INFO_NOT_FOUND, userProfile);
ResponseAssert.isTrue(CommonErrorCode.OPERATION_CONFLICT,
Objects.equals(expectedCountryId, userProfile.getCountryId()));
ResponseAssert.isTrue(UserErrorCode.USER_INFO_NOT_FOUND,
Objects.equals(userProfile.getOriginSys(), cmd.getSysOrigin()));
ResponseAssert.isFalse(TeamErrorCode.USER_ALREADY_TEAM_OWN,
@ -254,12 +330,7 @@ public class TeamProfileClientServiceImpl implements TeamProfileClientService {
);
teamProfile.setRemarks(Lists.newArrayList());
teamProfile.setSetting(cmd.getSetting());
teamProfile.setCountry(
new TeamCountry()
.setCountryId(userProfile.getCountryId())
.setCountryCode(userProfile.getCountryCode())
.setCountryName(userProfile.getCountryName())
);
teamProfile.setCountry(resolveTeamCountry(userProfile));
if (StringUtils.isBlank(teamProfile.getSysOrigin())) {
teamProfile.setSysOrigin(userProfile.getOriginSys());
@ -305,46 +376,38 @@ public class TeamProfileClientServiceImpl implements TeamProfileClientService {
businessDevelopmentBaseInfoService.incrMemberCount(Set.of(cmd.getBdUserId()));
}
// 创建团队
teamProfileService.create(teamInfraConvertor.toTeamProfile(teamProfile));
// 初始化账单
teamBillCycleService.createIfAbsent(
new TeamBillCmd()
.setSysOrigin(teamProfile.getSysOrigin())
.setTeamId(teamProfile.getId())
.setRegion(teamProfile.getRegion())
.setOperationTime(TimestampUtils.now())
.setOperationBackUser(teamProfile.getCreateUser())
);
teamMemberService.addMember(new TeamMember()
.setSortId(IdWorkerUtils.getId())
.setSysOrigin(teamProfile.getSysOrigin())
.setMemberId(userProfile.getId())
.setTeamId(teamProfile.getId())
.setRole(TeamMemberRole.OWN)
.setRemarks("")
.setCreateTime(TimestampUtils.now())
.setCreateUserOrigin(OperationUserOrigin.BACK.getValue())
.setCreateUser(cmd.getCreateUser())
.setUpdateTime(TimestampUtils.now())
.setUpdateUserOrigin(OperationUserOrigin.BACK.getValue())
.setUpdateUser(cmd.getCreateUser())
.setActiveTime(TimestampUtils.now())
);
// 处理等待消息, 全部标记拒绝
teamApplicationProcessService.changeUserTeamWaitReject(userProfile.getId());
// 标记身份
userHistoryIdentityService.saveAgentAndHost(userProfile.getId());
// 佩戴主播和代理徽章
userMqMessageService.sendBadgeOperateEvent(userProfile.getId(), BadgeKeyEnum.AGENCY.name(), BadgeOperateEvent.OperationType.WEAR.name());
userMqMessageService.sendBadgeOperateEvent(userProfile.getId(), BadgeKeyEnum.HOST.name(), BadgeOperateEvent.OperationType.WEAR.name());
redisService.unlock(key);
// 创建团队
teamProfileService.create(teamInfraConvertor.toTeamProfile(teamProfile));
return new TeamCreationContext(userProfile, teamProfile);
}
private TeamCountry resolveTeamCountry(UserProfile userProfile) {
if (userProfile.getCountryId() == null) {
return new TeamCountry()
.setCountryCode(userProfile.getCountryCode())
.setCountryName(userProfile.getCountryName());
}
return canonicalTeamCountry(userProfile.getCountryId());
}
private TeamCountry canonicalTeamCountry(Long countryId) {
// countryId 是稳定归属，团队冗余码与名称始终取国家主表的最新标准值.
SysCountryCode country = sysCountryCodeService.getById(countryId);
ResponseAssert.notNull(CommonErrorCode.NOT_FOUND_RECORD_INFO, country);
return new TeamCountry()
.setCountryId(country.getId())
.setCountryCode(country.getAlphaTwo())
.setCountryName(country.getCountryName());
}
private <T> T executeTeamCreationLocks(Long countryId, Long ownUserId, Supplier<T> action) {
Supplier<T> userLocked = () -> userCountryWriteLockService.execute(
ownUserId == null ? List.of() : List.of(ownUserId), action);
return countryId == null ? userLocked.get()
: countryDataSyncLockService.execute(countryId, userLocked);
}
private record TeamCreationContext(UserProfile userProfile, TeamProfileDTO teamProfile) {
}
private boolean existsAvailableTeamMember(Long userId) {
@ -373,17 +436,51 @@ public class TeamProfileClientServiceImpl implements TeamProfileClientService {
return businessDevelopmentBaseInfoService.getByUserId(param.getBdUserId());
}
@Override
public void updateTeam(TeamUpdateCmd cmd) {
TeamProfile oldTeamProfile = teamProfileService.getById(cmd.getId());
ResponseAssert.notNull(TeamErrorCode.TEAM_NOT_FOUND, oldTeamProfile);
teamProfileService.updateByIdSelective(teamInfraConvertor.toTeamProfile(cmd));
if (!Objects.equals(cmd.getRegion(), oldTeamProfile.getRegion())) {
SysRegionConfig newRegion = getSysRegionConfig(cmd.getRegion());
@Override
public void updateTeam(TeamUpdateCmd cmd) {
TeamProfile preflightTeam = teamProfileService.getById(cmd.getId());
ResponseAssert.notNull(TeamErrorCode.TEAM_NOT_FOUND, preflightTeam);
Long sourceCountryId = teamCountryId(preflightTeam.getCountry());
Long requestedCountryId = teamCountryId(cmd.getCountry());
Long targetCountryId = requestedCountryId == null ? sourceCountryId : requestedCountryId;
// 团队编辑与国家级联保持 country -> owner user 锁顺序，阻止迁移前旧表单在完成后回写旧码.
countryDataSyncLockService.execute(Arrays.asList(sourceCountryId, targetCountryId), () ->
userCountryWriteLockService.execute(
preflightTeam.getOwnUserId() == null ? List.of() : List.of(preflightTeam.getOwnUserId()),
() -> {
TeamProfile lockedTeam = teamProfileService.getById(cmd.getId());
ResponseAssert.notNull(TeamErrorCode.TEAM_NOT_FOUND, lockedTeam);
ResponseAssert.isTrue(CommonErrorCode.OPERATION_CONFLICT,
Objects.equals(sourceCountryId, teamCountryId(lockedTeam.getCountry())));
canonicalizeTeamCountry(cmd, targetCountryId);
updateTeamLocked(cmd, lockedTeam);
return null;
}));
}
private void canonicalizeTeamCountry(TeamUpdateCmd cmd, Long targetCountryId) {
if (targetCountryId == null) {
return;
}
// countryId 是团队国家归属，编码和名称始终重新取国家主表，不能信任页面旧快照.
cmd.setCountry(canonicalTeamCountry(targetCountryId));
}
private Long teamCountryId(TeamCountry country) {
return country == null ? null : country.getCountryId();
}
private void updateTeamLocked(TeamUpdateCmd cmd, TeamProfile oldTeamProfile) {
teamProfileService.updateByIdSelective(teamInfraConvertor.toTeamProfile(cmd));
String targetRegionId = StringUtils.isBlank(cmd.getRegion())
? oldTeamProfile.getRegion() : cmd.getRegion();
if (!Objects.equals(targetRegionId, oldTeamProfile.getRegion())) {
SysRegionConfig newRegion = getSysRegionConfig(targetRegionId);
SysRegionConfig oldRegion = getSysRegionConfig(oldTeamProfile.getRegion());
// 将之前的账单修改为历史存档状态

73
rc-service/rc-service-other/other-inner-endpoint/src/main/java/com/red/circle/other/app/inner/service/user/user/impl/UserProfileClientServiceImpl.java
View File

@ -8,19 +8,27 @@ import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.google.gson.TypeAdapter;
import com.google.gson.stream.JsonReader;
import com.google.gson.stream.JsonToken;
import com.google.gson.stream.JsonWriter;
import com.red.circle.framework.core.dto.PageCommand;
import com.google.gson.stream.JsonToken;
import com.google.gson.stream.JsonWriter;
import com.red.circle.framework.core.asserts.ResponseAssert;
import com.red.circle.framework.core.response.CommonErrorCode;
import com.red.circle.framework.core.response.ResponseErrorCode;
import com.red.circle.framework.core.dto.PageCommand;
import com.red.circle.framework.dto.PageResult;
import com.red.circle.other.app.service.user.user.UserAuthTypeService;
import com.red.circle.other.domain.model.user.UserProfile;
import com.red.circle.other.domain.gateway.user.UserProfileGateway;
import com.red.circle.other.infra.config.TimestampDeserializer;
import com.red.circle.other.domain.model.user.UserProfile;
import com.red.circle.other.domain.gateway.user.UserProfileGateway;
import com.red.circle.other.infra.common.sys.CountryDataSyncLockService;
import com.red.circle.other.infra.common.user.UserCountryWriteLockService;
import com.red.circle.other.infra.config.TimestampDeserializer;
import com.red.circle.other.infra.database.cache.service.user.UserCacheService;
import com.red.circle.other.infra.database.cache.service.user.UserRegionCacheService;
import com.red.circle.other.infra.database.cache.service.user.UserRunProfileCacheService;
import com.red.circle.other.infra.database.mongo.service.user.profile.UserRunProfileService;
import com.red.circle.other.infra.database.rds.service.user.user.BaseInfoService;
import com.red.circle.other.infra.database.cache.service.user.UserRegionCacheService;
import com.red.circle.other.infra.database.cache.service.user.UserRunProfileCacheService;
import com.red.circle.other.infra.database.mongo.service.user.profile.UserRunProfileService;
import com.red.circle.other.infra.database.rds.entity.sys.SysCountryCode;
import com.red.circle.other.infra.database.rds.entity.user.user.BaseInfo;
import com.red.circle.other.infra.database.rds.service.sys.SysCountryCodeService;
import com.red.circle.other.infra.database.rds.service.user.user.BaseInfoService;
import com.red.circle.other.infra.database.rds.service.user.user.RegisterInfoService;
import com.red.circle.other.infra.database.rds.service.user.user.RelationshipFriendService;
import com.red.circle.other.infra.database.rds.service.user.user.UserPhotoWallService;
@ -38,8 +46,8 @@ import com.red.circle.other.inner.model.dto.user.PetRelationshipFriendDTO;
import com.red.circle.other.inner.model.dto.user.UserPhotoWallDTO;
import com.red.circle.other.inner.model.dto.user.UserProfileDTO;
import com.red.circle.other.inner.model.dto.user.UserRegisterInfoDTO;
import com.red.circle.other.inner.model.dto.user.UserRunProfileDTO;
import com.red.circle.other.app.inner.service.user.user.UserProfileClientService;
import com.red.circle.other.inner.model.dto.user.UserRunProfileDTO;
import com.red.circle.other.app.inner.service.user.user.UserProfileClientService;
import java.io.IOException;
import java.sql.Timestamp;
@ -70,8 +78,11 @@ public class UserProfileClientServiceImpl implements UserProfileClientService {
private final UserProfileInnerConvertor userProfileInnerConvertor;
private final RelationshipFriendService relationshipFriendService;
private final UserAuthTypeService userAuthTypeService;
private final UserRunProfileCacheService userRunProfileCacheService;
private final UserRegionCacheService userRegionCacheService;
private final UserRunProfileCacheService userRunProfileCacheService;
private final UserRegionCacheService userRegionCacheService;
private final SysCountryCodeService sysCountryCodeService;
private final CountryDataSyncLockService countryDataSyncLockService;
private final UserCountryWriteLockService userCountryWriteLockService;
@Override
public void removeCacheAll(Long userId) {
@ -297,10 +308,36 @@ public class UserProfileClientServiceImpl implements UserProfileClientService {
}
@Override
public void updateSelectiveById(UserBaseInfoCmd cmd) {
userProfileGateway.updateSelectiveById(userProfileInnerConvertor.toUserProfile(cmd));
}
@Override
public void updateSelectiveById(UserBaseInfoCmd cmd) {
boolean countrySubmitted = cmd.getCountryId() != null || cmd.getCountryCode() != null
|| cmd.getCountryName() != null;
if (!countrySubmitted || cmd.getId() == null) {
userProfileGateway.updateSelectiveById(userProfileInnerConvertor.toUserProfile(cmd));
return;
}
BaseInfo preflight = baseInfoService.getById(cmd.getId());
ResponseAssert.notNull(CommonErrorCode.NOT_FOUND_RECORD_INFO, preflight);
Long sourceCountryId = preflight.getCountryId();
Long targetCountryId = cmd.getCountryId() == null ? sourceCountryId : cmd.getCountryId();
ResponseAssert.notNull(ResponseErrorCode.REQUEST_PARAMETER_ERROR, targetCountryId);
// 后台单用户改国家与App保持相同 country -> user 锁顺序，防止游标后迁入漏投影.
countryDataSyncLockService.execute(Arrays.asList(sourceCountryId, targetCountryId), () ->
userCountryWriteLockService.execute(List.of(cmd.getId()), () -> {
BaseInfo lockedUser = baseInfoService.getById(cmd.getId());
ResponseAssert.notNull(CommonErrorCode.NOT_FOUND_RECORD_INFO, lockedUser);
ResponseAssert.isTrue(CommonErrorCode.OPERATION_CONFLICT,
Objects.equals(sourceCountryId, lockedUser.getCountryId()));
SysCountryCode targetCountry = sysCountryCodeService.getById(targetCountryId);
ResponseAssert.notNull(CommonErrorCode.NOT_FOUND_RECORD_INFO, targetCountry);
// 管理页可能提交迁移前打开的旧快照，国家码和名称必须以稳定 countryId 的主表为准.
cmd.setCountryId(targetCountry.getId());
cmd.setCountryCode(targetCountry.getAlphaTwo());
cmd.setCountryName(targetCountry.getCountryName());
userProfileGateway.updateSelectiveById(userProfileInnerConvertor.toUserProfile(cmd));
return null;
}));
}
@Override
public void restoreDelById(Long id) {

30
sql/20260716_user_country_sync_indexes.sql Normal file
View File

@ -0,0 +1,30 @@
-- 国家资料级联同步按 (country_id, id) 游标分页；该索引避免国家更新时扫描整张用户表。
-- 生产表当前约 1.4 万行/4.5 MB，使用 INPLACE + LOCK=NONE，且限制元数据锁等待避免影响在线请求。
SET @current_schema = DATABASE();
SET SESSION lock_wait_timeout = 10;
SET @ddl = IF(
EXISTS (
SELECT 1
FROM INFORMATION_SCHEMA.STATISTICS first_column
WHERE first_column.TABLE_SCHEMA = @current_schema
AND first_column.TABLE_NAME = 'user_base_info'
AND first_column.SEQ_IN_INDEX = 1
AND first_column.COLUMN_NAME = 'country_id'
AND EXISTS (
SELECT 1
FROM INFORMATION_SCHEMA.STATISTICS second_column
WHERE second_column.TABLE_SCHEMA = first_column.TABLE_SCHEMA
AND second_column.TABLE_NAME = first_column.TABLE_NAME
AND second_column.INDEX_NAME = first_column.INDEX_NAME
AND second_column.SEQ_IN_INDEX = 2
AND second_column.COLUMN_NAME = 'id'
)
),
'SELECT 1',
'ALTER TABLE `user_base_info` ADD KEY `idx_country_id_id` (`country_id`, `id`), ALGORITHM=INPLACE, LOCK=NONE'
);
PREPARE stmt FROM @ddl;
EXECUTE stmt;
DEALLOCATE PREPARE stmt;

31
sql/20260716_user_country_sync_mongo_indexes.js Normal file
View File

@ -0,0 +1,31 @@
// 国家资料级联同步索引：批量迁移前执行，createIndex 对同名同定义索引幂等。
db.getCollection("team_profile").createIndex(
{"country.countryId": 1, _id: 1}, {name: "idx_country_id_cursor"});
db.getCollection("team_member").createIndex(
{teamId: 1, memberId: 1}, {name: "idx_team_member"});
db.getCollection("team_member").createIndex(
{memberId: 1, teamId: 1}, {name: "idx_member_team"});
db.getCollection("active_voice_room").createIndex(
{userId: 1}, {name: "idx_user_id"});
db.getCollection("room_profile_manager").createIndex(
{userId: 1}, {name: "idx_user_id"});
db.getCollection("user_daily_active_log").createIndex(
{userId: 1, activeDate: 1}, {name: "idx_uid_date", unique: true});
db.getCollection("user_daily_active_log").createIndex(
{activeDate: 1, sysOrigin: 1}, {name: "idx_date_origin"});
db.getCollection("user_daily_active_log").createIndex(
{activeDate: 1, registerCountryCode: 1}, {name: "idx_date_country"});
db.getCollection("team_bill_cycle").createIndex(
{billBelong: 1, status: 1}, {name: "idx_billBelong_status"});
db.getCollection("team_bill_cycle").createIndex(
{teamId: 1, status: 1}, {name: "idx_team_status"});
db.getCollection("team_member_target").createIndex(
{teamId: 1, history: 1}, {name: "idx_team_history"});
db.getCollection("team_member_target").createIndex(
{userId: 1, history: 1}, {name: "idx_user_history"});
db.getCollection("agent_activity_count").createIndex(
{teamId: 1, history: 1}, {name: "idx_team_history"});