<?php
/** php scripts/chat-worker.php start [-d]；通过 HTTPS 反向代理暴露 WSS。 */
require dirname(__DIR__).'/vendor/autoload.php';
require __DIR__.'/chat-realtime/vendor/autoload.php';

use Workerman\Worker;
use Workerman\Lib\Timer;
use app\mobile\service\ChatService;
use think\facade\Db;

if (!isset($app)) { $app = new \think\App(dirname(__DIR__).'/'); $app->initialize(); }
$chatListen = $app->config->get('chat.socket_listen');
if (!is_file(dirname(__DIR__).'/config/chat.php') || !is_string($chatListen) || !preg_match('#^websocket://[^/]+:[0-9]+$#', $chatListen)) {
    fwrite(STDERR, "Chat startup failed: missing config/chat.php or invalid CHAT.SOCKET_LISTEN. Run php scripts/chat-check.php\n");
    exit(1);
}
if (!class_exists(ChatService::class)) {
    fwrite(STDERR, "Chat startup failed: missing app/mobile/service/ChatService.php\n");
    exit(1);
}
$chatReportFailure = function ($area, $error) {
    $driver = 0;
    if ($error instanceof \think\db\exception\PDOException) {
        $info = $error->getData(); $driver = (int)($info['PDO Error Info']['Driver Error Code'] ?? 0);
    } elseif ($error instanceof \PDOException) $driver = (int)($error->errorInfo[1] ?? 0);
    $hints = [1146=>'Missing database tables; run scripts/chat-check.php',1054=>'Missing database columns; run scripts/chat-check.php',1045=>'Database authentication failed',2002=>'Cannot connect to MySQL',2003=>'Cannot connect to MySQL'];
    $hint = $hints[$driver] ?? ($error instanceof \RedisException ? 'Redis connection unavailable' : 'Run scripts/chat-check.php');
    error_log('chat '.$area.' failed: '.$hint.'; type='.get_class($error).'; driver='.$driver.'; origin='.basename($error->getFile()).':'.$error->getLine());
};
Worker::$pidFile = dirname(__DIR__).'/runtime/chat-worker.pid';
Worker::$logFile = dirname(__DIR__).'/runtime/chat-worker.log';
Worker::$stdoutFile = dirname(__DIR__).'/runtime/chat-worker-stdout.log';
$worker = new Worker($chatListen);
$worker->name = 'oms-chat';
$worker->count = 1;
// ponytail: 单个进程持有连接，Redis 每 200ms 汇总通知；跨实例时各自读取同一失效集合。
$worker->onWorkerStart = function ($worker) use ($app, $chatReportFailure) {
    $service = new ChatService($app); $lastPoll = ChatService::now();
    echo 'chat worker ready; pid='.getmypid().'; namespace='.$service->key('diagnostic').PHP_EOL;
    Timer::add(0.2, function () use ($worker, $service, &$lastPoll, $chatReportFailure) {
        try {
            $until = ChatService::now();
            $rooms = $service->redis()->zRangeByScore($service->key('events'), $lastPoll, $until);
            if (!$rooms) { $lastPoll = $until; return; }
            $active=[];foreach($worker->connections as $connection)if(isset($connection->chatRoom))$active[$connection->chatRoom]=true;
            $rooms=array_values(array_filter($rooms,function($id)use($active){return isset($active[$id]);}));
            $changed = array_fill_keys($rooms, true);
            $joined = [];
            foreach ($rooms as $roomId) $joined[$roomId] = array_fill_keys(Db::name('chat_member')->where(['room_id'=>$roomId,'status'=>'joined'])->column('user_id'), true);
            foreach ($worker->connections as $connection) if (isset($connection->chatUser, $connection->chatRoom) && isset($changed[$connection->chatRoom])) {
                if (!isset($joined[$connection->chatRoom][$connection->chatUser])) { $connection->send(json_encode(['type'=>'access-revoked'])); $connection->close(); }
                else $connection->send(json_encode(['type'=>'sync','roomId'=>$connection->chatRoom]));
            }
            $lastPoll = $until;
        } catch (\Throwable $error) { $chatReportFailure('event synchronization', $error); }
    });
    Timer::add(30, function () use ($worker, $service) {
        foreach ($worker->connections as $connection) {
            if (!isset($connection->chatUser) || time() - $connection->chatSeen > 60) { $connection->close(); continue; }
            try { $service->user($connection->chatUser); }
            catch (\Throwable $error) { $connection->close(); }
        }
    });
    Timer::add(30, function () use ($service, $chatReportFailure) {
        try { $service->maintain(false); }
        catch (\Throwable $error) { $chatReportFailure('expiration maintenance', $error); }
    });
    Timer::add(900, function () use ($service, $chatReportFailure) {
        try { $service->maintain(); }
        catch (\Throwable $error) { $chatReportFailure('history cleanup', $error); }
    });
};
$worker->onConnect = function ($connection) use ($app, $chatReportFailure) {
    $connection->chatSeen = time();
    $connection->maxPackageSize = 16384;
    $connection->maxSendBufferSize = 65536;
    $connection->onWebSocketConnect = function ($connection, $httpBuffer) use ($app, $chatReportFailure) {
        try {
            $ticket = $_GET['ticket'] ?? '';
            if (!preg_match('/^[a-f0-9]{64}$/', $ticket)) { $connection->close(); return; }
            $service = new ChatService($app);
            // Redis GET + DEL 原子消费一次性握手票据，避免重放。
            $raw = $service->redis()->eval("local v=redis.call('GET',KEYS[1]); if v then redis.call('DEL',KEYS[1]) end; return v", [$service->key('socket:'.hash('sha256',$ticket))], 1);
            $data = json_decode((string)$raw,true);
            if (!$data) { error_log('chat handshake rejected: ticket missing or expired; check worker Redis/database namespace'); $connection->close(); return; }
            $service->user($data['userId']);
            $member = Db::name('chat_member')->where(['room_id'=>$data['roomId'],'user_id'=>$data['userId'],'status'=>'joined'])->find();
            if (!$member) { error_log('chat handshake rejected: room membership unavailable'); $connection->close(); return; }
            $connection->chatUser = (int)$data['userId']; $connection->chatRoom = (int)$data['roomId'];
            $connection->send(json_encode(['type'=>'ready','sendViaSocket'=>true]));
        } catch (\Throwable $error) { $chatReportFailure('handshake',$error); $connection->close(); }
    };
};
$worker->onMessage = function ($connection, $body) use ($app, $chatReportFailure) {
    if (!isset($connection->chatUser)) { $connection->close(); return; }
    $connection->chatSeen = time();
    $data = json_decode($body,true);
    if (($data['type'] ?? '') === 'ping') $connection->send(json_encode(['type'=>'pong']));
    if (($data['type'] ?? '') !== 'send') return;
    $payload = $data['data'] ?? [];
    if (!is_array($payload) || !is_string($payload['clientId'] ?? null) || strlen($payload['clientId']) > 80) { $connection->close(); return; }
    $result = ['type'=>'send-result','clientId'=>$payload['clientId']];
    try {
        $service = new ChatService($app);
        $user = $service->authenticate($data['token'] ?? '');
        if ($user['id'] !== $connection->chatUser || (string)($payload['roomId'] ?? '') !== (string)$connection->chatRoom) {
            throw new \think\exception\HttpResponseException(json(['code'=>1,'message'=>'实时连接身份或房间不匹配'],403));
        }
        $result += ['code'=>0,'data'=>$service->perform('send',$user,$payload)];
    } catch (\think\exception\HttpResponseException $error) {
        $response = $error->getResponse();
        $result += ['code'=>1,'status'=>$response->getCode(),'message'=>$response->getData()['message'] ?? '消息发送失败'];
    } catch (\Throwable $error) {
        $chatReportFailure('message send',$error);
        $result += ['code'=>1,'status'=>503,'message'=>'消息保存失败，请稍后重试'];
    }
    $connection->send(json_encode($result));
};
Worker::runAll();