README

Symfony bundle that collects user behaviour events from any API service and dispatches them to RabbitMQ → S3 → Athena.

JWT claims contract

The bundle reads all user context exclusively from the JWT token payload. The following claims must be present in every token. If any are missing the collection event will be silently skipped and a warning will be logged — the request itself is never disrupted.

Claim	Type	Notes
`username`	string	Also mapped to `email` — intentionally the same value for now, kept separate for future use
`roles`	array
`zone`	string
`locale`	string
`user_id`	int
`org_id`	int
`role_id`	int
`partner_id`	string (UUID)
`first_name`	string
`last_name`	string

Installation

1. Install the bundle and the AMQP transport:

composer require rezilio/collector-bundle
composer require symfony/amqp-messenger

2. Ensure the php-amqp extension is installed at the system level:

# macOS (Homebrew)
pecl install amqp

# Ubuntu / Debian
apt install php-amqp

3. Register the bundle manually in config/bundles.php — Symfony Flex does not auto-register third-party bundles:

// config/bundles.php
return [
    Rezilio\CollectorBundle\CollectorBundle::class => ['all' => true],
];

1. Add the collection_queue transport to your messenger.yaml

Do NOT import the bundle's messenger.yaml. Instead add the transport and routing directly to your project's config/packages/messenger.yaml (and any environment overrides such as config/packages/dev/messenger.yaml):

# config/packages/messenger.yaml  (and each environment override)
framework:
  messenger:
    transports:
      # ... your existing transports ...
      collection_queue: '%env(MESSENGER_TRANSPORT_DSN)%%rabbitmq_prefix%-CollectionQueue'

    routing:
      # ... your existing routing ...
      'Rezilio\CollectorBundle\Message\CollectionResponse': collection_queue

Important — %rabbitmq_prefix% parameter: The DSN uses a %rabbitmq_prefix% parameter that must be defined in your config/services.yaml or config/packages/parameters.yaml. Without it, the transport will fail to connect with a cryptic error.
# config/services.yaml
parameters:
    rabbitmq_prefix: 'dev'   # change per environment: dev, staging, prod

Important — environment overrides: If your project has per-environment messenger files (e.g. config/packages/dev/messenger.yaml) they override the main file entirely. Make sure collection_queue transport and routing are present in every environment file that exists, not just the main one.

Important — RabbitMQ bindings: If you previously had a messages default queue, make sure it is not bound to the CollectionQueue exchange. Stale bindings will cause messages to land in both queues. Delete the binding via the RabbitMQ management UI or API if needed.

2. Register AppCollectorListener

AppCollectorListener extends the bundle's CollectorListener, which means Symfony's autowiring will attempt to register it twice if EventListener is included in the auto-discovery resource path. You must explicitly exclude EventListener from the App\ auto-discovery block before manually re-registering it, otherwise the service will be double-registered and events may fire twice.

# config/services.yaml
App\:
    resource: '../src/*'
    exclude: '../src/{DependencyInjection,DataProvider,DataMapping,Entity,Migrations,Tests,Kernel.php,EventListener}'

App\EventListener\AppCollectorListener:
    arguments:
        $factory: '@Rezilio\CollectorBundle\Service\CollectorServiceFactory'
        $logger:  '@logger'
        $routes:  []
    tags:
        - { name: kernel.event_subscriber }

Create the listener class. Use buildExtra() to attach any app-specific fields beyond the standard JWT claims — the bundle handles all standard claims automatically:

// src/EventListener/AppCollectorListener.php
namespace App\EventListener;

use Rezilio\CollectorBundle\EventListener\CollectorListener;

class AppCollectorListener extends CollectorListener
{
    protected function buildExtra(): array
    {
        return [
            // Add any app-specific fields here, e.g.:
            // 'mission_id' => $this->resolveMissionId(),
        ];
    }
}

3. Add a CollectionResponseHandler

Symfony Messenger requires a handler for every routed message class. Add a minimal stub that satisfies this requirement — the actual processing is done by the rezilio_collector worker project.

The implementation differs depending on your Symfony version:

Symfony 4.4 / 5.x / 6.x — implement MessageHandlerInterface:

// src/MessageHandler/CollectionResponseHandler.php
namespace App\MessageHandler;

use Rezilio\CollectorBundle\Message\CollectionResponse;
use Symfony\Component\Messenger\Handler\MessageHandlerInterface;

class CollectionResponseHandler implements MessageHandlerInterface
{
    public function __invoke(CollectionResponse $message)
    {
        // Handled by rezilio_collector worker
    }
}

Symfony 7.x — MessageHandlerInterface was removed. Use the #[AsMessageHandler] attribute instead:

// src/MessageHandler/CollectionResponseHandler.php
namespace App\MessageHandler;

use Rezilio\CollectorBundle\Message\CollectionResponse;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;

#[AsMessageHandler]
class CollectionResponseHandler
{
    public function __invoke(CollectionResponse $message)
    {
        // Handled by rezilio_collector worker
    }
}

4. Define which routes to collect

# config/packages/collector.yaml
collector:
    default_zone: 'ca'    # required — see section 5
    routes:
        - { path: '^/api/v1/events' }
        - { path: '^/api/v2/events' }
        - { path: '^/api/v2/content', controller: 'App\DataProvider\ContentCollectionDataProvider' }

5. Rejected-request collection (built-in)

The bundle ships a CollectorRejectedSubscriber that collects requests which were rejected by an authentication or authorization gate — 401 or 403 — before any controller ran, so those requests are not lost. It runs late on KernelEvents::RESPONSE and only acts when the route listener did not already collect the request, so every request yields exactly one record.

Rejected requests are collected with class = "AuthRejected" and carry an extra status_code field (and siteUuid when present in the query string or body). JWT claims are used when a token is present, even if it was invalid; when the token has no zone claim — or there is no token at all — the configured default_zone is used.

It is active by default and default_zone is required:

# config/packages/collector.yaml
collector:
    default_zone: 'ca'
    rejected_collection: true    # set to false to disable
    routes:
        - { path: '^/api/v1/events' }

Upgrading from a version with pre_auth_collection: remove that key and add default_zone. Leaving pre_auth_collection in place fails config validation with an "unrecognized option" error. The old subscriber dispatched a differently-shaped payload on every request; the new one uses the standard payload shape and only fires for rejected requests.

6. Dispatching directly (optional)

For cases where you need to dispatch a collection event without going through the listener — for example from a controller or CLI command — dispatch the message directly:

use Rezilio\CollectorBundle\Message\CollectionResponse;

$this->bus->dispatch(new CollectionResponse(
    $zone,
    ['event' => 'order.created', 'order_id' => $orderId]
));

From a CLI command:

php bin/console app:queue:publish <zone> '<json-data>'

# Example
php bin/console app:queue:publish ca '{"event":"order.created","orderId":42}'

7. Sanitize rules (optional)

Redact, hash, or measure fields per route:

collector:
    routes:
        - path: '^/api/v1/profile'
          sanitize:
              - { field: password, action: redact }

        - path: '^/api/v1/content'
          sanitize:
              - { field: content, action: hash }
              - { field: content, action: length }
              - { field: content, action: redact }

Actions: redact replaces value with -, hash adds {field}_md5, length adds {field}_length.

What is included

Class	Purpose
`CollectorBundle`	Bundle entry point
`CollectionResponse`	Messenger message (zone + data)
`CollectorService`	Builds and dispatches the payload from JWT claims
`CollectorServiceFactory`	Creates a `CollectorService` from a JWT payload array
`CollectorListener`	Kernel event subscriber — auto-collects matching routes after auth
`CollectorRejectedSubscriber`	Kernel event subscriber — collects 401/403 requests rejected before the controller
`DataProviderAbstract`	Optional base class for API Platform DataProviders
`Configuration`	Validates `collector.yaml` config

rezilio / collector-bundle

Package info

Security

Statistics