PHP code example of crumbls / fanout

    
        
<?php
require_once('vendor/autoload.php');

/* Start to develop here. Best regards https://php-download.com/ */

    

crumbls / fanout example snippets

// in your existing webhook controller / job
use Crumbls\Fanout\Facades\Fanout;

public function handle(Request $request): Response
{
    $payload = $request->all();

    // ...your existing production logic...

    Fanout::dispatch('stripe-prod', $payload, $request->headers->all());

    return response()->noContent();
}

'endpoints' => [
    'self'    => ['url' => 'https://prod.example.com/internal/handler', 'enabled' => true],
    'staging' => ['url' => env('STAGING_WEBHOOK_URL'),                   'enabled' => true],
    'dev'     => ['url' => env('DEV_WEBHOOK_URL'),                        'enabled' => env('FANOUT_DEV_ENABLED', false)],
],

// config/fanout.php
return [
    'route' => [
        'enabled'    => true,
        'prefix'     => env('FANOUT_ROUTE_PREFIX', 'fanout/in'),
        'middleware' => ['api'],
    ],

    'queue' => [
        'connection' => env('FANOUT_QUEUE_CONNECTION'),
        'queue'      => env('FANOUT_QUEUE', 'fanout'),
    ],

    'models' => [
        'event'    => Crumbls\Fanout\Models\FanoutEvent::class,
        'delivery' => Crumbls\Fanout\Models\FanoutDelivery::class,
    ],

    'profiles' => [

        'stripe-prod' => [
            'persist'                       => 'full',
            'validator'                     => Crumbls\Fanout\Validators\StripeSignatureValidator::class,
            'secret'                        => env('STRIPE_WEBHOOK_SECRET'),
            'signature_header'              => 'Stripe-Signature',
            'continue_on_endpoint_failure'  => true,

            'endpoints' => [
                'staging' => [
                    'url'              => env('STAGING_WEBHOOK_URL'),
                    'enabled'          => true,
                    'environment'      => 'staging',
                    'timeout'          => 10,
                    'headers'          => [
                        'X-Fanout-Source' => 'production',
                        'X-Fanout-Event'  => '{event.id}',
                    ],
                    'signer'           => Crumbls\Fanout\Signers\HmacSha256Signer::class,
                    'secret'           => env('STAGING_WEBHOOK_SECRET'),
                    'signature_header' => 'X-Fanout-Signature',
                    'retry'            => ['attempts' => 5, 'backoff' => 'exponential', 'base_seconds' => 5],
                    'rate_limit'       => ['per_minute' => 60],
                ],

                'dev' => [
                    'url'         => env('DEV_WEBHOOK_URL'),
                    'enabled'     => env('FANOUT_DEV_ENABLED', false),
                    'environment' => 'dev',
                ],
            ],
        ],

    ],
];

namespace App\Models;

use Crumbls\Fanout\Models\FanoutEvent as BaseEvent;

class FanoutEvent extends BaseEvent
{
    protected function casts(): array
    {
        return array_merge(parent::casts(), [
            'payload' => MyKmsEncryptedArrayCast::class,
            'headers' => MyKmsEncryptedArrayCast::class,
        ]);
    }
}

// config/fanout.php
'models' => [
    'event'    => App\Models\FanoutEvent::class,
    'delivery' => Crumbls\Fanout\Models\FanoutDelivery::class,
],

namespace App\Webhooks;

use Crumbls\Fanout\Contracts\SignatureValidator;

class ShopifyWebhookValidator implements SignatureValidator
{
    public function verify(string $rawBody, array $headers, array $config): bool
    {
        $secret = (string) ($config['secret'] ?? '');
        $provided = $headers['x-shopify-hmac-sha256'][0] ?? null;
        if ($secret === '' || $provided === null) {
            return false;
        }

        $expected = base64_encode(hash_hmac('sha256', $rawBody, $secret, true));

        return hash_equals($expected, $provided);
    }
}

'validator' => App\Webhooks\ShopifyWebhookValidator::class,
'secret'    => env('SHOPIFY_WEBHOOK_SECRET'),

namespace App\Webhooks;

use Crumbls\Fanout\Contracts\PayloadFilter;
use Crumbls\Fanout\Models\FanoutEvent;
use Crumbls\Fanout\Support\EndpointConfig;

class DropTestEvents implements PayloadFilter
{
    public function shouldDeliver(array $payload, EndpointConfig $endpoint, ?FanoutEvent $event): bool
    {
        // For Stripe-style sources: livemode=false means it's a test event
        return ! ($payload['livemode'] ?? true);
    }
}

namespace App\Webhooks;

use Crumbls\Fanout\Contracts\PayloadTransformer;
use Crumbls\Fanout\Models\FanoutEvent;
use Crumbls\Fanout\Support\EndpointConfig;

class StripPii implements PayloadTransformer
{
    public function transform(array $payload, EndpointConfig $endpoint, ?FanoutEvent $event): array
    {
        unset(
            $payload['data']['object']['email'],
            $payload['data']['object']['phone'],
            $payload['data']['object']['shipping']['address'],
        );

        return $payload;
    }
}

'filter'    => App\Webhooks\DropTestEvents::class,
'transform' => App\Webhooks\StripPii::class,

'headers' => [
    'X-Fanout-Source' => 'production',
    'X-Fanout-Event'  => '{event.id}',
    'X-Event-Type'    => '{event.type}',
],

'retry' => [
    'attempts'     => 5,
    'backoff'      => 'exponential', // 'fixed' | 'linear' | 'exponential'
    'base_seconds' => 5,
],

'rate_limit' => ['per_minute' => 60],

use Crumbls\Fanout\Facades\Fanout;

Fanout::replay($event);
Fanout::replay($eventId, endpoint: 'staging');
Fanout::replayFailed(profile: 'stripe-prod');

Fanout::dispatch('stripe-prod', $payload, $headers);

Schedule::command('fanout:purge')->daily();

'endpoints' => [
    'staging' => ['url' => env('STAGING_WEBHOOK_URL'), 'enabled' => true],
    'dev'     => ['url' => env('DEV_WEBHOOK_URL'),     'enabled' => env('FANOUT_DEV_ENABLED', false)],
],

class OnlyTheseEventTypes implements Crumbls\Fanout\Contracts\PayloadFilter
{
    public function __construct(private array $allowed) {}

    public function shouldDeliver(array $payload, $endpoint, $event): bool
    {
        return in_array($payload['type'] ?? null, $this->allowed, true);
    }
}

$this->app->bind(OnlyTheseEventTypes::class, fn () => new OnlyTheseEventTypes(
    explode(',', (string) env('FANOUT_DEV_EVENT_TYPES', '')),
));

'staging' => [
    'signer' => HmacSha256Signer::class,
    'secret' => env('STAGING_WEBHOOK_SECRET'),
],
'dev' => [
    'signer' => HmacSha256Signer::class,
    'secret' => env('DEV_WEBHOOK_SECRET'),
],

Fanout::dispatch('stripe-prod', [
    'type' => 'invoice.paid',
    'data' => ['object' => ['id' => 'in_123']],
]);

Fanout::replay($eventId, endpoint: 'dev');

bash
composer fanout:install
php artisan migrate

bash
php artisan fanout:purge          # removes rows past their purgeable_at
php artisan fanout:purge --dry-run

bash
php artisan queue:work --queue=fanout

bash
php artisan fanout:replay <event_id> --endpoint=dev