Pub/Sub

<?php
// ============================================================
// 1) In-process pub/sub (synchronous)
// ============================================================
class Topic {
    /** @var callable[] */
    private array $subscribers = [];

    public function subscribe(callable $handler): void {
        $this->subscribers[] = $handler;
    }

    public function publish(array $event): void {
        foreach ($this->subscribers as $h) {
            try { $h($event); }
            catch (Throwable $e) {
                error_log('handler failed: ' . $e->getMessage());
                // one handler failure should NOT block siblings
            }
        }
    }
}

$orderEvents = new Topic();

// Subscribers
$orderEvents->subscribe(function (array $event) {
    if ($event['type'] === 'OrderPaid') {
        echo "send receipt for order {$event['orderId']}
";
    }
});
$orderEvents->subscribe(function (array $event) {
    if ($event['type'] === 'OrderPaid') {
        echo "increment analytics for {$event['orderId']}
";
    }
});

// Publish
$orderEvents->publish([ 'type' => 'OrderPaid', 'orderId' => 'o1' ]);

// ============================================================
// 2) Synchronous vs asynchronous
// ============================================================
// Synchronous in-process pub/sub blocks the publisher until all handlers run.
// For slow side effects (HTTP, email), push to a queue and run async.

// ============================================================
// 3) Redis Streams — at-least-once durable fan-out
// ============================================================
// Producer
$r = new Redis();
$r->connect('127.0.0.1', 6379);

$r->xAdd('orders.events', '*', [
    'type' => 'OrderPaid', 'orderId' => 'o1', 'amount' => '4995',
]);

// One-time setup of consumer group
try { $r->xGroup('CREATE', 'orders.events', 'analytics', '$', true); } catch (Throwable $e) {}
try { $r->xGroup('CREATE', 'orders.events', 'mailer',    '$', true); } catch (Throwable $e) {}

// Worker — analytics consumer
function run_worker(string $group, string $consumer): void {
    $r = new Redis(); $r->connect('127.0.0.1', 6379);
    while (true) {
        $entries = $r->xReadGroup($group, $consumer, [ 'orders.events' => '>' ], 16, 5000);
        if (!$entries) continue;
        foreach ($entries['orders.events'] as $id => $fields) {
            try {
                handle_event($fields);
                $r->xAck('orders.events', $group, [$id]);
            } catch (Throwable $e) {
                // leave unacked -> XPENDING will redeliver after timeout
                error_log('handler failed: ' . $e->getMessage());
            }
        }
    }
}

function handle_event(array $fields): void {
    // do the work
}

// ============================================================
// 4) Patterns to internalise
// ============================================================
// - Publishers KNOW NOTHING about subscribers
// - Each subscriber has its own retry / failure policy
// - Use durable streams (Kafka, Streams) for inter-service events
// - In-process pub/sub for cross-cutting side effects
// - Idempotency on the consumer side (events may be redelivered)
// - Schema versioning: events are forever; design the payload conservatively

// ============================================================
// 5) Decision matrix
// ============================================================
// - Same process, lightweight side effects     -> in-process Topic
// - Cross-service, durable fan-out              -> Kafka / Redis Streams / SNS+SQS
// - Real-time browser updates                   -> WebSockets / Server-Sent Events
// - 'I need exactly-once'                        -> hard. Use idempotency keys + at-least-once.
// - 'I want to query past events'                -> Kafka with infinite retention, or an outbox table

// ============================================================
// 6) Pitfalls
// ============================================================
// - In-process pub/sub for slow handlers -> blocks the request
// - No idempotency on consumers -> double-charges, double-emails
// - Publishing inside a transaction that may roll back -> phantom events
//   (use the outbox pattern: insert event row in the same tx; worker reads + publishes)
// - Schema changes that break consumers -> version events; never repurpose fields

// Async, often cross-process. Publishers and subscribers don't know each other.
// Real-world: Redis pub/sub, NATS, Kafka, MQTT, Postgres LISTEN/NOTIFY.