429 lines
18 KiB
PHP
429 lines
18 KiB
PHP
<?php
|
|
|
|
declare(strict_types=1);
|
|
|
|
namespace app\common\service\qywx;
|
|
|
|
use RuntimeException;
|
|
use think\facade\Db;
|
|
|
|
/** 获客客户归因、会话统计与回调幂等落库。 */
|
|
class QywxCustomerAcquisitionCustomerService
|
|
{
|
|
private QywxCustomerAcquisitionApiService $api;
|
|
|
|
public function __construct(?QywxCustomerAcquisitionApiService $api = null)
|
|
{
|
|
$this->api = $api ?? new QywxCustomerAcquisitionApiService();
|
|
}
|
|
|
|
/**
|
|
* 同步一个远端获客链接的全部客户，远端列表字段采用覆盖语义。
|
|
* recv_msg_cnt 不在列表接口中返回，因此同步时保留本地值。
|
|
*
|
|
* @return array{scanned:int,created:int,updated:int,pages:int,truncated:bool}
|
|
*/
|
|
public function syncLink(string $remoteLinkId, int $maxCustomers = 20000): array
|
|
{
|
|
$remoteLinkId = trim($remoteLinkId);
|
|
if ($remoteLinkId === '') {
|
|
throw new RuntimeException('获客链接 ID 不能为空');
|
|
}
|
|
$cursor = '';
|
|
$scanned = 0;
|
|
$created = 0;
|
|
$updated = 0;
|
|
$pages = 0;
|
|
do {
|
|
$page = $this->api->listCustomers($remoteLinkId, $cursor, 1000);
|
|
$pages++;
|
|
foreach ($page['customer_list'] as $customer) {
|
|
if ($scanned >= $maxCustomers) {
|
|
break 2;
|
|
}
|
|
$scanned++;
|
|
$result = self::upsertCustomer($remoteLinkId, $customer, false);
|
|
$result === 'created' ? $created++ : $updated++;
|
|
}
|
|
$cursor = (string) ($page['next_cursor'] ?? '');
|
|
} while ($cursor !== '');
|
|
|
|
return compact('scanned', 'created', 'updated', 'pages') + ['truncated' => $cursor !== ''];
|
|
}
|
|
|
|
/**
|
|
* 处理 customer_acquisition 回调。相同事件只成功处理一次；失败事件保留审计并允许企微重试。
|
|
*
|
|
* @return array{duplicate:bool,status:string}
|
|
*/
|
|
public function handleCallback(array $message): array
|
|
{
|
|
$changeType = trim((string) ($message['ChangeType'] ?? $message['change_type'] ?? ''));
|
|
if (!in_array($changeType, ['customer_start_chat', 'message_from_customer'], true)) {
|
|
return ['duplicate' => false, 'status' => 'ignored'];
|
|
}
|
|
$chatKey = trim((string) ($message['ChatKey'] ?? $message['Chatkey'] ?? $message['chat_key'] ?? ''));
|
|
$eventTime = (int) ($message['CreateTime'] ?? $message['create_time'] ?? 0);
|
|
$eventKey = self::eventKey($message, $changeType, $chatKey, $eventTime);
|
|
$event = self::beginEvent($eventKey, $changeType, $chatKey, $eventTime, $message);
|
|
if (($event['duplicate'] ?? false) === true) {
|
|
return ['duplicate' => true, 'status' => 'success'];
|
|
}
|
|
|
|
$eventId = (int) ($event['id'] ?? 0);
|
|
try {
|
|
// customer_start_chat 仅能确认“客户已发起会话”，企业微信不保证该事件携带 ChatKey。
|
|
// 此时先落归因与聊天状态，精确消息数等待 message_from_customer 回调补齐。
|
|
if ($changeType === 'customer_start_chat' && $chatKey === '') {
|
|
$remoteLinkId = trim((string) (
|
|
$message['LinkID'] ?? $message['LinkId'] ?? $message['link_id'] ?? ''
|
|
));
|
|
$externalUserId = trim((string) (
|
|
$message['ExternalUserID'] ?? $message['ExternalUserId'] ?? $message['external_userid'] ?? ''
|
|
));
|
|
$userId = trim((string) ($message['UserID'] ?? $message['UserId'] ?? $message['userid'] ?? ''));
|
|
if ($remoteLinkId === '' || $externalUserId === '' || $userId === '') {
|
|
throw new RuntimeException('customer_start_chat 回调缺少 link_id / external_userid / userid');
|
|
}
|
|
self::upsertCustomer($remoteLinkId, [
|
|
'external_userid' => $externalUserId,
|
|
'userid' => $userId,
|
|
'chat_status' => 1,
|
|
'state' => (string) ($message['State'] ?? $message['state'] ?? ''),
|
|
'event_time' => $eventTime,
|
|
'snapshot' => $message,
|
|
], false);
|
|
self::finishEvent($eventId, 1, '', $remoteLinkId, $externalUserId, $userId);
|
|
|
|
return ['duplicate' => false, 'status' => 'success'];
|
|
}
|
|
if ($chatKey === '') {
|
|
self::finishEvent($eventId, 3, 'failed_invalid: message_from_customer 回调缺少 ChatKey');
|
|
throw new RuntimeException('message_from_customer 回调缺少 ChatKey');
|
|
}
|
|
$now = time();
|
|
if ($eventTime > 0 && ($now - $eventTime) >= 1800) {
|
|
throw new RuntimeException('获客回调 ChatKey 已超过 30 分钟有效期');
|
|
}
|
|
$chat = $this->api->getChatInfo($chatKey);
|
|
$chatInfo = is_array($chat['chat_info'] ?? null) ? $chat['chat_info'] : [];
|
|
$remoteLinkId = trim((string) (
|
|
$chatInfo['link_id'] ?? $message['LinkID'] ?? $message['LinkId'] ?? $message['link_id'] ?? ''
|
|
));
|
|
$externalUserId = trim((string) (
|
|
$chat['external_userid'] ?? $message['ExternalUserID'] ?? $message['ExternalUserId'] ?? ''
|
|
));
|
|
$userId = trim((string) ($chat['userid'] ?? $message['UserID'] ?? $message['UserId'] ?? ''));
|
|
if ($remoteLinkId === '' || $externalUserId === '' || $userId === '') {
|
|
throw new RuntimeException('get_chat_info 未返回完整的 link_id / external_userid / userid');
|
|
}
|
|
self::upsertCustomer($remoteLinkId, [
|
|
'external_userid' => $externalUserId,
|
|
'userid' => $userId,
|
|
'chat_status' => max(1, (int) ($message['ChatStatus'] ?? 1)),
|
|
'recv_msg_cnt' => max(0, (int) ($chatInfo['recv_msg_cnt'] ?? 0)),
|
|
'state' => (string) ($chatInfo['state'] ?? $message['State'] ?? ''),
|
|
'event_time' => $eventTime,
|
|
'snapshot' => $chat,
|
|
], true);
|
|
self::finishEvent($eventId, 1, '', $remoteLinkId, $externalUserId, $userId);
|
|
|
|
return ['duplicate' => false, 'status' => 'success'];
|
|
} catch (\Throwable $e) {
|
|
if ($eventId > 0 && str_contains($e->getMessage(), 'message_from_customer 回调缺少 ChatKey')) {
|
|
throw $e;
|
|
}
|
|
self::scheduleRetryOrExpire($eventId, $eventTime, $e->getMessage());
|
|
throw $e;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* 重试仍在 ChatKey 30 分钟有效期内的失败回调，并把到期记录明确标记 failed_expired。
|
|
*
|
|
* @return array{selected:int,success:int,failed:int,expired:int}
|
|
*/
|
|
public function retryPending(int $limit = 100): array
|
|
{
|
|
$now = time();
|
|
// 进程在 beginEvent 后异常退出时，处理中事件会卡在 status=0；一分钟后自动回收再试。
|
|
Db::name('qywx_customer_acquisition_event')
|
|
->where('status', 0)
|
|
->where('update_time', '<=', $now - 60)
|
|
->where('expire_time', '>', $now)
|
|
->update([
|
|
'status' => 2,
|
|
'next_retry' => $now,
|
|
'error_message' => 'watchdog_recovered: 上次处理未正常结束',
|
|
'update_time' => $now,
|
|
]);
|
|
$expired = (int) Db::name('qywx_customer_acquisition_event')
|
|
->whereIn('status', [0, 2])
|
|
->where('expire_time', '>', 0)
|
|
->where('expire_time', '<=', $now)
|
|
->update([
|
|
'status' => 3,
|
|
'next_retry' => 0,
|
|
'error_message' => 'failed_expired: ChatKey 已超过 30 分钟有效期',
|
|
'chat_key' => '',
|
|
'raw_json' => null,
|
|
'update_time' => $now,
|
|
]);
|
|
$rows = Db::name('qywx_customer_acquisition_event')
|
|
->where('status', 2)
|
|
->where('next_retry', '<=', $now)
|
|
->where('expire_time', '>', $now)
|
|
->order('next_retry', 'asc')
|
|
->limit(min(500, max(1, $limit)))
|
|
->select()->toArray();
|
|
$success = 0;
|
|
$failed = 0;
|
|
foreach ($rows as $row) {
|
|
$message = json_decode((string) ($row['raw_json'] ?? ''), true);
|
|
if (!is_array($message)) {
|
|
self::scheduleRetryOrExpire(
|
|
(int) $row['id'],
|
|
(int) ($row['event_time'] ?? 0),
|
|
'回调原始数据无法解析'
|
|
);
|
|
$failed++;
|
|
continue;
|
|
}
|
|
try {
|
|
$this->handleCallback($message);
|
|
$success++;
|
|
} catch (\Throwable) {
|
|
$failed++;
|
|
}
|
|
}
|
|
|
|
return ['selected' => count($rows), 'success' => $success, 'failed' => $failed, 'expired' => $expired];
|
|
}
|
|
|
|
public static function eventKey(array $message, string $changeType, string $chatKey, int $eventTime): string
|
|
{
|
|
$parts = [
|
|
(string) ($message['MsgId'] ?? $message['MsgID'] ?? ''),
|
|
$changeType,
|
|
$chatKey,
|
|
(string) $eventTime,
|
|
(string) ($message['LinkID'] ?? $message['LinkId'] ?? ''),
|
|
(string) ($message['ExternalUserID'] ?? $message['ExternalUserId'] ?? ''),
|
|
(string) ($message['UserID'] ?? $message['UserId'] ?? ''),
|
|
];
|
|
|
|
return hash('sha256', implode('|', $parts));
|
|
}
|
|
|
|
/** @return array{expire_time:int,next_retry:int,expired:bool} */
|
|
public static function retryDecision(int $eventTime, int $now, int $storedExpireTime = 0): array
|
|
{
|
|
$expireTime = $storedExpireTime > 0
|
|
? $storedExpireTime
|
|
: ($eventTime > 0 ? $eventTime + 1800 : $now + 1800);
|
|
$expired = $expireTime <= $now;
|
|
|
|
return [
|
|
'expire_time' => $expireTime,
|
|
'next_retry' => $expired ? 0 : min($expireTime - 1, $now + 30),
|
|
'expired' => $expired,
|
|
];
|
|
}
|
|
|
|
/** @return array{id:int,duplicate:bool} */
|
|
private static function beginEvent(
|
|
string $eventKey,
|
|
string $changeType,
|
|
string $chatKey,
|
|
int $eventTime,
|
|
array $message
|
|
): array {
|
|
$now = time();
|
|
$raw = self::encodeJson($message);
|
|
$expireTime = self::retryDecision($eventTime, $now)['expire_time'];
|
|
try {
|
|
$id = (int) Db::name('qywx_customer_acquisition_event')->insertGetId([
|
|
'event_key' => $eventKey,
|
|
'change_type' => $changeType,
|
|
'chat_key' => $chatKey,
|
|
'status' => 0,
|
|
'attempts' => 1,
|
|
'event_time' => max(0, $eventTime),
|
|
'expire_time' => $expireTime,
|
|
'next_retry' => 0,
|
|
'error_message' => '',
|
|
'raw_json' => $raw,
|
|
'create_time' => $now,
|
|
'update_time' => $now,
|
|
]);
|
|
|
|
return ['id' => $id, 'duplicate' => false];
|
|
} catch (\Throwable $e) {
|
|
$existing = Db::name('qywx_customer_acquisition_event')->where('event_key', $eventKey)->find();
|
|
if (!$existing) {
|
|
throw $e;
|
|
}
|
|
if ((int) ($existing['status'] ?? 0) === 1) {
|
|
return ['id' => (int) $existing['id'], 'duplicate' => true];
|
|
}
|
|
if ((int) ($existing['status'] ?? 0) !== 2) {
|
|
return ['id' => (int) $existing['id'], 'duplicate' => true];
|
|
}
|
|
$claimed = Db::name('qywx_customer_acquisition_event')
|
|
->where('id', (int) $existing['id'])
|
|
->where('status', 2)
|
|
->update([
|
|
'status' => 0,
|
|
'attempts' => (int) ($existing['attempts'] ?? 0) + 1,
|
|
'error_message' => '',
|
|
'raw_json' => $raw,
|
|
'update_time' => $now,
|
|
]);
|
|
if ($claimed <= 0) {
|
|
return ['id' => (int) $existing['id'], 'duplicate' => true];
|
|
}
|
|
|
|
return ['id' => (int) $existing['id'], 'duplicate' => false];
|
|
}
|
|
}
|
|
|
|
private static function finishEvent(
|
|
int $id,
|
|
int $status,
|
|
string $error = '',
|
|
string $remoteLinkId = '',
|
|
string $externalUserId = '',
|
|
string $userId = ''
|
|
): void {
|
|
if ($id <= 0) {
|
|
return;
|
|
}
|
|
Db::name('qywx_customer_acquisition_event')->where('id', $id)->update([
|
|
'status' => $status,
|
|
'link_id' => $remoteLinkId,
|
|
'external_userid' => $externalUserId,
|
|
'userid' => $userId,
|
|
'error_message' => mb_substr($error, 0, 1000),
|
|
'next_retry' => 0,
|
|
// ChatKey 是短时敏感凭证，终态后不再保留；原始回调也随之清理。
|
|
'chat_key' => '',
|
|
'raw_json' => null,
|
|
'update_time' => time(),
|
|
]);
|
|
}
|
|
|
|
private static function scheduleRetryOrExpire(int $id, int $eventTime, string $error): void
|
|
{
|
|
if ($id <= 0) {
|
|
return;
|
|
}
|
|
$now = time();
|
|
$expireTime = (int) (Db::name('qywx_customer_acquisition_event')
|
|
->where('id', $id)->value('expire_time') ?? 0);
|
|
$decision = self::retryDecision($eventTime, $now, $expireTime);
|
|
$expireTime = $decision['expire_time'];
|
|
$expired = $decision['expired'];
|
|
Db::name('qywx_customer_acquisition_event')->where('id', $id)->update([
|
|
'status' => $expired ? 3 : 2,
|
|
'expire_time' => $expireTime,
|
|
'next_retry' => $decision['next_retry'],
|
|
'error_message' => mb_substr(
|
|
$expired ? 'failed_expired: ' . $error : $error,
|
|
0,
|
|
1000
|
|
),
|
|
'chat_key' => $expired ? '' : Db::raw('chat_key'),
|
|
'raw_json' => $expired ? null : Db::raw('raw_json'),
|
|
'update_time' => $now,
|
|
]);
|
|
}
|
|
|
|
/** @return 'created'|'updated' */
|
|
private static function upsertCustomer(string $remoteLinkId, array $customer, bool $messageCountKnown): string
|
|
{
|
|
$externalUserId = trim((string) ($customer['external_userid'] ?? ''));
|
|
$userId = trim((string) ($customer['userid'] ?? ''));
|
|
if ($remoteLinkId === '' || $externalUserId === '' || $userId === '') {
|
|
throw new RuntimeException('获客客户数据缺少 link_id / external_userid / userid');
|
|
}
|
|
[$ownerAdminId, $deptId] = self::resolveOwner($userId);
|
|
$now = time();
|
|
$existing = Db::name('qywx_customer_acquisition_customer')
|
|
->where('link_id', $remoteLinkId)
|
|
->where('external_userid', $externalUserId)
|
|
->where('userid', $userId)
|
|
->find();
|
|
$snapshot = $customer['snapshot'] ?? $customer;
|
|
$incomingChatStatus = max(0, min(2, (int) ($customer['chat_status'] ?? 0)));
|
|
$data = [
|
|
'promotion_link_id' => (int) (Db::name('qywx_promotion_link')
|
|
->where('remote_link_id', $remoteLinkId)->value('id') ?? 0),
|
|
'owner_admin_id' => $ownerAdminId,
|
|
'dept_id' => $deptId,
|
|
'state' => mb_substr((string) ($customer['state'] ?? ''), 0, 255),
|
|
// 已确认发过消息后，列表同步返回的“未发/未知”不得把状态回退。
|
|
'chat_status' => $existing
|
|
? Db::raw('CASE WHEN chat_status = 1 OR ' . $incomingChatStatus . ' = 1 THEN 1 ELSE ' . $incomingChatStatus . ' END')
|
|
: $incomingChatStatus,
|
|
'last_sync_time' => $now,
|
|
'raw_snapshot' => self::encodeJson($snapshot),
|
|
'update_time' => $now,
|
|
];
|
|
if ($messageCountKnown) {
|
|
$remoteCount = max(0, (int) ($customer['recv_msg_cnt'] ?? 0));
|
|
// get_chat_info 返回累计值，必须 max/覆盖，绝不按回调次数累加。
|
|
$data['recv_msg_cnt'] = $existing
|
|
? Db::raw('GREATEST(recv_msg_cnt,' . $remoteCount . ')')
|
|
: $remoteCount;
|
|
$data['message_count_known'] = 1;
|
|
}
|
|
if ($incomingChatStatus === 1 || $messageCountKnown) {
|
|
$eventTime = max(0, (int) ($customer['event_time'] ?? $now));
|
|
$data['last_chat_time'] = $existing
|
|
? Db::raw('GREATEST(last_chat_time,' . $eventTime . ')')
|
|
: $eventTime;
|
|
}
|
|
if ($existing) {
|
|
Db::name('qywx_customer_acquisition_customer')->where('id', (int) $existing['id'])->update($data);
|
|
|
|
return 'updated';
|
|
}
|
|
$data += [
|
|
'link_id' => $remoteLinkId,
|
|
'external_userid' => $externalUserId,
|
|
'userid' => $userId,
|
|
'recv_msg_cnt' => $messageCountKnown ? max(0, (int) ($customer['recv_msg_cnt'] ?? 0)) : 0,
|
|
'message_count_known' => $messageCountKnown ? 1 : 0,
|
|
'first_acquired_time' => max(0, (int) ($customer['create_time'] ?? $customer['event_time'] ?? $now)),
|
|
'last_chat_time' => ($incomingChatStatus === 1 || $messageCountKnown)
|
|
? max(0, (int) ($customer['event_time'] ?? $now))
|
|
: 0,
|
|
'create_time' => $now,
|
|
];
|
|
Db::name('qywx_customer_acquisition_customer')->insert($data);
|
|
|
|
return 'created';
|
|
}
|
|
|
|
/** @return array{0:int,1:int} */
|
|
private static function resolveOwner(string $userId): array
|
|
{
|
|
$adminId = (int) (Db::name('admin')->where('work_wechat_userid', $userId)
|
|
->whereNull('delete_time')->value('id') ?? 0);
|
|
if ($adminId <= 0) {
|
|
return [0, 0];
|
|
}
|
|
$deptId = (int) (Db::name('admin_dept')->where('admin_id', $adminId)
|
|
->order('dept_id', 'asc')->value('dept_id') ?? 0);
|
|
|
|
return [$adminId, $deptId];
|
|
}
|
|
|
|
private static function encodeJson(mixed $value): string
|
|
{
|
|
$json = json_encode($value, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES);
|
|
|
|
return $json === false ? '{}' : $json;
|
|
}
|
|
}
|