ddd-message-handler-specialist

name	ddd-message-handler-specialist
description	Create Symfony Messenger message + handler pairs for async background processing in the mgamadeus/ddd framework. Covers message classes, handler classes, auth context propagation, workspace routing, logging, messenger.yaml transport/routing config, and supervisor consumers.
metadata	{"author":"mgamadeus","version":"1.0.0","framework":"mgamadeus/ddd"}

DDD Message Handler Specialist

Async background processing via Symfony Messenger within the DDD Core framework (mgamadeus/ddd).

When to Use

• Adding a new async background task processed by Symfony Messenger
• Adding a bool $async option to a service method and implementing dispatching
• Ensuring background work executes under the same account permissions as the triggering request
• Implementing heavy jobs with time/memory limits and consistent logging

Framework Base Classes

Class	Location	Purpose
AppMessage	src/Domain/Base/Entities/MessageHandlers/AppMessage.php	Base message class (extends ValueObject, implements SerializerInterface)
AppMessageHandler	src/Domain/Base/Entities/MessageHandlers/AppMessageHandler.php	Abstract base handler with auth, logging, error handling

AppMessage provides:

• dispatch() -- dispatches to Symfony Messenger bus, auto-captures accountId and dispatchedFromWorkspaceDir
• encodeForCommandline() / decodeFromCommandline() -- CLI transport (gzip + base64)
• persistToTempDir() / loadFromTempDir() -- temp file transport
• processOnWorkspace() / processOnWorkspaceIfNecessary() -- cross-workspace processing

AppMessageHandler provides:

• getLogger() -- returns injected messengerLogger or falls back to DDDService::instance()->getLogger()
• setAuthAccountFromMessage(AppMessage) -- restores auth context from message's accountId
• logShortException(LoggerInterface, string, Throwable) -- structured error log with top 3 stack trace frames
• logIssue(Throwable, AppMessage, ?string) -- comprehensive exception logging with message payload
• extractMessagePayload(AppMessage) -- extracts scalar properties for JSON logging

---

Conventions

1. Prefer IDs in Message Payload

Store only IDs and primitive values in the message. Re-fetch the entity in the handler.

Why: Reduces message size, avoids serialization issues, handler always loads the latest DB state.

Exception: When the object is not yet stored (no ID) or serialization is explicitly desired.

2. Handler is Ultra-Slim — ALL Logic Lives in the Service

The handler is a thin dispatcher. It MUST NOT contain business logic. Its only job is:

1. Set auth context
2. Check workspace routing
3. Load entity by ID
4. Call a single service method with async: false
5. Catch and log errors

NEVER put moderation logic, translation calls, notification dispatch, rate limiting, DB queries, or any other business logic directly in the handler. All of that belongs in the service method. The handler should be ~20-30 lines max.

Anti-pattern (WRONG):

// WRONG — handler doing business logic
class FooBarHandler extends AppMessageHandler {
    public function __invoke(FooBarMessage $message): void {
        // ...auth, workspace...
        $entity = FooBar::byId($message->id);
        // ❌ Business logic in handler:
        $this->doStep1($entity);
        $this->doStep2($entity);
        $this->doStep3($entity);
    }
    protected function doStep1(...) { /* 50 lines */ }
    protected function doStep2(...) { /* 80 lines */ }
}

Correct pattern (RIGHT):

// Handler — thin
class FooBarHandler extends AppMessageHandler {
    public function __invoke(FooBarMessage $message): void {
        // ...auth, workspace...
        $service = FooBars::getService();
        $entity = $service->find($message->id);
        $service->processFooBar($entity, async: false); // ✅ One call
    }
}

// Service — all logic here
class FooBarsService extends EntitiesService {
    public function processFooBar(FooBar $entity, bool $async = true): void {
        if ($async) { (new FooBarMessage($entity->id))->dispatch(); return; }
        $this->doStep1($entity);
        $this->doStep2($entity);
        $this->doStep3($entity);
    }
}

3. Service Method Pattern: bool $async = false

Business logic lives in a service method that accepts $async. The caller decides whether to run immediately or enqueue:

public function doSomething(int $entityId, bool $async = false): void
{
    if ($async) {
        (new DoSomethingMessage($entityId))->dispatch();
        return;
    }
    // synchronous execution
    // ...actual work...
}

Handler side: always calls the service method with async: false to avoid re-dispatch loops.

4. Always Propagate Auth Context

In every handler, call:

$this->setAuthAccountFromMessage($message);

This ensures the message is processed with the rights of the account that dispatched it.

5. Workspace Routing Guard

Handlers must check workspace routing before processing:

if ($message->processOnWorkspaceIfNecessary()) {
    return;
}

Order: set auth -> workspace guard -> run.

6. Logging Pattern

• Use $this->getLogger() for all logging -- never DDDService::instance()->getLogger() directly in handlers
• Log an info line before work starts with identifying IDs
• Wrap the body in try/catch (Throwable $t)
• In the catch block, use $this->logShortException() for structured error logging

$this->getLogger()->info("Processing FooBar for {$message->fooBarId}");
try {
    // ...work...
} catch (Throwable $t) {
    $this->logShortException(
        $this->getLogger(),
        "Processing FooBar for {$message->fooBarId}",
        $t
    );
}

7. Time/Memory Limits for Heavy Jobs

set_time_limit(120);           // 120s for standard tasks, 500s for heavy imports
ini_set('memory_limit', '1024M');

8. Admin Privilege Escalation (When Required)

Some async operations must run with admin privileges (e.g., cross-tenant data access):

$defaultAccount = DDDService::instance()->getDefaultAccountForCliOperations();
AuthService::instance()->setAccount($defaultAccount);

Use only when necessary. Still call setAuthAccountFromMessage() first for attribution.

---

Implementation Steps

Step 1: Choose Domain + Naming

Location: src/Domain/{Domain}/MessageHandlers/

Naming:

• FooBarMessage -- the message class
• FooBarHandler -- the handler class
• Transport name in snake_case: process_foo_bar

Step 2: Create the Message Class

<?php
declare(strict_types=1);

namespace {Namespace}\Domain\{Domain}\MessageHandlers;

use DDD\Domain\Base\Entities\MessageHandlers\AppMessage;

class FooBarMessage extends AppMessage
{
    public static string $messageHandler = FooBarHandler::class;

    public ?int $fooBarId = null;
    public bool $regenerate = false;

    public function __construct(?int $fooBarId = null, bool $regenerate = false)
    {
        parent::__construct();
        $this->fooBarId = $fooBarId;
        $this->regenerate = $regenerate;
    }
}

Rules:

• Extend AppMessage
• Define public static string $messageHandler = FooBarHandler::class;
• Use primitive payload (IDs, bools, strings) -- not entity objects
• Call parent::__construct() in constructor

Step 3: Create the Handler Class

Application-level handlers should extend a project-specific CustomAppMessageHandler (which extends AppMessageHandler) or AppMessageHandler directly:

<?php
declare(strict_types=1);

namespace {Namespace}\Domain\{Domain}\MessageHandlers;

use DDD\Domain\Base\Entities\MessageHandlers\AppMessageHandler;
use DDD\Infrastructure\Services\DDDService;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;
use Throwable;

#[AsMessageHandler(fromTransport: 'process_foo_bar')]
class FooBarHandler extends AppMessageHandler
{
    public function __invoke(FooBarMessage $message): void
    {
        set_time_limit(120);
        ini_set('memory_limit', '1024M');

        $this->setAuthAccountFromMessage($message);
        if ($message->processOnWorkspaceIfNecessary()) {
            return;
        }

        $this->getLogger()->info("Processing FooBar for {$message->fooBarId}");

        try {
            // 1. Load entity by ID
            $fooBar = FooBar::byId($message->fooBarId);
            if (!$fooBar) {
                return;
            }

            // 2. Call service method with async: false
            /** @var FooBarsService $fooBarsService */
            $fooBarsService = FooBars::getService();
            $fooBarsService->doSomething($fooBar->id, async: false);
        } catch (Throwable $t) {
            $this->logShortException(
                $this->getLogger(),
                "Processing FooBar for {$message->fooBarId}",
                $t
            );
        }
    }
}

Rules:

• Extend AppMessageHandler (or project-specific subclass)
• Add #[AsMessageHandler(fromTransport: 'transport_name')]
• Set auth context and workspace guard first
• Load entity by ID inside the handler (not from message payload)
• Call service method with async: false
• Wrap in try/catch with logShortException()

Step 4: Wire Messenger Transport + Routing

Update config/symfony/default/packages/messenger.yaml:

framework:
  messenger:
    transports:
      # Add transport (RabbitMQ queue)
      process_foo_bar: '%env(MESSENGER_TRANSPORT_DSN_RABBITMQ)%/process_foo_bar'

    routing:
      # Route message to transport
      '{Namespace}\Domain\{Domain}\MessageHandlers\FooBarMessage': process_foo_bar

Debug tip: For local troubleshooting, temporarily switch the transport to sync://.

Step 5: Add Supervisor Consumer

IMPORTANT: Search the project for an existing supervisor config file. Common locations:

config/system/supervisorWorkers.conf
config/supervisor/*.conf
docker/supervisor/*.conf

Use Glob with patterns like **/*supervisor* or **/*Workers.conf to find it.

Once found:

1. Read the existing file to learn the project's conventions — the command= path, user=, numprocs patterns, etc. Every project is different.
2. Copy an existing [program:] block as a template for the new entry.
3. Replace only the transport name (in [program:] and messenger:consume argument) with the new transport name from Step 4.
4. Append the new block to the end of the file.

General structure of a supervisor consumer block:

[program:{transport_name}]
command=php {path_from_existing_entries}/bin/console messenger:consume {transport_name} --no-debug --limit={N} --time-limit=3600 --memory-limit={M}
process_name=%(program_name)s_%(process_num)02d
numprocs={see below}
autostart=true
autorestart=true
startsecs=0
startretries=10
user={copy from existing entries}
redirect_stderr=true

Rules:

• Never hardcode paths — always copy the command= path prefix from an existing entry in the same file
• Never guess user= — copy from existing entries
• Transport name in [program:] and messenger:consume must match the transport name in messenger.yaml
• Set numprocs based on job cost:
  • IO-bound / short tasks: more workers (e.g., push notifications: 5-10)
  • Heavy CPU/memory: fewer workers (e.g., imports: 1)
  • Standard async processing: 1-2

Step 5.1: Worker recycling — --limit, --time-limit, --memory-limit

Long-running PHP workers accumulate static state across messages: Translatable's static currentLanguageCode / translationSettingsSnapshot, the entity registry / EntityCache, AuthService's account, opaque caches inside Firebase / external SDKs, etc. Symfony Messenger handlers DO NOT reset between messages — only between worker processes. State leaks from message N into message N+1 in the same worker, causing non-deterministic alternating bugs that are hard to reproduce.

The defence is to recycle worker processes regularly. Symfony Messenger ships three command-line limits that exit the worker after a threshold; supervisor's autorestart=true then immediately spawns a fresh process with clean static state.

Flag	Effect	Recommended baseline
--limit=N	Exit after N successfully-processed messages	per-transport, see below
--time-limit=3600	Exit after N seconds (defence against slow leaks)	3600 for every transport
--memory-limit=256M	Exit if memory exceeds threshold	match the transport's real workload

Sizing --limit:

Transport profile	Suggested --limit
State-sensitive, high-volume (push notifications, chat post-process)	100
State-sensitive, lower-volume (translations, embeddings, support post-process)	50
Long-running single jobs (Strava import, AI resolution, summary generation)	20
Heavy single jobs that pin a worker for minutes (track migration)	10
State-light high-volume (track geohashes)	200

Lower --limit = more frequent state hygiene at the cost of more PHP cold-start overhead. Cold start is ~1s; for a transport handling 10 msg/s, --limit=100 means a 1s pause every 10s of work — negligible.

Sizing --memory-limit: match the transport's real footprint. Push notifications: 256M. AI / translations / embeddings: 512M. Strava imports: 1024M. Track migration: 2048M. Don't set it lower than the actual job needs — otherwise you'll see workers killed mid-message and the message redelivered.

Why startsecs=0 + startretries=10: with --limit=100 and 100 fast messages, a worker can complete in <1 s. Supervisor's default startsecs=1 would treat that as a crash-loop and give up after startretries=3. Setting startsecs=0 accepts sub-second completions as legitimate; startretries=10 keeps the safety net for real failures.

Reload after editing: the user runs

sudo supervisorctl reread
sudo supervisorctl update

— this picks up the new config without bouncing all unaffected workers.

Step 5.2: Rebuild the container BEFORE (re)starting workers — --no-debug does NOT auto-rebuild

Symptom pattern: you change a transport (e.g. flip sync:// ↔ AMQP, add a new transport), supervisorctl start/restart the workers, and they're either absent, crash-looping, or behave as if the old config were still active — the queue is never declared, messages aren't consumed, or a publish runs inline instead of being queued.

Cause: supervisor consumer commands run with --no-debug (messenger:consume <transport> --no-debug …). --no-debug disables the debug kernel, which is the thing that rebuilds the compiled container on a config-file change. So a --no-debug worker reads the stale compiled container — it keeps seeing the OLD transport DSN (e.g. still sync://), so messenger:consume <transport> on a now-sync transport crashes immediately, or consumes the wrong transport. FPM and any messenger:consume run without --no-debug self-heal (they rebuild on change), which is why a foreground messenger:consume <transport> -vv "works" while the supervisor-managed --no-debug workers silently don't.

Rule: rebuild the container + reset the cache BEFORE every supervisor start/stop/restart that follows a code or transport change. Either:

• bin/console cache:clear (the kernel the workers run under), or
• start the worker once without --no-debug (messenger:consume <transport> -vv) — the debug kernel rebuilds the shared container, then Ctrl-C and let supervisor take over.

FPM and the workers share the same var/cache/<env> dir, so one rebuild fixes both the consume side (workers) and the publish side (the dispatching request). Diagnose a silently-failing worker by running it in the foreground without --no-debug — that prints the real exception (transport-not-found / does-not-support-receiving / AMQP error) instead of swallowing it into a supervisor logfile.

Step 6: Use From Service

public function doSomething(int $fooBarId, bool $async = false): void
{
    if ($async) {
        (new FooBarMessage($fooBarId))->dispatch();
        return;
    }

    // Synchronous execution
    $fooBar = FooBar::byId($fooBarId);
    // ...actual work...
}

---

Checklist

• Message extends AppMessage
• Message has public static string $messageHandler = FooBarHandler::class;
• Message payload uses IDs/primitives (not entity objects)
• Message constructor calls parent::__construct()
• Handler extends AppMessageHandler (or project-specific subclass)
• Handler has #[AsMessageHandler(fromTransport: '...')]
• Handler calls $this->setAuthAccountFromMessage($message) first
• Handler calls $message->processOnWorkspaceIfNecessary() and returns early
• Handler uses $this->getLogger() (never DDDService::instance()->getLogger())
• Handler uses $this->logShortException() in catch blocks
• Handler calls service method with async: false
• messenger.yaml has transport + routing entries
• Supervisor config has a consumer for the transport
• Service method has bool $async = false signature
• Never use private -- always protected

---

Cross-Reference

• The service the handler delegates to — all business logic and the bool $async = false method live in the service; see ddd-service-specialist.
• Loading the entity by ID in the handler — FooBar::byId(...) and the repository/lazy-load mechanics are owned by ddd-entity-specialist.
• Triggering a message from a console command — see ddd-cli-command-specialist for the dispatch-from-CLI entry point.