package com.ylx.massage.task;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.collection.CollectionUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.ylx.common.constant.MassageConstants;
import com.ylx.common.core.domain.entity.SysDept;
import com.ylx.common.utils.DateUtils;
import com.ylx.common.utils.StringUtils;
import com.ylx.massage.domain.*;
import com.ylx.massage.enums.*;
import com.ylx.point.domain.PointActivity;
import com.ylx.point.domain.PointUserLog;
import com.ylx.massage.mapper.TConsumptionLogMapper;
import com.ylx.massage.service.*;
import com.ylx.massage.utils.DateTimeUtils;
import com.ylx.massage.utils.LocationUtil;
import com.ylx.massage.utils.WeChatUtil;
import com.ylx.point.service.IPointActivityService;
import com.ylx.point.service.IPointUserLogService;
import com.ylx.system.service.ISysDeptService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.util.*;
import java.util.stream.Collectors;


/**
 * @author jianlong
 * @date 2024-04-19 14:11
 */
@Slf4j
@Component("massageTask")
public class massageTask {

    @Resource
    private TOrderService orderService;

    @Autowired
    private IProductOrderInfoService productOrderInfoService;

    @Resource
    private LocationUtil locationUtil;

    @Resource
    private TJsService jsService;

    @Resource
    private TSignService signService;

    @Autowired
    private ISysDeptService deptService;

    @Autowired
    private TRechargeService rechargeService;
    @Resource
    private TJsDayService jsDayService;

    @Resource
    private TConsumptionLogService consumptionLogService;

    @Resource
    private TConsumptionLogMapper consumptionLogMapper;

    @Resource
    private TWxUserService userService;

    @Resource
    private WeChatUtil weChatUtil;

    @Resource
    private IPointUserLogService pointUserLogService;
    @Resource
    private IPointActivityService pointActivityService;

    /**
     * 取消超时未支付的订单
     */
    public void cancelOrder() {
        Date nowDate = new Date();
        log.info("开始执行取消订单任务，当前时间：{}", nowDate);

        Date date = DateTimeUtils.addMinute(nowDate, -5);
        log.info("开始执行取消订单任务，当前时间减5分钟,{}", date);
        long total = 1L;
        while (total > 0L) {
            total = extracted(date);
        }
    }


    /**
     * 取消超时未支付的商品订单
     */
    public void cancelProductOrder() {
        Date nowDate = new Date();
        String currentTime = DateTimeUtils.formatDate(nowDate, "yyyy-MM-dd HH:mm:ss");
        log.info("开始执行取消商品订单任务，当前时间：{}", currentTime);

        Date date = DateTimeUtils.addMinute(nowDate, -5);
        String dateStr = DateTimeUtils.formatDate(date, "yyyy-MM-dd HH:mm:ss");
        log.info("开始执行取消商品订单任务，当前时间减5分钟：{}", dateStr);
        long total = 1L;
        while (total > 0L) {
            total = extractedProduct(date);
            log.info("取消商品订单任务，当前时间减5分钟，共取消{}条商品订单", total);
        }
    }

    /**
     * 取消超时未支付订单
     *
     * @param nowDate
     * @return Long
     */
    private Long extracted(Date nowDate) {
        log.info("extracted 开始执行取消订单任务，时间：{}", nowDate);
        LambdaQueryWrapper<TOrder> objectLambdaQueryWrapper = new LambdaQueryWrapper<>();
        objectLambdaQueryWrapper.eq(TOrder::getnStatus, OrderStatusEnum.WAIT_PAY.getCode()).
                le(TOrder::getDtCreateTime, nowDate).
                orderByAsc(TOrder::getDtCreateTime);

        Page<TOrder> page = new Page<>();
        page.setSize(MassageConstants.TWO_HUNDRED);

        Page<TOrder> resPage = orderService.page(page, objectLambdaQueryWrapper);
        if (CollectionUtil.isNotEmpty(resPage.getRecords())) {
            resPage.getRecords().forEach(order -> {
                TOrder newOrder = new TOrder();
                //todo 订单锁
                newOrder.setcId(order.getcId());
                newOrder.setnStatus(OrderStatusEnum.CANCEL.getCode());
                orderService.updateById(newOrder);
                log.info("取消超时未支付订单orderNo：{}", order.getOrderNo());
            });
        }
        return resPage.getTotal();
    }


    /**
     * 取消超时未支付的商品订单
     *
     * @param nowDate
     * @return Long
     */
    private Long extractedProduct(Date nowDate) {
        LambdaQueryWrapper<ProductOrderInfo> objectLambdaQueryWrapper = new LambdaQueryWrapper<>();
        objectLambdaQueryWrapper.eq(ProductOrderInfo::getOrderStatus, ProductOrderStatusEnum.WAIT_PAY.getCode()).
                eq(ProductOrderInfo::getPayStatus, PaymentStatusEnum.UNPAID.getCode()).
                lt(ProductOrderInfo::getPayExpireTime, LocalDateTime.now());

        Page<ProductOrderInfo> page = new Page<>();
        page.setSize(MassageConstants.TWO_HUNDRED);

        Page<ProductOrderInfo> resPage = productOrderInfoService.page(page, objectLambdaQueryWrapper);
        if (CollectionUtil.isNotEmpty(resPage.getRecords())) {
            resPage.getRecords().forEach(order -> {
                ProductOrderInfo productOrderInfo = new ProductOrderInfo();
                productOrderInfo.setId(order.getId());
                productOrderInfo.setOrderStatus(ProductOrderStatusEnum.CANCELLED.getCode());
                productOrderInfoService.updateById(productOrderInfo);
            });
        }
        return resPage.getTotal();
    }

    /**
     * 超时未接单
     */
    public void timeoutNotOrder() {
        Date nowDate = new Date();
        log.info("开始执行超时未接单任务,{}", nowDate);

        Date date = DateTimeUtils.addMinute(nowDate, -10);
        log.info("开始执行超时未接单任务当前时间减10分钟,{}", date);
        long total = 1L;
        while (total > 0L) {
            total = timeoutNotOrder(date);
        }
    }

    private Long timeoutNotOrder(Date nowDate) {
        LambdaQueryWrapper<TOrder> objectLambdaQueryWrapper = new LambdaQueryWrapper<>();
        objectLambdaQueryWrapper.eq(TOrder::getnStatus, OrderStatusEnum.WAIT_JD.getCode()).
                le(TOrder::getPayTime, nowDate).
                orderByAsc(TOrder::getPayTime);

        Page<TOrder> page = new Page<>();
        page.setSize(MassageConstants.TWO_HUNDRED);

        Page<TOrder> resPage = orderService.page(page, objectLambdaQueryWrapper);
        if (CollectionUtil.isNotEmpty(resPage.getRecords())) {
            resPage.getRecords().forEach(order -> {
                //调用拒绝接单
                order.setReasonRefusal("超时未接单");
                orderService.jujue(order);
            });
        }
        return resPage.getTotal();

    }


    public void isGetAmount() {
        Date nowDate = new Date();
        log.info("定时分配出账金额日期,{}", nowDate);

        Date date = DateTimeUtils.addDays(nowDate, -7);
        log.info("定时分配出账金额日期,{}", date);
        isGetAmount(date);
    }

    private void isGetAmount(Date date) {

        //查询所有部门
        List<SysDept> sysDepts = deptService.selectDeptList(new SysDept());
        //遍历部门技师
        sysDepts.forEach(sysDept -> {
            LambdaQueryWrapper<TJs> jsLambdaQueryWrapper = new LambdaQueryWrapper<>();
            jsLambdaQueryWrapper.eq(TJs::getnTong, JsStatusEnum.JS_PASS.getCode()).eq(TJs::getDeptId, sysDept.getDeptId());
            List<TJs> jsList = jsService.list(jsLambdaQueryWrapper);
            if (CollectionUtil.isNotEmpty(jsList)) {
                jsList.forEach(js -> {
                    //根据技师openId 查询未领取金额
                    LambdaQueryWrapper<TConsumptionLog> logLambdaQueryWrapper = new LambdaQueryWrapper<>();
                    logLambdaQueryWrapper.eq(TConsumptionLog::getIsGet, MassageConstants.INTEGER_ZERO).
                            eq(TConsumptionLog::getOpenId, js.getcOpenId()).
                            in(TConsumptionLog::getBillType, Arrays.asList(BillTypeEnum.INCOME.getCode(), BillTypeEnum.DISTRIBUTION.getCode())).
                            le(TConsumptionLog::getCreateTime, date).
                            orderByAsc(TConsumptionLog::getCreateTime);
                    List<TConsumptionLog> consumptionLogList = consumptionLogService.list(logLambdaQueryWrapper);

                    if (CollectionUtil.isNotEmpty(consumptionLogList)) {
                        //处理可提现金额
                        //1.修改余额记录状态
                        consumptionLogList.forEach(consumptionLog -> {
                            //修改状态未已到提现账户
                            consumptionLog.setIsGet(MassageConstants.INTEGER_ONE);
                        });
                        int i = consumptionLogMapper.insertOrUpdateBatch(consumptionLogList);
                        //2.修改技师提现账户
                        BigDecimal getAmount = consumptionLogList.stream().map(TConsumptionLog::getAmount).reduce(BigDecimal.ZERO, BigDecimal::add);
                        TWxUser user = userService.getByOpenId(js.getcOpenId());
                        user.setGetAmount(user.getGetAmount().add(getAmount));
                        userService.updateById(user);
                    }
                });
            }
        });
    }


    public void cancelNewJs() {

        Date nowDate = new Date();
        log.info("开始执行取消新技师标识任务当前时间,{}", nowDate);

        long flag = 1L;
        while (flag > 0L) {
            LambdaQueryWrapper<TJs> objectLambdaQueryWrapper = new LambdaQueryWrapper<>();
            Date date = DateTimeUtils.addDays(nowDate, -7);

            log.info("开始执行取新技师当前时间减7天,{}", date);
            objectLambdaQueryWrapper.eq(TJs::getnB2, MassageConstants.INTEGER_ONE).
                    le(TJs::getcTime, date).
                    gt(TJs::getId, String.valueOf(flag)).
                    orderByAsc(TJs::getId);

            Page<TJs> page = new Page<>();
            page.setSize(MassageConstants.TWO_HUNDRED);

            Page<TJs> page1 = jsService.page(page, objectLambdaQueryWrapper);

            if (CollectionUtil.isNotEmpty(page1.getRecords())) {
                page1.getRecords().forEach(tjs -> {
                    TJs js = new TJs();
                    js.setId(tjs.getId());
                    js.setnB2(MassageConstants.INTEGER_ZERO);
                    jsService.updateById(js);
                    log.info("取消新技师标识,{}", js.getcOpenId() + js.getcName());
                });
                Optional<String> maxId = page1.getRecords().stream().max(Comparator.comparing(TJs::getId)).map(TJs::getId);
                flag = Long.parseLong(maxId.orElse("0"));
            } else {
                flag = 0L;
            }
        }
    }

    public void cancelHotJs() {

        Date nowDate = new Date();
        log.info("开始执行取消技师热度标识任务当前时间,{}", nowDate);

        Date date = DateTimeUtils.addDays(nowDate, -3);
        long flag = 1L;
        while (flag > 0L) {
            LambdaQueryWrapper<TJs> objectLambdaQueryWrapper = new LambdaQueryWrapper<>();
            objectLambdaQueryWrapper.eq(TJs::getnB3, MassageConstants.INTEGER_ONE).gt(TJs::getId, String.valueOf(flag)).orderByAsc(TJs::getId);

            Page<TJs> page = new Page<>();
            page.setSize(MassageConstants.TWO_HUNDRED);
            Page<TJs> page1 = jsService.page(page, objectLambdaQueryWrapper);

            if (CollectionUtil.isNotEmpty(page1.getRecords())) {
                page1.getRecords().forEach(tjs -> {
                    List<TOrder> list = orderService.list(new LambdaQueryWrapper<TOrder>().eq(TOrder::getcJsId, tjs.getId()).ge(TOrder::getDtCreateTime, date));
                    if (CollectionUtil.isEmpty(list) || list.size() < 3) {
                        TJs js = new TJs();
                        js.setId(tjs.getId());
                        js.setnB3(MassageConstants.INTEGER_ZERO);
                        jsService.updateById(js);
                        log.info("取消热度技师标识,{}", js.getcOpenId() + js.getcName());
                    }
                });
                Optional<String> maxId = page1.getRecords().stream().max(Comparator.comparing(TJs::getId)).map(TJs::getId);
                flag = Long.parseLong(maxId.orElse("0"));
            } else {
                flag = 0L;
            }
        }
    }

    /**
     * 定时下岗
     * 12:01 查询前一天 是否有上岗记录
     * 如果有上岗记录 没有下岗记录
     * 插入一条(前一天)下岗记录(并计算在线时长) 和当天上岗记录
     *
     * @param date yyyy-MM-dd
     **/
    public void RegularLayoff(String date) {

        Date date2 = new Date();
        if (StringUtils.isNotBlank(date)) {
            date2 = DateTimeUtils.parseDate(date, DateTimeUtils.DATE_FORMAT);
        } else {
            date2 = DateTimeUtils.addDays(date2, -1);
        }
        Date startDate = DateTimeUtils.getStartDate(date2);
        Date endDate = DateTimeUtils.getEndDate(date2);
        long total = 1L;
        while (total > 0L) {
            Page<TSign> signPage = new Page<>();
            signPage.setSize(MassageConstants.TWO_HUNDRED);
            LambdaQueryWrapper<TSign> signLambdaQueryWrapper = new LambdaQueryWrapper<>();
            //上岗

            signLambdaQueryWrapper.isNull(TSign::getLayoffTime).ge(TSign::getSingTime, startDate).le(TSign::getSingTime, endDate);
            Page<TSign> page = signService.page(signPage, signLambdaQueryWrapper);

            if (CollectionUtil.isNotEmpty(page.getRecords())) {
                page.getRecords().forEach(sign -> {
                    sign.setLayoffTime(endDate);
                    Long layoff = DateTimeUtils.dateToStamp(sign.getLayoffTime());
                    Long sing = DateTimeUtils.dateToStamp(sign.getSingTime());
                    Long time = layoff - sing;
                    time = time / 1000;
                    time = time / 60;
                    sign.setOnlineTime(time.intValue());
                    signService.updateById(sign);
                    //增加上岗记录
                    TSign tSign = new TSign();
                    tSign.setSingTime(endDate);
                    tSign.setName(sign.getName());
                    tSign.setJsId(sign.getJsId());
                    tSign.setDeptId(sign.getDeptId());
                    tSign.setDeptName(sign.getDeptName());
                    tSign.setOpenId(sign.getOpenId());
                    signService.save(tSign);
                });
            }
            total = page.getTotal();
        }
        //查询所有部门
        List<SysDept> sysDepts = deptService.selectDeptList(new SysDept());

        sysDepts.forEach(sysDept -> {
            //查询所有技师
            LambdaQueryWrapper<TJs> jsLambdaQueryWrapper = new LambdaQueryWrapper<>();
            jsLambdaQueryWrapper.eq(TJs::getnTong, JsStatusEnum.JS_PASS.getCode()).eq(TJs::getDeptId, sysDept.getDeptId());
            List<TJs> jsList = jsService.list(jsLambdaQueryWrapper);
            if (CollectionUtil.isNotEmpty(jsList)) {
                jsList.forEach(js -> {
                    TJsDay tJsDay = new TJsDay();
                    tJsDay.setDeptName(js.getCity());
                    tJsDay.setDeptId(js.getDeptId());
                    tJsDay.setOpenId(js.getcOpenId());
                    tJsDay.setJsId(js.getId());
                    tJsDay.setName(js.getcName());
                    Integer onLine = signService.getOnLineTime(js.getId(), startDate, endDate);
                    tJsDay.setOnLine(onLine);
                    Integer orderNum = orderService.getOrderNum(js.getId(), startDate, endDate);
                    tJsDay.setOrderNum(orderNum);
                    Integer addNum = orderService.getAddNum(js.getId(), startDate, endDate);
                    tJsDay.setAddNum(addNum);
                    Integer upgradeNum = orderService.getUpgradeNum(js.getId(), startDate, endDate);
                    tJsDay.setUpgradeNum(upgradeNum);
                    BigDecimal turnover = orderService.getTurnover(js.getId(), startDate, endDate);
                    tJsDay.setTurnover(turnover);
                    BigDecimal recharge = rechargeService.getRecharge(js.getId(), startDate, endDate);
                    tJsDay.setRecharge(recharge);
                    tJsDay.setCountDate(DateTimeUtils.formatDate(startDate, DateTimeUtils.DATE_FORMAT));
                    jsDayService.save(tJsDay);
                });
            }
        });


    }

    public void sysJsWeiZhi() {
        log.info("开始执行同步技师位置时间,{}", DateUtils.getNowDate());
        long flag = 1L;
        while (flag > 0L) {
            LambdaQueryWrapper<TJs> objectLambdaQueryWrapper = new LambdaQueryWrapper<>();

            objectLambdaQueryWrapper.gt(TJs::getId, String.valueOf(flag)).orderByAsc(TJs::getId);

            Page<TJs> page = new Page<>();
            page.setSize(MassageConstants.TWO_HUNDRED);
            Page<TJs> page1 = jsService.page(page, objectLambdaQueryWrapper);

            if (CollectionUtil.isNotEmpty(page1.getRecords())) {
                page1.getRecords().forEach(tjs -> {
                    if (StringUtils.isNotBlank(tjs.getcPhone()) && null != tjs.getLongitude() && null != tjs.getLatitude()) {
                        locationUtil.geoAdd(LocationUtil.GEO_KEY, tjs.getcPhone(),
                                Double.parseDouble(tjs.getLongitude().toString()),
                                Double.parseDouble(tjs.getLatitude().toString()));
                    }
                });
                Optional<String> maxId = page1.getRecords().stream().max(Comparator.comparing(TJs::getId)).map(TJs::getId);
                flag = Long.parseLong(maxId.orElse("0"));
            } else {
                flag = 0L;
            }
        }
        log.info("结束执行同步技师位置时间,{}", DateUtils.getNowDate());
    }

    public void sysJsWeiTicke() {
        log.info("开始执行同步技师二维码,{}", DateUtils.getNowDate());
        long flag = 1L;
        while (flag > 0L) {
            LambdaQueryWrapper<TJs> objectLambdaQueryWrapper = new LambdaQueryWrapper<>();

            objectLambdaQueryWrapper.gt(TJs::getId, String.valueOf(flag)).orderByAsc(TJs::getId);

            Page<TJs> page = new Page<>();
            page.setSize(MassageConstants.TWO_HUNDRED);
            Page<TJs> page1 = jsService.page(page, objectLambdaQueryWrapper);

            if (CollectionUtil.isNotEmpty(page1.getRecords())) {
                page1.getRecords().forEach(tjs -> {
                    String token = weChatUtil.getToken();
                    //获取的二维码ticket
                    Map<?, ?> jsTicket = weChatUtil.getJsTicket(token, tjs.getId());
                    tjs.setTicket(jsTicket.get("ticket").toString());
                    jsService.updateById(tjs);
                });
                Optional<String> maxId = page1.getRecords().stream().max(Comparator.comparing(TJs::getId)).map(TJs::getId);
                flag = Long.parseLong(maxId.orElse("0"));
            } else {
                flag = 0L;
            }
        }
        log.info("结束执行同步技师二维码,{}", DateUtils.getNowDate());
    }

    /**
     * 定时任务入口：批量更新过期的积分记录
     */
    @Transactional(rollbackFor = Exception.class)
    public void expirePoints() {
        Date nowDate = new Date();
        log.info("开始执行积分过期任务，当前时间：{}", nowDate);

        // 建议添加分布式锁，防止多实例并发执行
        // String lockKey = "lock:points:expire:" + DateUtils.getDate();
        // if (redisLock.tryLock(lockKey)) { ... }

        try {
            // 使用游标模式，避免深分页问题
            Long lastId = 0L;
            int batchSize = 500; // 适当调大批次

            while (true) {
                // 获取最后一批处理的ID，作为下一批的起点
                Long currentLastId = processBatch(nowDate, lastId, batchSize);

                // 如果返回的ID没有变化，或者小于等于上一轮ID，说明没有更多数据了
                if (currentLastId <= lastId) {
                    break;
                }
                lastId = currentLastId;
            }
        } finally {
            // redisLock.unlock(lockKey);
            log.info("结束执行积分过期任务，当前时间：{}", DateUtils.getNowDate());
        }
    }

    /**
     * 定时任务入口：批量更新过期的活动记录
     */
    @Transactional(rollbackFor = Exception.class)
    public void expireActivity() {
        Date nowDate = new Date();
        log.info("开始执行活动过期任务，当前时间：{}", nowDate);

        // 建议添加分布式锁，防止多实例并发执行
        // String lockKey = "lock:points:expire:" + DateUtils.getDate();
        // if (redisLock.tryLock(lockKey)) { ... }

        try {
            // 使用游标模式，避免深分页问题
            Long lastId = 0L;
            int batchSize = 500; // 适当调大批次

            while (true) {
                // 获取最后一批处理的ID，作为下一批的起点
                Long currentLastId = processActivityBatch(nowDate, lastId, batchSize);

                // 如果返回的ID没有变化，或者小于等于上一轮ID，说明没有更多数据了
                if (currentLastId <= lastId) {
                    break;
                }
                lastId = currentLastId;
            }
        } finally {
            // redisLock.unlock(lockKey);
            log.info("结束执行活动过期任务，当前时间：{}", DateUtils.getNowDate());
        }
    }

    public Long processBatch(Date nowDate, Long lastId, int batchSize) {

        // 1. 查询：使用 list 代替 page，减少 count 查询开销
        LambdaQueryWrapper<PointUserLog> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(PointUserLog::getOpType, 1)          // 获取积分
                .eq(PointUserLog::getIsExpired, 0)           // 未过期
                .lt(PointUserLog::getExpireTime, nowDate)    // 已到期
                .isNotNull(PointUserLog::getExpireTime)      // 有有效期
                .gt(PointUserLog::getId, lastId)             // 游标推进
                .orderByAsc(PointUserLog::getId)             // 必须排序
                .last("LIMIT " + batchSize);                 // 物理限制数量

        List<PointUserLog> records = pointUserLogService.list(queryWrapper);

        if (CollUtil.isEmpty(records)) {
            return lastId;
        }

        // 2. 准备批量数据
        List<PointUserLog> insertList = new ArrayList<>(records.size());
        List<Long> updateIds = new ArrayList<>(records.size());

        for (PointUserLog pointLog : records) {
            PointUserLog expireLog = BeanUtil.copyProperties(pointLog, PointUserLog.class);

            // 重置主键，让数据库自增
            expireLog.setId(null);
            // 覆盖关键字段
            expireLog.setOpType(3);
            expireLog.setSourceLogId(pointLog.getId()); // 关联原流水ID
            expireLog.setIsExpired(1);

            insertList.add(expireLog);
            updateIds.add(pointLog.getId());
        }

        // 3. 批量执行
        try {
            // A. 批量插入过期流水
            // 注意：saveBatch 默认不开启 rewriteBatchedStatements，数据量大时建议在配置中开启
            boolean insertSuccess = pointUserLogService.saveBatch(insertList);

            if (!insertSuccess) {
                throw new RuntimeException("批量插入过期流水失败");
            }

            // B. 批量更新原记录状态
            LambdaUpdateWrapper<PointUserLog> updateWrapper = new LambdaUpdateWrapper<>();
            updateWrapper.in(PointUserLog::getId, updateIds)
                    .set(PointUserLog::getIsExpired, 1);

            boolean updateSuccess = pointUserLogService.update(updateWrapper);

            if (!updateSuccess) {
                throw new RuntimeException("批量更新原流水状态失败");
            }

            log.info("批次处理成功，处理条数：{}", records.size());

        } catch (Exception e) {
            log.error("批次处理失败，lastId: {}, error: {}", lastId, e.getMessage(), e);
            // 抛出异常触发 @Transactional 回滚，保证数据一致性
            throw e;
        }

        // 4. 返回游标位置
        return records.get(records.size() - 1).getId();
    }

    /**
     * 获取用户当前可用积分余额
     *
     * @param openId 用户openId
     * @return 当前余额
     */
    private Integer getBalance(String openId) {
        QueryWrapper<PointUserLog> queryWrapper = new QueryWrapper<>();
        queryWrapper.select("SUM(points)");
        queryWrapper.eq("open_id", openId);

        PointUserLog pointUserLog = pointUserLogService.getOne(queryWrapper);

        if (pointUserLog == null || pointUserLog.getPoints() == null) {
            return 0;
        }

        return pointUserLog.getPoints();
    }

    /**
     * 处理单批次活动过期逻辑
     *
     * @param nowDate   当前时间
     * @param lastId    上一批次最后处理的ID（游标起点）
     * @param batchSize 批次大小
     * @return 本批次最后处理的ID
     */
    public Long processActivityBatch(Date nowDate, Long lastId, int batchSize) {

        // 1. 构建分页查询对象
        // 注意：虽然传入了页码1，但实际数据定位由 queryWrapper 中的 gt(id, lastId) 控制
        Page<PointActivity> page = new Page<>(1, batchSize);

        // 2. 构建查询条件
        LambdaQueryWrapper<PointActivity> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.gt(PointActivity::getId, lastId)          // 核心游标：ID 大于上一批次的最大值
                .ne(PointActivity::getStatus, 2)           // 排除已经是“已结束”的
                .lt(PointActivity::getEndTime, nowDate)    // 核心条件：结束时间 < 当前时间
                .orderByAsc(PointActivity::getId);         // 必须按 ID 排序，保证游标连续性

        // 3. 执行分页查询
        // 使用 service 层自带的 page 方法
        Page<PointActivity> resultPage = pointActivityService.page(page, queryWrapper);

        List<PointActivity> records = resultPage.getRecords();

        // 4. 判断是否还有数据
        if (CollUtil.isEmpty(records)) {
            // 如果没有查到数据，返回当前的 lastId，外层循环检测到相等就会停止
            return lastId;
        }

        // 5. 提取 ID 列表进行批量更新
        Long currentLastId = records.get(records.size() - 1).getId();
        List<Long> ids = records.stream().map(PointActivity::getId).collect(Collectors.toList());

        UpdateWrapper<PointActivity> updateWrapper = new UpdateWrapper<>();
        updateWrapper.in("id", ids)
                .eq("status", 1); // 乐观检查：确保更新时状态仍为“进行中”，防止并发冲突

        PointActivity updateEntity = new PointActivity();
        updateEntity.setStatus(2);           // 状态改为 2 (已结束)
        updateEntity.setUpdateTime(nowDate); // 更新时间
        updateEntity.setUpdateBy("admin");   // 更新人

        // 6. 执行更新
        try {
            boolean success = pointActivityService.update(updateEntity, updateWrapper);

            if (success) {
                log.info("批次处理成功，更新ID范围: {} - {}, 数量: {}", ids.get(0), currentLastId, ids.size());
            } else {
                log.warn("批次处理部分失败或无更新，ID范围: {} - {}", ids.get(0), currentLastId);
            }
        } catch (Exception e) {
            log.error("批次处理失败，lastId: {}, error: {}", lastId, e.getMessage(), e);
            throw e;
        }

        // 7. 返回本批次最大的 ID，作为下一轮的游标起点
        return currentLastId;
    }

}
