fix: 优化生产环境收益稳定发放与队列可靠性，支持自动对账补偿与幂等防重

addons/shopro/console/PartnerSettle.php

@@ -7,6 +7,7 @@ use think\console\Output;
 use think\console\input\Option;
 use think\console\Command;
 use addons\shopro\service\partner\DailySettleService;
+use addons\shopro\service\partner\PartnerCompensateService;
 
 class PartnerSettle extends Command
 {
@@ -16,7 +17,9 @@ class PartnerSettle extends Command
             ->addOption('date', 'd', Option::VALUE_OPTIONAL, '结算日期 Y-m-d，指定日期时执行传统日结')
             ->addOption('rerun', 'r', Option::VALUE_NONE, '删除当天日结后重算')
             ->addOption('daily', null, Option::VALUE_NONE, '强制执行昨天标准日结')
-            ->setDescription('收益中心结算（默认1分钟dmepa增量结算与兑换券释放，亦支持指定日期日结）');
+            ->addOption('compensate', 'c', Option::VALUE_NONE, '执行未结算订单自动对账与补偿发放')
+            ->addOption('check', null, Option::VALUE_NONE, '收益中心与队列运行健康度体检诊断')
+            ->setDescription('收益中心结算（默认1分钟dmepa增量结算与兑换券释放，支持指定日期日结及未结算补偿）');
     }
 
     protected function execute(Input $input, Output $output)
@@ -24,12 +27,36 @@ class PartnerSettle extends Command
         $date = $input->getOption('date');
         $rerun = (bool) $input->getOption('rerun');
         $isDaily = (bool) $input->getOption('daily');
+        $isCompensate = (bool) $input->getOption('compensate');
+        $isCheck = (bool) $input->getOption('check');
 
-        // 指定日期或强制日结模式
+        // 1. 系统健康度与漏单诊断体检模式
+        if ($isCheck) {
+            $output->writeln('[' . date('Y-m-d H:i:s') . '] partner health check start...');
+            $health = (new PartnerCompensateService())->checkHealth();
+            $output->writeln('partner health check done: ' . json_encode($health, JSON_UNESCAPED_UNICODE | JSON_PRETTY_PRINT));
+            return 0;
+        }
+
+        // 2. 独立对账补偿模式
+        if ($isCompensate) {
+            $output->writeln('[' . date('Y-m-d H:i:s') . '] partner compensation start...');
+            $compensateReport = (new PartnerCompensateService())->compensateAll(200);
+            $output->writeln('partner compensation done: ' . json_encode($compensateReport, JSON_UNESCAPED_UNICODE | JSON_PRETTY_PRINT));
+            return 0;
+        }
+
+        // 3. 指定日期或强制日结模式（前置自动执行补偿，确保无漏单）
         if ($date || $isDaily || $rerun) {
             $settleDate = $date ?: date('Y-m-d', strtotime('-1 day'));
             $output->writeln('partner daily settle start ' . $settleDate . ($rerun ? ' rerun' : ''));
             try {
+                // 日结前置：扫描并补偿未结算即时奖订单，避免影响大盘 GMV 统计与分红池
+                $compensateReport = (new PartnerCompensateService())->compensateAll(100);
+                if (!empty($compensateReport['success_count'])) {
+                    $output->writeln("  [✓] 日结前置补偿已完成: 成功补发 {$compensateReport['success_count']} 笔订单即时奖");
+                }
+
                 $report = (new DailySettleService())->run($date, $rerun);
                 $output->writeln('partner daily settle done: ' . json_encode($report ? $report->toArray() : [], JSON_UNESCAPED_UNICODE));
                 return 0;
@@ -39,7 +66,7 @@ class PartnerSettle extends Command
             }
         }
 
-        // 默认执行 1 分钟dmepa增量结算与兑换券释放
+        // 4. 默认执行 1 分钟dmepa增量结算与兑换券释放
         $output->writeln('[' . date('Y-m-d H:i:s') . '] partner minute settle start');
         try {
             $result = (new DailySettleService())->runMinute();

addons/shopro/job/BaseJob.php

@@ -13,13 +13,6 @@ class BaseJob
 
     public function failed($data)
     {
-        // 失败队列，这里不用添加了， 后续程序会自动添加，这里可以发送邮件或者通知
-
-        // 记录日志
-        // \think\Db::name('shopro_failed_job')->insert([
-        //     'data' => json_encode($data),
-        //     'createtime' => time(),
-        //     'updatetime' => time()
-        // ]);
+        Log::record('[QUEUE FAILED JOB] ' . (is_string($data) ? $data : json_encode($data, JSON_UNESCAPED_UNICODE)), 'error');
     }
 }

addons/shopro/job/OrderPaid.php

@@ -23,39 +23,51 @@ class OrderPaid extends BaseJob
      */
     public function paid(Job $job, $data)
     {
+        $orderData = $data['order'] ?? null;
+        $userData = $data['user'] ?? null;
+        $orderId = is_array($orderData) ? ($orderData['id'] ?? 0) : ($orderData->id ?? 0);
+        $userId = is_array($userData) ? ($userData['id'] ?? 0) : ($userData->id ?? 0);
+
         try {
-            $order = $data['order'];
-            $user = $data['user'];
+            if ($orderId) {
+                Db::transaction(function () use ($orderId, $userId) {
+                    $order = Order::with('items')->where('id', $orderId)->lock(true)->find();
+                    if (!$order) {
+                        return;
+                    }
+                    $user = User::get($userId ?: $order->user_id);
+                    $ext = is_array($order->ext) ? $order->ext : (json_decode($order->ext ?? '', true) ?: []);
 
-            $order = Order::with('items')->where('id', $order['id'])->find();
-            $user = User::get($user['id']);
+                    // 幂等防重：检查是否已经执行过减库存与活动支付处理
+                    if (empty($ext['paid_forward_processed'])) {
+                        // 订单减库存
+                        $stockSale = new StockSale();
+                        $stockSale->forwardStockSale($order);
 
-            // 数据库删订单的问题常见，这里被删的订单直接把队列移除
-            if ($order) {
-                Db::transaction(function () use ($order, $user, $data) {
-                    // 订单减库存
-                    $stockSale = new StockSale();
-                    $stockSale->forwardStockSale($order);
-
-                    // 处理发票审核改为等待开具
-                    if ($order->invoice_status == 1) {
-                        $invoice = OrderInvoice::where('order_id', $order->id)->find();
-                        if ($invoice) {
-                            $invoice->status = 'waiting';
-                            $invoice->save();
+                        // 处理发票审核改为等待开具
+                        if ($order->invoice_status == 1) {
+                            $invoice = OrderInvoice::where('order_id', $order->id)->find();
+                            if ($invoice) {
+                                $invoice->status = 'waiting';
+                                $invoice->save();
+                            }
                         }
+
+                        // 处理活动，加入拼团，完成拼团，添加赠品记录等
+                        ActivityFacade::buyOk($order, $user);
+
+                        // 将订单参与活动信息改为已支付
+                        $orderOper = new OrderOper();
+                        $orderOper->activityOrderPaid($order);
+
+                        $ext['paid_forward_processed'] = time();
+                        $order->ext = $ext;
+                        $order->save();
                     }
 
-                    // 处理活动，加入拼团，完成拼团，添加赠品记录等
-                    ActivityFacade::buyOk($order, $user);
-
-                    // 将订单参与活动信息改为已支付
-                    $orderOper = new OrderOper();
-                    $orderOper->activityOrderPaid($order);
-                    
-                    // 触发订单支付完成事件
-                    $data = ['order' => $order, 'user' => $user];
-                    \think\Hook::listen('order_paid_after', $data);
+                    // 触发订单支付完成事件（下游结算服务内置原生幂等）
+                    $eventData = ['order' => $order, 'user' => $user];
+                    \think\Hook::listen('order_paid_after', $eventData);
                 });
             }
 
@@ -65,9 +77,11 @@ class OrderPaid extends BaseJob
             $data = $e->getResponse()->getData();
             $message = $data ? ($data['msg'] ?? '') : $e->getMessage();
             format_log_error($e, 'OrderPaid.paid.HttpResponseException', $message);
-        } catch (\Exception $e) {
-            // 队列执行失败
-            format_log_error($e, 'OrderPaid.paid');
+            $this->handleFailure($job, $orderId, 'paid', $message);
+        } catch (\Throwable $e) {
+            $message = $e->getMessage();
+            format_log_error($e, 'OrderPaid.paid', $message);
+            $this->handleFailure($job, $orderId, 'paid', $message);
         }
     }
 
@@ -76,26 +90,37 @@ class OrderPaid extends BaseJob
      */
     public function offline(Job $job, $data)
     {
+        $orderData = $data['order'] ?? null;
+        $userData = $data['user'] ?? null;
+        $orderId = is_array($orderData) ? ($orderData['id'] ?? 0) : ($orderData->id ?? 0);
+        $userId = is_array($userData) ? ($userData['id'] ?? 0) : ($userData->id ?? 0);
+
         try {
-            $order = $data['order'];
-            $user = $data['user'];
+            if ($orderId) {
+                Db::transaction(function () use ($orderId, $userId) {
+                    $order = Order::with('items')->where('id', $orderId)->lock(true)->find();
+                    if (!$order) {
+                        return;
+                    }
+                    $user = User::get($userId ?: $order->user_id);
+                    $ext = is_array($order->ext) ? $order->ext : (json_decode($order->ext ?? '', true) ?: []);
 
-            $order = Order::with('items')->where('id', $order['id'])->find();
-            $user = User::get($user['id']);
+                    if (empty($ext['offline_forward_processed'])) {
+                        // 订单减库存
+                        $stockSale = new StockSale();
+                        $stockSale->forwardStockSale($order);
 
-            // 数据库删订单的问题常见，这里被删的订单直接把队列移除
-            if ($order) {
-                Db::transaction(function () use ($order, $user, $data) {
-                    // 订单减库存
-                    $stockSale = new StockSale();
-                    $stockSale->forwardStockSale($order);
+                        // 处理活动，加入拼团，完成拼团，添加赠品记录等
+                        ActivityFacade::buyOk($order, $user);
 
-                    // 处理活动，加入拼团，完成拼团，添加赠品记录等
-                    ActivityFacade::buyOk($order, $user);
+                        $ext['offline_forward_processed'] = time();
+                        $order->ext = $ext;
+                        $order->save();
+                    }
 
                     // 触发订单选择线下支付(货到付款)完成事件
-                    $data = ['order' => $order, 'user' => $user];
-                    \think\Hook::listen('order_offline_after', $data);
+                    $eventData = ['order' => $order, 'user' => $user];
+                    \think\Hook::listen('order_offline_after', $eventData);
                 });
             }
 
@@ -105,9 +130,35 @@ class OrderPaid extends BaseJob
             $data = $e->getResponse()->getData();
             $message = $data ? ($data['msg'] ?? '') : $e->getMessage();
             format_log_error($e, 'OrderPaid.offline.HttpResponseException', $message);
-        } catch (\Exception $e) {
-            // 队列执行失败
-            format_log_error($e, 'OrderPaid.offline');
+            $this->handleFailure($job, $orderId, 'offline', $message);
+        } catch (\Throwable $e) {
+            $message = $e->getMessage();
+            format_log_error($e, 'OrderPaid.offline', $message);
+            $this->handleFailure($job, $orderId, 'offline', $message);
+        }
+    }
+
+    /**
+     * 生产级失败与阶梯重试处理
+     */
+    protected function handleFailure(Job $job, $orderId, $action, $errorMessage)
+    {
+        $attempts = $job->attempts();
+        $maxAttempts = 5;
+
+        if ($attempts >= $maxAttempts) {
+            \think\Log::record("[OrderPaid CRITICAL] 订单 ID:{$orderId} ({$action}) 处理连续失败 {$attempts} 次已达上限: {$errorMessage}", 'error');
+            $this->failed([
+                'job' => "OrderPaid@{$action}",
+                'order_id' => $orderId,
+                'error' => $errorMessage,
+                'attempts' => $attempts,
+                'failed_time' => time(),
+            ]);
+            $job->delete();
+        } else {
+            $delay = min(60, $attempts * 5);
+            $job->release($delay);
         }
     }
 }

addons/shopro/job/PartnerSettle.php

@@ -23,18 +23,53 @@ class PartnerSettle extends BaseJob
             $data = $e->getResponse()->getData();
             $message = $data ? ($data['msg'] ?? '') : $e->getMessage();
             format_log_error($e, 'PartnerSettle.HttpResponseException', $message);
-            if ($job->attempts() >= 3) {
-                $job->delete();
-            } else {
-                $job->release(5);
-            }
-        } catch (\Exception $e) {
-            format_log_error($e, 'PartnerSettle');
-            if ($job->attempts() >= 3) {
-                $job->delete();
-            } else {
-                $job->release(5);
+            $this->handleFailure($job, $orderId, $message);
+        } catch (\Throwable $e) {
+            $message = $e->getMessage();
+            format_log_error($e, 'PartnerSettle', $message);
+            $this->handleFailure($job, $orderId, $message);
+        }
+    }
+
+    /**
+     * 生产级失败与阶梯重试处理，达到上限时持久化标记到订单 ext 并写入告警
+     */
+    protected function handleFailure(Job $job, $orderId, $errorMessage)
+    {
+        $attempts = $job->attempts();
+        $maxAttempts = 5;
+
+        if ($attempts >= $maxAttempts) {
+            \think\Log::record("[PartnerSettle CRITICAL] 订单 ID:{$orderId} 收益结算连续失败 {$attempts} 次已达上限: {$errorMessage}", 'error');
+            if ($orderId) {
+                try {
+                    $freshOrder = Db::name('shopro_order')->where('id', $orderId)->find();
+                    if ($freshOrder) {
+                        $ext = is_array($freshOrder['ext']) ? $freshOrder['ext'] : (json_decode($freshOrder['ext'] ?? '', true) ?: []);
+                        $ext['reward_settle_failed'] = 1;
+                        $ext['reward_settle_error'] = mb_substr($errorMessage, 0, 500);
+                        $ext['reward_settle_failed_time'] = time();
+                        $ext['reward_settle_attempts'] = $attempts;
+                        Db::name('shopro_order')->where('id', $orderId)->update([
+                            'ext' => json_encode($ext, JSON_UNESCAPED_UNICODE),
+                        ]);
+                    }
+                } catch (\Throwable $ex) {
+                    \think\Log::record("[PartnerSettle] 记录订单 {$orderId} 结算失败标记异常: " . $ex->getMessage(), 'error');
+                }
             }
+            $this->failed([
+                'job' => 'PartnerSettle@order',
+                'order_id' => $orderId,
+                'error' => $errorMessage,
+                'attempts' => $attempts,
+                'failed_time' => time(),
+            ]);
+            $job->delete();
+        } else {
+            // 阶梯式延迟重试：5s, 10s, 15s, 20s
+            $delay = min(60, $attempts * 5);
+            $job->release($delay);
         }
     }
 }

addons/shopro/service/partner/PartnerCompensateService.php

@@ -0,0 +1,197 @@
+<?php
+
+namespace addons\shopro\service\partner;
+
+use think\Db;
+use app\admin\model\shopro\order\Order as OrderModel;
+use app\admin\model\shopro\partner\Reward as RewardModel;
+use app\admin\model\shopro\commission\CommissionGoods;
+
+/**
+ * 恒联数通 · 订单收益自动对账与补偿服务 (Production-Grade Auto-Compensation)
+ * 
+ * 职责：
+ * 1. 扫描线上已支付但由于队列崩溃、丢单、重试上限或Redis故障未完成收益结算的合格订单；
+ * 2. 具备绝对幂等性安全补偿发放，杜绝漏发与重复发放；
+ * 3. 提供收益中心健康度诊断与对账审计报表。
+ */
+class PartnerCompensateService
+{
+    /**
+     * 查询未结算收益的合格订单
+     *
+     * @param int $limit 查询数量上限
+     * @param int|null $startTime 起始支付时间戳
+     * @param int|null $endTime 结束支付时间戳
+     * @return \think\Collection
+     */
+    public function findUnsettledOrders($limit = 100, $startTime = null, $endTime = null)
+    {
+        $rateService = new RateService();
+        $query = $rateService->qualifiedOrderQuery($startTime, $endTime);
+
+        $orderTable = (new OrderModel())->getQuery()->getTable();
+        $rewardTable = (new RewardModel())->getQuery()->getTable();
+
+        // 核心排除：在 shopro_partner_reward 表中已存在该订单的有效流水或标记
+        $query->whereNotExists(function ($q) use ($rewardTable) {
+            $q->table($rewardTable)->alias('pr')
+                ->whereRaw('pr.order_id = o.id')
+                ->where('pr.order_id', '>', 0);
+        });
+
+        return $query->order('o.id', 'asc')->limit($limit)->select();
+    }
+
+    /**
+     * 查询结算标记失败的异常订单
+     *
+     * @param int $limit
+     * @return \think\Collection
+     */
+    public function findFailedMarkedOrders($limit = 50)
+    {
+        return OrderModel::alias('o')
+            ->where('o.type', '<>', 'score')
+            ->where('o.pay_fee', '>', 0)
+            ->whereIn('o.status', [OrderModel::STATUS_PAID, OrderModel::STATUS_COMPLETED])
+            ->where('o.ext', 'like', '%"reward_settle_failed":1%')
+            ->order('o.id', 'asc')
+            ->limit($limit)
+            ->select();
+    }
+
+    /**
+     * 执行自动对账与收益安全补偿
+     *
+     * @param int $limit 单次最大补偿订单数
+     * @param bool $dryRun 是否仅试算对账不写入
+     * @return array 对账与补偿执行报表
+     */
+    public function compensateAll($limit = 100, $dryRun = false)
+    {
+        $unsettled = $this->findUnsettledOrders($limit);
+        $rewardService = new PartnerRewardService();
+
+        $report = [
+            'total_found' => count($unsettled),
+            'dry_run' => $dryRun,
+            'success_count' => 0,
+            'failed_count' => 0,
+            'skipped_count' => 0,
+            'details' => [],
+            'time' => date('Y-m-d H:i:s'),
+        ];
+
+        foreach ($unsettled as $order) {
+            $orderId = intval($order->id);
+            $orderSn = $order->order_sn;
+            $payFee = (float) $order->pay_fee;
+
+            if ($dryRun) {
+                $report['details'][] = [
+                    'order_id' => $orderId,
+                    'order_sn' => $orderSn,
+                    'pay_fee' => $payFee,
+                    'status' => 'preview',
+                    'message' => '待补偿',
+                ];
+                continue;
+            }
+
+            try {
+                $result = Db::transaction(function () use ($rewardService, $orderId) {
+                    return $rewardService->settleOrder($orderId);
+                });
+
+                if ($result) {
+                    $report['success_count']++;
+                    $rewards = RewardModel::where('order_id', $orderId)->where('order_id', '>', 0)->select();
+                    $cashTotal = 0;
+                    $types = [];
+                    foreach ($rewards as $r) {
+                        if ((float) $r->amount > 0 && $r->status == 1) {
+                            $cashTotal = bcadd((string)$cashTotal, (string)$r->amount, 2);
+                        }
+                        $types[] = $r->reward_type;
+                    }
+
+                    $report['details'][] = [
+                        'order_id' => $orderId,
+                        'order_sn' => $orderSn,
+                        'pay_fee' => $payFee,
+                        'status' => 'success',
+                        'cash_distributed' => $cashTotal,
+                        'reward_types' => array_values(array_unique($types)),
+                    ];
+                } else {
+                    $report['skipped_count']++;
+                    $report['details'][] = [
+                        'order_id' => $orderId,
+                        'order_sn' => $orderSn,
+                        'pay_fee' => $payFee,
+                        'status' => 'skipped',
+                        'message' => '订单不满足计奖条件或已结',
+                    ];
+                }
+            } catch (\Throwable $e) {
+                $report['failed_count']++;
+                $errorMsg = $e->getMessage();
+                \think\Log::record("[PartnerCompensate] 订单 ID:{$orderId} 补偿执行异常: {$errorMsg}", 'error');
+                $report['details'][] = [
+                    'order_id' => $orderId,
+                    'order_sn' => $orderSn,
+                    'pay_fee' => $payFee,
+                    'status' => 'failed',
+                    'error' => $errorMsg,
+                ];
+            }
+        }
+
+        return $report;
+    }
+
+    /**
+     * 收益中心与队列运行健康度全景诊断
+     *
+     * @return array
+     */
+    public function checkHealth()
+    {
+        $queueConfig = \think\Config::get('queue');
+        $expire = $queueConfig['expire'] ?? null;
+        $selectDb = $queueConfig['select'] ?? null;
+
+        $unsettledCount = count($this->findUnsettledOrders(500));
+        $failedMarkCount = count($this->findFailedMarkedOrders(500));
+
+        $warnings = [];
+        if ($expire === 0) {
+            $warnings[] = "致命配置警告：queue.expire 为 0，会导致队列任务被反复超时重放与并发重复消费！";
+        } elseif ($expire === null || $expire < 60) {
+            $warnings[] = "配置建议：queue.expire 应设置为 >= 60 秒（推荐 180 秒）。";
+        }
+
+        if ($unsettledCount > 0) {
+            $warnings[] = "发现 {$unsettledCount} 笔已支付但未进行收益结算的订单，建议执行补偿操作。";
+        }
+
+        if ($failedMarkCount > 0) {
+            $warnings[] = "发现 {$failedMarkCount} 笔标记为结算重试失败的订单，需重点关注。";
+        }
+
+        return [
+            'status' => empty($warnings) ? 'healthy' : ($expire === 0 ? 'critical' : 'warning'),
+            'queue' => [
+                'connector' => $queueConfig['connector'] ?? 'unknown',
+                'expire_seconds' => $expire,
+                'redis_db' => $selectDb,
+                'default_queue' => $queueConfig['default'] ?? 'default',
+            ],
+            'unsettled_orders_count' => $unsettledCount,
+            'failed_marked_orders_count' => $failedMarkCount,
+            'warnings' => $warnings,
+            'check_time' => date('Y-m-d H:i:s'),
+        ];
+    }
+}

addons/shopro/service/partner/PartnerRewardService.php

@@ -35,6 +35,10 @@ class PartnerRewardService
             }
 
             $buyerId = intval($order['user_id']);
+            // 确保双向保活已更新（即使钩子丢失或补偿场景也能准确刷新）
+            $paidTime = !empty($order['paid_time']) ? intval($order['paid_time']) : time();
+            $this->touchOrderTimes($buyerId, $paidTime);
+
             $active = new ActiveService();
             $chainService = new ChainService();
             $parentChain = $chainService->collect($buyerId, false);
@@ -54,6 +58,14 @@ class PartnerRewardService
 
             (new VoucherService())->grant($order, $buyerId, $orderRules);
 
+            // 清理可能存在的历史重试失败标记
+            $ext = is_array($order->ext) ? $order->ext : (json_decode($order->ext ?? '', true) ?: []);
+            if (!empty($ext['reward_settle_failed'])) {
+                unset($ext['reward_settle_failed'], $ext['reward_settle_error'], $ext['reward_settle_failed_time'], $ext['reward_settle_attempts']);
+                $order->ext = $ext;
+                $order->save();
+            }
+
             return true;
         });
     }
@@ -275,14 +287,21 @@ class PartnerRewardService
         try {
             $row = RewardModel::create($payload);
             self::$lastWriteCreated = true;
+            if ($row) {
+                $row->_isNewCreated = true;
+            }
             return $row;
         } catch (\Exception $e) {
             $msg = $e->getMessage();
             if (stripos($msg, 'Duplicate') !== false || stripos($msg, '1062') !== false) {
-                return RewardModel::where('order_id', $payload['order_id'])
+                $existing = RewardModel::where('order_id', $payload['order_id'])
                     ->where('user_id', $payload['user_id'])
                     ->where('reward_type', $payload['reward_type'])
                     ->find();
+                if ($existing) {
+                    $existing->_isNewCreated = false;
+                }
+                return $existing;
             }
             throw $e;
         }
@@ -325,7 +344,8 @@ class PartnerRewardService
             'ext' => $ext,
             'status' => 1,
         ]);
-        if ($row && self::$lastWriteCreated && (float) $amount > 0 && $userId > 0 && $row->status == 1 && in_array($type, PartnerConfig::instantTypes())) {
+        $isNew = ($row && !empty($row->_isNewCreated)) || self::$lastWriteCreated;
+        if ($row && $isNew && (float) $amount > 0 && $userId > 0 && $row->status == 1 && in_array($type, PartnerConfig::instantTypes())) {
             Wallet::change($userId, $walletType, $amount, $type, [
                 'partner_reward_id' => $row->id,
                 'order_id' => $orderId,