feat: derive tenant header from inbound messages via #[WithTenantResolver]

[dgafka]

Why is this change proposed?

External messages (Kafka, AMQP, Scheduled poller) often don't carry the
multi-tenant header — the tenant has to be derived from something else,
like the originating topic or a payload field. Today this fails because
the connection-switching Around interceptors fire before any user
code can populate the header, throwing "Lack of context about tenant
in Message Headers".

This change introduces a first-class, declarative #[WithTenantResolver]
attribute that runs the user's expression against the inbound message
before any tenant-aware interceptor sees it.

Description of Changes

• New #[WithTenantResolver(expression: "...")] attribute (Apache, in
  ecotone/dbal). Place it on inbound channel adapter methods.
• New MultiTenantHeaderResolver Before interceptor (Apache):
  evaluates the expression against payload/headers, injects the
  configured tenant header into the message. Explicit tenant headers
  always win; null expression results are a no-op; non-scalar results
  throw a meaningful InvalidArgumentException.
• ScheduledModule now propagates method-level annotations to the
  inbound channel adapter gateway, mirroring KafkaModule/RabbitConsumerModule.
• Validation: #[WithTenantResolver] on a non-inbound-adapter method
  (CommandHandler, EventHandler, Asynchronous, etc.) throws a
  ConfigurationException at boot with an explanation of why internal
  Message Channels don't need a resolver.

All new components are Apache-licensed. Combining #[WithTenantResolver]
with an Apache adapter like #[Scheduled] requires no Enterprise
licence; combining it with #[KafkaConsumer] or #[RabbitConsumer]
requires Enterprise only because those adapters themselves do.

Usage

use Ecotone\Dbal\Attribute\WithTenantResolver;
use Ecotone\Kafka\Attribute\KafkaConsumer;
use Ecotone\Messaging\Attribute\Parameter\Headers;

final class OrderEventConsumer
{
    #[KafkaConsumer('orders', topics: ['orders_eu', 'orders_us'])]
    #[WithTenantResolver(expression: "headers['kafka_topic']")]
    public function handle(string $payload, #[Headers] array $headers): void
    {
        // headers['tenant'] is populated from the originating Kafka topic
    }
}

use Ecotone\Dbal\Attribute\WithTenantResolver;
use Ecotone\Messaging\Attribute\Scheduled;
use Ecotone\Messaging\Message;
use Ecotone\Messaging\Support\MessageBuilder;

final class ExternalEventPoller
{
    #[Scheduled(requestChannelName: 'externalArrived', endpointId: 'externalPoller')]
    #[WithTenantResolver(expression: "payload['tenantId']")]
    public function poll(): ?Message
    {
        $event = $this->source->next();
        return $event === null ? null : MessageBuilder::withPayload($event)->build();
    }
}

You can also reference services from the expression:

#[WithTenantResolver(expression: "reference('topicTenantMap').lookup(headers['kafka_topic'])")]

Use cases

1. Kafka topic per tenant — One #[KafkaConsumer] subscribes to many
   per-tenant topics; the resolver picks the tenant from kafka_topic.
2. External REST/webhook ingestion — A #[Scheduled] poller pulls
   pending events from a third-party API and tags each with its origin;
   the resolver translates origin → tenant before downstream handlers run.
3. AMQP per-tenant queues — #[AmqpConsumer] consumes from queues
   named after tenants; resolver reads the routing key.

Flow

sequenceDiagram
    participant Broker as External broker / scheduler
    participant Adapter as Inbound channel adapter gateway
    participant Resolver as MultiTenantHeaderResolver (Before)
    participant Channel as Request channel
    participant Handler as User handler
    participant Tenant as MultiTenantConnectionFactory

    Broker->>Adapter: inbound message
    Adapter->>Resolver: pointcut WithTenantResolver matches
    Resolver->>Resolver: evaluate expression on payload+headers
    Resolver-->>Adapter: tenant header injected
    Adapter->>Channel: forward message
    Channel->>Handler: dispatch
    Handler->>Tenant: getConnection() — tenant context resolved

Loading

Out of scope (for follow-up)

#[WithTenantResolver] on #[Asynchronous] handlers is rejected by the
new placement validation. Internal Message Channels already carry the
tenant context propagated from the originating bus call, so a resolver
there would have nothing meaningful to derive. If an asynchronous
handler is processing externally-arrived messages, the resolver belongs
on the inbound channel adapter that produces them.

Pull Request Contribution Terms

• I have read and agree to the contribution terms outlined in CONTRIBUTING.