Nodarx/server
1
0
Fork 0
Code Issues 1 Pull Requests 5 Actions Projects Releases Activity

refactor(kernel): unify HTTP and CLI application lifecycle

Browse Source 
Signed-off-by: Sebastian Krupinski <krupinski01@gmail.com>
This commit is contained in:
Sebastian Krupinski
2026-07-27 00:48:11 -04:00
parent 98826143a8
commit 65c1b5fb75
67 changed files with 2737 additions and 1070 deletions
Download Patch File Download Diff File Expand all files Collapse all files
shared/lib/Event/DeferredEventProcessorInterface.php
+14
View File
@@ -0,0 +1,14 @@
<?php
declare(strict_types=1);
namespace KTXF\Event;
interface DeferredEventProcessorInterface
{
public function beginExecution(string $executionId): void;
public function processDeferred(string $executionId): DeferredProcessingResult;
public function discardDeferred(string $executionId): void;
}
shared/lib/Event/DeferredProcessingResult.php
+16
View File
@@ -0,0 +1,16 @@
<?php
declare(strict_types=1);
namespace KTXF\Event;
final readonly class DeferredProcessingResult
{
public function __construct(
public int $processed,
public int $remaining,
public bool $deadlineExceeded,
public bool $limitExceeded = false,
) {
}
}
shared/lib/Event/DeliveryMode.php
+11
View File
@@ -0,0 +1,11 @@
<?php
declare(strict_types=1);
namespace KTXF\Event;
enum DeliveryMode: string
{
case Immediate = 'immediate';
case Deferred = 'deferred';
}
shared/lib/Event/EventBus.php
-186
View File
@@ -1,186 +0,0 @@
<?php
declare(strict_types=1);
namespace KTXF\Event;
/**
* Simple event bus for decoupled pub/sub communication between services
*
* Features:
* - Priority-based listener ordering
* - Synchronous and asynchronous (deferred) event handling
* - Event propagation control
*/
class EventBus
{
/** @var array<string, array<array{callback: callable, priority: int}>> */
private array $listeners = [];
/** @var array<string, array<callable>> */
private array $asyncListeners = [];
/** @var Event[] */
private array $deferredEvents = [];
/**
* Subscribe to an event with optional priority
* Higher priority listeners are called first
*/
public function subscribe(string $eventName, callable $listener, int $priority = 0): self
{
$this->listeners[$eventName][] = [
'callback' => $listener,
'priority' => $priority,
];
// Sort by priority (higher first)
usort(
$this->listeners[$eventName],
fn($a, $b) => $b['priority'] <=> $a['priority']
);
return $this;
}
/**
* Subscribe to an event for async/deferred processing
* These handlers run at the end of the request cycle
*/
public function subscribeAsync(string $eventName, callable $listener): self
{
$this->asyncListeners[$eventName][] = $listener;
return $this;
}
/**
* Unsubscribe a listener from an event
*/
public function unsubscribe(string $eventName, callable $listener): self
{
if (isset($this->listeners[$eventName])) {
$this->listeners[$eventName] = array_filter(
$this->listeners[$eventName],
fn($item) => $item['callback'] !== $listener
);
}
if (isset($this->asyncListeners[$eventName])) {
$this->asyncListeners[$eventName] = array_filter(
$this->asyncListeners[$eventName],
fn($item) => $item !== $listener
);
}
return $this;
}
/**
* Publish an event to all subscribers
*/
public function publish(Event $event): self
{
$eventName = $event->getName();
// Execute synchronous listeners
if (isset($this->listeners[$eventName])) {
foreach ($this->listeners[$eventName] as $listenerData) {
if ($event->isPropagationStopped()) {
break;
}
try {
call_user_func($listenerData['callback'], $event);
} catch (\Throwable $e) {
// Log error but don't break the chain
error_log(sprintf(
'Event listener error for %s: %s',
$eventName,
$e->getMessage()
));
}
}
}
// Queue for async processing if there are async listeners
if (isset($this->asyncListeners[$eventName]) && !empty($this->asyncListeners[$eventName])) {
$this->deferredEvents[] = $event;
}
return $this;
}
/**
* Process deferred/async events
* Call this at the end of the request cycle
*/
public function processDeferred(): int
{
$processed = 0;
foreach ($this->deferredEvents as $event) {
$eventName = $event->getName();
if (!isset($this->asyncListeners[$eventName])) {
continue;
}
foreach ($this->asyncListeners[$eventName] as $listener) {
try {
call_user_func($listener, $event);
$processed++;
} catch (\Throwable $e) {
// Log but don't fail - these are non-critical
error_log(sprintf(
'Async event handler error for %s: %s',
$eventName,
$e->getMessage()
));
}
}
}
$this->deferredEvents = [];
return $processed;
}
/**
* Check if an event has any listeners
*/
public function hasListeners(string $eventName): bool
{
return !empty($this->listeners[$eventName]) || !empty($this->asyncListeners[$eventName]);
}
/**
* Get count of listeners for an event
*/
public function getListenerCount(string $eventName): int
{
$sync = isset($this->listeners[$eventName]) ? count($this->listeners[$eventName]) : 0;
$async = isset($this->asyncListeners[$eventName]) ? count($this->asyncListeners[$eventName]) : 0;
return $sync + $async;
}
/**
* Get count of pending deferred events
*/
public function getDeferredCount(): int
{
return count($this->deferredEvents);
}
/**
* Clear all listeners (useful for testing)
*/
public function clear(): self
{
$this->listeners = [];
$this->asyncListeners = [];
$this->deferredEvents = [];
return $this;
}
}
shared/lib/Event/EventDispatcher.php
+124
View File
@@ -0,0 +1,124 @@
<?php
declare(strict_types=1);
namespace KTXF\Event;
use Psr\Container\ContainerInterface;
use Psr\Log\LoggerInterface;
final class EventDispatcher implements EventDispatcherInterface, DeferredEventProcessorInterface
{
/** @var array<string, list<Event>> */
private array $deferred = [];
private ?string $activeExecution = null;
private int $dispatchDepth = 0;
public function __construct(
private readonly EventListenerRegistry $registry,
private readonly ContainerInterface $container,
private readonly LoggerInterface $logger,
) {
}
public function dispatch(Event $event): void
{
if (++$this->dispatchDepth > 32) {
--$this->dispatchDepth;
throw new \RuntimeException('Event dispatch recursion limit exceeded.');
}
try {
$this->invoke($event, DeliveryMode::Immediate);
if ($this->registry->listeners($event->getName(), DeliveryMode::Deferred) !== []) {
if ($this->activeExecution === null) {
throw new \LogicException('Deferred events require an active execution scope.');
}
$this->deferred[$this->activeExecution][] = $event;
}
} finally {
--$this->dispatchDepth;
}
}
public function beginExecution(string $executionId): void
{
if ($this->activeExecution !== null) {
throw new \LogicException('An event execution scope is already active.');
}
$this->activeExecution = $executionId;
$this->deferred[$executionId] = [];
}
public function processDeferred(string $executionId): DeferredProcessingResult
{
if ($this->activeExecution !== $executionId) {
throw new \LogicException('Cannot process deferred events for an inactive execution.');
}
$processed = 0;
$deadline = microtime(true) + 1.0;
$deadlineExceeded = false;
$limitExceeded = false;
while (($event = array_shift($this->deferred[$executionId])) !== null) {
if ($processed >= 1000) {
$limitExceeded = true;
array_unshift($this->deferred[$executionId], $event);
break;
}
if (microtime(true) >= $deadline) {
$deadlineExceeded = true;
array_unshift($this->deferred[$executionId], $event);
break;
}
$processed += $this->invoke($event, DeliveryMode::Deferred);
}
$remaining = count($this->deferred[$executionId]);
unset($this->deferred[$executionId]);
$this->activeExecution = null;
return new DeferredProcessingResult(
$processed,
$remaining,
$deadlineExceeded,
$limitExceeded,
);
}
public function discardDeferred(string $executionId): void
{
unset($this->deferred[$executionId]);
if ($this->activeExecution === $executionId) {
$this->activeExecution = null;
}
}
private function invoke(Event $event, DeliveryMode $delivery): int
{
$processed = 0;
foreach ($this->registry->listeners($event->getName(), $delivery) as $listener) {
if ($event->isPropagationStopped()) {
break;
}
try {
$service = $this->container->get($listener->service);
$service->{$listener->method}($event);
$processed++;
} catch (\Throwable $error) {
$this->logger->error('Event listener failed.', [
'event' => $event->getName(),
'module' => $listener->module,
'listener' => $listener->service . '::' . $listener->method,
'exception' => $error,
]);
if ($listener->failurePolicy === FailurePolicy::Propagate) {
throw $error;
}
}
}
return $processed;
}
}
shared/lib/Event/EventDispatcherInterface.php
+10
View File
@@ -0,0 +1,10 @@
<?php
declare(strict_types=1);
namespace KTXF\Event;
interface EventDispatcherInterface
{
public function dispatch(Event $event): void;
}
shared/lib/Event/EventListenerDefinition.php
+22
View File
@@ -0,0 +1,22 @@
<?php
declare(strict_types=1);
namespace KTXF\Event;
final readonly class EventListenerDefinition
{
/**
* @param class-string $service
*/
public function __construct(
public string $module,
public string $event,
public string $service,
public string $method,
public DeliveryMode $delivery,
public int $priority,
public FailurePolicy $failurePolicy,
) {
}
}
shared/lib/Event/EventListenerRegistry.php
+104
View File
@@ -0,0 +1,104 @@
<?php
declare(strict_types=1);
namespace KTXF\Event;
use Psr\Container\ContainerInterface;
final class EventListenerRegistry
{
/** @var array<string, list<EventListenerDefinition>> */
private array $listeners = [];
/** @var array<string, true> */
private array $registrationIds = [];
private bool $frozen = false;
/**
* @param class-string $service
*/
public function listen(
string $module,
string $event,
string $service,
string $method,
DeliveryMode $delivery = DeliveryMode::Immediate,
int $priority = 0,
FailurePolicy $failurePolicy = FailurePolicy::Continue,
): void {
if ($this->frozen) {
throw new \LogicException('The event listener registry is frozen.');
}
if ($module === '' || $event === '' || $service === '' || $method === '') {
throw new \InvalidArgumentException('Event listener registrations require module, event, service, and method.');
}
if (!class_exists($service) || !method_exists($service, $method)) {
throw new \InvalidArgumentException("Invalid event listener {$service}::{$method}.");
}
if ($priority < -10000 || $priority > 10000) {
throw new \InvalidArgumentException('Event listener priority must be between -10000 and 10000.');
}
$id = implode('|', [$module, $event, $service, $method, $delivery->value]);
if (isset($this->registrationIds[$id])) {
throw new \LogicException("Duplicate event listener registration: {$id}.");
}
$this->registrationIds[$id] = true;
$this->listeners[$event][] = new EventListenerDefinition(
$module,
$event,
$service,
$method,
$delivery,
$priority,
$failurePolicy,
);
}
public function freeze(?ContainerInterface $container = null): void
{
foreach ($this->listeners as &$listeners) {
if ($container !== null) {
foreach ($listeners as $listener) {
if (!$container->has($listener->service)) {
throw new \LogicException(
"Event listener service is not resolvable: {$listener->service}.",
);
}
}
}
usort(
$listeners,
static fn(EventListenerDefinition $a, EventListenerDefinition $b): int =>
$b->priority <=> $a->priority,
);
}
unset($listeners);
$this->frozen = true;
}
/**
* @return list<EventListenerDefinition>
*/
public function listeners(string $event, DeliveryMode $delivery): array
{
return array_values(array_filter(
$this->listeners[$event] ?? [],
static fn(EventListenerDefinition $listener): bool => $listener->delivery === $delivery,
));
}
public function frozen(): bool
{
return $this->frozen;
}
/**
* @return list<EventListenerDefinition>
*/
public function definitions(): array
{
return array_merge(...array_values($this->listeners));
}
}
shared/lib/Event/FailurePolicy.php
+11
View File
@@ -0,0 +1,11 @@
<?php
declare(strict_types=1);
namespace KTXF\Event;
enum FailurePolicy: string
{
case Continue = 'continue';
case Propagate = 'propagate';
}