更新

server/app/common/service/qywx/QywxPromotionMemberSchedulerService.php

@@ -6,7 +6,7 @@ namespace app\common\service\qywx;
 
 use think\facade\Db;
 
-/** 根据实际获客回调记账，并为同一个官方链接规划下一名承接成员。 */
+/** 根据实际获客回调记账，并维护同一个官方链接的可用成员范围。 */
 class QywxPromotionMemberSchedulerService
 {
     public static function poolIdFromState(string $state): int
@@ -54,7 +54,7 @@ class QywxPromotionMemberSchedulerService
     }
 
     /**
-     * 管理端变更成员开关、权重或上限后，确保当前远端成员仍可用；必要时立即排队切换。
+     * 管理端变更成员开关、上限或有效期后，重新计算企业微信官方多人路由范围。
      *
      * @return array{pool_id:int,next_member_id:int,queued:bool,blocked:bool}
      */
@@ -66,55 +66,30 @@ class QywxPromotionMemberSchedulerService
             $linkId = self::promotionLinkId($poolId);
             $sync = self::lockedSyncRow($poolId);
             $members = self::lockedMembers($poolId);
+            $syncStatus = (int) ($sync['status'] ?? 0);
+            $syncError = (string) ($sync['last_error'] ?? '');
+            if ($syncStatus === 5 || ($syncStatus === 4 && str_starts_with($syncError, '企业微信官方获客链接删除失败'))) {
+                return ['pool_id' => $poolId, 'next_member_id' => 0, 'queued' => false, 'blocked' => true];
+            }
             if ($linkId <= 0 || $members === []) {
                 return ['pool_id' => $poolId, 'next_member_id' => 0, 'queued' => false, 'blocked' => true];
             }
 
-            $desiredId = (int) ($sync['desired_member_id'] ?? 0);
-            $desiredEligible = false;
-            foreach ($members as &$member) {
-                if ((string) ($member['today_date'] ?? '') !== $today) {
-                    $member['today_date'] = $today;
-                    $member['today_count'] = 0;
-                }
-                if ((int) $member['id'] === $desiredId) {
-                    $desiredEligible = QywxPromotionWeightedRandom::eligible($member, $now);
-                }
-            }
-            unset($member);
-            self::persistMemberCursors($members, $now);
-            if ($desiredEligible) {
-                if ((int) ($sync['status'] ?? 0) === 4) {
-                    Db::name('qywx_promotion_range_sync')->where('pool_id', $poolId)->update([
-                        'status' => 0,
-                        'next_retry' => 0,
-                        'last_error' => '',
-                        'update_time' => $now,
-                    ]);
-                }
-                return ['pool_id' => $poolId, 'next_member_id' => $desiredId, 'queued' => false, 'blocked' => false];
-            }
-
-            return self::selectAndQueueLocked($poolId, $linkId, $sync, $members, $today, $now);
+            return self::queueEligibleRangeLocked($poolId, $linkId, $sync, $members, $today, $now);
         });
     }
 
-    public static function initialisePool(int $poolId, int $promotionLinkId, string $selectedUserId): int
+    public static function initialisePool(int $poolId, int $promotionLinkId): int
     {
-        $memberId = (int) (Db::name('qywx_promotion_pool_member')
-            ->where('pool_id', $poolId)
-            ->where('userid', $selectedUserId)
-            ->whereNull('delete_time')
-            ->value('id') ?? 0);
-        if ($memberId <= 0 || $promotionLinkId <= 0) {
+        if ($poolId <= 0 || $promotionLinkId <= 0) {
             return 0;
         }
         $now = time();
         $existing = Db::name('qywx_promotion_range_sync')->where('pool_id', $poolId)->find();
         $data = [
             'promotion_link_id' => $promotionLinkId,
-            'desired_member_id' => $memberId,
-            'applied_member_id' => $memberId,
+            'desired_member_id' => 0,
+            'applied_member_id' => 0,
             'status' => 0,
             'attempts' => 0,
             'next_retry' => 0,
@@ -133,7 +108,55 @@ class QywxPromotionMemberSchedulerService
             Db::name('qywx_promotion_range_sync')->insert($data);
         }
 
-        return $memberId;
+        return $promotionLinkId;
+    }
+
+    /** 强制排队一次成员范围同步，也用于方案名称或免验证配置变化。 */
+    public static function requestPoolSync(int $poolId, int $promotionLinkId): void
+    {
+        if ($poolId <= 0 || $promotionLinkId <= 0) {
+            return;
+        }
+        Db::transaction(function () use ($poolId, $promotionLinkId): void {
+            $now = time();
+            $existing = self::lockedSyncRow($poolId);
+            if ($existing === null) {
+                Db::name('qywx_promotion_range_sync')->insert([
+                    'pool_id' => $poolId,
+                    'promotion_link_id' => $promotionLinkId,
+                    'desired_member_id' => 0,
+                    'desired_version' => 1,
+                    'applied_member_id' => 0,
+                    'applied_version' => 0,
+                    'status' => 1,
+                    'attempts' => 0,
+                    'next_retry' => $now,
+                    'lock_token' => '',
+                    'lock_until' => 0,
+                    'last_error' => '',
+                    'create_time' => $now,
+                    'update_time' => $now,
+                ]);
+
+                return;
+            }
+            $existingStatus = (int) ($existing['status'] ?? 0);
+            $leaseActive = in_array($existingStatus, [2, 5], true)
+                && (int) ($existing['lock_until'] ?? 0) > $now;
+            $data = [
+                'promotion_link_id' => $promotionLinkId,
+                'desired_member_id' => 0,
+                'desired_version' => max(1, (int) ($existing['desired_version'] ?? 0) + 1),
+                'applied_member_id' => 0,
+                'status' => $leaseActive ? $existingStatus : 1,
+                'next_retry' => $leaseActive ? (int) ($existing['next_retry'] ?? 0) : $now,
+                'update_time' => $now,
+            ];
+            if (!$leaseActive) {
+                $data += ['attempts' => 0, 'last_error' => ''];
+            }
+            Db::name('qywx_promotion_range_sync')->where('pool_id', $poolId)->update($data);
+        });
     }
 
     /** @return array{status:string,pool_id:int,member_id:int,next_member_id:int} */
@@ -167,7 +190,8 @@ class QywxPromotionMemberSchedulerService
                 return ['status' => 'ignored_member', 'pool_id' => $poolId, 'member_id' => 0, 'next_member_id' => 0];
             }
             $actualMemberId = (int) $actualMember['id'];
-            $eventKey = hash('sha256', $poolId . '|' . $userId . '|' . $externalUserId);
+            // 同一方案中的同一客户只计一次；后续更换跟进成员不能重复占用医助额度。
+            $eventKey = hash('sha256', $poolId . '|' . $externalUserId);
             try {
                 Db::name('qywx_promotion_dispatch_event')->insert([
                     'event_key' => $eventKey,
@@ -205,12 +229,11 @@ class QywxPromotionMemberSchedulerService
 
             $linkId = self::promotionLinkId($poolId);
             $sync = self::lockedSyncRow($poolId);
-            $desiredId = (int) ($sync['desired_member_id'] ?? 0);
-            if ($linkId <= 0 || ($desiredId > 0 && $desiredId !== $actualMemberId)) {
-                return ['status' => 'counted_stale', 'pool_id' => $poolId, 'member_id' => $actualMemberId, 'next_member_id' => $desiredId];
+            if ($linkId <= 0) {
+                return ['status' => 'counted_stale', 'pool_id' => $poolId, 'member_id' => $actualMemberId, 'next_member_id' => 0];
             }
 
-            $planned = self::selectAndQueueLocked($poolId, $linkId, $sync, $members, $today, $now);
+            $planned = self::queueEligibleRangeLocked($poolId, $linkId, $sync, $members, $today, $now);
 
             return [
                 'status' => $planned['blocked'] ? 'counted_blocked' : 'counted',
@@ -222,7 +245,7 @@ class QywxPromotionMemberSchedulerService
     }
 
     /** @return array{pool_id:int,next_member_id:int,queued:bool,blocked:bool} */
-    private static function selectAndQueueLocked(
+    private static function queueEligibleRangeLocked(
         int $poolId,
         int $linkId,
         ?array $sync,
@@ -230,40 +253,51 @@ class QywxPromotionMemberSchedulerService
         string $today,
         int $now
     ): array {
-        $selection = QywxPromotionWeightedRandom::select($members, $today, $now);
-        self::persistMemberCursors($selection['members'], $now);
-        $selectedId = (int) $selection['selected_id'];
-        if ($selectedId <= 0) {
-            self::upsertSync($poolId, $linkId, (int) ($sync['desired_member_id'] ?? 0), false, $sync, $now, '所有成员均已禁用、未生效或达到今日上限');
+        $range = QywxPromotionMemberRange::evaluate($members, $today, $now);
+        self::persistMemberCursors($range['members'], $now);
+        if ($range['userids'] === []) {
+            self::upsertSync($poolId, $linkId, false, $sync, $now, '所有成员均已禁用、未生效或达到今日上限');
 
             return ['pool_id' => $poolId, 'next_member_id' => 0, 'queued' => false, 'blocked' => true];
         }
-        $changed = $selectedId !== (int) ($sync['desired_member_id'] ?? 0);
-        self::upsertSync($poolId, $linkId, $selectedId, $changed, $sync, $now);
+        $appliedUserIds = self::linkRangeUserIds($linkId);
+        $changed = !QywxPromotionMemberRange::same($range['userids'], $appliedUserIds);
+        $alreadyPending = in_array((int) ($sync['status'] ?? 0), [1, 3], true);
+        $needsSync = $changed || $alreadyPending;
+        self::upsertSync($poolId, $linkId, $needsSync, $sync, $now);
 
-        return ['pool_id' => $poolId, 'next_member_id' => $selectedId, 'queued' => $changed, 'blocked' => false];
+        return ['pool_id' => $poolId, 'next_member_id' => 0, 'queued' => $needsSync, 'blocked' => false];
     }
 
     private static function upsertSync(
         int $poolId,
         int $linkId,
-        int $desiredMemberId,
         bool $pending,
         ?array $existing,
         int $now,
         string $error = ''
     ): void {
-        $version = max(1, (int) ($existing['desired_version'] ?? 0) + ($pending ? 1 : 0));
+        $desiredChanged = $pending || ($error !== '' && (int) ($existing['status'] ?? 0) !== 4);
+        $version = max(1, (int) ($existing['desired_version'] ?? 0) + ($desiredChanged ? 1 : 0));
+        $existingStatus = (int) ($existing['status'] ?? 0);
+        $leaseActive = in_array($existingStatus, [2, 5], true)
+            && (int) ($existing['lock_until'] ?? 0) > $now;
         $data = [
             'promotion_link_id' => $linkId,
-            'desired_member_id' => $desiredMemberId,
+            'desired_member_id' => 0,
             'desired_version' => $version,
-            'status' => $error !== '' ? 4 : ($pending ? 1 : (int) ($existing['status'] ?? 0)),
-            'next_retry' => $error !== '' ? strtotime('tomorrow', $now) : ($pending ? $now : 0),
-            'last_error' => mb_substr($error, 0, 500),
+            // 活跃租约不能被回调或分钟重算抢占；版本变化会让当前工作完成后继续同步。
+            'status' => $leaseActive ? $existingStatus : ($error !== '' ? 4 : ($pending ? 1 : 0)),
+            'next_retry' => $leaseActive
+                ? (int) ($existing['next_retry'] ?? 0)
+                : ($error !== '' ? strtotime('tomorrow', $now) : ($pending ? $now : 0)),
+            'last_error' => $leaseActive
+                ? (string) ($existing['last_error'] ?? '')
+                : mb_substr($error, 0, 500),
+            'applied_member_id' => 0,
             'update_time' => $now,
         ];
-        if ($pending) {
+        if (!$leaseActive) {
             $data['attempts'] = 0;
         }
         if ($existing) {
@@ -318,6 +352,18 @@ class QywxPromotionMemberSchedulerService
             ->value('id') ?? 0);
     }
 
+    /** @return list<string> */
+    private static function linkRangeUserIds(int $linkId): array
+    {
+        $json = (string) (Db::name('qywx_promotion_link')->where('id', $linkId)->value('range_user_json') ?? '[]');
+        $decoded = json_decode($json, true);
+
+        return array_values(array_filter(array_map(
+            static fn (mixed $value): string => trim((string) $value),
+            is_array($decoded) ? $decoded : []
+        ), static fn (string $value): bool => $value !== ''));
+    }
+
     /** @param list<array<string,mixed>> $members */
     private static function persistMemberCursors(array $members, int $now): void
     {

server/app/common/service/qywx/QywxPromotionRangeSyncService.php

@@ -7,7 +7,7 @@ namespace app\common\service\qywx;
 use RuntimeException;
 use think\facade\Db;
 
-/** 把回调调度出的单个目标成员同步到同一条企业微信官方获客链接。 */
+/** 把当前全部可用医助同步到同一条企业微信官方获客链接，由企微执行原生多人路由。 */
 class QywxPromotionRangeSyncService
 {
     private QywxCustomerAcquisitionApiService $api;
@@ -27,7 +27,7 @@ class QywxPromotionRangeSyncService
             }
             $status = (int) ($row['status'] ?? 0);
             $now = time();
-            if ($status === 0 || $status === 4 || ($status === 2 && (int) ($row['lock_until'] ?? 0) > $now)) {
+            if ($status === 0 || $status === 4 || $status === 5 || ($status === 2 && (int) ($row['lock_until'] ?? 0) > $now)) {
                 return null;
             }
             if ($status === 3 && (int) ($row['next_retry'] ?? 0) > $now) {
@@ -52,34 +52,55 @@ class QywxPromotionRangeSyncService
 
         $token = (string) $claim['lock_token'];
         $desiredVersion = (int) ($claim['desired_version'] ?? 0);
-        $memberId = (int) ($claim['desired_member_id'] ?? 0);
         try {
             $pool = Db::name('qywx_promotion_pool')->where('id', $poolId)->whereNull('delete_time')->find();
             $link = Db::name('qywx_promotion_link')
                 ->where('id', (int) ($claim['promotion_link_id'] ?? 0))
                 ->whereNull('delete_time')->find();
-            $member = Db::name('qywx_promotion_pool_member')
-                ->where('id', $memberId)->where('pool_id', $poolId)->whereNull('delete_time')->find();
-            if (!$pool || !$link || !$member || (int) ($member['enabled'] ?? 0) !== 1) {
-                throw new RuntimeException('分流方案、官方链接或目标成员已失效');
+            $members = Db::name('qywx_promotion_pool_member')
+                ->where('pool_id', $poolId)->whereNull('delete_time')->order('id', 'asc')->select()->toArray();
+            if (!$pool || !$link || $members === []) {
+                throw new RuntimeException('分流方案、官方链接或方案成员已失效');
             }
             $remoteLinkId = trim((string) ($link['remote_link_id'] ?? ''));
-            $userId = trim((string) ($member['userid'] ?? ''));
-            if ($remoteLinkId === '' || $userId === '') {
-                throw new RuntimeException('官方链接 ID 或目标成员 userid 为空');
+            if ($remoteLinkId === '') {
+                throw new RuntimeException('官方链接 ID 为空');
+            }
+            $range = QywxPromotionMemberRange::evaluate($members, date('Y-m-d'), time());
+            $desiredUserIds = $range['userids'];
+            if ($desiredUserIds === []) {
+                $message = '所有成员均已禁用、未生效或达到今日上限；企业微信官方链接至少需要保留一名成员';
+                Db::name('qywx_promotion_range_sync')
+                    ->where('pool_id', $poolId)->where('lock_token', $token)->update([
+                        'status' => 4,
+                        'desired_member_id' => 0,
+                        'applied_member_id' => 0,
+                        'next_retry' => strtotime('tomorrow'),
+                        'attempts' => 0,
+                        'lock_token' => '',
+                        'lock_until' => 0,
+                        'last_error' => $message,
+                        'update_time' => time(),
+                    ]);
+                Db::name('qywx_promotion_link')->where('id', (int) $link['id'])->update([
+                    'sync_error' => $message,
+                    'update_time' => time(),
+                ]);
+
+                return ['status' => 'blocked', 'pool_id' => $poolId, 'member_id' => 0];
             }
 
             $this->api->updateLink([
                 'link_id' => $remoteLinkId,
                 'link_name' => mb_substr((string) ($pool['name'] ?? '获客分流方案'), 0, 30),
-                'range' => ['user_list' => [$userId]],
+                'range' => ['user_list' => $desiredUserIds],
                 'skip_verify' => (int) ($link['skip_verify'] ?? 0) === 1,
             ]);
             $response = $this->api->getLink($remoteLinkId);
             $remote = QywxCustomerAcquisitionLinkService::normaliseRemoteResponse($response, $remoteLinkId);
             $actualUserIds = $remote['range_userids'];
-            if ($actualUserIds !== [$userId]) {
-                throw new RuntimeException('企业微信返回的成员范围与待同步成员不一致');
+            if (!QywxPromotionMemberRange::same($actualUserIds, $desiredUserIds)) {
+                throw new RuntimeException('企业微信返回的多人路由成员范围与方案可用医助不一致');
             }
             $url = $remote['url'];
             $snapshot = json_encode($remote['snapshot'], JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES);
@@ -101,7 +122,8 @@ class QywxPromotionRangeSyncService
                 ->where('lock_token', $token)
                 ->update([
                     'status' => $freshVersion === $desiredVersion ? 0 : 1,
-                    'applied_member_id' => $memberId,
+                    'desired_member_id' => 0,
+                    'applied_member_id' => 0,
                     'applied_version' => $desiredVersion,
                     'next_retry' => $freshVersion === $desiredVersion ? 0 : time(),
                     'attempts' => 0,
@@ -111,7 +133,7 @@ class QywxPromotionRangeSyncService
                     'update_time' => time(),
                 ]);
 
-            return ['status' => 'synced', 'pool_id' => $poolId, 'member_id' => $memberId];
+            return ['status' => 'synced', 'pool_id' => $poolId, 'member_id' => 0];
         } catch (\Throwable $e) {
             $attempts = max(1, (int) ($claim['attempts'] ?? 0) + 1);
             Db::name('qywx_promotion_range_sync')
@@ -136,16 +158,21 @@ class QywxPromotionRangeSyncService
     /** @return array{selected:int,synced:int,failed:int} */
     public function syncPending(int $limit = 100): array
     {
+        // 每分钟分页重算全部方案，避免有效期、跨日上限变化漏同步，也避免方案数超过批次后饥饿。
+        $lastPoolId = 0;
+        do {
+            $reconcilePoolIds = Db::name('qywx_promotion_range_sync')
+                ->where('pool_id', '>', $lastPoolId)
+                ->order('pool_id', 'asc')
+                ->limit(500)
+                ->column('pool_id');
+            foreach ($reconcilePoolIds as $reconcilePoolId) {
+                $lastPoolId = (int) $reconcilePoolId;
+                QywxPromotionMemberSchedulerService::reconcilePool($lastPoolId);
+            }
+        } while (count($reconcilePoolIds) === 500);
+
         $now = time();
-        $blockedPoolIds = Db::name('qywx_promotion_range_sync')
-            ->where('status', 4)
-            ->where('next_retry', '>', 0)
-            ->where('next_retry', '<=', $now)
-            ->limit(min(500, max(1, $limit)))
-            ->column('pool_id');
-        foreach ($blockedPoolIds as $blockedPoolId) {
-            QywxPromotionMemberSchedulerService::reconcilePool((int) $blockedPoolId);
-        }
         $poolIds = Db::name('qywx_promotion_range_sync')
             ->where(function ($query) use ($now): void {
                 $query->whereIn('status', [1, 3])->where('next_retry', '<=', $now)

server/app/common/service/qywx/QywxPromotionWeightedRandom.php

@@ -1,83 +0,0 @@
-<?php
-
-declare(strict_types=1);
-
-namespace app\common\service\qywx;
-
-use InvalidArgumentException;
-
-/** 获客方案成员的加权随机选择；数据库锁与持久化由调度服务负责。 */
-class QywxPromotionWeightedRandom
-{
-    /**
-     * @param list<array<string,mixed>> $members
-     * @param int|null $draw 测试用固定抽签值；生产环境留空并使用 random_int。
-     * @return array{selected_id:int,members:list<array<string,mixed>>,eligible_count:int,total_weight:int}
-     */
-    public static function select(array $members, string $today, int $now, ?int $draw = null): array
-    {
-        $eligibleIndexes = [];
-        $totalWeight = 0;
-        foreach ($members as $index => &$member) {
-            if ((string) ($member['today_date'] ?? '') !== $today) {
-                $member['today_date'] = $today;
-                $member['today_count'] = 0;
-            }
-            // current_weight 是旧版平滑轮询游标；随机模式不再使用，统一归零。
-            $member['current_weight'] = 0;
-            if (!self::eligible($member, $now)) {
-                continue;
-            }
-            $weight = max(1, (int) ($member['weight'] ?? 1));
-            $eligibleIndexes[] = ['index' => $index, 'weight' => $weight];
-            $totalWeight += $weight;
-        }
-        unset($member);
-
-        if ($eligibleIndexes === []) {
-            return [
-                'selected_id' => 0,
-                'members' => array_values($members),
-                'eligible_count' => 0,
-                'total_weight' => 0,
-            ];
-        }
-
-        $ticket = $draw ?? random_int(1, $totalWeight);
-        if ($ticket < 1 || $ticket > $totalWeight) {
-            throw new InvalidArgumentException('加权随机抽签值超出有效范围');
-        }
-        $cursor = 0;
-        $selectedIndex = (int) $eligibleIndexes[0]['index'];
-        foreach ($eligibleIndexes as $candidate) {
-            $cursor += (int) $candidate['weight'];
-            if ($ticket <= $cursor) {
-                $selectedIndex = (int) $candidate['index'];
-                break;
-            }
-        }
-
-        return [
-            'selected_id' => (int) ($members[$selectedIndex]['id'] ?? 0),
-            'members' => array_values($members),
-            'eligible_count' => count($eligibleIndexes),
-            'total_weight' => $totalWeight,
-        ];
-    }
-
-    /** @param array<string,mixed> $member */
-    public static function eligible(array $member, int $now): bool
-    {
-        if ((int) ($member['enabled'] ?? 0) !== 1 || (int) ($member['weight'] ?? 0) <= 0) {
-            return false;
-        }
-        $start = max(0, (int) ($member['active_start'] ?? 0));
-        $end = max(0, (int) ($member['active_end'] ?? 0));
-        if (($start > 0 && $start > $now) || ($end > 0 && $end < $now)) {
-            return false;
-        }
-        $limit = max(0, (int) ($member['daily_limit'] ?? 0));
-
-        return $limit === 0 || (int) ($member['today_count'] ?? 0) < $limit;
-    }
-}