ZYT/zyt
2
0
Fork 0
Code Issues Actions Packages Projects Releases Wiki Activity

更新

Browse Source
This commit is contained in:
Your Name
2026-09-01 11:26:47 +08:00
parent 5bd5eae62d
commit 398f9f3726
357 changed files with 733 additions and 343 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/app/common/service/qywx/QywxPromotionAutomationService.php
+28 -8
View File
@@ -32,9 +32,9 @@ class QywxPromotionAutomationService
/**
* 仅供验签解密后的回调调用。false代表沿用旧同步流程；已接管的入队错误必须返回HTTP500。
* 此处无网络请求，保证回调不等待客户详情、范围更新或素材上传。
* 默认仅入队；回调入口可启用即时通道，在当前请求内先发欢迎语并给正式客户打标。
*/
public function enqueueVerifiedEvent(array $event): bool
public function enqueueVerifiedEvent(array $event, bool $processImmediately = false): bool
{
$change = (string) ($event['ChangeType'] ?? '');
if (!in_array($change, ['add_external_contact', 'add_half_external_contact'], true)) {
@@ -85,7 +85,7 @@ class QywxPromotionAutomationService
'sync' => self::action($half ? 'skipped' : 'pending', $half ? 'half_contact' : ''),
];
$corp = (string) ($event['ToUserName'] ?? config('pay.wechat_work.corp_id', ''));
$this->store->enqueue([
$taskId = $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,
@@ -98,6 +98,14 @@ class QywxPromotionAutomationService
'status' => self::allTerminal($actions) ? 'done' : 'pending', 'next_retry' => 0,
'lock_token' => '', 'lock_until' => 0, 'create_time' => $now, 'update_time' => $now,
]);
if ($processImmediately) {
// 部署环境暂未启动常驻 worker 时仍要抢住 20 秒欢迎码窗口。
// 标签只在正式客户事件执行；其余慢动作仍由分钟补偿处理。
$this->consumeIds('welcome', [$taskId]);
if (!$half) {
$this->consumeIds('inline_metadata', [$taskId], ['tags']);
}
}
return true;
} catch (\Throwable) {
// 不附原异常，入库SQL可能包含密文和配置；回调层返回500触发企微重试。
@@ -130,9 +138,15 @@ class QywxPromotionAutomationService
}
private function consume(string $lane, int $limit): array
{
return $this->consumeIds($lane, $this->store->due($lane, $this->now(), $limit));
}
/** @param list<int> $ids @param list<string>|null $metadataActions */
private function consumeIds(string $lane, array $ids, ?array $metadataActions = null): array
{
$result = ['selected' => 0, 'processed' => 0, 'failed' => 0];
foreach ($this->store->due($lane, $this->now(), $limit) as $id) {
foreach ($ids as $id) {
++$result['selected'];
try {
$row = $this->store->claim($id, $lane, $this->now());
@@ -141,7 +155,8 @@ class QywxPromotionAutomationService
}
$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'])) {
// 即时标签与欢迎语共用同一任务，但不能把仍在 20 秒窗口内待重试的欢迎语判为过期。
if ($lane !== 'inline_metadata' && !self::terminal($actions['welcome']['status'])) {
if ($lane === 'welcome') {
$this->welcome($row, $actions, $config);
} else {
@@ -151,7 +166,7 @@ class QywxPromotionAutomationService
}
}
if ($lane !== 'welcome') {
$this->metadata($row, $actions, $config);
$this->metadata($row, $actions, $config, $metadataActions);
}
$row['lock_until'] = 0;
$row['update_time'] = $this->now();
@@ -225,10 +240,15 @@ class QywxPromotionAutomationService
$this->transition($row, $actions, 'welcome', $expired ? 'expired' : 'retry', $reason, $code, $this->now() + 1);
}
private function metadata(array &$row, array &$actions, array $config): void
/** @param list<string>|null $only */
private function metadata(array &$row, array &$actions, array $config, ?array $only = null): void
{
$names = null;
foreach (['tags', 'remark', 'description', 'dispatch', 'range', 'sync'] as $name) {
$namesToProcess = ['tags', 'remark', 'description', 'dispatch', 'range', 'sync'];
if ($only !== null) {
$namesToProcess = array_values(array_intersect($namesToProcess, $only));
}
foreach ($namesToProcess as $name) {
if (self::terminal($actions[$name]['status']) || (int) ($actions[$name]['next_retry'] ?? 0) > $this->now()) {
continue;
}