更新

2026-08-31 15:17:34 +08:00
parent ed48f8be31
commit 456dd667df
439 changed files with 5720 additions and 422 deletions
@@ -0,0 +1,363 @@
+<?php
+
+declare(strict_types=1);
+
+namespace app\common\service\qywx;
+
+use RuntimeException;
+
+/** 推广客户自动化：短时欢迎语与可补偿关系动作分开消费。 */
+class QywxPromotionAutomationService
+{
+    private QywxPromotionContactApiService $api;
+    private QywxPromotionMediaService $media;
+    private QywxPromotionAutomationStore $store;
+    private QywxPromotionCodeCipher $cipher;
+    private $clock;
+    private const TERMINAL = ['sent', 'success', 'skipped', 'expired', 'uncertain', 'failed'];
+
+    public function __construct(
+        ?QywxPromotionContactApiService $api = null,
+        ?QywxPromotionMediaService $media = null,
+        ?QywxPromotionAutomationStore $store = null,
+        ?QywxPromotionCodeCipher $cipher = null,
+        ?callable $clock = null
+    ) {
+        $this->api = $api ?? new QywxPromotionContactApiService();
+        $this->media = $media ?? new QywxPromotionMediaService($this->api);
+        $this->store = $store ?? new QywxPromotionAutomationStore();
+        $this->cipher = $cipher ?? new QywxPromotionCodeCipher();
+        $this->clock = $clock ?? static fn (): int => time();
+    }
+
+    /**
+     * 仅供验签解密后的回调调用。false代表沿用旧同步流程；已接管的入队错误必须返回HTTP500。
+     * 此处无网络请求，保证回调不等待客户详情、范围更新或素材上传。
+     */
+    public function enqueueVerifiedEvent(array $event): bool
+    {
+        $change = (string) ($event['ChangeType'] ?? '');
+        if (!in_array($change, ['add_external_contact', 'add_half_external_contact'], true)) {
+            return false;
+        }
+        $state = trim((string) ($event['State'] ?? ''));
+        $linkId = trim((string) ($event['LinkId'] ?? $event['LinkID'] ?? ''));
+        $userid = trim((string) ($event['UserID'] ?? $event['UserId'] ?? ''));
+        $external = trim((string) ($event['ExternalUserID'] ?? $event['ExternalUserId'] ?? ''));
+        if (($state === '' && $linkId === '') || $userid === '' || $external === '') {
+            return false;
+        }
+        try {
+            if (!$this->store->installed()) {
+                return false;
+            }
+            $attribution = $this->store->attribution($state, $linkId, $userid);
+            if ($attribution === null) {
+                return false;
+            }
+            $now = $this->now();
+            $eventTime = max(0, (int) ($event['CreateTime'] ?? 0));
+            $code = (string) ($event['WelcomeCode'] ?? '');
+            $config = $attribution['config'];
+            $half = $change === 'add_half_external_contact';
+            $welcomeStatus = 'pending';
+            $reason = '';
+            if (($config['welcome_mode'] ?? 'default') !== 'channel') {
+                $welcomeStatus = 'skipped';
+                $reason = 'mode_' . ($config['welcome_mode'] ?? 'default');
+            } elseif ($code === '') {
+                $welcomeStatus = 'skipped';
+                $reason = 'missing_welcome_code';
+            } elseif (strlen($code) > 1024) {
+                $welcomeStatus = 'failed';
+                $reason = 'invalid_welcome_code';
+            } elseif ($eventTime <= 0 || $eventTime > $now + 5 || $eventTime + 20 <= $now) {
+                $welcomeStatus = 'expired';
+                $reason = 'welcome_window_elapsed_or_invalid_event_time';
+            }
+            $actions = [
+                'welcome' => self::action($welcomeStatus, $reason),
+                'tags' => self::action(!$half && !empty($config['tags_enabled']) ? 'pending' : 'skipped', $half ? 'half_contact' : 'disabled'),
+                'remark' => self::action(!$half && !empty($config['remark_enabled']) ? 'pending' : 'skipped', $half ? 'half_contact' : 'disabled'),
+                'description' => self::action(!$half && !empty($config['description_enabled']) ? 'pending' : 'skipped', $half ? 'half_contact' : 'disabled'),
+                'dispatch' => self::action($half ? 'skipped' : 'pending', $half ? 'half_contact' : ''),
+                'range' => self::action($half ? 'skipped' : 'pending', $half ? 'half_contact' : ''),
+                'sync' => self::action($half ? 'skipped' : 'pending', $half ? 'half_contact' : ''),
+            ];
+            $corp = (string) ($event['ToUserName'] ?? config('pay.wechat_work.corp_id', ''));
+            $this->store->enqueue([
+                'event_key' => hash('sha256', implode('|', [$corp, $change, $userid, $external, (string) $eventTime])),
+                'pool_id' => $attribution['pool_id'], 'member_admin_id' => $attribution['member_admin_id'],
+                'change_type' => $change, 'userid' => $userid, 'external_userid' => $external,
+                'event_time' => $eventTime, 'received_at' => $now,
+                'config_json' => self::json($config), 'actions_json' => self::json($actions),
+                'welcome_cipher' => $welcomeStatus === 'pending' ? $this->cipher->encrypt($code) : '',
+                'welcome_code_hash' => $code !== '' ? hash('sha256', $code) : '',
+                'welcome_expires_at' => $eventTime > 0 ? min($eventTime + 20, $now + 20) : 0,
+                'welcome_status' => $welcomeStatus, 'welcome_next_retry' => 0,
+                'status' => self::allTerminal($actions) ? 'done' : 'pending', 'next_retry' => 0,
+                'lock_token' => '', 'lock_until' => 0, 'create_time' => $now, 'update_time' => $now,
+            ]);
+            return true;
+        } catch (\Throwable) {
+            // 不附原异常，入库SQL可能包含密文和配置；回调层返回500触发企微重试。
+            throw new QywxPromotionEnqueueException('推广自动化事件未能持久化，请检查数据库迁移和私有存储');
+        }
+    }
+
+    /** 常驻秒级worker仅处理欢迎语，不被范围/客户同步或大文件上传阻塞。 */
+    public function processWelcomes(int $limit = 100): array
+    {
+        return $this->consume('welcome', $limit);
+    }
+
+    /** 分钟补偿：过期欢迎语只记过期，绝不尝试补发。 */
+    public function retryPending(int $limit = 100): array
+    {
+        return $this->consume('metadata', $limit);
+    }
+
+    public static function selectWelcome(array $config, int $eventTime): array
+    {
+        if (!empty($config['welcome_schedule_enabled'])) {
+            foreach ((array) ($config['welcome_schedule'] ?? []) as $slot) {
+                if (QywxPromotionConfig::matches($slot, $eventTime)) {
+                    return ['text' => (string) ($slot['text'] ?? ''), 'attachments' => (array) ($slot['attachments'] ?? [])];
+                }
+            }
+        }
+        return ['text' => (string) ($config['welcome']['text'] ?? ''), 'attachments' => (array) ($config['welcome']['attachments'] ?? [])];
+    }
+
+    private function consume(string $lane, int $limit): array
+    {
+        $result = ['selected' => 0, 'processed' => 0, 'failed' => 0];
+        foreach ($this->store->due($lane, $this->now(), $limit) as $id) {
+            ++$result['selected'];
+            try {
+                $row = $this->store->claim($id, $lane, $this->now());
+                if ($row === null) {
+                    continue;
+                }
+                $actions = json_decode($row['actions_json'], true, 512, JSON_THROW_ON_ERROR);
+                $config = json_decode($row['config_json'], true, 512, JSON_THROW_ON_ERROR);
+                if (!self::terminal($actions['welcome']['status'])) {
+                    if ($lane === 'welcome') {
+                        $this->welcome($row, $actions, $config);
+                    } else {
+                        $running = $actions['welcome']['status'] === 'running';
+                        $this->transition($row, $actions, 'welcome', $running ? 'uncertain' : 'expired',
+                            $running ? 'worker_interrupted_after_send_started' : 'welcome_worker_not_available_in_window');
+                    }
+                }
+                if ($lane !== 'welcome') {
+                    $this->metadata($row, $actions, $config);
+                }
+                $row['lock_until'] = 0;
+                $row['update_time'] = $this->now();
+                $this->store->save($row);
+                ++$result['processed'];
+            } catch (\Throwable) {
+                // 失去DB/租约时保留running状态；欢迎语恢复时视为不确定，防止重复推送。
+                ++$result['failed'];
+            }
+        }
+        return $result;
+    }
+
+    private function welcome(array &$row, array &$actions, array $config): void
+    {
+        if ($actions['welcome']['status'] === 'running') {
+            $this->transition($row, $actions, 'welcome', 'uncertain', 'worker_interrupted_after_send_started');
+            return;
+        }
+        if ((int) $row['welcome_expires_at'] <= $this->now() + 1) {
+            $this->transition($row, $actions, 'welcome', 'expired', 'welcome_window_elapsed');
+            return;
+        }
+        $sendStarted = false;
+        try {
+            $message = self::selectWelcome($config, (int) $row['event_time']);
+            $text = $message['text'];
+            if (str_contains($text, '{customer_name}') || str_contains($text, '{employee_name}') || str_contains($text, '{add_time}')) {
+                $names = $this->names($row, $text, true);
+                $text = QywxPromotionConfig::render($text, $names['customer'], $names['employee'], (int) $row['event_time'], 1200);
+            }
+            $truncated = strlen($text) > 4000;
+            $text = mb_strcut($text, 0, 4000, 'UTF-8');
+            $attachments = $this->media->materialize($message['attachments'], $config);
+            $code = $this->cipher->decrypt($row['welcome_cipher']);
+            if ((int) $row['welcome_expires_at'] <= $this->now() + 1) {
+                $this->transition($row, $actions, 'welcome', 'expired', 'welcome_window_elapsed_during_prepare');
+                return;
+            }
+            // running先持久化：如果HTTP成功后进程/DB断开，恢复时绝不再次使用同一code。
+            $this->transition($row, $actions, 'welcome', 'running', 'send_started');
+            $sendStarted = true;
+            try {
+                $this->api->sendWelcome($code, $text, $attachments);
+                $this->transition($row, $actions, 'welcome', 'sent', $truncated ? 'sent_text_truncated_4000_bytes' : 'sent');
+            } catch (QywxPromotionContactApiException $e) {
+                if ($e->uncertain) {
+                    $this->transition($row, $actions, 'welcome', 'uncertain', 'network_result_unknown_do_not_resend', $e->getCode());
+                } elseif ($e->getCode() === 41051) {
+                    $this->transition($row, $actions, 'welcome', 'skipped', 'welcome_code_already_consumed', 41051);
+                } else {
+                    $this->welcomeRetry($row, $actions, 'explicit_api_rejection', $e->getCode());
+                }
+            } catch (\Throwable) {
+                $this->transition($row, $actions, 'welcome', 'uncertain', 'send_or_persist_result_unknown_do_not_resend');
+            } finally {
+                unset($code);
+            }
+        } catch (\Throwable $e) {
+            // 准备阶段没有执行发送，可以安全重试，且不会把错误原文/欢迎码写日志。
+            if ($sendStarted || $actions['welcome']['status'] === 'running') {
+                throw $e;
+            }
+            $this->welcomeRetry($row, $actions, 'prepare_failed_check_media_credentials_or_key', (int) $e->getCode());
+        }
+    }
+
+    private function welcomeRetry(array &$row, array &$actions, string $reason, int $code): void
+    {
+        $expired = (int) $row['welcome_expires_at'] <= $this->now() + 2;
+        $this->transition($row, $actions, 'welcome', $expired ? 'expired' : 'retry', $reason, $code, $this->now() + 1);
+    }
+
+    private function metadata(array &$row, array &$actions, array $config): void
+    {
+        $names = null;
+        foreach (['tags', 'remark', 'description', 'dispatch', 'range', 'sync'] as $name) {
+            if (self::terminal($actions[$name]['status']) || (int) ($actions[$name]['next_retry'] ?? 0) > $this->now()) {
+                continue;
+            }
+            $this->transition($row, $actions, $name, 'running', 'started');
+            try {
+                switch ($name) {
+                    case 'tags':
+                        $this->api->markTags($row['userid'], $row['external_userid'], (array) $config['tag_ids']);
+                        break;
+                    case 'remark':
+                        $names = $names ?? $this->names($row, (string) $config['remark_template'], false);
+                        $remark = QywxPromotionConfig::render($config['remark_template'], $names['customer'], $names['employee'], (int) $row['event_time'], 20);
+                        $this->api->remark($row['userid'], $row['external_userid'], ['remark' => $remark]);
+                        break;
+                    case 'description':
+                        $this->api->remark($row['userid'], $row['external_userid'], ['description' => (string) $config['description']]);
+                        break;
+                    case 'dispatch':
+                        $this->store->dispatch($row);
+                        break;
+                    case 'range':
+                        $this->store->syncRange($row);
+                        break;
+                    case 'sync':
+                        $this->store->syncCustomer($row);
+                        break;
+                }
+                $this->transition($row, $actions, $name, 'success', 'completed');
+            } catch (\Throwable $e) {
+                $attempt = (int) $actions[$name]['attempts'];
+                $failed = $attempt >= 10;
+                $this->transition($row, $actions, $name, $failed ? 'failed' : 'retry',
+                    $failed ? 'retry_limit_reached' : 'action_failed', (int) $e->getCode(),
+                    $this->now() + min(3600, 15 * (2 ** min(8, $attempt))));
+            }
+        }
+    }
+
+    private function names(array $row, string $template, bool $welcome): array
+    {
+        $names = ['customer' => '', 'employee' => ''];
+        try {
+            $names = $this->store->localNames($row);
+        } catch (\Throwable) {
+            // 本地资料失败不妨碍欢迎语使用明确的文案兜底。
+        }
+        $budget = fn (): bool => !$welcome || (int) $row['welcome_expires_at'] > $this->now() + 7;
+        if (str_contains($template, '{customer_name}') && $names['customer'] === ''
+            && $row['change_type'] !== 'add_half_external_contact' && $budget()) {
+            try {
+                $detail = $this->api->getExternalContact($row['external_userid']);
+                $names['customer'] = (string) ($detail['external_contact']['name'] ?? '');
+            } catch (\Throwable) {
+            }
+        }
+        if (str_contains($template, '{employee_name}') && $budget()) {
+            try {
+                $user = $this->api->getUser($row['userid']);
+                $names['employee'] = trim((string) ($user['name'] ?? '')) ?: $names['employee'];
+            } catch (\Throwable) {
+                // 通讯录姓名接口权限不足时回退后台成员称呼。
+            }
+        }
+        $names['customer'] = $names['customer'] !== '' ? $names['customer'] : '您';
+        $names['employee'] = $names['employee'] !== '' ? $names['employee'] : '客户顾问';
+        return $names;
+    }
+
+    private function transition(array &$row, array &$actions, string $name, string $status, string $reason, int $code = 0, int $retryAt = 0): void
+    {
+        $now = $this->now();
+        $action = $actions[$name];
+        if ($status === 'running' || ($name === 'welcome' && $status === 'retry' && $action['status'] !== 'running')) {
+            ++$action['attempts'];
+        }
+        $action = array_replace($action, ['status' => $status, 'reason' => $reason, 'error_code' => $code,
+            'next_retry' => $retryAt, 'update_time' => $now]);
+        if (self::terminal($status)) {
+            $action['finished_at'] = $now;
+        }
+        $actions[$name] = $action;
+        if ($name === 'welcome') {
+            $row['welcome_status'] = $status;
+            $row['welcome_next_retry'] = $retryAt;
+            if (self::terminal($status)) {
+                $row['welcome_cipher'] = '';
+            }
+        }
+        $row['status'] = self::allTerminal($actions) ? 'done' : 'pending';
+        $retry = [];
+        foreach ($actions as $key => $value) {
+            if ($key !== 'welcome' && !self::terminal($value['status'])) {
+                $retry[] = (int) ($value['next_retry'] ?? 0);
+            }
+        }
+        $row['next_retry'] = $retry === [] ? 0 : min($retry);
+        $row['actions_json'] = self::json($actions);
+        $row['update_time'] = $now;
+        $this->store->save($row, ['action' => $name, 'status' => $status, 'attempt' => $action['attempts'],
+            'reason' => $reason, 'error_code' => $code, 'create_time' => $now]);
+    }
+
+    private static function action(string $status, string $reason = ''): array
+    {
+        return ['status' => $status, 'reason' => $status === 'pending' ? '' : $reason, 'attempts' => 0, 'error_code' => 0, 'next_retry' => 0];
+    }
+
+    private static function allTerminal(array $actions): bool
+    {
+        foreach ($actions as $action) {
+            if (!self::terminal($action['status'])) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    private static function terminal(string $status): bool
+    {
+        return in_array($status, self::TERMINAL, true);
+    }
+
+    private static function json(array $value): string
+    {
+        return json_encode($value, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES | JSON_THROW_ON_ERROR);
+    }
+
+    private function now(): int
+    {
+        return (int) ($this->clock)();
+    }
+}