From 40f3229e01a95c8c17a83f46662c49453a76396c Mon Sep 17 00:00:00 2001 From: dzc <2388224167@qq.com> Date: Thu, 24 Sep 2026 17:34:36 +0800 Subject: [PATCH] =?UTF-8?q?feat:=E9=A2=84=E7=BA=A6=E5=85=85=E5=80=BC=20fea?= =?UTF-8?q?t:=E8=BF=81=E7=A7=BBthird=E6=A8=A1=E5=9D=97=E7=9A=84=E5=AE=9A?= =?UTF-8?q?=E6=97=B6=E4=BB=BB=E5=8A=A1=E5=88=B0quartz=E6=A8=A1=E5=9D=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- db/20260924_thirdparty推送任务迁移sys_job.txt | 78 +++ db/20260924_钱包预约执行定时任务.txt | 80 +-- local.env | 2 +- .../feign/RemoteThirdPushDeliveryService.java | 32 ++ .../vo/pay/ThirdPaymentTaskDispatchVo.java | 5 + .../bsd/pjcloud/thirdparty/ThirdPartyApp.java | 3 +- .../ThirdPushDeliveryInnerController.java | 50 ++ .../job/ThirdAggregateOrderPayJob.java | 176 ------- .../thirdparty/job/ThirdAppJgPushJob.java | 56 -- .../job/ThirdAppMessagePushJob.java | 49 -- .../thirdparty/job/ThirdAppPlusPayJob.java | 494 ------------------ .../impl/appplus/ThirdAppPlusServiceImpl.java | 22 +- .../impl/pay/ThirdPaymentTaskServiceImpl.java | 12 +- .../mapper/ThirdPaymentTaskMapper.xml | 8 +- .../job/ThirdAggregateOrderPayJobTest.java | 100 ---- .../job/ThirdAppPlusPayJobTest.java | 200 ------- .../daemon/quartz/task/ThirdPushTask.java | 85 +++ .../daemon/quartz/task/WalletReserveTask.java | 14 + .../src/main/resources/application.yml | 6 + .../src/main/resources/quartz-config.yml | 2 +- .../recharge/WalletRechargeApplyMapper.java | 11 +- .../WalletRechargeApplyServiceImpl.java | 51 +- .../WalletReserveExecServiceImpl.java | 4 +- 23 files changed, 403 insertions(+), 1137 deletions(-) create mode 100644 db/20260924_thirdparty推送任务迁移sys_job.txt create mode 100644 pjcloud-thirdparty/pjcloud-thirdparty-api/src/main/java/com/bsd/pjcloud/thirdparty/api/feign/RemoteThirdPushDeliveryService.java create mode 100644 pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/controller/appmessage/ThirdPushDeliveryInnerController.java delete mode 100644 pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/job/ThirdAggregateOrderPayJob.java delete mode 100644 pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/job/ThirdAppJgPushJob.java delete mode 100644 pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/job/ThirdAppMessagePushJob.java delete mode 100644 pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/job/ThirdAppPlusPayJob.java delete mode 100644 pjcloud-thirdparty/pjcloud-thirdparty-biz/src/test/java/com/bsd/pjcloud/thirdparty/job/ThirdAggregateOrderPayJobTest.java delete mode 100644 pjcloud-thirdparty/pjcloud-thirdparty-biz/src/test/java/com/bsd/pjcloud/thirdparty/job/ThirdAppPlusPayJobTest.java create mode 100644 pjcloud-visual/pjcloud-daemon-quartz/src/main/java/com/bsd/pjcloud/daemon/quartz/task/ThirdPushTask.java diff --git a/db/20260924_thirdparty推送任务迁移sys_job.txt b/db/20260924_thirdparty推送任务迁移sys_job.txt new file mode 100644 index 000000000..a0ccf77e6 --- /dev/null +++ b/db/20260924_thirdparty推送任务迁移sys_job.txt @@ -0,0 +1,78 @@ +======================================================================== +thirdparty App消息Push定时任务迁移 daemon-quartz - sys_job 手工添加清单 (2026-09-24) +======================================================================== + +说明：以下内容需要管理员在对应环境手工添加（Quartz 任务管理页或直接执行 SQL）。 +SQL 幂等，可重复执行。 +注意库名：sys_job 在 `pjcloudx_job` 库（与 qrtz_* / sys_job_log 同库）。 + +迁移背景： + thirdparty 原本地 @Scheduled 定时任务（ThirdAppMessagePushJob：每10秒Push投递扫描 + + 每日02:10过期清理）已迁移到 daemon-quartz 统一调度，集群部署下避免各实例重复触发， + 且可在任务管理页暂停/改cron/手动执行/看执行日志。 + thirdparty 侧新增内部端点 /third-push-delivery/inner/process-due 与 /inner/cleanup， + 由 daemon 的 thirdPushTask bean（class_name=thirdPushTask）通过 Feign 反射调用。 + +部署顺序（硬前提，顺序不能颠倒）： + 1. 部署 pjcloud-thirdparty-biz（新增内部端点、已移除本地 @Scheduled 与 @EnableScheduling）； + 2. 部署 pjcloud-daemon-quartz（新增 thirdPushTask bean）； + 3. 执行下方 sys_job INSERT； + 4. 重启 daemon-quartz 自动加载（或调 POST /sys-job/refresh-jobs，权限 job_sys_job_refresh_job）。 + +行为说明： + 原 fixedDelay=10s（上轮结束后再计时）改为固定 10 秒 cron 节奏；Quartz 任务带 + @DisallowConcurrentExecution，单轮未结束时下一触发自动顺延，不会重叠。 + 投递任务内部由条件更新（status CAS）领取，多节点/重复触发幂等安全。 + +------------------------------------------------------------------------ +一、sys_job：App消息Push投递扫描（每10秒） +------------------------------------------------------------------------ + +INSERT INTO `pjcloudx_job`.`sys_job` +(`job_id`, `job_name`, `job_group`, `job_order`, `job_type`, `execute_path`, `class_name`, `method_name`, + `method_params_value`, `cron_expression`, `misfire_policy`, `job_tenant_type`, `job_status`, + `job_execute_status`, `create_by`, `create_time`, `update_by`, `update_time`, `tenant_id`, `remark`) +SELECT (SELECT GREATEST(COALESCE(MAX(`job_id`), 0), 2026092400000000000) FROM `pjcloudx_job`.`sys_job`) + 3, + 'App消息Push投递扫描', 'THIRDPARTY', '1', '2', NULL, 'thirdPushTask', 'processDue', + NULL, '0/10 * * * * ?', '3', '1', '2', + '0', 'system', NOW(), 'system', NOW(), 1, '每10秒扫描到期App消息Push投递任务并投递' +FROM DUAL +WHERE NOT EXISTS ( + SELECT 1 FROM `pjcloudx_job`.`sys_job` + WHERE `job_name` = 'App消息Push投递扫描' AND `job_group` = 'THIRDPARTY' AND `tenant_id` = 1 +); + +------------------------------------------------------------------------ +二、sys_job：App消息Push过期清理（每日 02:10，与原本地任务一致） +------------------------------------------------------------------------ + +INSERT INTO `pjcloudx_job`.`sys_job` +(`job_id`, `job_name`, `job_group`, `job_order`, `job_type`, `execute_path`, `class_name`, `method_name`, + `method_params_value`, `cron_expression`, `misfire_policy`, `job_tenant_type`, `job_status`, + `job_execute_status`, `create_by`, `create_time`, `update_by`, `update_time`, `tenant_id`, `remark`) +SELECT (SELECT GREATEST(COALESCE(MAX(`job_id`), 0), 2026092400000000000) FROM `pjcloudx_job`.`sys_job`) + 4, + 'App消息Push过期清理', 'THIRDPARTY', '2', '2', NULL, 'thirdPushTask', 'cleanup', + NULL, '0 10 2 * * ?', '3', '1', '2', + '0', 'system', NOW(), 'system', NOW(), 1, '每日02:10清理超保留期的Push消息/终态投递/尝试日志' +FROM DUAL +WHERE NOT EXISTS ( + SELECT 1 FROM `pjcloudx_job`.`sys_job` + WHERE `job_name` = 'App消息Push过期清理' AND `job_group` = 'THIRDPARTY' AND `tenant_id` = 1 +); + +------------------------------------------------------------------------ +三、内部端点与任务对应关系（勿对外暴露） +------------------------------------------------------------------------ + +thirdPushTask.processDue: + POST /third-push-delivery/inner/process-due 扫描并处理到期Push投递（@Inner 鉴权） + +thirdPushTask.cleanup: + POST /third-push-delivery/inner/cleanup 清理超保留期记录（@Inner 鉴权） + +------------------------------------------------------------------------ +四、回滚说明 +------------------------------------------------------------------------ + 1. 停用任务：将两条 sys_job 的 job_status 改为 '4'（删除）或 '3'（暂停）并重启 daemon-quartz + （或调 POST /sys-job/refresh-jobs）； + 2. 恢复本地调度：回滚 thirdparty 代码（ThirdAppMessagePushJob + @EnableScheduling）并重新部署。 diff --git a/db/20260924_钱包预约执行定时任务.txt b/db/20260924_钱包预约执行定时任务.txt index b2e82ce21..a8704398f 100644 --- a/db/20260924_钱包预约执行定时任务.txt +++ b/db/20260924_钱包预约执行定时任务.txt @@ -4,6 +4,7 @@ 说明：以下内容需要管理员在对应环境手工添加（ Quartz 任务管理页或直接执行 SQL、 UPMS 管理端「菜单管理」）。SQL 按租户幂等，风格与 db/20260920_饭卡卡面模板菜单权限.txt 一致。 +注意库名：sys_job 在 `pjcloudx_job` 库（与 qrtz_* / sys_job_log 同库）；sys_menu 在 `pjcloudx` 库。 前置条件： 1. 先执行 db/20260924_add_wallet_recharge_apply_reserve_exec.sql（加列 + 扫描索引 + 对账差异表）； @@ -22,17 +23,17 @@ UPMS 管理端「菜单管理」）。SQL 按租户幂等，风格与 db/2026092 一、sys_job：钱包预约充值到期执行（job_type=2，反射调 daemon 内 spring bean） ------------------------------------------------------------------------ -INSERT INTO `pjcloudx`.`sys_job` +INSERT INTO `pjcloudx_job`.`sys_job` (`job_id`, `job_name`, `job_group`, `job_order`, `job_type`, `execute_path`, `class_name`, `method_name`, `method_params_value`, `cron_expression`, `misfire_policy`, `job_tenant_type`, `job_status`, `job_execute_status`, `create_by`, `create_time`, `update_by`, `update_time`, `tenant_id`, `remark`) -SELECT (SELECT GREATEST(COALESCE(MAX(`job_id`), 0), 2026092400000000000) FROM `pjcloudx`.`sys_job`) + 1, +SELECT (SELECT GREATEST(COALESCE(MAX(`job_id`), 0), 2026092400000000000) FROM `pjcloudx_job`.`sys_job`) + 1, '钱包预约充值到期执行', 'WALLET', '1', '2', NULL, 'walletReserveTask', 'executeDueApplies', NULL, '0 */10 * * * ?', '3', '1', '2', '0', 'system', NOW(), 'system', NOW(), 1, '每10分钟扫描到期预约充值/扣款申请并幂等执行' FROM DUAL WHERE NOT EXISTS ( - SELECT 1 FROM `pjcloudx`.`sys_job` + SELECT 1 FROM `pjcloudx_job`.`sys_job` WHERE `job_name` = '钱包预约充值到期执行' AND `job_group` = 'WALLET' AND `tenant_id` = 1 ); @@ -40,93 +41,92 @@ WHERE NOT EXISTS ( 二、sys_job：钱包预约充值对账（同 bean 第二个方法，错开 5 分钟） ------------------------------------------------------------------------ -INSERT INTO `pjcloudx`.`sys_job` +INSERT INTO `pjcloudx_job`.`sys_job` (`job_id`, `job_name`, `job_group`, `job_order`, `job_type`, `execute_path`, `class_name`, `method_name`, `method_params_value`, `cron_expression`, `misfire_policy`, `job_tenant_type`, `job_status`, `job_execute_status`, `create_by`, `create_time`, `update_by`, `update_time`, `tenant_id`, `remark`) -SELECT (SELECT GREATEST(COALESCE(MAX(`job_id`), 0), 2026092400000000000) FROM `pjcloudx`.`sys_job`) + 2, +SELECT (SELECT GREATEST(COALESCE(MAX(`job_id`), 0), 2026092400000000000) FROM `pjcloudx_job`.`sys_job`) + 2, '钱包预约充值对账', 'WALLET', '2', '2', NULL, 'walletReserveTask', 'reconcile', NULL, '5 */10 * * * ?', '3', '1', '2', '0', 'system', NOW(), 'system', NOW(), 1, '每10分钟复核近7天预约单明细与钱包流水一致性' FROM DUAL WHERE NOT EXISTS ( - SELECT 1 FROM `pjcloudx`.`sys_job` + SELECT 1 FROM `pjcloudx_job`.`sys_job` WHERE `job_name` = '钱包预约充值对账' AND `job_group` = 'WALLET' AND `tenant_id` = 1 ); ------------------------------------------------------------------------ -三、按钮权限：充值/扣款申请「立即执行」（挂在充值申请页面菜单下，人工兜底） +三、按钮权限：充值/扣款申请「预约执行」「立即执行」（挂在充值申请页面菜单下） ------------------------------------------------------------------------ +「预约执行」控制新增/编辑表单里的预约执行开关与时间控件（前端显隐 + 后端字段校验双重控制， +无权限用户提交的预约设置会被后端忽略或强制关闭）。「立即执行」是待执行/失败单的人工兜底入口。 + INSERT INTO `pjcloudx`.`sys_menu` (`menu_id`, `name`, `permission`, `parent_id`, `visible`, `sort_order`, `keep_alive`, `embedded`, `menu_type`, `create_by`, `create_time`, `del_flag`, `tenant_id`) -SELECT (SELECT GREATEST(COALESCE(MAX(`menu_id`), 0), 2026092400000000000) FROM `pjcloudx`.`sys_menu`) + ROW_NUMBER() OVER (ORDER BY page_menu.`tenant_id`, page_menu.`menu_id`), - '立即执行', 'wallet_walletRechargeApply_execute', page_menu.`menu_id`, - '1', 60, '0', '0', '2', 'system', NOW(), '0', page_menu.`tenant_id` +SELECT (SELECT GREATEST(COALESCE(MAX(`menu_id`), 0), 2026092400000000000) FROM `pjcloudx`.`sys_menu`) + ROW_NUMBER() OVER (ORDER BY page_menu.`tenant_id`, page_menu.`menu_id`, item.`sort`), + item.`name`, item.`permission`, page_menu.`menu_id`, + '1', item.`sort`, '0', '0', '2', 'system', NOW(), '0', page_menu.`tenant_id` FROM `pjcloudx`.`sys_menu` page_menu +JOIN ( + SELECT '预约执行' AS `name`, 'wallet_walletRechargeApply_reserve' AS `permission`, 59 AS `sort` + UNION ALL SELECT '立即执行', 'wallet_walletRechargeApply_execute', 60 +) item ON item.`name` IS NOT NULL WHERE page_menu.`permission` = 'wallet_walletRechargeApply_view' AND page_menu.`menu_type` = '0' AND page_menu.`del_flag` = '0' AND NOT EXISTS ( SELECT 1 FROM `pjcloudx`.`sys_menu` existing WHERE existing.`tenant_id` = page_menu.`tenant_id` - AND existing.`permission` = 'wallet_walletRechargeApply_execute' AND existing.`del_flag` = '0' + AND existing.`permission` = item.`permission` AND existing.`del_flag` = '0' ); ------------------------------------------------------------------------ -四、页面菜单：预约执行对账（与充值/扣款申请同级，复用其父菜单） +四、按钮权限：预约执行对账 tab 与「标记已处理」（挂在充值/扣款申请页面菜单下） ------------------------------------------------------------------------ -前端页面：src/views/wallet/wallet-recharge-exec-check/index.vue - -INSERT INTO `pjcloudx`.`sys_menu` -(`menu_id`, `name`, `permission`, `path`, `component`, `parent_id`, `icon`, `visible`, `sort_order`, `keep_alive`, `embedded`, `menu_type`, `create_by`, `create_time`, `update_by`, `update_time`, `del_flag`, `tenant_id`) -SELECT (SELECT GREATEST(COALESCE(MAX(`menu_id`), 0), 2026092400000000000) FROM `pjcloudx`.`sys_menu`) + ROW_NUMBER() OVER (ORDER BY sibling.`tenant_id`, sibling.`menu_id`), - '预约执行对账', 'wallet_walletRechargeExecCheck_view', '/wallet/wallet-recharge-exec-check', 'wallet/wallet-recharge-exec-check/index', - sibling.`parent_id`, NULL, '1', 55, '0', '0', '0', - 'system', NOW(), 'system', NOW(), '0', sibling.`tenant_id` -FROM `pjcloudx`.`sys_menu` sibling -WHERE sibling.`permission` = 'wallet_walletRechargeApply_view' AND sibling.`menu_type` = '0' AND sibling.`del_flag` = '0' - AND NOT EXISTS ( - SELECT 1 FROM `pjcloudx`.`sys_menu` existing - WHERE existing.`tenant_id` = sibling.`tenant_id` - AND existing.`permission` = 'wallet_walletRechargeExecCheck_view' AND existing.`del_flag` = '0' - ); - ------------------------------------------------------------------------- -五、按钮权限：预约执行对账「标记已处理」（挂在对账页面菜单下） ------------------------------------------------------------------------- +对账入口集成在充值/扣款申请列表页（tab 形式，无独立页面/路由）： + src/views/wallet/wallet-recharge-apply/index.vue + 持有 wallet_walletRechargeExecCheck_view 权限的用户才会看到「预约执行对账」tab。 INSERT INTO `pjcloudx`.`sys_menu` (`menu_id`, `name`, `permission`, `parent_id`, `visible`, `sort_order`, `keep_alive`, `embedded`, `menu_type`, `create_by`, `create_time`, `del_flag`, `tenant_id`) -SELECT (SELECT GREATEST(COALESCE(MAX(`menu_id`), 0), 2026092400000000000) FROM `pjcloudx`.`sys_menu`) + ROW_NUMBER() OVER (ORDER BY page_menu.`tenant_id`, page_menu.`menu_id`), - '标记已处理', 'wallet_walletRechargeExecCheck_handle', page_menu.`menu_id`, - '1', 10, '0', '0', '2', 'system', NOW(), '0', page_menu.`tenant_id` +SELECT (SELECT GREATEST(COALESCE(MAX(`menu_id`), 0), 2026092400000000000) FROM `pjcloudx`.`sys_menu`) + ROW_NUMBER() OVER (ORDER BY page_menu.`tenant_id`, page_menu.`menu_id`, item.`sort`), + item.`name`, item.`permission`, page_menu.`menu_id`, + '1', item.`sort`, '0', '0', '2', 'system', NOW(), '0', page_menu.`tenant_id` FROM `pjcloudx`.`sys_menu` page_menu -WHERE page_menu.`permission` = 'wallet_walletRechargeExecCheck_view' AND page_menu.`menu_type` = '0' AND page_menu.`del_flag` = '0' +JOIN ( + SELECT '预约执行对账' AS `name`, 'wallet_walletRechargeExecCheck_view' AS `permission`, 61 AS `sort` + UNION ALL SELECT '标记已处理', 'wallet_walletRechargeExecCheck_handle', 62 +) item ON item.`name` IS NOT NULL +WHERE page_menu.`permission` = 'wallet_walletRechargeApply_view' AND page_menu.`menu_type` = '0' AND page_menu.`del_flag` = '0' AND NOT EXISTS ( SELECT 1 FROM `pjcloudx`.`sys_menu` existing WHERE existing.`tenant_id` = page_menu.`tenant_id` - AND existing.`permission` = 'wallet_walletRechargeExecCheck_handle' AND existing.`del_flag` = '0' + AND existing.`permission` = item.`permission` AND existing.`del_flag` = '0' ); ------------------------------------------------------------------------ -六、后端权限字符串与接口对应关系（前端页面取用） +五、后端权限字符串与接口对应关系（前端页面取用） ------------------------------------------------------------------------ +wallet_walletRechargeApply_reserve: + 前端表单「预约执行」开关与时间控件显隐（v-if）；后端新增接口无权限强制关闭预约、 + 修改接口无权限忽略预约字段保持原值，防止绕过前端越权设置 + wallet_walletRechargeApply_execute: POST /wallet/wallet-recharge-apply/execute 立即执行预约单（待执行/待重试/失败终态均可触发） wallet_walletRechargeExecCheck_view: - GET /wallet/wallet-recharge-exec-check/page 对账差异分页 + GET /wallet/wallet-recharge-exec-check/page 对账差异分页（同时控制列表页「预约执行对账」tab 显隐） wallet_walletRechargeExecCheck_handle: - PUT /wallet/wallet-recharge-exec-check 标记差异已处理（填处理备注） + PUT /wallet/wallet-recharge-exec-check/handle 标记差异已处理（填处理备注） 内部端点（daemon-quartz 定时任务专用，@Inner 鉴权，勿对外）： POST /wallet/wallet-recharge-apply/inner/execute-due-reserve 扫描并执行到期预约单 POST /wallet/wallet-recharge-apply/inner/reconcile 预约单对账复核 ------------------------------------------------------------------------ -七、回滚说明 +六、回滚说明 ------------------------------------------------------------------------ 停用任务：将两条 sys_job 的 job_status 改为 '4'（删除）或 '3'（暂停）并重启 daemon-quartz （或调 POST /sys-job/refresh-jobs）； diff --git a/local.env b/local.env index e71fa9059..1cb6967d7 100644 --- a/local.env +++ b/local.env @@ -5,7 +5,7 @@ MYSQL_HOST=123.157.250.162 MYSQL_PORT=28816 -MYSQL_USER=root@sys +MYSQL_USER=root MYSQL_PWD=bsd.1799 SERVICE_MYSQL_USER=jhkjx diff --git a/pjcloud-thirdparty/pjcloud-thirdparty-api/src/main/java/com/bsd/pjcloud/thirdparty/api/feign/RemoteThirdPushDeliveryService.java b/pjcloud-thirdparty/pjcloud-thirdparty-api/src/main/java/com/bsd/pjcloud/thirdparty/api/feign/RemoteThirdPushDeliveryService.java new file mode 100644 index 000000000..374847e41 --- /dev/null +++ b/pjcloud-thirdparty/pjcloud-thirdparty-api/src/main/java/com/bsd/pjcloud/thirdparty/api/feign/RemoteThirdPushDeliveryService.java @@ -0,0 +1,32 @@ +package com.bsd.pjcloud.thirdparty.api.feign; + +import com.bsd.pjcloud.common.core.constant.ServiceNameConstants; +import com.bsd.pjcloud.common.core.util.R; +import com.bsd.pjcloud.common.feign.annotation.NoToken; +import org.springframework.cloud.openfeign.FeignClient; +import org.springframework.web.bind.annotation.PostMapping; + +/** + * daemon-quartz 定时触发 thirdparty App消息Push投递的内部 Feign 契约。 + */ +@FeignClient(contextId = "remoteThirdPushDeliveryService", value = ServiceNameConstants.THIRDPARTY_SERVICE) +public interface RemoteThirdPushDeliveryService { + + /** + * 扫描并处理所有租户到期的投递任务；条件更新领取保证重复触发幂等。 + * + * @return 本轮成功领取并处理的任务数量 + */ + @NoToken + @PostMapping("/third-push-delivery/inner/process-due") + R processDueDeliveries(); + + /** + * 分批物理清理超过保留期的收件箱、终态投递和供应商尝试日志。 + * + * @return 本轮实际清理记录数量 + */ + @NoToken + @PostMapping("/third-push-delivery/inner/cleanup") + R cleanupExpiredRecords(); +} diff --git a/pjcloud-thirdparty/pjcloud-thirdparty-api/src/main/java/com/bsd/pjcloud/thirdparty/vo/pay/ThirdPaymentTaskDispatchVo.java b/pjcloud-thirdparty/pjcloud-thirdparty-api/src/main/java/com/bsd/pjcloud/thirdparty/vo/pay/ThirdPaymentTaskDispatchVo.java index d76336087..940a144b6 100644 --- a/pjcloud-thirdparty/pjcloud-thirdparty-api/src/main/java/com/bsd/pjcloud/thirdparty/vo/pay/ThirdPaymentTaskDispatchVo.java +++ b/pjcloud-thirdparty/pjcloud-thirdparty-api/src/main/java/com/bsd/pjcloud/thirdparty/vo/pay/ThirdPaymentTaskDispatchVo.java @@ -1,13 +1,18 @@ package com.bsd.pjcloud.thirdparty.vo.pay; +import lombok.AllArgsConstructor; import lombok.Builder; import lombok.Data; +import lombok.NoArgsConstructor; /** * thirdparty 任务分发结果，供 Quartz 判断服务是否接受本轮触发。 */ @Data @Builder +// Feign 响应经 Jackson 反序列化必须有无参构造器；@Builder 依赖全参构造器，两者都要显式声明 +@NoArgsConstructor +@AllArgsConstructor public class ThirdPaymentTaskDispatchVo { /** 已接收任务数。 */ private int acceptedCount; diff --git a/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/ThirdPartyApp.java b/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/ThirdPartyApp.java index b245c00f9..825b3ea35 100644 --- a/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/ThirdPartyApp.java +++ b/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/ThirdPartyApp.java @@ -6,7 +6,6 @@ import com.bsd.pjcloud.common.swagger.annotation.EnableOpenApi; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.cloud.client.discovery.EnableDiscoveryClient; -import org.springframework.scheduling.annotation.EnableScheduling; /** * @author archetype auto @@ -17,7 +16,7 @@ import org.springframework.scheduling.annotation.EnableScheduling; @EnablePjcloudFeignClients @EnableDiscoveryClient @EnablePjcloudResourceServer -@EnableScheduling +// 周期调度已统一迁移到 daemon-quartz（sys_job 注册 thirdPushTask / 支付任务租约表），集群下避免各实例重复触发 @SpringBootApplication public class ThirdPartyApp { public static void main(String[] args) { diff --git a/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/controller/appmessage/ThirdPushDeliveryInnerController.java b/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/controller/appmessage/ThirdPushDeliveryInnerController.java new file mode 100644 index 000000000..877e0bbee --- /dev/null +++ b/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/controller/appmessage/ThirdPushDeliveryInnerController.java @@ -0,0 +1,50 @@ +package com.bsd.pjcloud.thirdparty.controller.appmessage; + +import com.bsd.pjcloud.common.core.util.R; +import com.bsd.pjcloud.common.feign.annotation.NoToken; +import com.bsd.pjcloud.common.security.annotation.Inner; +import com.bsd.pjcloud.thirdparty.service.appmessage.ThirdPushDeliveryService; +import io.swagger.v3.oas.annotations.Operation; +import lombok.RequiredArgsConstructor; +import org.springframework.web.bind.annotation.PostMapping; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RestController; + +/** + * App业务消息Push投递内部调度接口，供 daemon-quartz 定时任务通过 Feign 触发。 + * 投递任务由条件更新领取，Quartz 集群重复触发安全。 + */ +@RestController +@RequiredArgsConstructor +@RequestMapping("/third-push-delivery") +public class ThirdPushDeliveryInnerController { + + /** 投递、重试、超时恢复和保留期清理服务。 */ + private final ThirdPushDeliveryService deliveryService; + + /** + * 扫描并处理所有租户到期的投递任务，必须携带内部调用标识。 + * + * @return 本轮成功领取并处理的任务数量 + */ + @Operation(summary = "内部触发Push投递扫描") + @Inner + @NoToken + @PostMapping("/inner/process-due") + public R processDueDeliveries() { + return R.ok(deliveryService.processDueDeliveries()); + } + + /** + * 分批物理清理超过保留期的收件箱、终态投递和供应商尝试日志。 + * + * @return 本轮实际清理记录数量 + */ + @Operation(summary = "内部触发Push过期记录清理") + @Inner + @NoToken + @PostMapping("/inner/cleanup") + public R cleanupExpiredRecords() { + return R.ok(deliveryService.cleanupExpiredRecords()); + } +} diff --git a/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/job/ThirdAggregateOrderPayJob.java b/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/job/ThirdAggregateOrderPayJob.java deleted file mode 100644 index 28cf6ebb2..000000000 --- a/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/job/ThirdAggregateOrderPayJob.java +++ /dev/null @@ -1,176 +0,0 @@ -package com.bsd.pjcloud.thirdparty.job; - -import cn.hutool.core.util.ObjectUtil; -import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; -import com.bsd.pjcloud.common.data.tenant.TenantBroker; -import com.bsd.pjcloud.thirdparty.constant.pay.ThirdAggregatePaySessionStatusEnum; -import com.bsd.pjcloud.thirdparty.entity.pay.ThirdAggregatePaySessionEntity; -import com.bsd.pjcloud.thirdparty.service.ThirdAggregateOrderPayService; -import com.bsd.pjcloud.thirdparty.service.ThirdAggregatePaySessionService; -import com.bsd.pjcloud.thirdparty.service.ThirdPayPollingCoordinator; -import lombok.RequiredArgsConstructor; -import lombok.extern.slf4j.Slf4j; -import org.springframework.stereotype.Component; - -import java.time.LocalDateTime; -import java.util.List; - -/** - * 支付宝与微信聚合订单码处理器。周期入口已迁移到 Quartz 支付任务表。 - */ -@Component -@Slf4j -@RequiredArgsConstructor -public class ThirdAggregateOrderPayJob { - - /** - * 单轮最多处理的聚合会话数量，限制第三方网关慢调用对调度线程的占用。 - */ - private static final int BATCH_SIZE = 100; - - /** - * 到期后的关单补偿窗口，单位分钟；超过该窗口不再后台调用第三方支付平台。 - */ - private static final long CLOSE_COMPENSATION_MINUTES = 10L; - - /** - * 聚合会话服务，用于跨租户分批读取仍可能支付的会话。 - */ - private final ThirdAggregatePaySessionService aggregatePaySessionService; - - /** - * 聚合支付编排服务，负责平台查单、撤销、关单和原子状态同步。 - */ - private final ThirdAggregateOrderPayService aggregateOrderPayService; - - /** - * 旧兼容入口的进程内门控器；不创建线程，也不是支付周期调度来源。 - */ - private final ThirdPayPollingCoordinator pollingCoordinator; - - /** - * 分批处理等待扫码、平台创建中和支付中的聚合会话。 - * - *

该方法没有 {@code @Scheduled}，仅保留为显式诊断兼容入口。单条处理前使用记录自身 tenantId - * 恢复租户上下文；生产环境由 Quartz 分发任务表记录，不调用本方法形成 thirdparty 兜底调度。

- */ - public void handlePendingSessions() { - if (!pollingCoordinator.tryStartAggregatePendingPolling()) { - return; - } - boolean hasPendingSessions = true; - try { - LocalDateTime historyCutoff = LocalDateTime.now() - .minusMinutes(ThirdPayPollingCoordinator.MAX_AUTO_PROCESSING_MINUTES); - List sessions = aggregatePaySessionService.list( - new LambdaQueryWrapper() - .in(ThirdAggregatePaySessionEntity::getStatus, - ThirdAggregatePaySessionStatusEnum.WAIT_SCAN.getType(), - ThirdAggregatePaySessionStatusEnum.PLATFORM_CREATING.getType(), - ThirdAggregatePaySessionStatusEnum.PAYING.getType()) - .ge(ThirdAggregatePaySessionEntity::getCreateTime, historyCutoff) - .orderByAsc(ThirdAggregatePaySessionEntity::getCreateTime) - .last("limit " + BATCH_SIZE)); - hasPendingSessions = ObjectUtil.isNotEmpty(sessions); - if (!hasPendingSessions) { - return; - } - for (ThirdAggregatePaySessionEntity session : sessions) { - if (session.getTenantId() == null) { - log.warn("聚合支付会话缺少持久化租户ID,sessionId:{}", session.getId()); - continue; - } - try { - TenantBroker.runAs(session.getTenantId(), tenantId -> aggregateOrderPayService.processPendingSession(session.getId())); - } catch (Exception exception) { - log.error("处理聚合支付会话异常,sessionId:{},tenantId:{}", session.getId(), session.getTenantId(), exception); - } - } - } finally { - pollingCoordinator.finishAggregatePendingPolling(hasPendingSessions); - } - } - - /** - * 补偿处理已到期但子订单尚未确认关闭的聚合会话，并继续核对超过一小时的历史记录。 - * - *

补偿窗口从会话原始失效时间开始计算，最多持续十分钟；因此客户端退出、断网或不再查询状态时， - * 服务端仍会独立完成查单和关单。该兼容方法没有 {@code @Scheduled}，周期分发完全由 Quartz 和 - * {@code third_payment_task} 承担；存在渠道子订单的历史记录也不能本地盲关。

- */ - public void handleExpiredSessionCompensation() { - if (!pollingCoordinator.tryStartAggregateCompensationPolling()) { - return; - } - boolean hasExpiredSessions = true; - try { - LocalDateTime now = LocalDateTime.now(); - List historicalSessions = aggregatePaySessionService.list( - new LambdaQueryWrapper() - .in(ThirdAggregatePaySessionEntity::getStatus, - ThirdAggregatePaySessionStatusEnum.WAIT_SCAN.getType(), - ThirdAggregatePaySessionStatusEnum.PLATFORM_CREATING.getType(), - ThirdAggregatePaySessionStatusEnum.PAYING.getType(), - ThirdAggregatePaySessionStatusEnum.EXPIRED.getType()) - .and(timeWrapper -> timeWrapper - .isNull(ThirdAggregatePaySessionEntity::getCreateTime) - .or() - .lt(ThirdAggregatePaySessionEntity::getCreateTime, - now.minusMinutes(ThirdPayPollingCoordinator.MAX_AUTO_PROCESSING_MINUTES))) - .orderByAsc(ThirdAggregatePaySessionEntity::getCreateTime) - .last("limit " + BATCH_SIZE)); - closeHistoricalSessions(historicalSessions); - List sessions = aggregatePaySessionService.list( - new LambdaQueryWrapper() - .eq(ThirdAggregatePaySessionEntity::getStatus, ThirdAggregatePaySessionStatusEnum.EXPIRED.getType()) - .isNotNull(ThirdAggregatePaySessionEntity::getPayNo) - .isNotNull(ThirdAggregatePaySessionEntity::getExpireTime) - .between(ThirdAggregatePaySessionEntity::getExpireTime, - now.minusMinutes(CLOSE_COMPENSATION_MINUTES), now) - .orderByDesc(ThirdAggregatePaySessionEntity::getExpireTime) - .last("limit " + BATCH_SIZE)); - hasExpiredSessions = ObjectUtil.isNotEmpty(historicalSessions) || ObjectUtil.isNotEmpty(sessions); - for (ThirdAggregatePaySessionEntity session : sessions) { - if (session.getTenantId() == null) { - log.warn("到期聚合支付会话缺少持久化租户ID,sessionId:{}", session.getId()); - continue; - } - try { - TenantBroker.runAs(session.getTenantId(), tenantId -> aggregateOrderPayService.processPendingSession(session.getId())); - } catch (Exception exception) { - log.error("补偿关闭聚合支付会话异常,sessionId:{},tenantId:{}", session.getId(), session.getTenantId(), exception); - } - } - } finally { - pollingCoordinator.finishAggregateCompensationPolling(hasExpiredSessions); - } - } - - /** - * 处理超过一小时仍未收敛的聚合会话。 - * - *

方法只保留为旧诊断入口，不承担周期调度。每条记录仍交给聚合支付编排服务，后者对有渠道子单的 - * 会话执行查单或关单，对未知结果抛出异常并保留待处理状态。

- * - * @param sessions 超过一小时且仍处于活动或过期状态的聚合会话 - */ - private void closeHistoricalSessions(List sessions) { - if (ObjectUtil.isEmpty(sessions)) { - return; - } - for (ThirdAggregatePaySessionEntity session : sessions) { - if (session.getTenantId() == null) { - log.warn("历史聚合支付会话缺少持久化租户ID,sessionId:{}", session.getId()); - continue; - } - try { - TenantBroker.runAs(session.getTenantId(), tenantId -> { - aggregateOrderPayService.processPendingSession(session.getId()); - }); - } catch (Exception exception) { - log.warn("关闭历史聚合支付会话失败,sessionId:{},tenantId:{},exceptionType:{}", - session.getId(), session.getTenantId(), exception.getClass().getSimpleName()); - } - } - } -} diff --git a/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/job/ThirdAppJgPushJob.java b/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/job/ThirdAppJgPushJob.java deleted file mode 100644 index 3e126e34a..000000000 --- a/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/job/ThirdAppJgPushJob.java +++ /dev/null @@ -1,56 +0,0 @@ -package com.bsd.pjcloud.thirdparty.job; - -import com.bsd.pjcloud.admin.api.feign.RemoteUserService; -import com.bsd.pjcloud.thirdparty.config.JgConfig; -import com.bsd.pjcloud.thirdparty.dto.jg.JgPusDto; -import com.bsd.pjcloud.thirdparty.service.ThirdJgPushService; -import lombok.RequiredArgsConstructor; -import lombok.extern.slf4j.Slf4j; -import org.springframework.stereotype.Component; - -/** - * App 极光推送定时任务，测试推送对象与设备清理策略共用同一份配置名单。 - */ -@Component -@Slf4j -@RequiredArgsConstructor -public class ThirdAppJgPushJob { - - /** - * 极光消息推送服务。 - */ - private final ThirdJgPushService thirdJgPushService; - - /** - * 用户远程服务，供后续恢复全量用户推送任务时读取用户范围。 - */ - private final RemoteUserService remoteUserService; - - /** - * 极光配置，提供统一维护的测试用户ID列表。 - */ - private final JgConfig jgConfig; - - -// @PostConstruct -// public void test() { -//// pushToAllUsers("早餐提醒", "早餐时间就要到了,请尽快预约哦!!"); -//// pushToAllUsers("午餐提醒", "午餐间就要到了,请尽快预约哦!!"); -//// pushToAllUsers("晚餐提醒", "晚餐时间就要到了,请尽快预约哦!!"); -// } - - /** - * 向配置中的测试用户发送定时提醒，避免测试名单在任务和设备策略中重复维护。 - * - * @param title 通知标题 - * @param subtitle 通知副标题 - */ - private void pushToAllUsers(String title, String subtitle) { - JgPusDto jgPusDto = new JgPusDto(); - jgPusDto.setTitle(title); - jgPusDto.setSubtitle(subtitle); - jgPusDto.setUserIds(jgConfig.getJgPush().getTestUserIds()); - thirdJgPushService.push(jgPusDto); - } - -} diff --git a/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/job/ThirdAppMessagePushJob.java b/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/job/ThirdAppMessagePushJob.java deleted file mode 100644 index 91cebf3f1..000000000 --- a/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/job/ThirdAppMessagePushJob.java +++ /dev/null @@ -1,49 +0,0 @@ -package com.bsd.pjcloud.thirdparty.job; - -import com.bsd.pjcloud.thirdparty.service.appmessage.ThirdPushDeliveryService; -import lombok.RequiredArgsConstructor; -import lombok.extern.slf4j.Slf4j; -import org.springframework.scheduling.annotation.Scheduled; -import org.springframework.stereotype.Component; - -/** - * App业务消息Push投递扫描任务，服务多实例时由条件领取避免同一任务并发执行。 - */ -@Slf4j -@Component -@RequiredArgsConstructor -public class ThirdAppMessagePushJob { - - /** 投递、重试和超时恢复服务。 */ - private final ThirdPushDeliveryService deliveryService; - - /** - * 每十秒扫描一次到期任务；每轮限制数量，避免阻塞既有定时广播任务。 - */ - @Scheduled(fixedDelay = 10000L, initialDelay = 15000L) - public void processDueDeliveries() { - try { - int processed = deliveryService.processDueDeliveries(); - if (processed > 0) { - log.info("App业务消息Push投递扫描完成,processed:{}", processed); - } - } catch (Exception exception) { - log.error("App业务消息Push投递扫描异常", exception); - } - } - - /** - * 每日凌晨分批清理超过保留期的消息、终态投递和供应商尝试日志。 - */ - @Scheduled(cron = "0 10 2 * * ?") - public void cleanupExpiredRecords() { - try { - int deleted = deliveryService.cleanupExpiredRecords(); - if (deleted > 0) { - log.info("App业务消息到期记录清理完成,deleted:{}", deleted); - } - } catch (Exception exception) { - log.error("App业务消息到期记录清理异常", exception); - } - } -} diff --git a/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/job/ThirdAppPlusPayJob.java b/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/job/ThirdAppPlusPayJob.java deleted file mode 100644 index 301cf3906..000000000 --- a/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/job/ThirdAppPlusPayJob.java +++ /dev/null @@ -1,494 +0,0 @@ -package com.bsd.pjcloud.thirdparty.job; - - -import cn.hutool.core.util.ObjectUtil; -import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; -import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper; -import com.bsd.pjcloud.common.core.constant.enums.YesNoEnum; -import com.bsd.pjcloud.common.data.tenant.TenantBroker; -import com.bsd.pjcloud.thirdparty.constant.miniplatform.AppPlusPlatformTypeEnum; -import com.bsd.pjcloud.thirdparty.constant.fz.ThirdFzOrderStatusEnum; -import com.bsd.pjcloud.thirdparty.constant.pay.ThirdPayOrderBizTypeEnum; -import com.bsd.pjcloud.thirdparty.constant.pay.ThirdPayOrderStatusEnum; -import com.bsd.pjcloud.thirdparty.entity.pay.ThirdPayOrderEntity; -import com.bsd.pjcloud.thirdparty.entity.pay.ThirdFzOrderEntity; -import com.bsd.pjcloud.thirdparty.service.ThirdAppPlusAliPayService; -import com.bsd.pjcloud.thirdparty.service.ThirdAppPlusService; -import com.bsd.pjcloud.thirdparty.service.ThirdAppPlusWechatFzService; -import com.bsd.pjcloud.thirdparty.service.ThirdFzOrderService; -import com.bsd.pjcloud.thirdparty.service.ThirdPayOrderService; -import com.bsd.pjcloud.thirdparty.service.ThirdPayPollingCoordinator; -import lombok.RequiredArgsConstructor; -import lombok.extern.slf4j.Slf4j; -import org.springframework.stereotype.Component; - -import java.time.LocalDateTime; -import java.util.List; - -/** - * 第三方支付订单处理器。周期入口已迁移到 Quartz 支付任务表，本类只保留既有批处理方法供诊断兼容。 - */ -@Component -@Slf4j -@RequiredArgsConstructor -public class ThirdAppPlusPayJob { - - /** - * 单轮订单码轮询的最大处理数量，避免慢支付网关调用长期占用调度线程。 - */ - private static final int ORDER_CODE_BATCH_SIZE = 100; - - /** - * 单轮 App 支付、退款和分账补偿最大处理数量，避免第三方网络调用占满调度线程。 - */ - private static final int PAYMENT_COMPENSATION_BATCH_SIZE = 100; - - /** - * 支付成功后等待资金进入可结算状态的最短时间，单位秒。 - */ - private static final long FZ_MIN_DELAY_SECONDS = 30L; - - /** - * 原生订单码到期后的关单补偿窗口，单位分钟；超过窗口后停止后台查询。 - */ - private static final long ORDER_CODE_CLOSE_COMPENSATION_MINUTES = 10L; - - /** - * 支付订单服务，用于分批读取仍未支付的订单码记录。 - */ - private final ThirdPayOrderService thirdPayOrderService; - - /** - * 支付编排服务，负责按平台查询并执行带状态条件的成功或关闭更新。 - */ - private final ThirdAppPlusService thirdAppPlusService; - - /** - * 分账订单服务，用于扫描微信及支付宝待确认的分账状态。 - */ - private final ThirdFzOrderService thirdFzOrderService; - - /** - * 微信分账服务，用于创建及查询分账、分账退款。 - */ - private final ThirdAppPlusWechatFzService thirdAppPlusWechatFzService; - - /** - * 支付宝服务，用于直付通确认结算和平台抽佣补偿。 - */ - private final ThirdAppPlusAliPayService thirdAppPlusAliPayService; - - /** - * 旧兼容入口的进程内门控器；不创建线程，也不承担生产支付调度。 - */ - private final ThirdPayPollingCoordinator pollingCoordinator; - - /** - * 补偿支付或退款平台终态已经落库、但跨服务业务后置尚未完成的订单。 - * - *

不限制创建时间，确保服务长期停机后仍能恢复；单轮限制数量并在订单租户上下文中执行。

- */ - public void handleUnBizCallbackOrder() { - List list = thirdPayOrderService.list(new LambdaQueryWrapper() - .and(wrapper -> wrapper.isNull(ThirdPayOrderEntity::getIsBizCallback) - .or().eq(ThirdPayOrderEntity::getIsBizCallback, YesNoEnum.NO.getCode())) - .in(ThirdPayOrderEntity::getStatus, ThirdPayOrderStatusEnum.PAY_SUCCESS.getType(), - ThirdPayOrderStatusEnum.REFUND_SUCCESS.getType()) - .orderByAsc(ThirdPayOrderEntity::getUpdateTime) - .last("limit " + PAYMENT_COMPENSATION_BATCH_SIZE)); - for (ThirdPayOrderEntity item : list) { - runWithOrderTenant(item, () -> { - ThirdPayOrderEntity latest = thirdPayOrderService.getById(item.getId()); - thirdAppPlusService.queryPayResult(latest); - touchPendingBusinessCallback(item.getId(), item.getStatus()); - }, "业务回调补偿"); - } - } - - /** - * 分批处理支付宝和微信云消费机订单码，最长处理到订单过期后十分钟。 - * - *

该方法没有 {@code @Scheduled}，仅保留为显式诊断兼容入口。处理单条订单前通过 TenantBroker - * 恢复订单所属租户；生产周期调度只由 Quartz 分发 {@code third_payment_task}。

- */ - public void handleOrderCodeOrders() { - if (!pollingCoordinator.tryStartOrderCodePolling()) { - return; - } - boolean hasPendingOrders = true; - try { - LocalDateTime earliestExpireTime = LocalDateTime.now().minusMinutes(ORDER_CODE_CLOSE_COMPENSATION_MINUTES); - List orders = thirdPayOrderService.list(new LambdaQueryWrapper() - .in(ThirdPayOrderEntity::getPayPlatformType, - AppPlusPlatformTypeEnum.ALI_PAY.getType(), AppPlusPlatformTypeEnum.WECHAT.getType()) - .eq(ThirdPayOrderEntity::getBizType, ThirdPayOrderBizTypeEnum.YXFJ_PAY.getType()) - .eq(ThirdPayOrderEntity::getStatus, ThirdPayOrderStatusEnum.PAY_UNPAID.getType()) - .isNotNull(ThirdPayOrderEntity::getExpireTime) - .ge(ThirdPayOrderEntity::getExpireTime, earliestExpireTime) - .notExists("SELECT 1 FROM third_aggregate_pay_session s WHERE s.pay_no = third_pay_order.pay_no AND s.del_flag = '0'") - .orderByAsc(ThirdPayOrderEntity::getCreateTime) - .last("limit " + ORDER_CODE_BATCH_SIZE)); - hasPendingOrders = ObjectUtil.isNotEmpty(orders); - if (!hasPendingOrders) { - return; - } - for (ThirdPayOrderEntity order : orders) { - if (order.getTenantId() == null) { - log.warn("订单码缺少租户ID,orderId:{},platform:{}", order.getId(), order.getPayPlatformType()); - continue; - } - try { - TenantBroker.runAs(order.getTenantId(), tenantId -> { - ThirdPayOrderEntity latestOrder = thirdPayOrderService.getById(order.getId()); - if (latestOrder != null && ObjectUtil.equals(latestOrder.getStatus(), ThirdPayOrderStatusEnum.PAY_UNPAID.getType())) { - thirdAppPlusService.queryPayResult(latestOrder); - } - }); - } catch (Exception exception) { - log.error("处理订单码状态异常,orderId:{},tenantId:{},platform:{}", - order.getId(), order.getTenantId(), order.getPayPlatformType(), exception); - } - } - } finally { - pollingCoordinator.finishOrderCodePolling(hasPendingOrders); - } - } - - /** - * 轮询 App 支付订单，覆盖微信或支付宝回调丢失、服务重启及多实例部署场景。 - */ - public void handleAppPaymentOrders() { - if (!pollingCoordinator.tryStartAppPaymentPolling()) { - return; - } - boolean hasPendingOrders = true; - try { - List orders = thirdPayOrderService.list(new LambdaQueryWrapper() - .in(ThirdPayOrderEntity::getPayPlatformType, - AppPlusPlatformTypeEnum.ALI_PAY.getType(), AppPlusPlatformTypeEnum.WECHAT.getType()) - .in(ThirdPayOrderEntity::getBizType, List.of( - ThirdPayOrderBizTypeEnum.APP_SHOP_PAY.getType(), - ThirdPayOrderBizTypeEnum.MINI_APP_WALLET_RECHARGE.getType(), - ThirdPayOrderBizTypeEnum.GUARDIAN_WARD_WALLET_RECHARGE.getType())) - .eq(ThirdPayOrderEntity::getStatus, ThirdPayOrderStatusEnum.PAY_UNPAID.getType()) - .orderByAsc(ThirdPayOrderEntity::getUpdateTime) - .last("limit " + PAYMENT_COMPENSATION_BATCH_SIZE)); - hasPendingOrders = ObjectUtil.isNotEmpty(orders); - for (ThirdPayOrderEntity order : orders) { - runWithOrderTenant(order, () -> { - ThirdPayOrderEntity latest = thirdPayOrderService.getById(order.getId()); - if (latest != null && ObjectUtil.equals(latest.getStatus(), ThirdPayOrderStatusEnum.PAY_UNPAID.getType())) { - thirdAppPlusService.queryPayResult(latest); - touchPendingPayment(order.getId()); - } - }, "App支付查单"); - } - } finally { - pollingCoordinator.finishAppPaymentPolling(hasPendingOrders); - } - } - - /** - * 轮询微信和支付宝普通退款状态；支付宝退款至少间隔十秒后查询，避免把受理响应误判为终态。 - */ - public void handleRefundOrders() { - if (!pollingCoordinator.tryStartRefundPolling()) { - return; - } - boolean hasPendingRefunds = true; - try { - List orders = thirdPayOrderService.list(new LambdaQueryWrapper() - .in(ThirdPayOrderEntity::getPayPlatformType, - AppPlusPlatformTypeEnum.WECHAT.getType(), AppPlusPlatformTypeEnum.ALI_PAY.getType()) - .eq(ThirdPayOrderEntity::getStatus, ThirdPayOrderStatusEnum.REFUNDING.getType()) - .isNotNull(ThirdPayOrderEntity::getRefundNo) - .le(ThirdPayOrderEntity::getRefundTime, LocalDateTime.now().minusSeconds(10)) - .orderByAsc(ThirdPayOrderEntity::getUpdateTime) - .last("limit " + PAYMENT_COMPENSATION_BATCH_SIZE)); - hasPendingRefunds = ObjectUtil.isNotEmpty(orders); - for (ThirdPayOrderEntity order : orders) { - runWithOrderTenant(order, () -> { - thirdAppPlusService.queryRefundResult(thirdPayOrderService.getById(order.getId())); - touchPendingRefund(order.getId(), order.getRefundNo()); - }, "退款查单"); - } - } finally { - pollingCoordinator.finishRefundPolling(hasPendingRefunds); - } - } - - /** - * 创建支付成功后缺失的直付通阶段记录，并轮询微信分账、微信分账退款及支付宝确认结算/可选抽佣状态。 - */ - public void handleFzOrders() { - if (!pollingCoordinator.tryStartFzPolling()) { - return; - } - boolean hasPendingFz = true; - try { - createMissingFzOrders(); - reconcileSuccessfulFzOrders(); - List fzOrders = thirdFzOrderService.list(new LambdaQueryWrapper() - .in(ThirdFzOrderEntity::getStatus, ThirdFzOrderStatusEnum.FZ_UNPAID.getType(), - ThirdFzOrderStatusEnum.FZ_REFUNDING.getType()) - // 关闭平台抽佣的支付宝订单确认结算成功后已经完成全部资金处理，不能继续占用旧轮询批次。 - .notExists("SELECT 1 FROM third_pay_order p WHERE p.tenant_id = third_fz_order.tenant_id" - + " AND p.pay_no = third_fz_order.pay_no AND p.del_flag = '0'" - + " AND p.pay_platform_type = '20' AND (p.enable_fz IS NULL OR p.enable_fz <> '1')" - + " AND JSON_UNQUOTE(JSON_EXTRACT(COALESCE(third_fz_order.create_res, '{}'), '$.settleConfirm.status')) = 'SUCCESS'") - .orderByAsc(ThirdFzOrderEntity::getUpdateTime) - .last("limit " + PAYMENT_COMPENSATION_BATCH_SIZE)); - hasPendingFz = ObjectUtil.isNotEmpty(fzOrders); - for (ThirdFzOrderEntity fzOrder : fzOrders) { - if (fzOrder.getTenantId() == null) { - log.warn("分账订单缺少租户ID,fzOrderId:{}", fzOrder.getId()); - continue; - } - try { - TenantBroker.runAs(fzOrder.getTenantId(), tenantId -> pollFzOrder(fzOrder.getId())); - } catch (Exception exception) { - log.warn("分账轮询异常,fzOrderId:{},exceptionType:{}", fzOrder.getId(), exception.getClass().getSimpleName()); - } - } - } finally { - pollingCoordinator.finishFzPolling(hasPendingFz); - } - } - - /** - * 为支付成功满 30 秒且尚无阶段记录的支付宝直付通订单或已开启分账的微信订单创建稳定请求。 - */ - private void createMissingFzOrders() { - List orders = thirdPayOrderService.list(new LambdaQueryWrapper() - .eq(ThirdPayOrderEntity::getStatus, ThirdPayOrderStatusEnum.PAY_SUCCESS.getType()) - .and(wrapper -> wrapper.eq(ThirdPayOrderEntity::getPayPlatformType, AppPlusPlatformTypeEnum.ALI_PAY.getType()) - .or(wechat -> wechat.eq(ThirdPayOrderEntity::getPayPlatformType, AppPlusPlatformTypeEnum.WECHAT.getType()) - .eq(ThirdPayOrderEntity::getEnableFz, YesNoEnum.YES.getCode()))) - .le(ThirdPayOrderEntity::getPayTime, LocalDateTime.now().minusSeconds(FZ_MIN_DELAY_SECONDS)) - .notExists("SELECT 1 FROM third_fz_order f WHERE f.pay_no = third_pay_order.pay_no AND f.del_flag = '0'") - .orderByAsc(ThirdPayOrderEntity::getUpdateTime) - .last("limit " + PAYMENT_COMPENSATION_BATCH_SIZE)); - for (ThirdPayOrderEntity order : orders) { - runWithOrderTenant(order, () -> { - ThirdPayOrderEntity latest = thirdPayOrderService.getById(order.getId()); - if (latest == null || thirdFzOrderService.getByPayNo(latest.getPayNo()) != null) { - return; - } - if (ObjectUtil.equals(latest.getPayPlatformType(), AppPlusPlatformTypeEnum.WECHAT.getType())) { - thirdAppPlusWechatFzService.create(latest); - } else { - thirdAppPlusAliPayService.processDirectPayLifecycle(latest); - } - if (thirdFzOrderService.getByPayNo(latest.getPayNo()) == null) { - touchPayOrderAwaitingFzCreation(latest.getId()); - } - }, "创建直付通阶段记录"); - } - } - - /** - * 补偿分账终态已落库但支付订单最终抽佣标记尚未写入的窄窗口。 - * - *

分账记录与支付订单不在同一条原子更新中，多实例或进程退出可能发生在两次写入之间。 - * 本方法只信任 {@code third_fz_order} 的最终成功状态，并使用支付订单原状态条件重复写入， - * 不会重新发起支付宝或微信分账请求。

- */ - private void reconcileSuccessfulFzOrders() { - List orders = thirdPayOrderService.list(new LambdaQueryWrapper() - .eq(ThirdPayOrderEntity::getStatus, ThirdPayOrderStatusEnum.PAY_SUCCESS.getType()) - .eq(ThirdPayOrderEntity::getEnableFz, YesNoEnum.YES.getCode()) - .and(wrapper -> wrapper.isNull(ThirdPayOrderEntity::getIsFzSuccess) - .or().ne(ThirdPayOrderEntity::getIsFzSuccess, YesNoEnum.YES.getCode())) - .exists("SELECT 1 FROM third_fz_order f WHERE f.pay_no = third_pay_order.pay_no AND f.del_flag = '0' AND f.status = '" - + ThirdFzOrderStatusEnum.FZ_SUCCESS.getType() + "'") - .orderByAsc(ThirdPayOrderEntity::getUpdateTime) - .last("limit " + PAYMENT_COMPENSATION_BATCH_SIZE)); - for (ThirdPayOrderEntity order : orders) { - runWithOrderTenant(order, () -> { - ThirdPayOrderEntity latest = thirdPayOrderService.getById(order.getId()); - if (latest == null || ObjectUtil.equals(latest.getIsFzSuccess(), YesNoEnum.YES.getCode())) { - return; - } - ThirdFzOrderEntity fzOrder = thirdFzOrderService.getByPayNo(latest.getPayNo()); - if (fzOrder == null || ObjectUtil.notEqual(fzOrder.getStatus(), ThirdFzOrderStatusEnum.FZ_SUCCESS.getType())) { - return; - } - ThirdPayOrderEntity update = new ThirdPayOrderEntity(); - update.setFzPrice(fzOrder.getPrice()); - update.setIsFzSuccess(YesNoEnum.YES.getCode()); - thirdPayOrderService.update(update, new LambdaUpdateWrapper() - .eq(ThirdPayOrderEntity::getId, latest.getId()) - .eq(ThirdPayOrderEntity::getStatus, ThirdPayOrderStatusEnum.PAY_SUCCESS.getType()) - .and(wrapper -> wrapper.isNull(ThirdPayOrderEntity::getIsFzSuccess) - .or().ne(ThirdPayOrderEntity::getIsFzSuccess, YesNoEnum.YES.getCode()))); - }, "分账成功标记补偿"); - } - } - - /** - * 按分账订单平台查询并推进状态。 - * - * @param fzOrderId 分账订单主键 - */ - private void pollFzOrder(Long fzOrderId) { - ThirdFzOrderEntity fzOrder = thirdFzOrderService.getById(fzOrderId); - if (fzOrder == null) { - return; - } - ThirdPayOrderEntity payOrder = thirdPayOrderService.getByPayNo(fzOrder.getPayNo()); - if (payOrder == null) { - return; - } - if (ObjectUtil.equals(payOrder.getPayPlatformType(), AppPlusPlatformTypeEnum.WECHAT.getType())) { - if (ObjectUtil.equals(fzOrder.getStatus(), ThirdFzOrderStatusEnum.FZ_REFUNDING.getType())) { - thirdAppPlusWechatFzService.pollFzRefundOrder(fzOrder); - } else { - thirdAppPlusWechatFzService.pollFzOrder(fzOrder); - } - } else if (List.of(ThirdFzOrderStatusEnum.FZ_UNPAID.getType(), - ThirdFzOrderStatusEnum.FZ_REFUNDING.getType()).contains(fzOrder.getStatus())) { - thirdAppPlusAliPayService.pollDirectPayLifecycle(fzOrder, payOrder); - } - touchPendingFzOrder(fzOrder.getId(), fzOrder.getStatus()); - } - - /** - * 将仍未支付的 App 订单移动到下一轮批次队尾，避免长期 NOTPAY 记录阻塞较新的订单。 - * - * @param orderId 支付订单主键 - */ - private void touchPendingPayment(Long orderId) { - ThirdPayOrderEntity update = new ThirdPayOrderEntity(); - update.setUpdateTime(LocalDateTime.now()); - thirdPayOrderService.update(update, new LambdaUpdateWrapper() - .eq(ThirdPayOrderEntity::getId, orderId) - .eq(ThirdPayOrderEntity::getStatus, ThirdPayOrderStatusEnum.PAY_UNPAID.getType())); - } - - /** - * 将仍在退款中的订单移动到批次队尾，退款号条件确保不会触碰并发发起的另一笔退款。 - * - * @param orderId 支付订单主键 - * @param refundNo 本轮查询使用的稳定退款号 - */ - private void touchPendingRefund(Long orderId, String refundNo) { - ThirdPayOrderEntity update = new ThirdPayOrderEntity(); - update.setUpdateTime(LocalDateTime.now()); - thirdPayOrderService.update(update, new LambdaUpdateWrapper() - .eq(ThirdPayOrderEntity::getId, orderId) - .eq(ThirdPayOrderEntity::getStatus, ThirdPayOrderStatusEnum.REFUNDING.getType()) - .eq(ThirdPayOrderEntity::getRefundNo, refundNo)); - } - - /** - * 将业务回调仍未完成的终态订单移动到队尾，使单个跨服务故障不会占满固定批次。 - * - * @param orderId 支付订单主键 - * @param status 本轮读取的支付或退款成功状态 - */ - private void touchPendingBusinessCallback(Long orderId, String status) { - ThirdPayOrderEntity update = new ThirdPayOrderEntity(); - update.setUpdateTime(LocalDateTime.now()); - thirdPayOrderService.update(update, new LambdaUpdateWrapper() - .eq(ThirdPayOrderEntity::getId, orderId) - .eq(ThirdPayOrderEntity::getStatus, status) - .and(wrapper -> wrapper.isNull(ThirdPayOrderEntity::getIsBizCallback) - .or().eq(ThirdPayOrderEntity::getIsBizCallback, YesNoEnum.NO.getCode()))); - } - - /** - * 直付通阶段记录因历史配置异常尚未创建时轮转支付订单，防止它持续占据缺失记录扫描首批。 - * - * @param orderId 支付订单主键 - */ - private void touchPayOrderAwaitingFzCreation(Long orderId) { - ThirdPayOrderEntity update = new ThirdPayOrderEntity(); - update.setUpdateTime(LocalDateTime.now()); - thirdPayOrderService.update(update, new LambdaUpdateWrapper() - .eq(ThirdPayOrderEntity::getId, orderId) - .eq(ThirdPayOrderEntity::getStatus, ThirdPayOrderStatusEnum.PAY_SUCCESS.getType()) - .and(wrapper -> wrapper.eq(ThirdPayOrderEntity::getEnableFz, YesNoEnum.YES.getCode()) - .or().eq(ThirdPayOrderEntity::getPayPlatformType, AppPlusPlatformTypeEnum.ALI_PAY.getType()))); - } - - /** - * 将仍处于同一处理中状态的分账记录移动到队尾；终态更新会因原状态条件不匹配而保持不变。 - * - * @param fzOrderId 分账订单主键 - * @param status 本轮读取的待分账或退分账状态 - */ - private void touchPendingFzOrder(Long fzOrderId, String status) { - if (!List.of(ThirdFzOrderStatusEnum.FZ_UNPAID.getType(), - ThirdFzOrderStatusEnum.FZ_REFUNDING.getType()).contains(status)) { - return; - } - ThirdFzOrderEntity update = new ThirdFzOrderEntity(); - update.setUpdateTime(LocalDateTime.now()); - thirdFzOrderService.update(update, new LambdaUpdateWrapper() - .eq(ThirdFzOrderEntity::getId, fzOrderId) - .eq(ThirdFzOrderEntity::getStatus, status)); - } - - /** - * 在支付订单所属租户上下文中执行补偿逻辑。 - * - * @param order 支付订单 - * @param action 补偿动作 - * @param actionName 日志动作名称 - */ - private void runWithOrderTenant(ThirdPayOrderEntity order, Runnable action, String actionName) { - if (order.getTenantId() == null) { - log.warn("{}缺少租户ID,orderId:{}", actionName, order.getId()); - return; - } - try { - TenantBroker.runAs(order.getTenantId(), tenantId -> action.run()); - } catch (Exception exception) { - log.warn("{}异常,orderId:{},exceptionType:{}", actionName, order.getId(), exception.getClass().getSimpleName()); - } - } - - /** - * 核对超过一小时仍未支付的原生支付宝、微信订单码。 - * - *

该方法没有 {@code @Scheduled}，只保留为显式诊断入口。历史年龄不能替代渠道终态， - * 因此仍复用支付编排服务按订单配置快照查单或关单，未知结果保持未支付。

- */ - public void closeHistoricalOrderCodeOrders() { - LocalDateTime cutoff = LocalDateTime.now() - .minusMinutes(ThirdPayPollingCoordinator.MAX_AUTO_PROCESSING_MINUTES); - List orders = thirdPayOrderService.list(new LambdaQueryWrapper() - .in(ThirdPayOrderEntity::getPayPlatformType, - AppPlusPlatformTypeEnum.ALI_PAY.getType(), AppPlusPlatformTypeEnum.WECHAT.getType()) - .eq(ThirdPayOrderEntity::getBizType, ThirdPayOrderBizTypeEnum.YXFJ_PAY.getType()) - .eq(ThirdPayOrderEntity::getStatus, ThirdPayOrderStatusEnum.PAY_UNPAID.getType()) - .isNotNull(ThirdPayOrderEntity::getExpireTime) - .and(timeWrapper -> timeWrapper - .isNull(ThirdPayOrderEntity::getCreateTime) - .or() - .lt(ThirdPayOrderEntity::getCreateTime, cutoff)) - .notExists("SELECT 1 FROM third_aggregate_pay_session s WHERE s.pay_no = third_pay_order.pay_no AND s.del_flag = '0'") - .orderByAsc(ThirdPayOrderEntity::getCreateTime) - .last("limit " + ORDER_CODE_BATCH_SIZE)); - for (ThirdPayOrderEntity order : orders) { - if (order.getTenantId() == null) { - log.warn("历史订单码缺少租户ID,orderId:{},platform:{}", order.getId(), order.getPayPlatformType()); - continue; - } - try { - TenantBroker.runAs(order.getTenantId(), tenantId -> { - ThirdPayOrderEntity latest = thirdPayOrderService.getById(order.getId()); - if (latest != null && ObjectUtil.equals( - latest.getStatus(), ThirdPayOrderStatusEnum.PAY_UNPAID.getType())) { - thirdAppPlusService.queryPayResult(latest); - } - }); - } catch (Exception exception) { - log.warn("关闭历史订单码失败,orderId:{},tenantId:{},platform:{},exceptionType:{}", - order.getId(), order.getTenantId(), order.getPayPlatformType(), - exception.getClass().getSimpleName()); - } - } - } - -} diff --git a/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/service/impl/appplus/ThirdAppPlusServiceImpl.java b/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/service/impl/appplus/ThirdAppPlusServiceImpl.java index c4026f5fe..cf454dfab 100644 --- a/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/service/impl/appplus/ThirdAppPlusServiceImpl.java +++ b/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/service/impl/appplus/ThirdAppPlusServiceImpl.java @@ -137,6 +137,11 @@ public class ThirdAppPlusServiceImpl implements ThirdAppPlusService { */ private static final String ALIPAY_CANCEL_RETRY = "Y"; + /** + * 渠道未回传截止时间的未支付订单按创建时间兜底的过期小时数，超时后查单确认并走渠道关单。 + */ + private static final long UNPAID_CLOSE_FALLBACK_HOURS = 2L; + /** * 仅当 Redis 锁值仍属于当前请求时删除锁，避免超时后误删后继请求的锁。 */ @@ -1433,13 +1438,22 @@ public class ThirdAppPlusServiceImpl implements ThirdAppPlusService { /** * 判断支付订单是否已经超过渠道下单时使用的截止时间。 * + *

渠道未回传截止时间的历史订单（expire_time 为空）按创建时间兜底：超过 + * {@link #UNPAID_CLOSE_FALLBACK_HOURS} 小时仍未支付视为到期，交由查单确认后走渠道关单收敛， + * 避免放弃支付的订单被查单任务无限轮询。

+ * * @param payOrder 当前内部支付订单 - * @return 订单仍未支付、保存了截止时间且已经到期时返回 {@code true} + * @return 订单仍未支付且已按截止时间或创建时间兜底判定到期时返回 {@code true} */ private boolean isExpiredUnpaidPayOrder(ThirdPayOrderEntity payOrder) { - return payOrder != null && payOrder.getExpireTime() != null - && ObjectUtil.equals(payOrder.getStatus(), ThirdPayOrderStatusEnum.PAY_UNPAID.getType()) - && !payOrder.getExpireTime().isAfter(LocalDateTime.now()); + if (payOrder == null || !ObjectUtil.equals(payOrder.getStatus(), ThirdPayOrderStatusEnum.PAY_UNPAID.getType())) { + return false; + } + if (payOrder.getExpireTime() != null) { + return !payOrder.getExpireTime().isAfter(LocalDateTime.now()); + } + return payOrder.getCreateTime() != null + && payOrder.getCreateTime().isBefore(LocalDateTime.now().minusHours(UNPAID_CLOSE_FALLBACK_HOURS)); } /** diff --git a/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/service/impl/pay/ThirdPaymentTaskServiceImpl.java b/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/service/impl/pay/ThirdPaymentTaskServiceImpl.java index 0a171cb48..9ee8dc863 100644 --- a/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/service/impl/pay/ThirdPaymentTaskServiceImpl.java +++ b/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/java/com/bsd/pjcloud/thirdparty/service/impl/pay/ThirdPaymentTaskServiceImpl.java @@ -77,6 +77,9 @@ public class ThirdPaymentTaskServiceImpl extends ServiceImpl 0 && attempts >= maxAttempts) { + // 不可收敛任务停止轮询转终态，保留最后错误供人工排查后 manualRetry 恢复 + baseMapper.fail(task.getId(), task.getLeaseToken(), safeCode, safeMessage, LocalDateTime.now()); + return; + } long seconds = Math.min(maxBackoffSeconds, 5L * (1L << Math.min(attempts, 8))); String status = attempts >= alertAttempts ? ThirdPaymentTaskStatusEnum.ALERT.getCode() : ThirdPaymentTaskStatusEnum.RETRY.getCode(); baseMapper.retry(task.getId(), task.getLeaseToken(), status, LocalDateTime.now().plusSeconds(seconds), diff --git a/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/resources/mapper/ThirdPaymentTaskMapper.xml b/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/resources/mapper/ThirdPaymentTaskMapper.xml index ee5c6ae05..3fecaa5ac 100644 --- a/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/resources/mapper/ThirdPaymentTaskMapper.xml +++ b/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/main/resources/mapper/ThirdPaymentTaskMapper.xml @@ -202,8 +202,12 @@ SELECT p.tenant_id, #{taskType} AS task_type, CONCAT('PAY_CLOSE:', p.pay_no) AS task_key, p.id AS pay_order_id, p.pay_no, 10 AS priority, #{now} AS next_run_time FROM third_pay_order p - WHERE p.del_flag = '0' AND p.status = '10' AND p.biz_type = '20' - AND p.expire_time IS NOT NULL AND (p.create_time IS NULL OR p.create_time < DATE_SUB(#{now}, INTERVAL 60 MINUTE)) + WHERE p.del_flag = '0' AND p.status = '10' + AND ( + (p.expire_time IS NOT NULL AND p.expire_time <= #{now}) + OR (p.expire_time IS NULL AND p.create_time IS NOT NULL + AND p.create_time < DATE_SUB(#{now}, INTERVAL 120 MINUTE)) + ) AND NOT EXISTS ( SELECT 1 FROM third_aggregate_pay_session s WHERE s.tenant_id = p.tenant_id AND s.pay_no = p.pay_no AND s.del_flag = '0') diff --git a/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/test/java/com/bsd/pjcloud/thirdparty/job/ThirdAggregateOrderPayJobTest.java b/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/test/java/com/bsd/pjcloud/thirdparty/job/ThirdAggregateOrderPayJobTest.java deleted file mode 100644 index c1160fa8e..000000000 --- a/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/test/java/com/bsd/pjcloud/thirdparty/job/ThirdAggregateOrderPayJobTest.java +++ /dev/null @@ -1,100 +0,0 @@ -package com.bsd.pjcloud.thirdparty.job; - -import com.baomidou.mybatisplus.core.conditions.Wrapper; -import com.bsd.pjcloud.common.data.tenant.TenantContextHolder; -import com.bsd.pjcloud.thirdparty.constant.pay.ThirdAggregatePaySessionStatusEnum; -import com.bsd.pjcloud.thirdparty.entity.pay.ThirdAggregatePaySessionEntity; -import com.bsd.pjcloud.thirdparty.service.ThirdAggregateOrderPayService; -import com.bsd.pjcloud.thirdparty.service.ThirdAggregatePaySessionService; -import com.bsd.pjcloud.thirdparty.service.ThirdPayPollingCoordinator; -import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.extension.ExtendWith; -import org.mockito.InjectMocks; -import org.mockito.Mock; -import org.mockito.junit.jupiter.MockitoExtension; - -import java.time.LocalDateTime; -import java.util.List; - -import static org.mockito.ArgumentMatchers.any; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.when; - -/** - * Thirdparty 聚合订单码历史终态和补偿窗口测试。 - */ -@ExtendWith(MockitoExtension.class) -class ThirdAggregateOrderPayJobTest { - - /** - * 聚合会话服务 Mock，用于返回超过一小时的历史会话并验证终态更新。 - */ - @Mock - private ThirdAggregatePaySessionService aggregatePaySessionService; - - /** - * 聚合支付编排服务 Mock，用于确认历史记录继续交由可信查单路径处理。 - */ - @Mock - private ThirdAggregateOrderPayService aggregateOrderPayService; - - /** - * 自适应轮询协调器 Mock，用于允许本轮补偿扫描。 - */ - @Mock - private ThirdPayPollingCoordinator pollingCoordinator; - - /** - * 被测 Thirdparty 聚合订单码任务。 - */ - @InjectMocks - private ThirdAggregateOrderPayJob job; - - /** - * 清理测试期间由 TenantBroker 设置的线程租户上下文。 - */ - @AfterEach - void tearDown() { - TenantContextHolder.setTenantId(null); - } - - /** - * 验证超过一小时且没有渠道单的会话仍委托业务编排服务，由业务层决定本地终态。 - */ - @Test - void compensationShouldDelegateHistoricalSessionWithoutChannelOrder() { - ThirdAggregatePaySessionEntity session = new ThirdAggregatePaySessionEntity(); - session.setId(201L); - session.setTenantId(66L); - session.setStatus(ThirdAggregatePaySessionStatusEnum.EXPIRED.getType()); - session.setCreateTime(LocalDateTime.now().minusHours(2)); - when(pollingCoordinator.tryStartAggregateCompensationPolling()).thenReturn(true); - when(aggregatePaySessionService.list(any(Wrapper.class))).thenReturn(List.of(session), List.of()); - - job.handleExpiredSessionCompensation(); - - verify(aggregateOrderPayService).processPendingSession(session.getId()); - verify(pollingCoordinator).finishAggregateCompensationPolling(true); - } - - /** - * 验证历史会话已绑定渠道单时不会在 Job 中盲关，而是委托编排服务核对渠道事实。 - */ - @Test - void compensationShouldDelegateHistoricalSessionWithChannelOrder() { - ThirdAggregatePaySessionEntity session = new ThirdAggregatePaySessionEntity(); - session.setId(202L); - session.setTenantId(66L); - session.setPayNo("historical-child-pay-no"); - session.setStatus(ThirdAggregatePaySessionStatusEnum.EXPIRED.getType()); - session.setCreateTime(LocalDateTime.now().minusHours(2)); - when(pollingCoordinator.tryStartAggregateCompensationPolling()).thenReturn(true); - when(aggregatePaySessionService.list(any(Wrapper.class))).thenReturn(List.of(session), List.of()); - - job.handleExpiredSessionCompensation(); - - verify(aggregateOrderPayService).processPendingSession(session.getId()); - verify(pollingCoordinator).finishAggregateCompensationPolling(true); - } -} diff --git a/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/test/java/com/bsd/pjcloud/thirdparty/job/ThirdAppPlusPayJobTest.java b/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/test/java/com/bsd/pjcloud/thirdparty/job/ThirdAppPlusPayJobTest.java deleted file mode 100644 index 0f1d35898..000000000 --- a/pjcloud-thirdparty/pjcloud-thirdparty-biz/src/test/java/com/bsd/pjcloud/thirdparty/job/ThirdAppPlusPayJobTest.java +++ /dev/null @@ -1,200 +0,0 @@ -package com.bsd.pjcloud.thirdparty.job; - -import com.baomidou.mybatisplus.core.conditions.Wrapper; -import com.bsd.pjcloud.common.data.tenant.TenantContextHolder; -import com.bsd.pjcloud.common.core.constant.enums.YesNoEnum; -import com.bsd.pjcloud.thirdparty.constant.fz.ThirdFzOrderStatusEnum; -import com.bsd.pjcloud.thirdparty.constant.pay.ThirdPayOrderStatusEnum; -import com.bsd.pjcloud.thirdparty.entity.pay.ThirdFzOrderEntity; -import com.bsd.pjcloud.thirdparty.entity.pay.ThirdPayOrderEntity; -import com.bsd.pjcloud.thirdparty.service.ThirdAppPlusAliPayService; -import com.bsd.pjcloud.thirdparty.service.ThirdAppPlusService; -import com.bsd.pjcloud.thirdparty.service.ThirdAppPlusWechatFzService; -import com.bsd.pjcloud.thirdparty.service.ThirdFzOrderService; -import com.bsd.pjcloud.thirdparty.service.ThirdPayOrderService; -import com.bsd.pjcloud.thirdparty.service.ThirdPayPollingCoordinator; -import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.Assertions; -import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.extension.ExtendWith; -import org.mockito.InjectMocks; -import org.mockito.Mock; -import org.mockito.junit.jupiter.MockitoExtension; -import org.springframework.test.util.ReflectionTestUtils; - -import java.time.LocalDateTime; -import java.util.List; - -import static org.mockito.ArgumentMatchers.any; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.verifyNoInteractions; -import static org.mockito.Mockito.when; - -/** - * 原生订单码服务端轮询任务的租户恢复和自适应扫描测试。 - */ -@ExtendWith(MockitoExtension.class) -class ThirdAppPlusPayJobTest { - - /** - * 支付订单服务 Mock，用于返回跨租户待处理订单。 - */ - @Mock - private ThirdPayOrderService thirdPayOrderService; - - /** - * 支付编排服务 Mock，用于验证恢复租户后执行平台查单。 - */ - @Mock - private ThirdAppPlusService thirdAppPlusService; - - /** - * 分账订单服务 Mock，用于验证最终分账标记的重启恢复。 - */ - @Mock - private ThirdFzOrderService thirdFzOrderService; - - /** - * 微信分账服务 Mock，补齐被测任务的构造依赖。 - */ - @Mock - private ThirdAppPlusWechatFzService thirdAppPlusWechatFzService; - - /** - * 支付宝直付通服务 Mock，补齐被测任务的构造依赖。 - */ - @Mock - private ThirdAppPlusAliPayService thirdAppPlusAliPayService; - - /** - * 自适应轮询协调器 Mock，用于控制本轮是否允许访问数据库。 - */ - @Mock - private ThirdPayPollingCoordinator pollingCoordinator; - - /** - * 被测原生订单码轮询任务。 - */ - @InjectMocks - private ThirdAppPlusPayJob job; - - /** - * 清理测试期间由 TenantBroker 设置的线程租户上下文。 - */ - @AfterEach - void tearDown() { - TenantContextHolder.setTenantId(null); - } - - /** - * 验证取得扫描资格后按订单租户恢复上下文，并把非空结果反馈给协调器。 - */ - @Test - void handleOrderCodeOrdersShouldRestoreEachOrderTenant() { - ThirdPayOrderEntity order = new ThirdPayOrderEntity(); - order.setId(99L); - order.setTenantId(66L); - order.setStatus(ThirdPayOrderStatusEnum.PAY_UNPAID.getType()); - order.setExpireTime(LocalDateTime.now().plusMinutes(1)); - when(pollingCoordinator.tryStartOrderCodePolling()).thenReturn(true); - when(thirdPayOrderService.list(any(Wrapper.class))).thenReturn(List.of(order)); - when(thirdPayOrderService.getById(order.getId())).thenAnswer(invocation -> { - Assertions.assertEquals(66L, TenantContextHolder.getTenantId()); - return order; - }); - - job.handleOrderCodeOrders(); - - verify(thirdAppPlusService).queryPayResult(order); - verify(pollingCoordinator).finishOrderCodePolling(true); - Assertions.assertNull(TenantContextHolder.getTenantId()); - } - - /** - * 验证空闲退避尚未到期时任务直接返回，不执行任何数据库查询。 - */ - @Test - void handleOrderCodeOrdersShouldSkipDatabaseWhenBackoffIsActive() { - when(pollingCoordinator.tryStartOrderCodePolling()).thenReturn(false); - - job.handleOrderCodeOrders(); - - verifyNoInteractions(thirdPayOrderService, thirdAppPlusService); - } - - /** - * 验证超过一小时的原生订单码只做本地关闭标记，不再调用支付平台查单流程。 - */ - @Test - void closeHistoricalOrderCodeOrdersShouldNotQueryPaymentPlatform() { - ThirdPayOrderEntity order = new ThirdPayOrderEntity(); - order.setId(100L); - order.setTenantId(66L); - order.setStatus(ThirdPayOrderStatusEnum.PAY_UNPAID.getType()); - order.setCreateTime(LocalDateTime.now().minusHours(2)); - order.setExpireTime(LocalDateTime.now().minusHours(2)); - when(thirdPayOrderService.list(any(Wrapper.class))).thenReturn(List.of(order)); - - job.closeHistoricalOrderCodeOrders(); - - verify(thirdPayOrderService).update(any(ThirdPayOrderEntity.class), any(Wrapper.class)); - verifyNoInteractions(thirdAppPlusService); - } - - /** - * 验证 App 待支付订单不受一小时创建窗口限制，并在仍未支付时更新时间轮转到批次队尾。 - */ - @Test - void handleAppPaymentOrdersShouldRecoverHistoricalOrderAndRotatePendingRecord() { - ThirdPayOrderEntity order = new ThirdPayOrderEntity(); - order.setId(102L); - order.setTenantId(66L); - order.setStatus(ThirdPayOrderStatusEnum.PAY_UNPAID.getType()); - order.setCreateTime(LocalDateTime.now().minusHours(2)); - order.setUpdateTime(LocalDateTime.now().minusHours(2)); - when(pollingCoordinator.tryStartAppPaymentPolling()).thenReturn(true); - when(thirdPayOrderService.list(any(Wrapper.class))).thenReturn(List.of(order)); - when(thirdPayOrderService.getById(order.getId())).thenReturn(order); - - job.handleAppPaymentOrders(); - - verify(thirdAppPlusService).queryPayResult(order); - org.mockito.ArgumentCaptor updateCaptor = - org.mockito.ArgumentCaptor.forClass(ThirdPayOrderEntity.class); - verify(thirdPayOrderService).update(updateCaptor.capture(), any(Wrapper.class)); - Assertions.assertNotNull(updateCaptor.getValue().getUpdateTime()); - verify(pollingCoordinator).finishAppPaymentPolling(true); - Assertions.assertNull(TenantContextHolder.getTenantId()); - } - - /** - * 验证服务在分账终态落库后退出，下一轮仍能把最终抽佣金额和成功标记补写到支付订单。 - */ - @Test - void reconcileSuccessfulFzOrdersShouldRecoverInterruptedFinalMarker() { - ThirdPayOrderEntity order = new ThirdPayOrderEntity(); - order.setId(101L); - order.setTenantId(66L); - order.setPayNo("pay-101"); - order.setStatus(ThirdPayOrderStatusEnum.PAY_SUCCESS.getType()); - order.setEnableFz(YesNoEnum.YES.getCode()); - order.setIsFzSuccess(YesNoEnum.NO.getCode()); - ThirdFzOrderEntity fzOrder = new ThirdFzOrderEntity(); - fzOrder.setId(order.getId()); - fzOrder.setPayNo(order.getPayNo()); - fzOrder.setPrice(123); - fzOrder.setStatus(ThirdFzOrderStatusEnum.FZ_SUCCESS.getType()); - when(thirdPayOrderService.list(any(Wrapper.class))).thenReturn(List.of(order)); - when(thirdPayOrderService.getById(order.getId())).thenReturn(order); - when(thirdFzOrderService.getByPayNo(order.getPayNo())).thenReturn(fzOrder); - - ReflectionTestUtils.invokeMethod(job, "reconcileSuccessfulFzOrders"); - - org.mockito.ArgumentCaptor updateCaptor = - org.mockito.ArgumentCaptor.forClass(ThirdPayOrderEntity.class); - verify(thirdPayOrderService).update(updateCaptor.capture(), any(Wrapper.class)); - Assertions.assertEquals(123, updateCaptor.getValue().getFzPrice()); - Assertions.assertEquals(YesNoEnum.YES.getCode(), updateCaptor.getValue().getIsFzSuccess()); - Assertions.assertNull(TenantContextHolder.getTenantId()); - } -} diff --git a/pjcloud-visual/pjcloud-daemon-quartz/src/main/java/com/bsd/pjcloud/daemon/quartz/task/ThirdPushTask.java b/pjcloud-visual/pjcloud-daemon-quartz/src/main/java/com/bsd/pjcloud/daemon/quartz/task/ThirdPushTask.java new file mode 100644 index 000000000..65009e1a9 --- /dev/null +++ b/pjcloud-visual/pjcloud-daemon-quartz/src/main/java/com/bsd/pjcloud/daemon/quartz/task/ThirdPushTask.java @@ -0,0 +1,85 @@ +package com.bsd.pjcloud.daemon.quartz.task; + + +import com.bsd.pjcloud.common.core.util.R; +import com.bsd.pjcloud.daemon.quartz.constants.PjcloudQuartzEnum; +import com.bsd.pjcloud.thirdparty.api.feign.RemoteThirdPushDeliveryService; +import lombok.RequiredArgsConstructor; +import lombok.SneakyThrows; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; + +import java.time.LocalDateTime; + +/** + * App业务消息Push投递任务。sys_job 通过 spring bean 反射调用（job_type=2， + * class_name=thirdPushTask），bean 内走 Feign 调 thirdparty 内部端点执行， + * 投递任务由条件更新领取，Quartz 集群重复触发安全。 + */ +@Slf4j +@Component("thirdPushTask") +@RequiredArgsConstructor +public class ThirdPushTask { + + private final RemoteThirdPushDeliveryService remoteThirdPushDeliveryService; + + /** + * 无参重载：sys_job 未配置 method_params_value 时，SpringBeanTaskInvok 按无参方法反射调用。 + */ + public String processDue() { + return processDue(null); + } + + /** + * 扫描到期Push投递任务，每 10 秒触发一次。 + */ + @SneakyThrows + public String processDue(String para) { + log.info("App消息Push投递扫描开始:{},输入参数:{}", LocalDateTime.now(), para); + try { + R result = remoteThirdPushDeliveryService.processDueDeliveries(); + if (result.isOk()) { + if (result.getData() != null && result.getData() > 0) { + log.info("App消息Push投递扫描完成,本轮处理:{}条", result.getData()); + } + } else { + log.error("App消息Push投递扫描失败,code:{},msg:{}", result.getCode(), result.getMsg()); + return PjcloudQuartzEnum.JOB_LOG_STATUS_FAIL.getType(); + } + } catch (Exception e) { + log.error("App消息Push投递扫描错误error:{},时间:{}", e.getMessage(), LocalDateTime.now(), e); + return PjcloudQuartzEnum.JOB_LOG_STATUS_FAIL.getType(); + } + return PjcloudQuartzEnum.JOB_LOG_STATUS_SUCCESS.getType(); + } + + /** + * 无参重载：sys_job 未配置 method_params_value 时，SpringBeanTaskInvok 按无参方法反射调用。 + */ + public String cleanup() { + return cleanup(null); + } + + /** + * 清理超过保留期的Push消息、终态投递和供应商尝试日志，每日 02:10 触发一次。 + */ + @SneakyThrows + public String cleanup(String para) { + log.info("App消息Push过期记录清理开始:{},输入参数:{}", LocalDateTime.now(), para); + try { + R result = remoteThirdPushDeliveryService.cleanupExpiredRecords(); + if (result.isOk()) { + if (result.getData() != null && result.getData() > 0) { + log.info("App消息Push过期记录清理完成,本轮清理:{}条", result.getData()); + } + } else { + log.error("App消息Push过期记录清理失败,code:{},msg:{}", result.getCode(), result.getMsg()); + return PjcloudQuartzEnum.JOB_LOG_STATUS_FAIL.getType(); + } + } catch (Exception e) { + log.error("App消息Push过期记录清理错误error:{},时间:{}", e.getMessage(), LocalDateTime.now(), e); + return PjcloudQuartzEnum.JOB_LOG_STATUS_FAIL.getType(); + } + return PjcloudQuartzEnum.JOB_LOG_STATUS_SUCCESS.getType(); + } +} diff --git a/pjcloud-visual/pjcloud-daemon-quartz/src/main/java/com/bsd/pjcloud/daemon/quartz/task/WalletReserveTask.java b/pjcloud-visual/pjcloud-daemon-quartz/src/main/java/com/bsd/pjcloud/daemon/quartz/task/WalletReserveTask.java index aa4657434..b56e32f78 100644 --- a/pjcloud-visual/pjcloud-daemon-quartz/src/main/java/com/bsd/pjcloud/daemon/quartz/task/WalletReserveTask.java +++ b/pjcloud-visual/pjcloud-daemon-quartz/src/main/java/com/bsd/pjcloud/daemon/quartz/task/WalletReserveTask.java @@ -23,6 +23,13 @@ public class WalletReserveTask { private final RemoteWalletRechargeApplyService remoteWalletRechargeApplyService; + /** + * 无参重载：sys_job 未配置 method_params_value 时，SpringBeanTaskInvok 按无参方法反射调用。 + */ + public String executeDueApplies() { + return executeDueApplies(null); + } + /** * 扫描并执行到期的预约充值/扣款申请，每 10 分钟触发一次。 */ @@ -44,6 +51,13 @@ public class WalletReserveTask { return PjcloudQuartzEnum.JOB_LOG_STATUS_SUCCESS.getType(); } + /** + * 无参重载：sys_job 未配置 method_params_value 时，SpringBeanTaskInvok 按无参方法反射调用。 + */ + public String reconcile() { + return reconcile(null); + } + /** * 复核近期预约申请明细与钱包流水一致性，每 10 分钟触发一次，与执行任务错开 5 分钟。 */ diff --git a/pjcloud-visual/pjcloud-daemon-quartz/src/main/resources/application.yml b/pjcloud-visual/pjcloud-daemon-quartz/src/main/resources/application.yml index 9364a3abe..23d5ce798 100644 --- a/pjcloud-visual/pjcloud-daemon-quartz/src/main/resources/application.yml +++ b/pjcloud-visual/pjcloud-daemon-quartz/src/main/resources/application.yml @@ -4,6 +4,12 @@ server: quartz: enabled: true +# Quartz 集群模式下每次触发会对 QRTZ_ 内部表执行十几条状态流转 SQL， +# 属于高频框架噪音，跳过 DruidSqlLogFilter 的 SQL 日志打印 +pjcloud: + mybatis: + skip-table: [qrtz_] + spring: application: name: @artifactId@ diff --git a/pjcloud-visual/pjcloud-daemon-quartz/src/main/resources/quartz-config.yml b/pjcloud-visual/pjcloud-daemon-quartz/src/main/resources/quartz-config.yml index 6a25742bc..db6cf554a 100644 --- a/pjcloud-visual/pjcloud-daemon-quartz/src/main/resources/quartz-config.yml +++ b/pjcloud-visual/pjcloud-daemon-quartz/src/main/resources/quartz-config.yml @@ -14,7 +14,7 @@ spring: isClustered: true # 支付任务只由Quartz集群触发，生产实例必须共享同一套QRTZ表 clusterCheckinInterval: 10000 # 集群心跳，节点失联后由存活实例接管 misfireThreshold: 60000 # 允许一分钟短暂抖动，恢复后按misfire策略继续触发 - acquireTriggersWithinLock: true # 多节点批量获取Trigger时在数据库锁内完成 + acquireTriggersWithinLock: false # 触发器扫描不在数据库锁内执行，缩短QRTZ_LOCKS持锁时间；多节点重复竞争由"仅WAITING可置ACQUIRED"的条件更新裁决，不会重复触发 useProperties: false threadPool: class: org.quartz.simpl.SimpleThreadPool # 线程池 diff --git a/pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/mapper/recharge/WalletRechargeApplyMapper.java b/pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/mapper/recharge/WalletRechargeApplyMapper.java index 4651509a7..54a7e6e36 100644 --- a/pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/mapper/recharge/WalletRechargeApplyMapper.java +++ b/pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/mapper/recharge/WalletRechargeApplyMapper.java @@ -57,27 +57,30 @@ public interface WalletRechargeApplyMapper extends PjcloudBaseMapper listDueReserveApplies(@Param("limit") int limit); + List listDueReserveApplies(@Param("now") LocalDateTime now, @Param("limit") int limit); /** * 跨租户筛选待对账的预约单：近期已执行/重试耗尽，或长期停留在未完成状态（卡死检测）。 diff --git a/pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/service/impl/recharge/WalletRechargeApplyServiceImpl.java b/pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/service/impl/recharge/WalletRechargeApplyServiceImpl.java index acdc12445..991ed25ca 100644 --- a/pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/service/impl/recharge/WalletRechargeApplyServiceImpl.java +++ b/pjcloud-wallet/pjcloud-wallet-biz/src/main/java/com/bsd/pjcloud/wallet/service/impl/recharge/WalletRechargeApplyServiceImpl.java @@ -22,6 +22,7 @@ import com.bsd.pjcloud.common.core.util.CollectionUtils; import com.bsd.pjcloud.common.core.util.R; import com.bsd.pjcloud.common.data.datascope.DataScope; import com.bsd.pjcloud.common.data.tenant.TenantContextHolder; +import com.bsd.pjcloud.common.security.component.PermissionService; import com.bsd.pjcloud.common.security.service.PjcloudUser; import com.bsd.pjcloud.common.security.util.SecurityUtils; import com.bsd.pjcloud.flow.task.api.feign.RemoteFlowEngineService; @@ -82,11 +83,15 @@ public class WalletRechargeApplyServiceImpl extends ServiceImpl 0) { log.warn("|{}|复活执行中卡死的预约单,count:{}", TAG, revived); } - List dueList = applyMapper.listDueReserveApplies(SCAN_LIMIT); + List dueList = applyMapper.listDueReserveApplies(now, SCAN_LIMIT); int executed = 0; for (WalletRechargeApplyEntity due : dueList) { if (due.getTenantId() == null) {