fdesk/app/inquiry/service/InquiryAssignService.php

<?php

namespace app\inquiry\service;

use support\Db;
use support\Redis;
use support\exception\BusinessException;
use Webman\RedisQueue\Redis as RedisQueueRedis;

class InquiryAssignService
{
    /**
     * 按渠道分配 inquiry_log 中待分配的记录，轮询业务员，并推送到 CRM 队列
     *
     * @param int[]  $salesUserIds   默认参与轮询的业务员ID列表（该渠道默认池）
     * @param string $roundRobinKey  默认轮询分组key（Redis记指针）
     * @param int    $channelId      渠道ID（对应 inquiry_log.channel）
     * @param array  $assignRules    渠道内规则：按国家切换业务员池（可选）
     * @param int    $limit          每次最多分配多少条
     *
     * @return int 实际分配的条数
     */
    public function assignBatch(
        array $salesUserIds,
        string $roundRobinKey,
        int $channelId,
        array $assignRules = [],
        int $limit = 100
    ): int {
        // 默认池（渠道默认业务员列表）
        $defaultUserIds = $this->normalizeUserIds($salesUserIds);
        if (empty($defaultUserIds)) {
            throw new BusinessException('分配失败：业务员列表为空');
        }

        $inquiries = Db::table('inquiry_log')
            ->where('status', 0)
            ->where('recipient', 0)
            ->where('channel', $channelId)
            ->orderBy('id')
            ->limit($limit)
            ->get();

        if ($inquiries->isEmpty()) {
            return 0;
        }

        $now   = time();
        $count = 0;

        foreach ($inquiries as $row) {
            $inquiryId = (int)$row->id;

            // ✅ 关键：按规则选业务员池（命中规则就用规则池，否则用默认池）
            [$poolUserIds, $rrKey] = $this->pickSalesPoolByRules(
                (int)($row->country ?? 0),
                $defaultUserIds,
                $roundRobinKey,
                $assignRules
            );

            // ✅ 用“对应池子自己的 rrKey”轮询拿业务员
            $recipientUserId = $this->nextRecipientId($poolUserIds, $rrKey);

            Db::transaction(function () use ($inquiryId, $recipientUserId, $now, &$count) {
                $affected = Db::table('inquiry_log')
                    ->where('id', $inquiryId)
                    ->update([
                        'recipient'  => $recipientUserId,
                        'status'     => 2,
                        'r_time'     => $now,
                        't_time'     => $now,
                        'updated_at' => $now,
                    ]);

                if ($affected === 0) {
                    // 可能被其他进程抢先处理了，直接跳过
                    return;
                }

                $res = [
                    'id'            => $inquiryId,
                    'recipient'     => $recipientUserId,
                    'dispatched_at' => $now,
                    'source'        => 'fb_leads_inquiry',
                ];

                RedisQueueRedis::send('inquiry-to-crm-new', $res);

                $count++;
            });
        }

        return $count;
    }

    /**
     * 轮询获取下一个业务员ID
     */
    protected function nextRecipientId(array $userIds, string $groupKey): int
    {
        $userIds = $this->normalizeUserIds($userIds);
        if (empty($userIds)) {
            throw new BusinessException('轮询分配失败：业务员列表为空');
        }

        $count = count($userIds);
        $key   = 'inquiry:assign:round_robin:' . $groupKey;

        $val   = Redis::incr($key); // 原子自增
        $index = ($val - 1) % $count;

        return $userIds[$index];
    }

    /**
     * 按国家规则挑选“业务员池 + 轮询key”
     * 命中 rules 则返回规则池，否则返回默认池
     *
     * rules 示例：
     * [
     *   [
     *     'countries' => [1,2,3],
     *     'sales_user_ids' => [63,64],
     *     'round_robin_key' => 'fb_xxx_ru',
     *   ],
     * ]
     *
     * @return array{0:int[], 1:string} [userIds, rrKey]
     */
    protected function pickSalesPoolByRules(
        int $countryId,
        array $defaultUserIds,
        string $defaultRrKey,
        array $rules
    ): array {
        foreach ($rules as $rule) {
            if (!is_array($rule)) continue;

            $countries = $rule['countries'] ?? [];
            if (!is_array($countries) || empty($countries)) continue;

            if (!in_array($countryId, array_map('intval', $countries), true)) {
                continue;
            }

            $userIds = $this->normalizeUserIds($rule['sales_user_ids'] ?? []);
            if (!empty($userIds)) {
                // 规则池必须使用自己的 rrKey，否则会和默认池“抢指针”
                $rrKey = (string)($rule['round_robin_key'] ?? ($defaultRrKey . '_rule_' . md5(json_encode($countries))));
                return [$userIds, $rrKey];
            }
        }

        return [$this->normalizeUserIds($defaultUserIds), $defaultRrKey];
    }

    /**
     * 统一清洗业务员ID数组：转 int、去重、过滤 <=0
     */
    protected function normalizeUserIds(array $ids): array
    {
        $ids = array_values(array_filter(array_map('intval', $ids), fn($v) => $v > 0));
        $ids = array_values(array_unique($ids));
        return $ids;
    }
}