log user tag（create session and save user）

doc/sql/update-20230630.sql

@@ -50,8 +50,11 @@ CREATE TABLE `xyqb_user`.`user_tag` (
   `id` BIGINT(20) NOT NULL AUTO_INCREMENT,
   `user_id` BIGINT(20) NOT NULL COMMENT '用户表id',
   `registered_from` BIGINT(20) NULL COMMENT '来源（之前字段）',
+  `tenant_id` INT NOT NULL COMMENT '租户id',
   `created_at` TIMESTAMP NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
   `updated_at` TIMESTAMP NULL DEFAULT CURRENT_TIMESTAMP  ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
+  UNIQUE `udx_user_id` (`user_id` ASC),
+  INDEX `idx_user_phone` (`registered_from` ASC),
   PRIMARY KEY (`id`));
 
 

src/main/java/cn/quantgroup/xyqb/config/AutoConfiguration.java

@@ -3,13 +3,19 @@ package cn.quantgroup.xyqb.config;
 import cn.quantgroup.xyqb.util.DefaultSequencer;
 import cn.quantgroup.xyqb.util.IdentitySequencer;
 import cn.quantgroup.xyqb.util.ServerUtils;
+import com.sensorsdata.analytics.javasdk.ISensorsAnalytics;
+import com.sensorsdata.analytics.javasdk.SensorsAnalytics;
+import com.sensorsdata.analytics.javasdk.consumer.BatchConsumer;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.core.env.Environment;
 
+import java.io.IOException;
+
 /**
  * Created by hechao on 2020/1/22.
  */
@@ -38,4 +44,16 @@ public class AutoConfiguration {
             return new DefaultSequencer(workId, sequencerProperties);
         }
     }
+
+    @Value("${sc.url}")
+    private String dataUrl;
+
+    @Bean(destroyMethod = "shutdown")
+    public ISensorsAnalytics init() throws IOException {
+//            //本地日志模式（此模式会在指定路径生成相应的日志文件）
+//            return new SensorsAnalytics(new ConcurrentLoggingConsumer("/Users/hongzhi/scEvent/"));
+        //网络批量发送模式（此模式在容器关闭的时候，如果存在数据还没有发送完毕，就会丢失未发送的数据！！！）
+        return new SensorsAnalytics(new BatchConsumer(dataUrl));
+    }
+
 }
\ No newline at end of file

src/main/java/cn/quantgroup/xyqb/controller/external/UserController.java

@@ -800,6 +800,7 @@ public class UserController implements IBaseController {
         if (geetestLogId != null) {
             geetestLogService.updateByUidGeetestLog(geetestLogId, user.getId());
         }
+
         //更新session
         return new JsonResult(sessionService.createSession(user, loginProperties, LoginType.ACCOUNTPASSWORD.ordinal(),tenantId));
     }

src/main/java/cn/quantgroup/xyqb/entity/UserTag.java

+package cn.quantgroup.xyqb.entity;
+
+import lombok.Builder;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+import javax.persistence.Column;
+import javax.persistence.Entity;
+import javax.persistence.Table;
+import java.io.Serializable;
+
+@EqualsAndHashCode(callSuper = true)
+@Data
+@Entity
+@Table(name = "user_tag")
+@Builder
+public class UserTag extends BaseEntity implements Serializable {
+    @Column(name = "user_id")
+    private Long userId;
+
+    //第一次用户来源 channel_id
+    @Column(name = "registered_from")
+    private Long registeredFrom;
+
+    @Column(name = "tenant_id")
+    private Integer tenantId;
+}

src/main/java/cn/quantgroup/xyqb/event/UserLoginEvent.java

 package cn.quantgroup.xyqb.event;
 
-import com.sensorsdata.analytics.javasdk.ISensorsAnalytics;
-import com.sensorsdata.analytics.javasdk.SensorsAnalytics;
-import com.sensorsdata.analytics.javasdk.consumer.BatchConsumer;
-import org.springframework.beans.factory.annotation.Value;
-import org.springframework.context.annotation.Bean;
+import cn.quantgroup.xyqb.entity.UserTag;
+import lombok.Builder;
+import lombok.Data;
+import lombok.Getter;
+import lombok.Setter;
+import org.springframework.context.ApplicationEvent;
 import org.springframework.context.annotation.Configuration;
 
-import java.io.IOException;
-
 /**
  * @author ：hongzhi
  */
-@Configuration
-public class UserLoginEvent {
-        @Value("${sc.url}")
-        private String dataUrl;
 
-        @Bean(destroyMethod = "shutdown")
-        public ISensorsAnalytics init() throws IOException {
-//            //本地日志模式（此模式会在指定路径生成相应的日志文件）
-//            return new SensorsAnalytics(new ConcurrentLoggingConsumer("/Users/hongzhi/scEvent/"));
-            //网络批量发送模式（此模式在容器关闭的时候，如果存在数据还没有发送完毕，就会丢失未发送的数据！！！）
-            return new SensorsAnalytics(new BatchConsumer(dataUrl));
+@Getter
+@Setter
+public class UserLoginEvent extends ApplicationEvent {
+
+    private UserTag userTag;
+
+    public UserLoginEvent(Object source) {
+        super(source);
     }
 
+    public UserLoginEvent(Object source, UserTag userTag) {
+        super(source);
+        this.userTag = userTag;
+    }
 }

src/main/java/cn/quantgroup/xyqb/event/UserTagLoginEventListener.java

+package cn.quantgroup.xyqb.event;
+
+import cn.quantgroup.xyqb.entity.UserTag;
+import cn.quantgroup.xyqb.repository.IUserTagRepository;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.ApplicationListener;
+import org.springframework.stereotype.Component;
+
+import java.time.LocalDateTime;
+
+/**
+ * user_tag表，用来监听
+ */
+@Component
+public class UserTagLoginEventListener implements ApplicationListener<UserLoginEvent> {
+
+    @Autowired
+    private IUserTagRepository userTagRepository;
+
+    /**
+     * 逻辑 每次登录发送UserLoginEvent，落user_tag表
+     * 如果没有就新增，如果有就更新
+     * @param userLoginEvent UserLoginEvent
+     */
+    @Override
+    public void onApplicationEvent(UserLoginEvent userLoginEvent) {
+        if (userLoginEvent.getUserTag() != null) {
+            UserTag userTag = userLoginEvent.getUserTag();
+            UserTag preTag = userTagRepository.findByUserIdAndTenantId(userTag.getUserId(), userTag.getTenantId());
+
+            if (preTag == null) {
+                preTag = userTag;
+            }
+
+            preTag.setCreatedAt(LocalDateTime.now());
+            preTag.setUpdatedAt(LocalDateTime.now());
+
+
+            userTagRepository.save(preTag);
+        }
+    }
+}

src/main/java/cn/quantgroup/xyqb/repository/IUserTagRepository.java

+package cn.quantgroup.xyqb.repository;
+
+import cn.quantgroup.xyqb.entity.UserTag;
+import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.data.jpa.repository.JpaSpecificationExecutor;
+
+public interface IUserTagRepository extends JpaRepository<UserTag, Long>, JpaSpecificationExecutor<UserTag> {
+    UserTag findByUserIdAndTenantId(Long userId,Integer tenantId);
+
+    void deleteByUserIdAndTenantId(Long userId,Integer tenantId);
+}

src/main/java/cn/quantgroup/xyqb/service/session/impl/SessionServiceImpl.java

@@ -3,6 +3,8 @@ package cn.quantgroup.xyqb.service.session.impl;
 import cn.quantgroup.xyqb.constant.enums.RecordType;
 import cn.quantgroup.xyqb.Constants;
 import cn.quantgroup.xyqb.entity.User;
+import cn.quantgroup.xyqb.entity.UserTag;
+import cn.quantgroup.xyqb.event.UserLoginEvent;
 import cn.quantgroup.xyqb.model.AuthBean;
 import cn.quantgroup.xyqb.model.LoginProperties;
 import cn.quantgroup.xyqb.model.UserStatistics;
@@ -17,9 +19,11 @@ import com.alibaba.fastjson.JSON;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.BeanUtils;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.cache.annotation.CacheEvict;
 import org.springframework.cache.annotation.Caching;
+import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.stereotype.Service;
 import org.springframework.util.CollectionUtils;
@@ -47,6 +51,9 @@ public class SessionServiceImpl implements ISessionService {
     @Value("${token.prefix}")
     private String prefix;
 
+    @Autowired
+    private ApplicationEventPublisher applicationEventPublisher;
+
 
     /**
      * 更新session
@@ -85,6 +92,10 @@ public class SessionServiceImpl implements ISessionService {
         // 添加登陆日志
         loginRecordService.saveLoginRecord(user.getId(), RecordType.LOGINRECORD.getName(), loginType);
 
+        //更新user_tag记录
+        applicationEventPublisher.publishEvent(new UserLoginEvent(this, UserTag.builder()
+                .userId(user.getId()).registeredFrom(user.getRegisteredFrom()).tenantId(user.getTenantId()).build()));
+
         return authBean;
     }
 

src/main/java/cn/quantgroup/xyqb/service/user/impl/UserServiceImpl.java

@@ -8,7 +8,9 @@ import cn.quantgroup.xyqb.controller.internal.user.resp.UserFullResp;
 import cn.quantgroup.xyqb.entity.Merchant;
 import cn.quantgroup.xyqb.entity.User;
 import cn.quantgroup.xyqb.entity.UserInfoEntity;
+import cn.quantgroup.xyqb.entity.UserTag;
 import cn.quantgroup.xyqb.event.PhoneNoUpdateEvent;
+import cn.quantgroup.xyqb.event.UserLoginEvent;
 import cn.quantgroup.xyqb.exception.DataException;
 import cn.quantgroup.xyqb.exception.UserNotExistException;
 import cn.quantgroup.xyqb.exception.UserRegisterLoginException;
@@ -86,6 +88,7 @@ public class UserServiceImpl implements IUserService, IBaseController {
     @Autowired
     private ISessionService sessionService;
 
+
     @Resource
     private ILockIpv4Service lockIpv4Service;
 
@@ -95,6 +98,9 @@ public class UserServiceImpl implements IUserService, IBaseController {
     @Autowired
     private IUserAttachedRepository userAttachedRepository;
 
+
+    @Autowired
+    private IUserTagRepository userTagRepository;
     @Autowired
     private IUserBtRegisterRepository userBtRegisterRepository;
 
@@ -165,7 +171,11 @@ public class UserServiceImpl implements IUserService, IBaseController {
     @Override
     @CacheEvict(value = "usercache", key = "'xyqbuser' + #user.phoneNo", cacheManager = "cacheManager")
     public User saveUser(User user) {
-        return userRepository.saveAndFlush(user);
+        user = userRepository.saveAndFlush(user);
+        //每次新增用户的时候，需要同步增长user_tag表
+        applicationEventPublisher.publishEvent(new UserLoginEvent(this, UserTag.builder()
+                .userId(user.getId()).registeredFrom(user.getRegisteredFrom()).tenantId(user.getTenantId()).build()));
+        return user;
     }
 
     @Override
@@ -197,9 +207,9 @@ public class UserServiceImpl implements IUserService, IBaseController {
         if (StringUtils.isBlank(phone)) {
             return null;
         }
-        User user = userRepository.findByEncryptedPhoneNoAndTenantId(phone,tenantId);
+        User user = userRepository.findByEncryptedPhoneNoAndTenantId(phone, tenantId);
         if (user == null) {
-            user = userRepository.findByPhoneNoAndTenantId(phone,tenantId);
+            user = userRepository.findByPhoneNoAndTenantId(phone, tenantId);
         }
         return user;
     }
@@ -207,12 +217,12 @@ public class UserServiceImpl implements IUserService, IBaseController {
     @Override
 //    @Cacheable(value = "usercache", key = "'xyqbuser' + #uuid", unless = "#result == null", cacheManager = "cacheManager")
     public User findByUuidWithCache(String uuid, Integer tenantId) {
-        return userRepository.findByUuidAndTenantId(uuid,tenantId);
+        return userRepository.findByUuidAndTenantId(uuid, tenantId);
     }
 
     @Override
     public boolean exist(String phoneNo, Integer tenantId) {
-        return userRepository.findByPhoneNoAndTenantId(phoneNo,tenantId) != null;
+        return userRepository.findByPhoneNoAndTenantId(phoneNo, tenantId) != null;
     }
 
     /**
@@ -235,26 +245,26 @@ public class UserServiceImpl implements IUserService, IBaseController {
         user.setCipherPassword(BctyptPasswordUtil.BCryptWithSalt(password));
         user = userRepository.save(user);
         stringRedisTemplate.expire("usercache:xyqbuser" + phoneNo, 1L, TimeUnit.MILLISECONDS);
-        sessionService.deleteByUserId(user.getId(),tenantId);
+        sessionService.deleteByUserId(user.getId(), tenantId);
         return BctyptPasswordUtil.BCryptCheckPw(password, user.getCipherPassword());
     }
 
     @Override
     public List<User> findRegisterUserByTime(String beginTime, String endTime, Integer tenantId) {
-        return userRepository.findRegisterUserByTimeAndTenantId(beginTime, endTime,tenantId);
+        return userRepository.findRegisterUserByTimeAndTenantId(beginTime, endTime, tenantId);
     }
 
     @Override
     @CacheEvict(value = "usercache", key = "'xyqbuser' + #phoneNo", cacheManager = "cacheManager")
     public int forbiddenUser(Boolean enable, String phoneNo, Integer tenantId) {
-        return userRepository.forbiddenUserAndTenantId(enable, phoneNo,tenantId);
+        return userRepository.forbiddenUserAndTenantId(enable, phoneNo, tenantId);
     }
 
 
     @Override
     public List<UserInfo> findUserInfosByPhones(List<String> phones, Integer tenantId) {
 
-        List<User> users = findByPhones(phones,tenantId);
+        List<User> users = findByPhones(phones, tenantId);
         if (CollectionUtils.isEmpty(phones)) {
             return Collections.emptyList();
         }
@@ -284,13 +294,13 @@ public class UserServiceImpl implements IUserService, IBaseController {
     @Transactional(rollbackFor = Exception.class)
     public User modifyPhoneNo(String oldPhoneNo, String newPhoneNo, Integer tenantId) {
         //1. 判断新手机号是否存在
-        User newPhoneUser = userRepository.findByPhoneNoAndTenantId(newPhoneNo,tenantId);
+        User newPhoneUser = userRepository.findByPhoneNoAndTenantId(newPhoneNo, tenantId);
         UserInfoEntity newPhoneUserDetail = userInfoRepository.findByPhoneNoAndTenantId(newPhoneNo, tenantId);
         if (Objects.nonNull(newPhoneUser) || Objects.nonNull(newPhoneUserDetail)) {
             //新手机号已存在
             throw new DataException("新手机号存在, 用户修改手机号后新手机号注册了。");
         }
-        User oldPhoneUser = userRepository.findByPhoneNoAndTenantId(oldPhoneNo,tenantId);
+        User oldPhoneUser = userRepository.findByPhoneNoAndTenantId(oldPhoneNo, tenantId);
         if (Objects.isNull(oldPhoneUser)) {
             //这不是扯了.旧手机号不存在.
             throw new DataException("旧手机号不存在, 可能已经修改成功了。");
@@ -327,9 +337,9 @@ public class UserServiceImpl implements IUserService, IBaseController {
     @Override
     @RedisLock(prefix = "lock:login:fast:", key = "#this[8]")
     public JsonResult loginFast(Long channelId, String appChannel, Long createdFrom, Long btRegisterChannelId,
-                                String dimension, String clickId, HttpServletRequest request, Merchant merchant, String phoneNo, Integer tenantId, Long geetestLogId,String appId) {
+                                String dimension, String clickId, HttpServletRequest request, Merchant merchant, String phoneNo, Integer tenantId, Long geetestLogId, String appId) {
         Boolean register = false;
-        User user = findByPhoneWithCache(phoneNo,tenantId);
+        User user = findByPhoneWithCache(phoneNo, tenantId);
         if (user != null && !user.getEnable()) {
             log.info("用户不存在，或者已经注销，phoneNo:{}", phoneNo);
             return JsonResult.buildErrorStateResult(USER_ERROR_OR_ENABLE_ERROR, null);
@@ -344,7 +354,7 @@ public class UserServiceImpl implements IUserService, IBaseController {
             MqUtils.sendRegisterMessageForGdt(phoneNo, clickId);
             register = true;
         }
-        if (!wechatRelateUserIfNecessary(user, request,appId,tenantId)) {
+        if (!wechatRelateUserIfNecessary(user, request, appId, tenantId)) {
             return JsonResult.buildErrorStateResult("登录时微信关联失败", null);
         }
 
@@ -391,7 +401,7 @@ public class UserServiceImpl implements IUserService, IBaseController {
             geetestLogService.updateByUidGeetestLog(geetestLogId, user.getId());
         }
         LoginProperties loginProperties = new LoginProperties("", 3, channelId, createdFrom, appChannel, merchant.getId(), merchant.getName(), tenantId);
-        AuthBean session = sessionService.createSession(user, loginProperties, LoginType.VERIFICATIONCODE.ordinal(),tenantId);
+        AuthBean session = sessionService.createSession(user, loginProperties, LoginType.VERIFICATIONCODE.ordinal(), tenantId);
         session.setRegister(register);
         lockIpv4Service.unLockPhone(phoneNo);
         return new JsonResult(session);
@@ -404,7 +414,7 @@ public class UserServiceImpl implements IUserService, IBaseController {
      * @param request - 当前请求
      * @return true - 继续登录，false - 微信关联失败，重新登录
      */
-    private boolean wechatRelateUserIfNecessary(User user, HttpServletRequest request,String appId ,Integer tenantId) {
+    private boolean wechatRelateUserIfNecessary(User user, HttpServletRequest request, String appId, Integer tenantId) {
         Objects.requireNonNull(request, "无效请求");
         String clientIp = IpUtil.getRemoteIP(request);
         Set<String> paramKeys = request.getParameterMap().keySet();
@@ -418,7 +428,7 @@ public class UserServiceImpl implements IUserService, IBaseController {
         Long userId = user.getId();
         String phoneNo = user.getPhoneNo();
         try {
-            int rows = wechatService.relateUser(userId, phoneNo, request.getParameter(Constants.WECHAT_OPEN_ID),appId,tenantId);
+            int rows = wechatService.relateUser(userId, phoneNo, request.getParameter(Constants.WECHAT_OPEN_ID), appId, tenantId);
             return rows > 0;
         } catch (Exception e) {
             log.error("微信关联失败，user:{}, request-Header:{}", user, JSON.toJSONString(getRequestHeaderMap(request)), e);
@@ -434,7 +444,7 @@ public class UserServiceImpl implements IUserService, IBaseController {
      */
     @Override
     public UserFullResp findUserFullSearchByUserId(Long userId, Integer tenantId) {
-        User user = userRepository.findByIdAndTenantId(userId,tenantId);
+        User user = userRepository.findByIdAndTenantId(userId, tenantId);
         UserInfoEntity userDetail = userInfoRepository.findByUserIdAndTenantId(userId, tenantId);
 
         UserFullResp userFullResp = UserFullResp.builder().userId(userId).build();
@@ -462,22 +472,22 @@ public class UserServiceImpl implements IUserService, IBaseController {
             List<Long> collect = vals.stream()
                     .map(Long::valueOf)
                     .collect(Collectors.toList());
-            users = userRepository.findByIdInAndTenantId(collect,tenantId);
+            users = userRepository.findByIdInAndTenantId(collect, tenantId);
         } else { //不是1 就是  uuids
-            users = userRepository.findByUuidInAndTenantId(vals,tenantId);
+            users = userRepository.findByUuidInAndTenantId(vals, tenantId);
         }
 
         return users;
     }
 
     @Override
-    public void logout(String token,Integer tenantId) {
-        sessionService.deleteSession(token,tenantId);
+    public void logout(String token, Integer tenantId) {
+        sessionService.deleteSession(token, tenantId);
     }
 
     @Override
     public void kdspLogout(Long userId, LoginProperties loginProperties, Integer tenantId) {
-        sessionService.kdspDeleteSession(userId, loginProperties,tenantId);
+        sessionService.kdspDeleteSession(userId, loginProperties, tenantId);
     }
 
 
@@ -485,7 +495,7 @@ public class UserServiceImpl implements IUserService, IBaseController {
     @Override
     public void deregister(Long userId, Integer tenantId) {
 
-        User user = userRepository.findByIdAndTenantId(userId,tenantId);
+        User user = userRepository.findByIdAndTenantId(userId, tenantId);
 
         if (Objects.isNull(user)) {
             throw new UserRegisterLoginException("用户不存在");
@@ -494,7 +504,9 @@ public class UserServiceImpl implements IUserService, IBaseController {
         /* 保存用户销户记录 */
         userDeregisterService.save(user);
         /* 删除用户 */
-        userRepository.deleteByIdAndTenantId(userId,tenantId);
+        userRepository.deleteByIdAndTenantId(userId, tenantId);
+        /* 删除用户tag*/
+        userTagRepository.deleteByUserIdAndTenantId(userId, tenantId);
 
         /* 删除用户附加信息 */
         userAttachedRepository.deleteByUserId(userId);
@@ -502,24 +514,24 @@ public class UserServiceImpl implements IUserService, IBaseController {
         userBtRegisterRepository.deleteByUserId(userId);
 
         /* 清空session */
-        sessionService.deleteByUserId(user.getId(),tenantId);
+        sessionService.deleteByUserId(user.getId(), tenantId);
         /* 清空缓存 */
-        sessionService.deleteUserCatch(user,tenantId);
+        sessionService.deleteUserCatch(user, tenantId);
         /* 禁用微信 Or 删除?*/
 //        wechatService.forbiddenXyqbAndWuxiUserByUserId(user.getId());
-        weChatUserRepository.deleteByUserIdAndTenantId(userId,tenantId);
+        weChatUserRepository.deleteByUserIdAndTenantId(userId, tenantId);
     }
 
     @Override
     @Transactional(rollbackFor = Exception.class)
     public User submitModifyPhone(String prevPhoneNo, String curPhoneNo, Integer tenantId) {
         //1. 判断新手机号是否存在
-        User newPhoneUser = findByPhoneInDb(curPhoneNo,tenantId);
+        User newPhoneUser = findByPhoneInDb(curPhoneNo, tenantId);
         if (Objects.nonNull(newPhoneUser)) {
             //新手机号已存在
             throw new DataException("新手机号存在, 用户修改手机号后新手机号注册了。");
         }
-        User oldPhoneUser = findByPhoneInDb(prevPhoneNo,tenantId);
+        User oldPhoneUser = findByPhoneInDb(prevPhoneNo, tenantId);
         if (Objects.isNull(oldPhoneUser)) {
             throw new DataException("旧手机号不存在, 可能已经修改成功了。");
         }
@@ -541,7 +553,7 @@ public class UserServiceImpl implements IUserService, IBaseController {
      * @param user
      * @param loginFrom
      */
-    public void channelUserSignContract(User user, Long loginFrom,Integer tenantId) {
+    public void channelUserSignContract(User user, Long loginFrom, Integer tenantId) {
         if (Objects.isNull(user) || StringUtils.isBlank(channelTemplate)) {
             return;
         }