feat: inject attributes declared on handler into closure expressions (Enterprise)

packages/Dbal/tests/Fixture/ClosureInAttribute/DedupPolicy.php

@@ -0,0 +1,18 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Test\Ecotone\Dbal\Fixture\ClosureInAttribute;
+
+use Attribute;
+
+#[Attribute(Attribute::TARGET_METHOD)]
+/**
+ * licence Apache-2.0
+ */
+final class DedupPolicy
+{
+    public function __construct(public string $scope)
+    {
+    }
+}

packages/Dbal/tests/Fixture/ClosureInAttribute/PolicyDrivenDeduplicatedHandler.php

@@ -0,0 +1,51 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Test\Ecotone\Dbal\Fixture\ClosureInAttribute;
+
+use Ecotone\Messaging\Attribute\Deduplicated;
+use Ecotone\Messaging\Attribute\Parameter\Header;
+use Ecotone\Modelling\Attribute\CommandHandler;
+use Ecotone\Modelling\Attribute\QueryHandler;
+
+/**
+ * licence Apache-2.0
+ */
+final class PolicyDrivenDeduplicatedHandler
+{
+    private array $handledPerCustomer = [];
+    private array $handledPerOrder = [];
+
+    #[DedupPolicy(scope: 'customer')]
+    #[Deduplicated(expression: static function (DedupPolicy $policy, #[Header('customerId')] string $customerId, #[Header('orderId')] string $orderId): string {
+        return $policy->scope === 'customer' ? $customerId : $orderId;
+    })]
+    #[CommandHandler('policyDedup.perCustomer', endpointId: 'policyDedupPerCustomerEndpoint')]
+    public function handlePerCustomer(#[Header('orderId')] string $orderId): void
+    {
+        $this->handledPerCustomer[] = $orderId;
+    }
+
+    #[DedupPolicy(scope: 'order')]
+    #[Deduplicated(expression: static function (DedupPolicy $policy, #[Header('customerId')] string $customerId, #[Header('orderId')] string $orderId): string {
+        return $policy->scope === 'customer' ? $customerId : $orderId;
+    })]
+    #[CommandHandler('policyDedup.perOrder', endpointId: 'policyDedupPerOrderEndpoint')]
+    public function handlePerOrder(#[Header('orderId')] string $orderId): void
+    {
+        $this->handledPerOrder[] = $orderId;
+    }
+
+    #[QueryHandler('policyDedup.handledPerCustomer')]
+    public function handledPerCustomer(): array
+    {
+        return $this->handledPerCustomer;
+    }
+
+    #[QueryHandler('policyDedup.handledPerOrder')]
+    public function handledPerOrder(): array
+    {
+        return $this->handledPerOrder;
+    }
+}

packages/Dbal/tests/Integration/ClosureInAttribute/ClosureExpressionDbalTest.php

@@ -24,6 +24,7 @@
 use Test\Ecotone\Dbal\DbalMessagingTestCase;
 use Test\Ecotone\Dbal\Fixture\ClosureInAttribute\ClosureDeduplicatedHandler;
 use Test\Ecotone\Dbal\Fixture\ClosureInAttribute\PersonClosureParameterApi;
+use Test\Ecotone\Dbal\Fixture\ClosureInAttribute\PolicyDrivenDeduplicatedHandler;
 use Test\Ecotone\Dbal\Fixture\ClosureInAttribute\TenantClosurePoller;
 use Test\Ecotone\Dbal\Fixture\MultiTenant\FakeConnectionFactory;
 
@@ -53,6 +54,36 @@ classesToResolve: [ClosureDeduplicatedHandler::class],
         $this->assertEquals(2, $ecotoneLite->sendQueryWithRouting('closureDedup.getCallCount'));
     }
 
+    public function test_deduplication_closure_expression_receives_attribute_declared_on_handler(): void
+    {
+        $handler = new PolicyDrivenDeduplicatedHandler();
+        $ecotoneLite = EcotoneLite::bootstrapFlowTesting(
+            classesToResolve: [PolicyDrivenDeduplicatedHandler::class],
+            containerOrAvailableServices: [$handler, DbalConnectionFactory::class => $this->getConnectionFactory(true)],
+            configuration: ServiceConfiguration::createWithDefaults()
+                ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::DBAL_PACKAGE])),
+            licenceKey: LicenceTesting::VALID_LICENCE,
+        );
+
+        $ecotoneLite->sendCommandWithRoutingKey('policyDedup.perCustomer', 'test', metadata: ['customerId' => 'customer-1', 'orderId' => 'order-1']);
+        $ecotoneLite->sendCommandWithRoutingKey('policyDedup.perCustomer', 'test', metadata: ['customerId' => 'customer-1', 'orderId' => 'order-2']);
+
+        $this->assertSame(
+            ['order-1'],
+            $ecotoneLite->sendQueryWithRouting('policyDedup.handledPerCustomer'),
+            'DedupPolicy with customer scope was injected into closure, so both orders of same customer deduplicate to one'
+        );
+
+        $ecotoneLite->sendCommandWithRoutingKey('policyDedup.perOrder', 'test', metadata: ['customerId' => 'customer-2', 'orderId' => 'order-3']);
+        $ecotoneLite->sendCommandWithRoutingKey('policyDedup.perOrder', 'test', metadata: ['customerId' => 'customer-2', 'orderId' => 'order-4']);
+
+        $this->assertSame(
+            ['order-3', 'order-4'],
+            $ecotoneLite->sendQueryWithRouting('policyDedup.handledPerOrder'),
+            'Same closure with order scoped DedupPolicy deduplicates per order, proving the injected attribute drives the key'
+        );
+    }
+
     public function test_deduplication_closure_expression_throws_licensing_exception_on_bootstrap_without_enterprise_licence(): void
     {
         $this->expectException(LicensingException::class);

packages/Dbal/tests/Integration/MultiTenant/ConsoleCommandTenantPropagationTest.php

@@ -0,0 +1,139 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Test\Ecotone\Dbal\Integration\MultiTenant;
+
+use Doctrine\DBAL\Connection;
+use Ecotone\Dbal\Attribute\MultiTenantConnection;
+use Ecotone\Dbal\Configuration\DbalConfiguration;
+use Ecotone\Dbal\MultiTenant\MultiTenantConfiguration;
+use Ecotone\Lite\EcotoneLite;
+use Ecotone\Lite\Test\FlowTestSupport;
+use Ecotone\Messaging\Attribute\ConsoleCommand;
+use Ecotone\Messaging\Attribute\Parameter\Reference;
+use Ecotone\Messaging\Config\ModulePackageList;
+use Ecotone\Messaging\Config\ServiceConfiguration;
+use Ecotone\Messaging\Handler\MethodInvocationException;
+use Ecotone\Modelling\Attribute\CommandHandler;
+use Ecotone\Modelling\CommandBus;
+use Test\Ecotone\Dbal\DbalMessagingTestCase;
+
+/**
+ * Proves that a tenant header can be propagated into a `#[ConsoleCommand]` call
+ * the same way it is propagated for inbound channel adapters (see
+ * ScheduledTenantResolverTest), so that MultiTenantConnectionFactory resolves
+ * the correct per-tenant connection when a console command is executed with
+ * `--header="tenant:tenant_a"` (or, in tests, `runConsoleCommand(..., ['header' => ['tenant:tenant_a']])`).
+ */
+/**
+ * licence Apache-2.0
+ * @internal
+ */
+final class ConsoleCommandTenantPropagationTest extends DbalMessagingTestCase
+{
+    public const MARKER_TABLE = 'tenant_marker';
+
+    public function setUp(): void
+    {
+        parent::setUp();
+
+        foreach ([$this->connectionForTenantA(), $this->connectionForTenantB()] as $connectionFactory) {
+            $connection = $connectionFactory->createContext()->getDbalConnection();
+            $connection->executeStatement('DROP TABLE IF EXISTS ' . self::MARKER_TABLE);
+            $connection->executeStatement('CREATE TABLE ' . self::MARKER_TABLE . ' (marker INTEGER)');
+        }
+    }
+
+    public function test_console_command_without_tenant_header_throws_when_no_default_connection(): void
+    {
+        $ecotoneLite = $this->bootstrapEcotone();
+
+        $this->expectException(MethodInvocationException::class);
+        $this->expectExceptionMessage('Lack of context about tenant in Message Headers');
+
+        $ecotoneLite->runConsoleCommand('multi_tenant:record_marker', []);
+    }
+
+    public function test_console_command_tenant_header_routes_to_correct_tenant_connection(): void
+    {
+        $ecotoneLite = $this->bootstrapEcotone();
+
+        $ecotoneLite->runConsoleCommand('multi_tenant:record_marker', ['header' => ['tenant:tenant_a']]);
+
+        $this->assertSame(1, $this->countMarkerRows($this->connectionForTenantA()), 'tenant_a should have received the marker');
+        $this->assertSame(0, $this->countMarkerRows($this->connectionForTenantB()), 'tenant_b must be untouched - console command routed to tenant_a only');
+
+        $ecotoneLite->runConsoleCommand('multi_tenant:record_marker', ['header' => ['tenant:tenant_b']]);
+
+        $this->assertSame(1, $this->countMarkerRows($this->connectionForTenantA()));
+        $this->assertSame(1, $this->countMarkerRows($this->connectionForTenantB()));
+    }
+
+    public function test_tenant_header_from_console_command_propagates_to_command_bus_sub_flow(): void
+    {
+        $ecotoneLite = $this->bootstrapEcotone();
+
+        $ecotoneLite->runConsoleCommand('multi_tenant:record_marker_via_command_bus', ['header' => ['tenant:tenant_b']]);
+
+        $this->assertSame(0, $this->countMarkerRows($this->connectionForTenantA()));
+        $this->assertSame(1, $this->countMarkerRows($this->connectionForTenantB()), 'tenant header must propagate from console command into the Command Bus sub-flow');
+    }
+
+    private function countMarkerRows(object $connectionFactory): int
+    {
+        return (int) $connectionFactory->createContext()->getDbalConnection()
+            ->executeQuery('SELECT COUNT(*) FROM ' . self::MARKER_TABLE)
+            ->fetchOne();
+    }
+
+    private function newTenantMarkerRecorder(): object
+    {
+        return new class () {
+            #[ConsoleCommand('multi_tenant:record_marker')]
+            public function record(#[MultiTenantConnection] Connection $connection): void
+            {
+                $connection->executeStatement('INSERT INTO ' . ConsoleCommandTenantPropagationTest::MARKER_TABLE . ' (marker) VALUES (1)');
+            }
+
+            #[ConsoleCommand('multi_tenant:record_marker_via_command_bus')]
+            public function recordViaCommandBus(#[Reference] CommandBus $commandBus): void
+            {
+                $commandBus->sendWithRouting('multi_tenant.record_marker');
+            }
+
+            #[CommandHandler('multi_tenant.record_marker')]
+            public function recordMarker(#[MultiTenantConnection] Connection $connection): void
+            {
+                $connection->executeStatement('INSERT INTO ' . ConsoleCommandTenantPropagationTest::MARKER_TABLE . ' (marker) VALUES (1)');
+            }
+        };
+    }
+
+    private function bootstrapEcotone(): FlowTestSupport
+    {
+        $recorder = $this->newTenantMarkerRecorder();
+
+        return EcotoneLite::bootstrapFlowTesting(
+            [$recorder::class],
+            [
+                $recorder,
+                'tenant_a_connection' => $this->connectionForTenantA(),
+                'tenant_b_connection' => $this->connectionForTenantB(),
+            ],
+            ServiceConfiguration::createWithDefaults()
+                ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([ModulePackageList::DBAL_PACKAGE]))
+                ->withExtensionObjects([
+                    MultiTenantConfiguration::create(
+                        tenantHeaderName: 'tenant',
+                        tenantToConnectionMapping: [
+                            'tenant_a' => 'tenant_a_connection',
+                            'tenant_b' => 'tenant_b_connection',
+                        ],
+                    ),
+                    DbalConfiguration::createWithDefaults()
+                        ->withDeduplication(false),
+                ]),
+        );
+    }
+}

packages/Ecotone/src/Messaging/Handler/ClosureExpression/AttributeExpressionExecutorCompiler.php

@@ -16,13 +16,16 @@
 use Ecotone\Messaging\Handler\InterfaceParameter;
 use Ecotone\Messaging\Handler\InterfaceToCall;
 use Ecotone\Messaging\Handler\ParameterConverterBuilder;
+use Ecotone\Messaging\Handler\Processor\MethodInvoker\Converter\AttributeBuilder;
 use Ecotone\Messaging\Handler\Processor\MethodInvoker\Converter\MessageConverterBuilder;
 use Ecotone\Messaging\Handler\Processor\MethodInvoker\Converter\PayloadBuilder;
 use Ecotone\Messaging\Handler\Processor\MethodInvoker\Converter\ReferenceBuilder;
 use Ecotone\Messaging\Handler\Processor\MethodInvoker\Converter\ValueBuilder;
 use Ecotone\Messaging\Handler\Type;
 use Ecotone\Messaging\Support\Assert;
+use ReflectionClass;
 use ReflectionFunction;
+use ReflectionMethod;
 use ReflectionParameter;
 
 /**
@@ -156,6 +159,8 @@ private static function executorDefinition(AttributeDefinition $attributeArgumen
             $closureParameterResolvers = self::parameterResolverDefinitions(
                 $reflectionParameters,
                 self::closureInterfaceToCall($ownerClassName, $ownerMethodName, $reflectionParameters),
+                $ownerClassName,
+                $ownerMethodName,
             );
         }
 
@@ -179,7 +184,7 @@ private static function matchesAttributeClass(string $annotationClassName, strin
      * @param ReflectionParameter[] $reflectionParameters
      * @return Definition[]
      */
-    private static function parameterResolverDefinitions(array $reflectionParameters, InterfaceToCall $interfaceToCall): array
+    private static function parameterResolverDefinitions(array $reflectionParameters, InterfaceToCall $interfaceToCall, string $ownerClassName, ?string $ownerMethodName): array
     {
         $parameterResolvers = [];
         foreach ($reflectionParameters as $index => $reflectionParameter) {
@@ -188,6 +193,12 @@ private static function parameterResolverDefinitions(array $reflectionParameters
 
             $converterBuilder = ParameterConverterAnnotationFactory::getConverterFor($interfaceParameter, $interfaceToCall);
             $resolvesFromAdditionalContext = $converterBuilder === null || $converterBuilder instanceof MessageConverterBuilder;
+            if ($converterBuilder === null) {
+                $converterBuilder = self::declaredAttributeConverterBuilderFor($interfaceParameter, $ownerClassName, $ownerMethodName);
+                if ($converterBuilder !== null) {
+                    $resolvesFromAdditionalContext = false;
+                }
+            }
             if ($converterBuilder === null) {
                 $converterBuilder = self::defaultConverterBuilderFor($interfaceParameter, $index === 0);
             }
@@ -249,6 +260,35 @@ private static function ensureNoNestedClosureExpression(InterfaceParameter $inte
         }
     }
 
+    /**
+     * Resolves closure parameter type hinted with an Attribute declared on the owning method or class,
+     * so expression may adapt its behaviour to configuration declared next to the endpoint.
+     */
+    private static function declaredAttributeConverterBuilderFor(InterfaceParameter $interfaceParameter, string $ownerClassName, ?string $ownerMethodName): ?ParameterConverterBuilder
+    {
+        if (! $interfaceParameter->isAnnotation()) {
+            return null;
+        }
+
+        $parameterType = $interfaceParameter->getTypeDescriptor()->withoutNull();
+
+        if ($ownerMethodName !== null) {
+            foreach ((new ReflectionMethod($ownerClassName, $ownerMethodName))->getAttributes() as $attribute) {
+                if (Type::object($attribute->getName())->equals($parameterType)) {
+                    return new AttributeBuilder($interfaceParameter->getName(), $attribute->newInstance(), $ownerClassName, $ownerMethodName);
+                }
+            }
+        }
+
+        foreach ((new ReflectionClass($ownerClassName))->getAttributes() as $attribute) {
+            if (Type::object($attribute->getName())->equals($parameterType)) {
+                return new AttributeBuilder($interfaceParameter->getName(), $attribute->newInstance(), $ownerClassName, null);
+            }
+        }
+
+        return null;
+    }
+
     private static function defaultConverterBuilderFor(InterfaceParameter $interfaceParameter, bool $isFirstParameter): ?ParameterConverterBuilder
     {
         if ($isFirstParameter) {

packages/Ecotone/tests/Messaging/Fixture/Handler/ClosureInAttribute/PolicyDrivenTokenService.php

@@ -0,0 +1,33 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Test\Ecotone\Messaging\Fixture\Handler\ClosureInAttribute;
+
+use Ecotone\Messaging\Attribute\Parameter\Header;
+use Ecotone\Modelling\Attribute\CommandHandler;
+use Ecotone\Modelling\Attribute\QueryHandler;
+
+/**
+ * licence Apache-2.0
+ */
+final class PolicyDrivenTokenService
+{
+    private array $tokens = [];
+
+    #[TokenPolicy(casing: 'upper')]
+    #[CommandHandler('policyToken.store')]
+    public function store(
+        #[Header('token', expression: static function (TokenPolicy $policy, #[Header('token')] string $token): string {
+            return $policy->casing === 'upper' ? strtoupper($token) : strtolower($token);
+        })] string $token,
+    ): void {
+        $this->tokens[] = $token;
+    }
+
+    #[QueryHandler('policyToken.getTokens')]
+    public function getTokens(): array
+    {
+        return $this->tokens;
+    }
+}

packages/Ecotone/tests/Messaging/Fixture/Handler/ClosureInAttribute/TokenPolicy.php

@@ -0,0 +1,18 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Test\Ecotone\Messaging\Fixture\Handler\ClosureInAttribute;
+
+use Attribute;
+
+#[Attribute(Attribute::TARGET_METHOD)]
+/**
+ * licence Apache-2.0
+ */
+final class TokenPolicy
+{
+    public function __construct(public string $casing)
+    {
+    }
+}