diff --git a/docs/package-extension-seams-plan.md b/docs/package-extension-seams-plan.md index f53254a..7d888c6 100644 --- a/docs/package-extension-seams-plan.md +++ b/docs/package-extension-seams-plan.md @@ -6,7 +6,7 @@ One important part of this plan has now started landing: - constructor parameter attributes can provide their own dependency value through a generic `resolveParameterValue()` seam - `core` no longer hardcodes `#[InjectRepository]` -- queue injection is now using the same seam too +- queue injection now uses the registered resolver seam and the application `QueueFactory` - `core` now also exposes `ParameterResolverInterface` - modules can now configure injector extensions before provider resolution begins - `OrmModule` uses that module-level seam to register repository resolution without teaching `core` about ORM classes @@ -22,7 +22,8 @@ Today, some package behavior leaks into `core`. The clearest example is constructor parameter resolution: - `orm` currently depends on `core` knowing about `#[InjectRepository]` -- queue support currently depends on `core` knowing about `#[InjectQueue]` + +Queue injection is a framework-owned facility and now resolves its metadata through the same generic resolver registry used by package extensions. That works in the short term, but it creates a scaling problem: @@ -188,7 +189,7 @@ The context would carry things like: ### How packages would use it - `orm` ships a resolver for `#[InjectRepository]` -- queue support ships a resolver for `#[InjectQueue]` +- core queue support registers a resolver for `#[InjectQueue]` - future packages can ship resolvers for their own attributes `core` only loops through registered resolvers. It does not need to recognize each package decorator by name. @@ -325,12 +326,12 @@ Add to `core`: Keep existing hardcoded branches temporarily so current apps do not break. -### Phase 3: move package logic out of `core` +### Phase 3: move specialized logic behind resolvers Move: - `InjectRepository` handling into `orm` -- `InjectQueue` handling into queue support +- `InjectQueue` handling into the dedicated core queue resolver and `QueueFactory` (complete) This should happen behind compatibility shims first. diff --git a/docs/queues-and-background-jobs.md b/docs/queues-and-background-jobs.md index 10e38e2..030644a 100644 --- a/docs/queues-and-background-jobs.md +++ b/docs/queues-and-background-jobs.md @@ -163,6 +163,30 @@ That is a very Assegai-style pattern: - the service stays injectable - the queue connection is resolved through attributes and configuration +Queue resolution does not contact the broker. Core keeps one configured queue instance per named connection in the application graph, while the transport opens its connection only when `add()`, `process()`, or another broker operation first needs it. An unavailable queue backend therefore does not prevent the HTTP application from booting; the operation that needs the backend reports the connection failure. + +When an application needs to choose a connection programmatically, inject `QueueFactory`: + +```php +queues->connection('rabbitmq.notifications')->add($job); + } +} +``` + ## Use a controller to hand off work quickly ```php @@ -290,21 +314,11 @@ If you do not want to use the CLI worker yet, you can still run a simple PHP scr declare(strict_types=1); -use Assegai\Common\Interfaces\Queues\QueueInterface; +use Assegai\Core\Queues\QueueFactory; require __DIR__ . '/../vendor/autoload.php'; -$driverClass = config('queues.drivers.beanstalk'); -$queueConfig = config('queues.connections.beanstalk.notifications'); - -if (!is_string($driverClass) || !class_exists($driverClass) || !is_array($queueConfig)) { - throw new RuntimeException('Queue configuration for beanstalk.notifications is missing.'); -} - -$queueConfig['name'] ??= 'notifications'; - -/** @var QueueInterface $queue */ -$queue = $driverClass::create($queueConfig); +$queue = (new QueueFactory())->connection('beanstalk.notifications'); while (true) { $result = $queue->process(function (object $job): void { diff --git a/docs/releases/0.10.1-release-notes.md b/docs/releases/0.10.1-release-notes.md new file mode 100644 index 0000000..960bd7b --- /dev/null +++ b/docs/releases/0.10.1-release-notes.md @@ -0,0 +1,24 @@ +# AssegaiPHP Core 0.10.1 + +Core 0.10.1 decouples application bootstrap from queue backend availability. + +## What changed + +- adds an injectable `QueueFactory` for resolving configured `driver.connection` queue names programmatically +- keeps one queue instance per named connection for the lifetime of the application graph +- resolves `#[InjectQueue]` through the container's parameter-resolver seam instead of performing work inside the attribute constructor +- keeps queue attributes as metadata and delegates driver validation and creation to the named factory +- retains the application queue factory across request-scope refreshes in reusable runtimes +- documents that queue drivers must defer broker I/O until an operation needs the transport + +## Compatibility + +- requires `assegaiphp/common:^0.10.0` +- lazy broker availability requires `assegaiphp/rabbitmq:^1.1.2` or `assegaiphp/beanstalkd:^1.1.2` +- existing `#[InjectQueue('driver.connection')] QueueInterface` constructor injection remains source compatible + +## Upgrade notes + +Applications can continue using `#[InjectQueue]` without changes. Code that manually reads `queues.php` and calls a driver's static `create()` method can instead inject `QueueFactory` and call `connection()`. + +Queue connection failures now occur at the first broker-backed operation rather than while the application graph is constructed. Handle enqueue failures at the domain boundary where the application can decide whether to return an error, persist an outbox record, or apply another explicit fallback. diff --git a/phpstan-baseline.neon b/phpstan-baseline.neon index 74d614d..100a9aa 100644 --- a/phpstan-baseline.neon +++ b/phpstan-baseline.neon @@ -1222,16 +1222,6 @@ parameters: count: 1 path: src/Pipes/ValidationPipe.php - - - message: "#^Method Assegai\\\\Core\\\\Queues\\\\Attributes\\\\InjectQueue\\:\\:resolveParameterValue\\(\\) return type with generic interface Assegai\\\\Common\\\\Interfaces\\\\Queues\\\\QueueInterface does not specify its types\\: T$#" - count: 1 - path: src/Queues/Attributes/InjectQueue.php - - - - message: "#^Property Assegai\\\\Core\\\\Queues\\\\Attributes\\\\InjectQueue\\:\\:\\$queue with generic interface Assegai\\\\Common\\\\Interfaces\\\\Queues\\\\QueueInterface does not specify its types\\: T$#" - count: 1 - path: src/Queues/Attributes/InjectQueue.php - - message: "#^Method Assegai\\\\Core\\\\RPC\\\\RpcArgumentsHost\\:\\:getArgs\\(\\) return type has no value type specified in iterable type array\\.$#" count: 1 diff --git a/src/App.php b/src/App.php index 04469eb..f05393a 100644 --- a/src/App.php +++ b/src/App.php @@ -49,6 +49,7 @@ use Assegai\Core\Interfaces\OnModuleInitInterface; use Assegai\Core\Rendering\Engines\DefaultTemplateEngine; use Assegai\Core\Rendering\Interfaces\TemplateEngineInterface; +use Assegai\Core\Queues\QueueFactory; use Assegai\Core\Routing\Router; use Assegai\Core\Runtimes\PhpHttpRuntime; use Assegai\Core\Runtimes\RuntimeContext; @@ -456,6 +457,7 @@ protected function resetRequestScopedDependencies(): void ControllerManager::class, ModuleManager::class, Injector::class, + QueueFactory::class, TemplateEngineInterface::class, LoggerInterface::class, DefaultTemplateEngine::class, diff --git a/src/Injector.php b/src/Injector.php index db50f05..5010e4c 100644 --- a/src/Injector.php +++ b/src/Injector.php @@ -29,7 +29,8 @@ use Assegai\Core\Interfaces\IEntryNotFoundException; use Assegai\Core\Interfaces\ParameterResolverInterface; use Assegai\Core\Interfaces\ITokenStoreOwner; -use Assegai\Core\Queues\Attributes\InjectQueue; +use Assegai\Core\Queues\QueueFactory; +use Assegai\Core\Queues\QueueParameterResolver; use Assegai\Core\Runtimes\RuntimeContext; use Assegai\Core\Util\TypeManager; use Codeception\Attribute\Skip; @@ -90,6 +91,9 @@ final class Injector implements ITokenStoreOwner, IContainer private final function __construct() { $this->logger = new ConsoleLogger(new ConsoleOutput()); + $queueFactory = new QueueFactory(); + $this->add(QueueFactory::class, $queueFactory); + $this->registerParameterResolver(new QueueParameterResolver($queueFactory)); } /** diff --git a/src/Queues/Attributes/InjectQueue.php b/src/Queues/Attributes/InjectQueue.php index f51555f..0284c92 100644 --- a/src/Queues/Attributes/InjectQueue.php +++ b/src/Queues/Attributes/InjectQueue.php @@ -2,16 +2,11 @@ namespace Assegai\Core\Queues\Attributes; -use Assegai\Common\Interfaces\Queues\QueueInterface; -use Assegai\Core\Queues\QueueFactory; use Attribute; -use InvalidArgumentException; #[Attribute(Attribute::TARGET_PARAMETER)] readonly class InjectQueue { - public QueueInterface $queue; - /** * InjectQueue constructor. * @@ -20,37 +15,5 @@ public function __construct( public string $path = '' ) { - $queueConfig = config("queues.connections.$path"); - - if (empty($queueConfig)) { - throw new InvalidArgumentException("Queue configuration for '$path' not found."); - } - - [$driver, $name] = explode('.', $path, 2) + [null, null]; - - if (is_null($name)) { - throw new InvalidArgumentException("Invalid queue path '$path'. Expected format: 'driver.name'."); - } - - if (!isset($queueConfig['name'])) { - $queueConfig['name'] = $name; - } - - $driverClass = config("queues.drivers.$driver"); - - if (!$driverClass || !class_exists($driverClass)) { - throw new InvalidArgumentException("Queue driver '$driver' is not configured or does not exist."); - } - - if (!is_subclass_of($driverClass, QueueInterface::class)) { - throw new InvalidArgumentException("Queue driver '$driver' must implement the QueueInterface."); - } - - $this->queue = $driverClass::create($queueConfig); - } - - public function resolveParameterValue(): QueueInterface - { - return $this->queue; } } diff --git a/src/Queues/QueueFactory.php b/src/Queues/QueueFactory.php new file mode 100644 index 0000000..37503c6 --- /dev/null +++ b/src/Queues/QueueFactory.php @@ -0,0 +1,56 @@ +> */ + private array $connections = []; + + /** + * Resolves one configured queue for the lifetime of the application graph. + * + * Driver creation must remain configuration-only. Transport implementations + * establish their broker connection when an operation first needs it. + * + * @return QueueInterface + */ + public function connection(string $path): QueueInterface + { + $path = trim($path); + + if (isset($this->connections[$path])) { + return $this->connections[$path]; + } + + [$driver, $name] = explode('.', $path, 2) + [null, null]; + + if (!is_string($driver) || $driver === '' || !is_string($name) || $name === '') { + throw new InvalidArgumentException("Invalid queue path '$path'. Expected format: 'driver.name'."); + } + + $queueConfig = config("queues.connections.$path"); + + if (!is_array($queueConfig)) { + throw new InvalidArgumentException("Queue configuration for '$path' not found."); + } + + $queueConfig['name'] ??= $name; + $driverClass = config("queues.drivers.$driver"); + + if (!is_string($driverClass) || !class_exists($driverClass)) { + throw new InvalidArgumentException("Queue driver '$driver' is not configured or does not exist."); + } + + if (!is_subclass_of($driverClass, QueueInterface::class)) { + throw new InvalidArgumentException("Queue driver '$driver' must implement the QueueInterface."); + } + + return $this->connections[$path] = $driverClass::create($queueConfig); + } +} diff --git a/src/Queues/QueueParameterResolver.php b/src/Queues/QueueParameterResolver.php new file mode 100644 index 0000000..fb5f355 --- /dev/null +++ b/src/Queues/QueueParameterResolver.php @@ -0,0 +1,28 @@ +getAttributes(InjectQueue::class) !== []; + } + + public function resolve(ReflectionParameter $parameter, Injector $injector): mixed + { + /** @var InjectQueue $attribute */ + $attribute = $parameter->getAttributes(InjectQueue::class)[0]->newInstance(); + + return $this->factory->connection($attribute->path); + } +} diff --git a/tests/Mocks/MockInjector.php b/tests/Mocks/MockInjector.php index f2d0edd..4168269 100644 --- a/tests/Mocks/MockInjector.php +++ b/tests/Mocks/MockInjector.php @@ -2,6 +2,8 @@ namespace Mocks; +use Assegai\Common\Interfaces\Queues\QueueInterface; +use Assegai\Common\Interfaces\Queues\QueueProcessResultInterface; use Assegai\Core\Attributes\Injectable; use Assegai\Core\Attributes\Modules\Module; use Assegai\Core\Config\ProjectConfig; @@ -17,6 +19,8 @@ use Assegai\Core\Interfaces\AssegaiModuleInterface; use Assegai\Core\Interfaces\ConfiguresInjectorInterface; use Assegai\Core\Interfaces\ParameterResolverInterface; +use Assegai\Core\Queues\Attributes\InjectQueue; +use Assegai\Core\Queues\QueueFactory; use Assegai\Core\Session; use Attribute; use ReflectionParameter; @@ -99,6 +103,60 @@ public function __construct( } } +final class FakeQueue implements QueueInterface +{ + public static int $creations = 0; + + /** @param array $config */ + public function __construct(private readonly array $config) + { + } + + public function add(object $job, object|array|null $options = null): void + { + } + + public function process(callable $callback): QueueProcessResultInterface + { + throw new \LogicException('Not used in this test.'); + } + + public function getName(): string + { + return (string) $this->config['name']; + } + + public function getTotalJobs(): int + { + return 0; + } + + public static function create(array $config): self + { + self::$creations++; + + return new self($config); + } +} + +#[Injectable] +final class QueueInjectedService +{ + public function __construct( + #[InjectQueue('fake.notifications')] + public QueueInterface $queue, + ) { + } +} + +#[Injectable] +final class QueueFactoryAwareService +{ + public function __construct(public QueueFactory $queueFactory) + { + } +} + class ResolverResolvedValue { public function __construct( @@ -511,6 +569,7 @@ class LibraryFeatureAppModule RequestCapturingService::class, ExplicitRequestScopedService::class, AttributeResolvedService::class, + QueueFactoryAwareService::class, ], controllers: [], imports: [], diff --git a/tests/Unit/InjectorCest.php b/tests/Unit/InjectorCest.php index 3b995f4..3ef2062 100644 --- a/tests/Unit/InjectorCest.php +++ b/tests/Unit/InjectorCest.php @@ -16,6 +16,7 @@ use Assegai\Core\Injector; use Assegai\Core\Interfaces\AppInterface; use Assegai\Core\ModuleManager; +use Assegai\Core\Queues\QueueFactory; use Assegai\Core\Runtimes\RuntimeContext; use Assegai\Core\Routing\Router; use Assegai\Core\Session; @@ -24,6 +25,7 @@ use Mocks\ChildUsesParentPrivateService; use Mocks\ExplicitRequestScopedService; use Mocks\ExportVisibilityAppModule; +use Mocks\FakeQueue; use Mocks\FrameworkAwareAppModule; use Mocks\FrameworkAwareContractsService; use Mocks\FrameworkAwareService; @@ -36,6 +38,8 @@ use Mocks\ParentPrivateModule; use Mocks\ParentPrivateVisibilityAppModule; use Mocks\ProviderOwnershipOrderingAppModule; +use Mocks\QueueFactoryAwareService; +use Mocks\QueueInjectedService; use Mocks\RequestCapturingService; use Mocks\ResolverAwareAppModule; use Mocks\ResolverOnlyAwareAppModule; @@ -56,11 +60,13 @@ class InjectorCest private string $logsDirectory = ''; private string $configDirectory = ''; private string $appConfigFilename = ''; + private string $queueConfigFilename = ''; private string $projectConfigFilename = ''; private bool $createdComposerConfig = false; private bool $createdSourceDirectory = false; private bool $createdConfigDirectory = false; private bool $createdAppConfig = false; + private bool $createdQueueConfig = false; private bool $createdProjectConfig = false; private array $previousServer = []; private array $previousGet = []; @@ -93,6 +99,7 @@ public function _before(): void $this->logsDirectory = getcwd() . '/logs'; $this->configDirectory = getcwd() . '/config'; $this->appConfigFilename = $this->configDirectory . '/default.php'; + $this->queueConfigFilename = $this->configDirectory . '/queues.php'; $this->projectConfigFilename = getcwd() . '/assegai.json'; $this->createdComposerConfig = false; $this->createdSourceDirectory = false; @@ -122,6 +129,23 @@ public function _before(): void file_put_contents($this->appConfigFilename, "createdAppConfig = true; + file_put_contents($this->queueConfigFilename, <<<'PHP' + [ + 'fake' => \Mocks\FakeQueue::class, + ], + 'connections' => [ + 'fake' => [ + 'notifications' => [], + ], + ], +]; +PHP); + $this->createdQueueConfig = true; + FakeQueue::$creations = 0; + file_put_contents($this->projectConfigFilename, json_encode([], JSON_PRETTY_PRINT | JSON_UNESCAPED_SLASHES)); $this->createdProjectConfig = true; @@ -194,6 +218,10 @@ public function _after(): void unlink($this->appConfigFilename); } + if ($this->createdQueueConfig && is_file($this->queueConfigFilename)) { + unlink($this->queueConfigFilename); + } + if ($this->createdProjectConfig && is_file($this->projectConfigFilename)) { unlink($this->projectConfigFilename); } @@ -266,6 +294,47 @@ public function testParameterAttributesCanResolveDependenciesWithoutInjectorHard $I->assertSame('attribute-seam', $service->value->value); } + public function testQueueInjectionUsesTheApplicationQueueFactory(UnitTester $I): void + { + $injector = Injector::getInstance(); + $service = $injector->resolve(QueueInjectedService::class); + $factoryAwareService = $injector->resolve(QueueFactoryAwareService::class); + + $I->assertInstanceOf(QueueInjectedService::class, $service); + $I->assertInstanceOf(QueueFactoryAwareService::class, $factoryAwareService); + $I->assertSame('notifications', $service->queue->getName()); + $I->assertSame($service->queue, $factoryAwareService->queueFactory->connection('fake.notifications')); + $I->assertSame(1, FakeQueue::$creations); + } + + public function testModuleProvidersCanInjectTheApplicationQueueFactory(UnitTester $I): void + { + $moduleManager = ModuleManager::getInstance(); + $moduleManager->setRootModuleClass(FrameworkAwareAppModule::class); + $moduleManager->buildModuleTokensList(FrameworkAwareAppModule::class); + $moduleManager->buildProviderTokensList(); + + $service = Injector::getInstance()->resolve(QueueFactoryAwareService::class); + + $I->assertInstanceOf(QueueFactoryAwareService::class, $service); + $I->assertSame(Injector::getInstance()->get(QueueFactory::class), $service->queueFactory); + } + + public function testQueueFactoryRejectsInvalidAndMissingConnectionPaths(UnitTester $I): void + { + $factory = Injector::getInstance()->get(QueueFactory::class); + + $I->assertInstanceOf(QueueFactory::class, $factory); + $I->expectThrowable( + \InvalidArgumentException::class, + static fn () => $factory->connection('notifications'), + ); + $I->expectThrowable( + \InvalidArgumentException::class, + static fn () => $factory->connection('fake.missing'), + ); + } + public function testImportedModulesCanRegisterPackageParameterResolversBeforeProviderResolution(UnitTester $I): void { $app = AssegaiFactory::create(ResolverAwareAppModule::class); @@ -275,7 +344,7 @@ public function testImportedModulesCanRegisterPackageParameterResolversBeforePro $I->assertInstanceOf(App::class, $app); $I->assertInstanceOf(ResolverResolvedService::class, $service); $I->assertSame('resolved-by-registry', $service->value->value); - $I->assertCount(1, Injector::getInstance()->getParameterResolvers()); + $I->assertCount(2, Injector::getInstance()->getParameterResolvers()); } public function testInjectorExtensionsCanBeConfiguredByImportedClassesWithoutTheModuleInterface(UnitTester $I): void @@ -287,7 +356,7 @@ public function testInjectorExtensionsCanBeConfiguredByImportedClassesWithoutThe $I->assertInstanceOf(App::class, $app); $I->assertInstanceOf(ResolverResolvedService::class, $service); $I->assertSame('resolved-by-registry', $service->value->value); - $I->assertCount(1, Injector::getInstance()->getParameterResolvers()); + $I->assertCount(2, Injector::getInstance()->getParameterResolvers()); } public function testRefreshingRequestScopeRebindsRequestScopedServices(UnitTester $I): void @@ -358,6 +427,19 @@ public function testRequestCapturingServicesResolveAsRequestScoped(UnitTester $I $I->assertNull($injector->get(ExplicitRequestScopedService::class)); } + public function testRefreshingRequestScopeRetainsTheApplicationQueueFactory(UnitTester $I): void + { + $app = AssegaiFactory::create(FrameworkAwareAppModule::class); + $injector = Injector::getInstance(); + $factory = $injector->get(QueueFactory::class); + + $refreshRequestScope = new \ReflectionMethod($app, 'refreshRequestScope'); + $refreshRequestScope->invoke($app); + + $I->assertInstanceOf(QueueFactory::class, $factory); + $I->assertSame($factory, $injector->get(QueueFactory::class)); + } + public function testChildModulesCanInjectParentModuleExports(UnitTester $I): void { $app = AssegaiFactory::create(ParentExportVisibilityAppModule::class);