191 lines
9.3 KiB
PHP
191 lines
9.3 KiB
PHP
<?php
|
|
|
|
declare(strict_types=1);
|
|
|
|
namespace app\common\service\qywx;
|
|
|
|
use app\adminapi\logic\qywx\CustomerLogic;
|
|
use RuntimeException;
|
|
use think\facade\Db;
|
|
|
|
/** DB 存储与既有同步边界；单测替换此类后不初始化业务数据库。 */
|
|
class QywxPromotionAutomationStore
|
|
{
|
|
public function installed(): bool
|
|
{
|
|
return QywxPromotionConfig::installed();
|
|
}
|
|
|
|
/** State只能定位，必须再核验真实方案、正式官方链接与实际成员关系。 */
|
|
public function attribution(string $state, string $linkId, string $userId): ?array
|
|
{
|
|
if ($state !== '') {
|
|
if (!preg_match('/^zyt_pool:([1-9][0-9]{0,9})$/', $state, $match)) {
|
|
return null;
|
|
}
|
|
$poolId = (int) $match[1];
|
|
} elseif ($linkId !== '') {
|
|
$poolId = (int) Db::name('qywx_promotion_link')->where('remote_link_id', $linkId)
|
|
->where('remote_status', 1)->whereNull('delete_time')->value('pool_id');
|
|
} else {
|
|
return null;
|
|
}
|
|
$pool = Db::name('qywx_promotion_pool')->where('id', $poolId)->where('status', 1)->whereNull('delete_time')->find();
|
|
$member = Db::name('qywx_promotion_pool_member')->where('pool_id', $poolId)->where('userid', $userId)
|
|
->whereNull('delete_time')->find();
|
|
$links = Db::name('qywx_promotion_link')->where('pool_id', $poolId)->where('remote_status', 1)
|
|
->where('remote_link_id', '<>', '')->whereNull('delete_time');
|
|
if ($linkId !== '') {
|
|
$links->where('remote_link_id', $linkId);
|
|
}
|
|
// 不用 enabled/当日额度验证：真实回调可能比排班切换晚到，不能漏掉已归属该方案的成员。
|
|
if (!$pool || !$member || !$links->find()) {
|
|
return null;
|
|
}
|
|
$configRow = Db::name('qywx_promotion_config')->where('pool_id', $poolId)->find();
|
|
if (!$configRow) {
|
|
// 尚未保存新增配置的旧方案仍保持原同步链路，不强制依赖新worker。
|
|
return null;
|
|
}
|
|
return ['pool_id' => $poolId, 'member_admin_id' => (int) $member['admin_id'],
|
|
'config' => QywxPromotionConfig::decode($configRow['config_json'])];
|
|
}
|
|
|
|
public function enqueue(array $row): int
|
|
{
|
|
$row['welcome_code_hash'] = $row['welcome_code_hash'] ?: null;
|
|
// 同一code可能同时出现在half/add：唯一索引把欢迎语消费权固定在第一次任务。
|
|
for ($attempt = 0; $attempt < 2; $attempt++) {
|
|
if ($row['welcome_code_hash'] !== null
|
|
&& Db::name('qywx_promotion_automation_task')->where('welcome_code_hash', $row['welcome_code_hash'])->find()) {
|
|
$actions = json_decode($row['actions_json'], true, 512, JSON_THROW_ON_ERROR);
|
|
$actions['welcome']['status'] = 'skipped';
|
|
$actions['welcome']['reason'] = 'same_welcome_code_already_queued';
|
|
$row['actions_json'] = json_encode($actions, JSON_UNESCAPED_UNICODE | JSON_THROW_ON_ERROR);
|
|
$row['welcome_status'] = 'skipped';
|
|
$row['welcome_cipher'] = '';
|
|
$row['welcome_code_hash'] = null;
|
|
$pending = array_filter($actions, static fn (array $a): bool => in_array($a['status'], ['pending', 'retry', 'running'], true));
|
|
$row['status'] = $pending === [] ? 'done' : 'pending';
|
|
}
|
|
try {
|
|
return (int) Db::name('qywx_promotion_automation_task')->insertGetId($row);
|
|
} catch (\Throwable $e) {
|
|
$existing = Db::name('qywx_promotion_automation_task')->where('event_key', $row['event_key'])->value('id');
|
|
if ($existing) {
|
|
return (int) $existing;
|
|
}
|
|
if ($attempt === 1 || $row['welcome_code_hash'] === null) {
|
|
throw $e;
|
|
}
|
|
}
|
|
}
|
|
throw new RuntimeException('推广任务入队失败');
|
|
}
|
|
|
|
/** 两条消费通道：常驻worker只发欢迎语，分钟任务不锁住尚有时效的欢迎语任务。 */
|
|
public function due(string $lane, int $now, int $limit): array
|
|
{
|
|
$query = Db::name('qywx_promotion_automation_task')->where('status', '<>', 'done')
|
|
->where('lock_until', '<=', $now);
|
|
if ($lane === 'welcome') {
|
|
$query->whereIn('welcome_status', ['pending', 'retry', 'running'])->where('welcome_next_retry', '<=', $now)
|
|
->order('welcome_expires_at', 'asc');
|
|
} else {
|
|
$query->where('next_retry', '<=', $now)->where(function ($q) use ($now) {
|
|
$q->whereNotIn('welcome_status', ['pending', 'retry', 'running'])
|
|
->whereOr('welcome_expires_at', '<=', $now);
|
|
})->order('id', 'asc');
|
|
}
|
|
return array_map('intval', $query->limit(max(1, min(500, $limit)))->column('id'));
|
|
}
|
|
|
|
public function claim(int $id, string $lane, int $now): ?array
|
|
{
|
|
return Db::transaction(function () use ($id, $lane, $now): ?array {
|
|
$row = Db::name('qywx_promotion_automation_task')->where('id', $id)->lock(true)->find();
|
|
if (!$row || $row['status'] === 'done' || (int) $row['lock_until'] > $now) {
|
|
return null;
|
|
}
|
|
$pendingWelcome = in_array($row['welcome_status'], ['pending', 'retry', 'running'], true);
|
|
if (($lane === 'welcome' && (!$pendingWelcome || (int) $row['welcome_next_retry'] > $now))
|
|
|| ($lane === 'metadata' && (($pendingWelcome && (int) $row['welcome_expires_at'] > $now) || (int) $row['next_retry'] > $now))
|
|
|| ($lane === 'inline_metadata' && (int) $row['next_retry'] > $now)) {
|
|
return null;
|
|
}
|
|
$row['lock_token'] = bin2hex(random_bytes(16));
|
|
$row['lock_until'] = $now + (in_array($lane, ['welcome', 'inline_metadata'], true) ? 30 : 600);
|
|
Db::name('qywx_promotion_automation_task')->where('id', $id)->update([
|
|
'lock_token' => $row['lock_token'], 'lock_until' => $row['lock_until'], 'update_time' => $now,
|
|
]);
|
|
return $row;
|
|
});
|
|
}
|
|
|
|
public function save(array $row, ?array $log = null): void
|
|
{
|
|
Db::transaction(function () use ($row, $log): void {
|
|
$fields = array_intersect_key($row, array_flip([
|
|
'actions_json', 'welcome_status', 'welcome_cipher', 'welcome_next_retry', 'status',
|
|
'next_retry', 'lock_until', 'update_time',
|
|
]));
|
|
// 租约令牌校验不能依赖affected rows：同秒同值更新在MySQL可能返回0。
|
|
$current = Db::name('qywx_promotion_automation_task')->where('id', $row['id'])->lock(true)->find();
|
|
if (!$current || !hash_equals((string) $current['lock_token'], (string) $row['lock_token'])) {
|
|
throw new RuntimeException('推广任务处理租约已失效');
|
|
}
|
|
Db::name('qywx_promotion_automation_task')->where('id', $row['id'])->update($fields);
|
|
if ($log !== null) {
|
|
Db::name('qywx_promotion_automation_action_log')->insert($log + ['task_id' => $row['id']]);
|
|
}
|
|
});
|
|
|
|
// 标签动作成功/终止后，以任务入队时的配置冻结事件渠道；重复调用由完成标记幂等保护。
|
|
QywxExternalContactEventTagSnapshotService::captureFromPromotionTask($row);
|
|
}
|
|
|
|
public function localNames(array $task): array
|
|
{
|
|
return [
|
|
'customer' => (string) (Db::name('qywx_external_contact')->where('external_userid', $task['external_userid'])->value('name') ?? ''),
|
|
'employee' => (string) (Db::name('admin')->where('id', $task['member_admin_id'])->value('name') ?? ''),
|
|
];
|
|
}
|
|
|
|
public function dispatch(array $task): void
|
|
{
|
|
$result = QywxPromotionMemberSchedulerService::recordFromState('zyt_pool:' . $task['pool_id'],
|
|
$task['userid'], $task['external_userid'], (int) $task['event_time'], 'external_contact');
|
|
if (!in_array($result['status'] ?? '', ['counted', 'counted_blocked', 'counted_stale', 'duplicate'], true)) {
|
|
throw new RuntimeException('推广成员记账未完成');
|
|
}
|
|
}
|
|
|
|
public function syncRange(array $task): void
|
|
{
|
|
// range服务自身有持久重试与版本保护；此调用负责触发。
|
|
(new QywxPromotionRangeSyncService())->syncPool((int) $task['pool_id']);
|
|
}
|
|
|
|
public function syncCustomer(array $task): void
|
|
{
|
|
$started = time();
|
|
$eventId = (int) Db::name('qywx_external_contact_event')
|
|
->where('change_type', 'add_external_contact')
|
|
->where('user_id', (string) $task['userid'])
|
|
->where('external_userid', (string) $task['external_userid'])
|
|
->where('event_time', (int) $task['event_time'])
|
|
->value('id');
|
|
CustomerLogic::upsertSingleExternalContactFromApi(
|
|
$task['external_userid'],
|
|
$eventId,
|
|
(string) $task['userid']
|
|
);
|
|
// 旧方法在API空结果时只log并返回void；必须核验本地实际更新，避免把未同步记为成功。
|
|
$updated = (int) Db::name('qywx_external_contact')->where('external_userid', $task['external_userid'])->value('update_time');
|
|
if ($updated < $started) {
|
|
throw new RuntimeException('推广客户资料尚未同步到本地');
|
|
}
|
|
}
|
|
}
|