Communication Architecture

Version: 1.0.0

This document describes how notifito services communicate, including the event bus abstraction, queue configuration, and the transactional outbox pattern.

Overview

notifito uses asynchronous communication by default. Services communicate through an event bus (Redis queues) unless a human is waiting for the answer.

Communication Patterns

Pattern Use Case Example
Asynchronous Background processing Matching, sending emails
Synchronous Human waiting Widget submit, dashboard API

Event Bus Abstraction

The EventBus interface in packages/contracts provides a thin abstraction over Laravel queues. This exists so that moving to RabbitMQ later is a configuration change, not a refactor.

Interface

// packages/contracts/src/Bus/EventBus.php
interface EventBus
{
    public function publish(DomainEvent $event): void;

    /** @param iterable<DomainEvent> $events */
    public function publishAll(iterable $events): void;
}

Implementations

Implementation Purpose
QueueEventBus Production: publishes to Laravel queues
FakeEventBus Testing: in-memory event collection

Domain Event Interface

// packages/contracts/src/Bus/DomainEvent.php
interface DomainEvent
{
    public function topic(): string;
    public function correlationId(): string;

    /** @return array<string, mixed> */
    public function toArray(): array;

    /** @param array<string, mixed> $payload */
    public static function fromArray(array $payload): static;
}

Topic Routing

Events are routed to specific queues based on their topic.

Routing Table

Topic Producer Consumer Queue Description
availability.received edge, dashboard core core-matching New availability event
subscription.confirmation_requested core dispatch dispatch-send Send confirmation email
subscription.matched core dispatch dispatch-send Send notification email
delivery.status_changed dispatch core core-delivery Delivery status update

Queue Configuration

// apps/core/config/queue.php
'connections' => [
    'redis' => [
        'driver' => 'redis',
        'connection' => 'default',
        'queue' => 'core-matching,core-delivery',
        'retry_after' => 300,
    ],
],

Event Payloads

AvailabilityEventReceived

Emitted when a seller signals availability.

// packages/contracts/src/Events/AvailabilityEventReceived.php
final readonly class AvailabilityEventReceived implements DomainEvent
{
    public function __construct(
        public string $eventId,
        public string $tenantId,
        public string $itemTypeId,
        public ?string $itemId,
        public array $attributes,
        public ?int $capacity,
        public ?string $availableFrom,
        public ?string $availableUntil,
        public AvailabilitySource $source,
        public string $idempotencyKey,
        public string $correlationId,
        public string $occurredAt,
    ) {}

    public function topic(): string
    {
        return 'availability.received';
    }
}

Wire Format:

{
    "event_id": "evt-123",
    "tenant_id": "ten-456",
    "item_type_id": "it-789",
    "item_id": null,
    "attributes": {"size": "42"},
    "capacity": 3,
    "available_from": null,
    "available_until": null,
    "source": "push_api",
    "idempotency_key": "unique-key",
    "correlation_id": "corr-abc",
    "occurred_at": "2026-08-28T10:00:00+00:00"
}

ConfirmationRequested

Emitted when a subscription needs email confirmation.

// packages/contracts/src/Events/ConfirmationRequested.php
final readonly class ConfirmationRequested implements DomainEvent
{
    public function __construct(
        public string $tenantId,
        public string $tenantName,
        public string $subscriptionId,
        public string $contactPointId,
        public string $channel,
        public string $contactValue,
        public string $confirmUrl,
        public string $itemTypeName,
        public ?string $itemName,
        public array $attributes,
        public string $correlationId,
    ) {}

    public function topic(): string
    {
        return 'subscription.confirmation_requested';
    }
}

SubscriptionMatched

Emitted when availability matches subscriptions (batched).

// packages/contracts/src/Events/SubscriptionMatched.php
final readonly class SubscriptionMatched implements DomainEvent
{
    public function __construct(
        public string $tenantId,
        public string $tenantName,
        public string $availabilityEventId,
        public string $itemTypeName,
        public ?string $itemName,
        public array $eventAttributes,
        public array $matches,  // List of MatchedSubscription
        public string $correlationId,
    ) {}

    public function topic(): string
    {
        return 'subscription.matched';
    }
}

DeliveryStatusChanged

Emitted when delivery status updates.

// packages/contracts/src/Events/DeliveryStatusChanged.php
final readonly class DeliveryStatusChanged implements DomainEvent
{
    public function __construct(
        public string $tenantId,
        public string $deliveryId,
        public ?string $subscriptionId,
        public string $contactPointId,
        public string $status,
        public ?string $detail,
        public ?string $providerMessageId,
        public string $occurredAt,
        public string $correlationId,
    ) {}

    public function topic(): string
    {
        return 'delivery.status_changed';
    }
}

Transactional Outbox

The outbox pattern ensures events are never lost between commit and publish.

Problem

Without the outbox: 1. Domain change commits 2. Event is lost here (process crash, network failure) 3. Event is published

Solution

With the outbox: 1. Domain change and outbox row commit in same transaction 2. Relay publishes unpublished rows 3. Row is marked as published

Implementation

// apps/core/app/Domain/Outbox/OutboxEventBus.php
class OutboxEventBus implements EventBus
{
    public function publish(DomainEvent $event): void
    {
        OutboxMessage::query()->create([
            'topic' => $event->topic(),
            'payload' => $event->toArray(),
            'correlation_id' => $event->correlationId(),
        ]);
    }
}

// apps/core/app/Domain/Outbox/OutboxRelay.php
class OutboxRelay
{
    public function relay(): void
    {
        OutboxMessage::query()
            ->whereNull('published_at')
            ->each(function (OutboxMessage $message) {
                try {
                    $this->bus->publish(
                        $this->reconstruct($message)
                    );
                    $message->update(['published_at' => now()]);
                } catch (\Throwable $e) {
                    $message->update(['error' => $e->getMessage()]);
                }
            });
    }
}

Relay Command

# Run the outbox relay
./bin/artisan core outbox:relay

# Run continuously
./bin/artisan core outbox:relay --watch

Correlation IDs

Every request carries a correlation ID from ingest through send, enabling end-to-end tracing.

Flow

sequenceDiagram participant S as Seller participant E as edge participant C as core participant D as dispatch S->>E: POST /v1/availabilityX-Request-ID: abc-123 E->>C: AvailabilityEventReceivedcorrelation_id: abc-123 C->>D: SubscriptionMatchedcorrelation_id: abc-123 D->>D: Send emailcorrelation_id: abc-123

Implementation

// edge generates or propagates correlation ID
$correlationId = $request->header('X-Request-ID') 
    ?? Str::uuid()->toString();

// Carried through all events
$event = new AvailabilityEventReceived(
    correlationId: $correlationId,
    // ...
);

Queue Configuration

Horizon Supervisors

// apps/core/config/horizon.php
'environments' => [
    'production' => [
        'supervisor-matching' => [
            'connection' => 'redis',
            'queue' => ['core-matching'],
            'balance' => 'auto',
            'minProcesses' => 1,
            'maxProcesses' => 6,
            'tries' => 5,
            'timeout' => 120,
        ],
        'supervisor-delivery' => [
            'connection' => 'redis',
            'queue' => ['core-delivery'],
            'balance' => 'auto',
            'minProcesses' => 1,
            'maxProcesses' => 3,
            'tries' => 5,
            'timeout' => 60,
        ],
    ],
],

Retry Configuration

Queue Tries Backoff Timeout
core-matching 5 Default 120s
core-delivery 5 Default 60s
dispatch-send 6 60, 300, 900, 3600, 10800s 60s

Synchronous Communication

Used only where a human is waiting for the answer.

edge → core

Widget subscription submit:

// edge validates and forwards to core
$response = Http::withHeaders([
    'X-Notifito-Internal-Secret' => config('notifito.internal_secret'),
])
->post('http://core:8000/internal/subscriptions', $payload);

dashboard → core

All dashboard API calls:

// dashboard/app/Support/CoreClient.php
private function request(): PendingRequest
{
    return Http::baseUrl(config('notifito.core_url'))
        ->withHeaders([
            'X-Notifito-Internal-Secret' => config('notifito.internal_secret'),
        ]);
}

Internal Authentication

All internal HTTP calls require the X-Notifito-Internal-Secret header.

// Middleware in core
public function handle(Request $request, Closure $next)
{
    $secret = $request->header('X-Notifito-Internal-Secret');

    if (!hash_equals(config('notifito.internal_secret'), $secret)) {
        return response()->json(['message' => 'Unauthorized'], 401);
    }

    return $next($request);
}

Error Handling

Lost Events

Prevented by the transactional outbox pattern. See Transactional Outbox.

Double Sends

Prevented by database constraints:

-- Availability event idempotency
ALTER TABLE availability_events 
ADD CONSTRAINT availability_events_idempotency_key_unique 
UNIQUE (idempotency_key);

-- Notification deduplication
ALTER TABLE notifications 
ADD CONSTRAINT notifications_subscription_event_unique 
UNIQUE (subscription_id, availability_event_id);

Provider Outages

Exponential backoff with dead-letter queue:

// apps/dispatch/app/Jobs/SendDeliveryJob.php
public int $tries = 6;

public function backoff(): array
{
    return [60, 300, 900, 3600, 10800]; // ~6 attempts over several hours
}

Related Documentation