<?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('推广客户资料尚未同步到本地');
        }
    }
}