<?php

declare(strict_types=1);

require __DIR__ . '/vendor/autoload.php';

use Workerman\Timer;
use Workerman\Worker;

$wsWorker = new Worker('websocket://0.0.0.0:2346');
$wsWorker->count = 1;

$connections = [];
$queueFile = __DIR__ . '/storage/ws/events.log';
$offsetFile = __DIR__ . '/storage/ws/events.offset';

$wsWorker->onConnect = function ($connection) use (&$connections) {
    $connections[$connection->id] = $connection;
};

$wsWorker->onClose = function ($connection) use (&$connections) {
    unset($connections[$connection->id]);
};

$wsWorker->onWorkerStart = function () use (&$connections, $queueFile, $offsetFile) {
    if (!is_dir(dirname($queueFile))) {
        mkdir(dirname($queueFile), 0775, true);
    }

    if (!file_exists($queueFile)) {
        touch($queueFile);
    }

    if (!file_exists($offsetFile)) {
        file_put_contents($offsetFile, '0');
    }

    Timer::add(1, function () use (&$connections, $queueFile, $offsetFile) {
        clearstatcache(true, $queueFile);

        $size = filesize($queueFile);
        $offset = (int) @file_get_contents($offsetFile);

        if ($size === false || $size <= $offset) {
            return;
        }

        $fp = fopen($queueFile, 'r');
        if (!$fp) {
            return;
        }

        fseek($fp, $offset);

        while (($line = fgets($fp)) !== false) {
            $line = trim($line);
            if ($line === '') {
                continue;
            }

            foreach ($connections as $connection) {
                $connection->send($line);
            }
        }

        $newOffset = ftell($fp);
        fclose($fp);

        file_put_contents($offsetFile, (string) $newOffset, LOCK_EX);
    });
};

Worker::runAll();