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

<?php

declare(strict_types=1);

namespace app\common\service\qywx;

use RuntimeException;
use think\facade\Db;

/** 把回调调度出的单个目标成员同步到同一条企业微信官方获客链接。 */
class QywxPromotionRangeSyncService
{
    private QywxCustomerAcquisitionApiService $api;

    public function __construct(?QywxCustomerAcquisitionApiService $api = null)
    {
        $this->api = $api ?? new QywxCustomerAcquisitionApiService();
    }

    /** @return array{status:string,pool_id:int,member_id:int} */
    public function syncPool(int $poolId): array
    {
        $claim = Db::transaction(function () use ($poolId): ?array {
            $row = Db::name('qywx_promotion_range_sync')->where('pool_id', $poolId)->lock(true)->find();
            if (!$row) {
                return null;
            }
            $status = (int) ($row['status'] ?? 0);
            $now = time();
            if ($status === 0 || $status === 4 || ($status === 2 && (int) ($row['lock_until'] ?? 0) > $now)) {
                return null;
            }
            if ($status === 3 && (int) ($row['next_retry'] ?? 0) > $now) {
                return null;
            }
            $token = bin2hex(random_bytes(16));
            Db::name('qywx_promotion_range_sync')->where('pool_id', $poolId)->update([
                'status' => 2,
                'lock_token' => $token,
                'lock_until' => $now + 90,
                'attempts' => (int) ($row['attempts'] ?? 0) + 1,
                'last_error' => '',
                'update_time' => $now,
            ]);
            $row['lock_token'] = $token;

            return $row;
        });
        if ($claim === null) {
            return ['status' => 'noop', 'pool_id' => $poolId, 'member_id' => 0];
        }

        $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('分流方案、官方链接或目标成员已失效');
            }
            $remoteLinkId = trim((string) ($link['remote_link_id'] ?? ''));
            $userId = trim((string) ($member['userid'] ?? ''));
            if ($remoteLinkId === '' || $userId === '') {
                throw new RuntimeException('官方链接 ID 或目标成员 userid 为空');
            }

            $this->api->updateLink([
                'link_id' => $remoteLinkId,
                'link_name' => mb_substr((string) ($pool['name'] ?? '获客分流方案'), 0, 30),
                'range' => ['user_list' => [$userId]],
                '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('企业微信返回的成员范围与待同步成员不一致');
            }
            $url = $remote['url'];
            $snapshot = json_encode($remote['snapshot'], JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES);
            Db::name('qywx_promotion_link')->where('id', (int) $link['id'])->update([
                'wecom_url' => $url,
                'remote_status' => 1,
                'range_user_json' => json_encode($actualUserIds, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES),
                'range_department_json' => json_encode($remote['range_department_ids'], JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES),
                'remote_snapshot' => $snapshot === false ? null : $snapshot,
                'last_sync_time' => time(),
                'sync_error' => '',
                'update_time' => time(),
            ]);

            $freshVersion = (int) (Db::name('qywx_promotion_range_sync')
                ->where('pool_id', $poolId)->value('desired_version') ?? 0);
            Db::name('qywx_promotion_range_sync')
                ->where('pool_id', $poolId)
                ->where('lock_token', $token)
                ->update([
                    'status' => $freshVersion === $desiredVersion ? 0 : 1,
                    'applied_member_id' => $memberId,
                    'applied_version' => $desiredVersion,
                    'next_retry' => $freshVersion === $desiredVersion ? 0 : time(),
                    'attempts' => 0,
                    'lock_token' => '',
                    'lock_until' => 0,
                    'last_error' => '',
                    'update_time' => time(),
                ]);

            return ['status' => 'synced', 'pool_id' => $poolId, 'member_id' => $memberId];
        } catch (\Throwable $e) {
            $attempts = max(1, (int) ($claim['attempts'] ?? 0) + 1);
            Db::name('qywx_promotion_range_sync')
                ->where('pool_id', $poolId)
                ->where('lock_token', $token)
                ->update([
                    'status' => 3,
                    'next_retry' => time() + min(300, 15 * $attempts),
                    'lock_token' => '',
                    'lock_until' => 0,
                    'last_error' => mb_substr($e->getMessage(), 0, 500),
                    'update_time' => time(),
                ]);
            Db::name('qywx_promotion_link')
                ->where('id', (int) ($claim['promotion_link_id'] ?? 0))
                ->update(['sync_error' => mb_substr($e->getMessage(), 0, 500), 'update_time' => time()]);

            throw $e;
        }
    }

    /** @return array{selected:int,synced:int,failed:int} */
    public function syncPending(int $limit = 100): array
    {
        $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)
                    ->whereOr(function ($running) use ($now): void {
                        $running->where('status', 2)->where('lock_until', '<=', $now);
                    });
            })
            ->order('next_retry', 'asc')
            ->limit(min(500, max(1, $limit)))
            ->column('pool_id');
        $synced = 0;
        $failed = 0;
        foreach ($poolIds as $poolId) {
            try {
                $result = $this->syncPool((int) $poolId);
                if ($result['status'] === 'synced') {
                    $synced++;
                }
            } catch (\Throwable) {
                $failed++;
            }
        }

        return ['selected' => count($poolIds), 'synced' => $synced, 'failed' => $failed];
    }
}