bot-28/app/Services/SiteMessageService.php

<?php

namespace App\Services;

use App\Support\LocalizedContent;
use Carbon\Carbon;
use Illuminate\Auth\Access\AuthorizationException;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Log;
use Illuminate\Support\Facades\Validator;
use Illuminate\Validation\Rule;

class SiteMessageService
{
    public const CATEGORIES = ['discount' => '优惠活动', 'site_notice' => '网站公告', 'internal' => '站内信', 'promotion' => '活动宣传'];
    public const TARGETS = ['all_users' => '所有人', 'specified_users' => '指定用户', 'all_agents' => '所有代理', 'specified_agents' => '指定代理'];
    public const STATES = ['scheduled' => '待发送', 'sent' => '已发送', 'retracted' => '已撤回'];

    public static function canManageAgents(): bool
    {
        return in_array((int) (request()->user->id ?? 0), config('agent.admin_ids', [1]), true);
    }

    public function options(): array
    {
        return ['categories' => $this->optionsFor(self::CATEGORIES), 'recipient_types' => $this->optionsFor(self::TARGETS),
            'languages' => $this->optionsFor(LocalizedContent::LANGUAGES), 'can_manage_agents' => self::canManageAgents(), 'timezone' => 'Asia/Shanghai'];
    }

    public function recipients(string $kind, array $input): array
    {
        if ($kind === 'agent') $this->requireAgentAccess();
        $params = Validator::make($input, $this->pageRules() + ['q' => ['nullable', 'string', 'max:100']])->validate();
        $query = $this->eligible($kind);
        if (isset($params['q']) && $params['q'] !== '') {
            $q = $params['q'];
            $query->where(function ($query) use ($kind, $q) {
                foreach ($kind === 'agent' ? ['username', 'real_name'] : ['account', 'username', 'first_name'] as $field) $query->orWhere('u.' . $field, 'like', '%' . $q . '%');
                if (ctype_digit($q)) {
                    $query->orWhere('u.id', $q);
                    if ($kind === 'user') $query->orWhere('u.user_id', $q)->orWhere('u.member_id', $q);
                }
            });
        }
        $total = (clone $query)->count();
        $fields = $kind === 'agent' ? ['u.id', 'u.username', 'u.real_name', 'u.language', 'u.status']
            : ['u.id', 'u.user_id', 'u.member_id', 'u.account', 'u.username', 'u.first_name', 'u.language'];
        $rows = $query->select($fields)->orderByDesc('u.id')->forPage((int) ($params['page'] ?? 1), (int) ($params['limit'] ?? 20))->get();
        return ['total' => $total, 'data' => $rows->map(function ($row) { $row->id = (int) $row->id; return $row; })->all()];
    }

    public function send(array $input): array
    {
        foreach (['sender_id', 'sender_name', 'sender_display'] as $key) if (array_key_exists($key, $input)) LocalizedContent::invalid($key, '发送者由登录管理员生成');
        $params = Validator::make($input, [
            'request_id' => ['required', 'string', 'regex:/^[A-Za-z0-9][A-Za-z0-9._:-]{0,63}$/D'],
            'category' => ['sometimes', Rule::in(array_keys(self::CATEGORIES))],
            'recipient_type' => ['required', Rule::in(array_keys(self::TARGETS))],
            'recipient_ids' => ['nullable', 'array', 'max:500'], 'recipient_ids.*' => ['required', 'integer', 'min:1'],
            'send_at' => ['nullable', 'date_format:Y-m-d H:i:s'], 'translations' => ['required', 'array'],
        ])->validate();
        if (empty($params['send_at'])) $params['send_at'] = null;
        $type = $params['recipient_type']; $kind = str_ends_with($type, 'agents') ? 'agent' : 'user';
        if ($kind === 'agent') $this->requireAgentAccess();
        $ids = array_values(array_unique(array_map('intval', $params['recipient_ids'] ?? []))); sort($ids);
        if (str_starts_with($type, 'specified_') && !$ids) LocalizedContent::invalid('recipient_ids', '请选择收件人');
        if (str_starts_with($type, 'all_') && $ids) LocalizedContent::invalid('recipient_ids', '群发全部账号时请清空指定收件人');
        $translations = LocalizedContent::parse($params['translations'], 500, 5000, true);
        $actor = OperationAuditService::actor();
        if (!$actor['operator_id'] || $actor['operator_name'] === '') LocalizedContent::invalid('sender_id', '请使用后台管理员身份');
        $hash = hash('sha256', json_encode([$params['category'] ?? 'internal', $type, $ids, $params['send_at'] ?? null, $translations], JSON_THROW_ON_ERROR));

        return DB::transaction(function () use ($params, $type, $kind, $ids, $translations, $actor, $hash) {
            $now = $this->now();
            // The unique key serializes concurrent retries before any receipts or audit records are created.
            $created = DB::table('site_messages')->insertOrIgnore([
                'category' => $params['category'] ?? 'internal', 'recipient_type' => $type,
                'recipient_ids' => json_encode($ids, JSON_THROW_ON_ERROR), 'send_at' => $params['send_at'] ?? $now,
                'status' => 'scheduled', 'sender_id' => $actor['operator_id'], 'sender_name' => $actor['operator_name'],
                'request_id' => $params['request_id'], 'request_hash' => $hash, 'created_at' => $now, 'updated_at' => $now,
            ]);
            $message = DB::table('site_messages')->where('sender_id', $actor['operator_id'])->where('request_id', $params['request_id'])->lockForUpdate()->first();
            if (!$message) throw new \RuntimeException('信息创建失败');
            if ($message->request_hash !== $hash) LocalizedContent::invalid('request_id', '请求标识已用于其他内容，请重新生成');
            if (!$created) return $this->detail((int) $message->id);
            if ($ids && $this->eligible($kind)->whereIn('u.id', $ids)->count() !== count($ids)) LocalizedContent::invalid('recipient_ids', '部分账号不存在或不在可发送范围，请重新选择');
            foreach ($translations as $language => $block) DB::table('site_message_translations')->insert($block + ['message_id' => $message->id, 'language' => $language]);
            if ($message->send_at <= $now) $this->deliver((int) $message->id, $actor);
            else OperationAuditService::record('site_message', (int) $message->id, 'schedule', [], $this->snapshot((int) $message->id), $now);
            return $this->detail((int) $message->id);
        }, 3);
    }

    public function deliver(int $id, ?array $actor = null): bool
    {
        return DB::transaction(function () use ($id, $actor) {
            $message = DB::table('site_messages')->where('id', $id)->lockForUpdate()->first();
            $now = $this->now();
            if (!$message || $message->status !== 'scheduled' || $message->send_at > $now) return false;
            $before = $this->snapshot($id);
            $kind = str_ends_with($message->recipient_type, 'agents') ? 'agent' : 'user';
            $ids = json_decode($message->recipient_ids, true, 512, JSON_THROW_ON_ERROR);
            $base = $this->eligible($kind);
            if ($ids) $base->whereIn('u.id', $ids);
            $maximum = (int) (clone $base)->max('u.id');
            $base->where('u.id', '<=', $maximum);
            $targetCount = $ids ? count($ids) : (int) (clone $base)->count();
            // One INSERT SELECT chooses each recipient's language once; no per-language re-scans or partial publication.
            $source = (clone $base)->join('site_message_translations as t', function ($join) use ($id) {
                $join->on('t.language', '=', 'u.language')->where('t.message_id', '=', $id);
            })->selectRaw('? AS message_id, ? AS recipient_kind', [$id, $kind])
                ->addSelect('u.id as recipient_id', 't.language')->selectRaw('? AS created_at', [$now]);
            DB::table('site_message_recipients')->insertUsing(['message_id', 'recipient_kind', 'recipient_id', 'language', 'created_at'], $source);
            $count = DB::table('site_message_recipients')->where('message_id', $id)->count();
            DB::table('site_messages')->where('id', $id)->update(['status' => 'sent', 'sent_at' => $now,
                'recipient_count' => $count, 'skipped_count' => max(0, $targetCount - $count), 'updated_at' => $now]);
            OperationAuditService::record('site_message', $id, 'send', $before, $this->snapshot($id), $now,
                $actor ?? ['operator_id' => null, 'operator_name' => '系统']);
            SiteRealtimeService::changed('message', $id, 'send');
            return true;
        }, 3);
    }

    public function dispatchDue(): array
    {
        $result = ['sent' => 0, 'failed' => 0];
        DB::table('site_messages')->where('status', 'scheduled')->where('send_at', '<=', $this->now())->orderBy('id')->chunkById(100, function ($rows) use (&$result) {
            foreach ($rows as $row) {
                try { if ($this->deliver((int) $row->id)) ++$result['sent']; }
                catch (\Throwable $e) { ++$result['failed']; Log::error('site_message_delivery_failed', ['id' => $row->id, 'exception' => $e]); }
            }
        });
        return $result;
    }

    public function retract(int $id): array
    {
        return DB::transaction(function () use ($id) {
            $found = DB::table('site_messages')->where('id', $id)->lockForUpdate()->first();
            if (!$found) LocalizedContent::invalid('id', '信息不存在');
            $this->checkAccess($found->recipient_type);
            if ($found->status === 'retracted') return $this->detail($id);
            $before = $this->snapshot($id); $now = $this->now();
            DB::table('site_messages')->where('id', $id)->update(['status' => 'retracted', 'retracted_at' => $now, 'updated_at' => $now]);
            OperationAuditService::record('site_message', $id, 'retract', $before, $this->snapshot($id), $now);
            SiteRealtimeService::changed('message', $id, 'retract');
            return $this->detail($id);
        }, 3);
    }

    public function listing(array $input): array
    {
        $params = Validator::make($input, $this->pageRules() + [
            'category' => ['nullable', Rule::in(array_keys(self::CATEGORIES))], 'language' => ['nullable', Rule::in(array_keys(LocalizedContent::LANGUAGES))],
            'title' => ['nullable', 'string', 'max:500'],
            'start_date' => ['nullable', 'required_with:end_date', 'date_format:Y-m-d'],
            'end_date' => ['nullable', 'required_with:start_date', 'date_format:Y-m-d', 'after_or_equal:start_date'],
        ])->validate();
        $query = DB::table('site_messages as m');
        if (!self::canManageAgents()) $query->whereIn('m.recipient_type', ['all_users', 'specified_users']);
        if (!empty($params['category'])) $query->where('m.category', $params['category']);
        if (!empty($params['start_date'])) $query->where('m.send_at', '>=', $params['start_date'] . ' 00:00:00')->where('m.send_at', '<=', $params['end_date'] . ' 23:59:59');
        if (!empty($params['language']) || (isset($params['title']) && $params['title'] !== '')) {
            $query->whereExists(function ($q) use ($params) {
                $q->selectRaw('1')->from('site_message_translations as t')->whereColumn('t.message_id', 'm.id');
                if (!empty($params['language'])) $q->where('t.language', $params['language']);
                if (isset($params['title']) && $params['title'] !== '') $q->where('t.title', 'like', '%' . $params['title'] . '%');
            });
        }
        $total = (clone $query)->count();
        $rows = $query->select('m.*')->orderByDesc('m.send_at')->orderByDesc('m.id')->forPage((int) ($params['page'] ?? 1), (int) ($params['limit'] ?? 20))->get();
        $translations = [];
        if ($rows->isNotEmpty()) foreach (DB::table('site_message_translations')->whereIn('message_id', $rows->pluck('id')->all())->get(['message_id', 'language', 'title']) as $text) {
            $translations[$text->message_id][$text->language] = ['title' => $text->title];
        }
        return ['total' => $total, 'data' => $rows->map(fn ($row) => $this->dto((array) $row, false, $translations[$row->id] ?? []))->all()];
    }

    public function detail(int $id): array
    {
        $row = DB::table('site_messages')->where('id', $id)->first();
        if (!$row) LocalizedContent::invalid('id', '信息不存在');
        $this->checkAccess($row->recipient_type);
        $item = $this->dto((array) $row, true);
        $item['delivered_by_language'] = DB::table('site_message_recipients')->where('message_id', $id)->select('language')->selectRaw('COUNT(*) AS total')->groupBy('language')->pluck('total', 'language')->map(fn ($n) => (int) $n)->all();
        return $item;
    }

    public function logs(array $input): array
    {
        return LocalizedAuditService::listing('site_message', $input, function ($q) {
            $q->join('site_messages as m', 'm.id', '=', 'a.resource_id');
            if (!self::canManageAgents()) $q->whereIn('m.recipient_type', ['all_users', 'specified_users']);
        });
    }

    private function snapshot(int $id): array
    {
        $row = (array) DB::table('site_messages')->where('id', $id)->first();
        unset($row['request_hash']);
        $row['recipient_ids'] = json_decode($row['recipient_ids'], true, 512, JSON_THROW_ON_ERROR);
        $row['translations'] = $this->translations($id);
        return $row;
    }

    private function dto(array $row, bool $full, ?array $translations = null): array
    {
        $translations = $translations ?? $this->translations((int) $row['id']);
        unset($row['request_hash']);
        foreach (['id', 'sender_id', 'recipient_count', 'skipped_count'] as $key) $row[$key] = (int) $row[$key];
        $row['recipient_ids'] = json_decode($row['recipient_ids'], true, 512, JSON_THROW_ON_ERROR);
        $row['titles'] = (object) array_map(fn ($b) => $b['title'], $translations); $row['languages'] = array_keys($translations);
        $row['category_name'] = self::CATEGORIES[$row['category']]; $row['recipient_label'] = self::TARGETS[$row['recipient_type']];
        $row['status_name'] = self::STATES[$row['status']]; $row['sender_display'] = '批核者：' . $row['sender_name'];
        if ($full) $row['translations'] = $translations;
        return $row;
    }

    private function translations(int $id): array
    {
        $result = [];
        foreach (DB::table('site_message_translations')->where('message_id', $id)->orderBy('language')->get() as $row) $result[$row->language] = ['title' => $row->title, 'subtitle' => $row->subtitle, 'content' => $row->content];
        return $result;
    }

    private function eligible(string $kind)
    {
        $query = DB::table(($kind === 'agent' ? 'agents' : 'users') . ' as u');
        if ($kind === 'user') $query->where('u.from', 1)->whereNull('u.delete_time');
        return $query;
    }
    private function checkAccess(string $type): void { if (str_ends_with($type, 'agents')) $this->requireAgentAccess(); }
    private function requireAgentAccess(): void { if (!self::canManageAgents()) throw new AuthorizationException('无代理管理权限'); }
    private function optionsFor(array $values): array { $result = []; foreach ($values as $value => $label) $result[] = compact('value', 'label'); return $result; }
    private function pageRules(): array { return ['page' => ['nullable', 'integer', 'min:1'], 'limit' => ['nullable', 'integer', 'min:1', 'max:200']]; }
    private function now(): string { return Carbon::now('Asia/Shanghai')->format('Y-m-d H:i:s'); }
}