<?php

declare(strict_types=1);

namespace base\tools\controller\timer;

use app\common\controller\TimerController;
use app\common\service\NoticeService;
use base\admin\model\SystemNoticeTargetBase;
use think\facade\Db;
use think\facade\Log;

/**
 * 通知异步投递消费者（Base 层）.
 *
 * 消费 system_notice_target 中 status=0（待发送）的 email/sms/wechat 渠道行：
 * inmail 行由 NoticeService 直接写成功终态永不入队，故候选查询无需渠道过滤。
 *
 * 消费模式（ulthon-queue 模式 A：DB 原子抢占）：
 *   1. 候选查询走 (status, id) 复合索引 idx_status_id，只读前 BATCH_SIZE 条
 *   2. 原子抢占：读和标记压进一条条件 UPDATE（WHERE 带 status=0），
 *      concurrency>1 时被抢先的行 affected=0，竞态在此消解；不按
 *      concurrencyId 划分数据（反模式：id % N 取余分片导致索引失效）
 *   3. 逐条投递：调 NoticeService::channels() 对应驱动 send(target, notice)，
 *      单条失败不影响同批其他行
 *   4. 回写幂等：WHERE 含 claim_token + status=1，防回写他人抢占行；
 *      失败走指数退避重试（不删失败任务），第 MAX_ATTEMPTS 次尝试失败置终态
 *
 * 时间窗口模式（ulthon-timer）：do() 内 maxRunTime 窗口循环 doBatch()，
 * 超时自动退出由定时器下次调度继续；frequency >= maxRunTime * 2。
 *
 * 调度与防刷：$frequency = null 显式关闭控制器侧防刷（手动触发不受
 * "请不要频繁请求"拦截），调度节流完全由定时器侧 Cache 节流承担。
 *
 * 依赖倒置：渠道映射经 app\common\service\NoticeService 入口获取
 * （双层双向注入惯例：base 层引用 app 入口类）；业务侧如需定制消费
 * 行为，重写 app/tools/controller/timer/NoticeConsumer.php 对应方法即可。
 */
class NoticeConsumerBase extends TimerController
{
    /**
     * 防刷间隔：null = 关闭控制器侧防刷（调度由定时器侧 frequency 节流）.
     */
    protected $frequency = null;

    /**
     * 并发上限 cap：2 分片（id ∈ [0, 1]），竞争消费由原子抢占消解.
     */
    protected $concurrency = 2;

    /**
     * 时间窗口（秒）：3 配 frequency=10s，满足 frequency >= maxRunTime * 2.
     */
    protected $maxRunTime = 3;

    /**
     * 单批候选数：走 (status, id) 索引区间扫描，与表总量无关.
     */
    protected const BATCH_SIZE = 20;

    /**
     * 最大尝试次数：首次 + 4 次重试 = 共 5 次，第 5 次失败置 STATUS_FAILED 终态.
     */
    protected const MAX_ATTEMPTS = 5;

    /**
     * 失败退避基准（秒）：next_retry_time = now + 基准 * 2^retry_count
     * （retry_count 为自增后的值），最大 60 * 2^4 = 960 秒.
     */
    protected const RETRY_BASE_SECONDS = 60;

    public function do()
    {
        $max_time = time() + $this->maxRunTime;

        do {
            $has_work = $this->doBatch();

            if (!$has_work) {
                sleep(1);          // 队列空时休眠，防空转；有数据时不要 sleep
                break;
            }

            if (time() >= $max_time) {
                break;
            }
        } while (true);

        return 'done';
    }

    /**
     * 消费一批候选（抢占 + 逐条投递）.
     *
     * @return bool 是否有工作量（无候选返回 false，do() 据此休眠退出）
     */
    protected function doBatch(): bool
    {
        // 候选：status=0 且退避时间已到，走 (status, id) 索引只读前 20 条
        $candidates = Db::name('system_notice_target')
            ->where('status', SystemNoticeTargetBase::STATUS_PENDING)
            ->where('next_retry_time', '<=', time())
            ->order('id')
            ->limit(self::BATCH_SIZE)
            ->column('id');

        if (empty($candidates)) {
            return false;
        }

        // 原子抢占：WHERE 带 status=0，被其他消费者抢先的行 affected=0，竞态在此消解
        $token = md5(uniqid('', true) . '-' . mt_rand());
        Db::name('system_notice_target')
            ->where('id', 'in', $candidates)
            ->where('status', SystemNoticeTargetBase::STATUS_PENDING)
            ->update([
                'status'      => SystemNoticeTargetBase::STATUS_SENDING,
                'claim_token' => $token,
                'claim_time'  => time(),
                'update_time' => time(),
            ]);

        // 查回真正抢到的：必须带 id IN（主键定位），只按 claim_token 查会全表扫描
        $mine = Db::name('system_notice_target')
            ->where('id', 'in', $candidates)
            ->where('claim_token', $token)
            ->select()
            ->toArray();

        if (empty($mine)) {
            return false;
        }

        // 批量取通知主体（title/content 驱动报文所需；notice 表无 claim 概念）
        $notices = Db::name('system_notice')
            ->whereIn('id', array_values(array_unique(array_column($mine, 'notice_id'))))
            ->select()
            ->toArray();
        $noticeMap = array_column($notices, null, 'id');

        $channels = NoticeService::channels();

        foreach ($mine as $item) {
            try {
                $driverClass = $channels[$item['channel']] ?? null;
                if ($driverClass === null) {
                    throw new \RuntimeException('未注册的投递渠道: ' . $item['channel']);
                }

                $notice = $noticeMap[$item['notice_id']] ?? null;
                if ($notice === null) {
                    throw new \RuntimeException('通知主体不存在: notice_id=' . $item['notice_id']);
                }

                $driver = new $driverClass();
                $driver->send($item, $notice);

                // 成功幂等回写：WHERE 含 claim_token（防回写他人抢占行）+ status=1 兜底
                Db::name('system_notice_target')
                    ->where('id', $item['id'])
                    ->where('claim_token', $token)
                    ->where('status', SystemNoticeTargetBase::STATUS_SENDING)
                    ->update([
                        'status'      => SystemNoticeTargetBase::STATUS_SUCCESS,
                        'send_time'   => time(),
                        'update_time' => time(),
                    ]);
            } catch (\Throwable $e) {
                $this->handleFailure($item, $token, $e);
            }
        }

        return true;
    }

    /**
     * 单条失败处理：指数退避重试，第 MAX_ATTEMPTS 次尝试失败置终态（不删失败任务）.
     *
     * 退避公式：next_retry_time = now + 60 * 2^retry_count（retry_count 为自增后值）；
     * 回写 WHERE 含 claim_token + status=1，幂等兜底重复消费。
     *
     * @param array     $item  抢占到的投递行
     * @param string    $token 本轮抢占令牌
     * @param \Throwable $e    投递异常
     */
    protected function handleFailure(array $item, string $token, \Throwable $e): void
    {
        $retry_count = (int) $item['retry_count'] + 1;
        $error_message = mb_substr($e->getMessage(), 0, 500);

        Log::error('notice consumer: 投递失败 target_id=' . $item['id']
            . ' channel=' . $item['channel'] . ' retry=' . $retry_count . ' - ' . $error_message);

        if ($retry_count >= self::MAX_ATTEMPTS) {
            // 第 5 次尝试失败：终态
            Db::name('system_notice_target')
                ->where('id', $item['id'])
                ->where('claim_token', $token)
                ->where('status', SystemNoticeTargetBase::STATUS_SENDING)
                ->update([
                    'status'        => SystemNoticeTargetBase::STATUS_FAILED,
                    'retry_count'   => $retry_count,
                    'error_message' => $error_message,
                    'update_time'   => time(),
                ]);

            return;
        }

        Db::name('system_notice_target')
            ->where('id', $item['id'])
            ->where('claim_token', $token)
            ->where('status', SystemNoticeTargetBase::STATUS_SENDING)
            ->update([
                'status'          => SystemNoticeTargetBase::STATUS_PENDING,
                'retry_count'     => $retry_count,
                'next_retry_time' => time() + self::RETRY_BASE_SECONDS * (2 ** $retry_count),
                'error_message'   => $error_message,
                'update_time'     => time(),
            ]);
    }
}