kiri-http-server/Abstracts/FileWatcher.php

<?php

namespace Kiri\Server\Abstracts;

use Kiri\Di\HotReloadState;
use Kiri\Di\Inject\Container;
use Kiri\Error\StdoutLogger;
use Kiri\Events\EventProvider;
use Kiri\Router\Router;
use Kiri\Server\Events\OnWorkerStart;
use Kiri\Server\Processes\AbstractProcess;
use Kiri\Server\Processes\OnProcessInterface;
use Psr\Log\LoggerInterface;
use Swoole\Coroutine;
use Swoole\Event;
use Swoole\Process;
use Swoole\Timer;
use const SIGKILL;
use const SIGTERM;
use const SIGUSR1;

class FileWatcher extends AbstractProcess implements OnProcessInterface
{
	use ReloadWorkers;

	protected mixed $pipe;

	#[Container(LoggerInterface::class)]
	public StdoutLogger|LoggerInterface $logger;

	protected bool $enable_coroutine = false;

	protected array $watches = [];

	protected bool $enable_queue = false;

	protected bool $reloading = false;

	private array  $watchPaths      = [];
	private array  $excludePatterns = [];
	private array  $extensions      = ['php'];
	private string $strategy;
	private bool   $running         = false;
	private bool   $scanning        = false;

	private mixed $inotifyFd           = null;
	private array $inotifyWatchMap     = [];
	private mixed $fswatchProcess      = null;
	private array $fswatchPipes        = [];
	private int   $pollInterval        = 2;
	private array $fileSnapshot        = [];
	private int   $pollTimer           = -1;
	private int   $debounceTimer       = -1;
	private int   $debounceMs          = 200;
	private int   $minReloadIntervalMs = 1000;
	private int   $lastReloadAtMs      = 0;

	private const string STRATEGY_INOTIFY = 'inotify';
	private const string STRATEGY_FSWATCH = 'fswatch';
	private const string STRATEGY_POLL    = 'poll';

	public function __construct()
	{
		$this->watchPaths      = config('servers.reload.listen', []);
		$this->excludePatterns = config('servers.reload.scan.skip_patterns', []);
		$this->extensions      = config('servers.reload.scan.extensions', ['php']);

		di(EventProvider::class)->on(OnWorkerStart::class, [di(Router::class), 'scan_build_route']);

		$this->debounceMs          = (int)config('servers.reload.debounce_ms', $this->debounceMs);
		$this->minReloadIntervalMs = (int)config('servers.reload.min_interval_ms', $this->minReloadIntervalMs);
		$this->strategy            = $this->detectBestStrategy();
	}

	public function getName(): string
	{
		return 'hotReload';
	}

	public function onSigterm(): void
	{
		$this->stop();
	}

	private function detectBestStrategy(): string
	{
		$forced = config('servers.reload.scan.strategy', 'auto');
		if (in_array($forced, [self::STRATEGY_INOTIFY, self::STRATEGY_FSWATCH, self::STRATEGY_POLL], true)) {
			return $forced;
		}

		if ($this->isMountedWindowsPath()) {
			return $this->isFswatchAvailable() ? self::STRATEGY_FSWATCH : self::STRATEGY_POLL;
		}

		if (extension_loaded('inotify')) {
			return self::STRATEGY_INOTIFY;
		}

		if ($this->isFswatchAvailable()) {
			return self::STRATEGY_FSWATCH;
		}

		return self::STRATEGY_POLL;
	}

	private function isMountedWindowsPath(): bool
	{
		foreach ($this->watchPaths as $path) {
			$real = realpath($path) ?: (string)$path;
			$real = str_replace('\\', '/', $real);
			if (str_starts_with($real, '/mnt/')) {
				return true;
			}
		}

		return false;
	}

	private function isFswatchAvailable(): bool
	{
		$output    = [];
		$returnVar = 0;
		exec('which fswatch 2>/dev/null', $output, $returnVar);
		return $returnVar === 0 && !empty($output[0]);
	}

	public function setDebounce(int $milliseconds): self
	{
		$this->debounceMs = $milliseconds;
		return $this;
	}

	public function setPollInterval(int $seconds): self
	{
		$this->pollInterval = $seconds;
		return $this;
	}

	public function process(Process|null $process): void
	{
		if (class_exists('\\Swoole\\Coroutine')) {
			Coroutine::set(['enable_deadlock_check' => false]);
		}

		if ($this->running) {
			return;
		}

		$this->running = true;

		switch ($this->strategy) {
			case self::STRATEGY_INOTIFY:
				$this->startInotify();
				break;
			case self::STRATEGY_FSWATCH:
				$this->startFswatch();
				break;
			case self::STRATEGY_POLL:
				$this->startPolling();
				break;
		}

		Event::wait();
	}

	public function stop(): void
	{
		$this->running = false;

		if ($this->inotifyFd && is_resource($this->inotifyFd)) {
			@Event::del($this->inotifyFd);
			@fclose($this->inotifyFd);
			$this->inotifyFd = null;
		}

		foreach ($this->fswatchPipes as $pipe) {
			if (is_resource($pipe)) {
				@Event::del($pipe);
				@fclose($pipe);
			}
		}
		$this->fswatchPipes = [];
		$this->stopFswatchProcess();

		if ($this->pollTimer > 0) {
			Timer::clear($this->pollTimer);
			$this->pollTimer = -1;
		}

		if ($this->debounceTimer > 0) {
			Timer::clear($this->debounceTimer);
			$this->debounceTimer = -1;
		}

		@Event::exit();
	}

	private function stopFswatchProcess(): void
	{
		if (!$this->fswatchProcess || !is_resource($this->fswatchProcess)) {
			$this->fswatchProcess = null;
			return;
		}

		$status = @proc_get_status($this->fswatchProcess);
		if (($status['running'] ?? false) === true) {
			@proc_terminate($this->fswatchProcess, SIGTERM);
			for ($i = 0; $i < 10; $i++) {
				usleep(100000);
				$status = @proc_get_status($this->fswatchProcess);
				if (($status['running'] ?? false) !== true) {
					break;
				}
			}
		}

		$status = @proc_get_status($this->fswatchProcess);
		if (($status['running'] ?? false) === true) {
			@proc_terminate($this->fswatchProcess, SIGKILL);
			for ($i = 0; $i < 5; $i++) {
				usleep(100000);
				$status = @proc_get_status($this->fswatchProcess);
				if (($status['running'] ?? false) !== true) {
					break;
				}
			}
		}

		$status = @proc_get_status($this->fswatchProcess);
		if (($status['running'] ?? false) !== true) {
			@proc_close($this->fswatchProcess);
		}
		$this->fswatchProcess = null;
	}

	private function isExcluded(string $path): bool
	{
		foreach ($this->excludePatterns as $pattern) {
			if (strpos($path, $pattern) !== false) {
				return true;
			}
		}

		return false;
	}

	private function hasValidExtension(string $path): bool
	{
		if (empty($this->extensions)) {
			return true;
		}

		$ext = pathinfo($path, PATHINFO_EXTENSION);
		return in_array($ext, $this->extensions, true);
	}

	private function triggerCallback(array $changedFiles): void
	{
		$changedFiles = array_filter($changedFiles, function ($file) {
				return !$this->isExcluded($file) && $this->hasValidExtension($file);
			})
		                |> array_unique(...)
		                |> array_values(...);

		if (empty($changedFiles)) {
			return;
		}

		if ($this->debounceTimer > 0) {
			Timer::clear($this->debounceTimer);
		}

		$now     = intdiv(hrtime(true), 1000000);
		$elapsed = $this->lastReloadAtMs > 0 ? $now - $this->lastReloadAtMs : $this->minReloadIntervalMs;
		$delay   = max($this->debounceMs, $this->minReloadIntervalMs - $elapsed);

		$this->debounceTimer = Timer::after($delay, function () use ($changedFiles) {
			$this->debounceTimer = -1;
			$this->reload($changedFiles);
		});
	}

	private function reload(array $changedFiles): void
	{
		if ($this->reloading) {
			return;
		}

		$this->reloading = true;

		try {
			$this->lastReloadAtMs = intdiv(hrtime(true), 1000000);
			$this->invalidateChangedFiles($changedFiles);
			$preview              = implode(', ', array_slice($changedFiles, 0, 3));
			if (count($changedFiles) > 3) {
				$preview .= ' ...';
			}

			di(HotReloadState::class)->store($changedFiles);
			di(StdoutLogger::class)->println('detected file changes, reloading server: ' . $preview);

			if (!$this->reloadWorkers()) {
				di(StdoutLogger::class)->println('reload server failed: master pid not found or signal failed');
			}
		} finally {
			$this->reloading = false;
		}
	}

	private function invalidateChangedFiles(array $changedFiles): void
	{
		if (!function_exists('opcache_invalidate')) {
			return;
		}

		foreach ($changedFiles as $file) {
			if (!is_string($file) || !str_ends_with($file, '.php')) {
				continue;
			}

			$path = realpath($file) ?: $file;
			if (is_file($path)) {
				@opcache_invalidate($path, true);
			}
		}
	}
	private function isMasterProcessCommand(string $command): bool
	{
		$prefix = '[' . config('site.id', 'system-service') . '].Master[';
		return str_contains($command, $prefix);
	}

	private function isSameAppProcess(int $pid): bool
	{
		$cwd = @readlink('/proc/' . $pid . '/cwd');
		if ($cwd === false || $cwd === '') {
			return false;
		}

		return $this->normalizePath($cwd) === $this->normalizePath(APP_PATH);
	}

	private function normalizePath(string $path): string
	{
		$real = realpath($path) ?: $path;
		return rtrim(str_replace('\\', '/', $real), '/');
	}


	private function isProcessAlive(int $pid): bool
	{
		return $pid > 1 && @Process::kill($pid, 0);
	}

	private function startInotify(): void
	{
		$this->inotifyFd = inotify_init();
		stream_set_blocking($this->inotifyFd, false);

		foreach ($this->watchPaths as $path) {
			$this->addInotifyWatchRecursive($path);
		}

		Event::add($this->inotifyFd, function ($fd) {
			if (!$this->running) {
				return;
			}

			$events = inotify_read($fd);
			if ($events === false) {
				return;
			}

			$changedFiles = [];
			foreach ($events as $event) {
				$wd       = $event['wd'];
				$dir      = $this->inotifyWatchMap[$wd] ?? '';
				$name     = $event['name'] ?? '';
				$filePath = $name === '' ? $dir : $dir . '/' . $name;

				if (($event['mask'] & IN_CREATE) && $filePath !== '' && is_dir($filePath) && !$this->isExcluded($filePath)) {
					$this->addInotifyWatchRecursive($filePath);
				}

				if ($filePath !== '') {
					$changedFiles[] = $filePath;
				}
			}

			$this->triggerCallback($changedFiles);
		});
	}

	private function addInotifyWatchRecursive(string $path): void
	{
		if ($this->isExcluded($path)) {
			return;
		}

		if (is_file($path)) {
			if ($this->hasValidExtension($path)) {
				$wd                         = inotify_add_watch($this->inotifyFd, $path, IN_MODIFY | IN_CREATE | IN_DELETE | IN_MOVE | IN_CLOSE_WRITE);
				$this->inotifyWatchMap[$wd] = dirname($path);
			}
			return;
		}

		if (!is_dir($path)) {
			return;
		}

		$wd                         = inotify_add_watch($this->inotifyFd, $path, IN_MODIFY | IN_CREATE | IN_DELETE | IN_MOVE | IN_CLOSE_WRITE);
		$this->inotifyWatchMap[$wd] = $path;

		$items = scandir($path);
		foreach ($items as $item) {
			if ($item === '.' || $item === '..') {
				continue;
			}

			$full = $path . '/' . $item;
			if (is_dir($full)) {
				$this->addInotifyWatchRecursive($full);
			}
		}
	}

	private function startFswatch(): void
	{
		$descriptorspec = [
			0 => ['pipe', 'r'],
			1 => ['pipe', 'w'],
			2 => ['pipe', 'w'],
		];

		$excludeArgs = [];
		foreach ($this->excludePatterns as $pattern) {
			$excludeArgs[] = '--exclude';
			$excludeArgs[] = '.*' . preg_quote($pattern, '/') . '.*';
		}

		$filterArgs = [];
		foreach ($this->extensions as $ext) {
			$filterArgs[] = '--include';
			$filterArgs[] = '\.' . $ext . '$';
		}

		$cmd = 'fswatch -0 -r ' . implode(' ', array_merge($excludeArgs, $filterArgs)) . ' ' . implode(' ', array_map('escapeshellarg', $this->watchPaths));

		$this->fswatchProcess = proc_open($cmd, $descriptorspec, $this->fswatchPipes);
		if (!is_resource($this->fswatchProcess)) {
			throw new \RuntimeException('Failed to start fswatch process');
		}

		stream_set_blocking($this->fswatchPipes[1], false);

		Event::add($this->fswatchPipes[1], function ($pipe) {
			if (!$this->running) {
				return;
			}

			$data = fread($pipe, 8192);
			if ($data === false || $data === '') {
				return;
			}

			$files        = explode("\0", trim($data, "\0"));
			$changedFiles = array_values(array_filter($files, function ($file) {
				return $file !== '';
			}));

			$this->triggerCallback($changedFiles);
		});
	}

	private function startPolling(): void
	{
		$this->buildFileSnapshot();

		$intervalMs      = $this->pollInterval * 1000;
		$this->pollTimer = Timer::tick($intervalMs, function () {
			if (!$this->running || $this->scanning) {
				return;
			}

			$this->scanning = true;
			try {
				$newSnapshot  = [];
				$changedFiles = $this->compareSnapshots($newSnapshot);
				if (!$this->running) {
					return;
				}
				if (!empty($changedFiles)) {
					$this->triggerCallback($changedFiles);
				}

				$this->fileSnapshot = $newSnapshot;
			} finally {
				$this->scanning = false;
			}
		});
	}

	private function buildFileSnapshot(): void
	{
		$this->fileSnapshot = [];
		foreach ($this->watchPaths as $path) {
			$this->scanDirectory($path, $this->fileSnapshot);
		}
	}

	private function scanDirectory(string $path, array &$snapshot): void
	{
		if ($this->isExcluded($path)) {
			return;
		}

		if (is_file($path)) {
			if ($this->hasValidExtension($path)) {
				$snapshot[$path] = filemtime($path);
			}
			return;
		}

		if (!is_dir($path)) {
			return;
		}

		$items = scandir($path);
		foreach ($items as $item) {
			if ($item === '.' || $item === '..') {
				continue;
			}

			$full = $path . '/' . $item;
			$this->scanDirectory($full, $snapshot);
		}
	}

	private function compareSnapshots(array &$newSnapshot): array
	{
		$changed = [];

		foreach ($this->watchPaths as $path) {
			$this->scanDirectory($path, $newSnapshot);
		}

		foreach ($newSnapshot as $file => $mtime) {
			if (!isset($this->fileSnapshot[$file])) {
				$changed[] = $file;
			} elseif ($this->fileSnapshot[$file] !== $mtime) {
				$changed[] = $file;
			}
		}

		foreach ($this->fileSnapshot as $file => $mtime) {
			if (!isset($newSnapshot[$file])) {
				$changed[] = $file;
			}
		}

		return $changed;
	}
}