Accepting Data You Don't Control

Something outside your control needs to send you data: webhooks from a payment provider, callbacks from a partner integration, telemetry from devices in the field, analytics events from a front end team, domain events from a service you did not write. Whatever it is, you didn’t design the payload, you can’t version it on your own schedule, and you’ll hear about changes to it after they ship rather than before.

That last part is what makes an ingest server a coupling problem rather than a modelling one, and it’s where most of them go wrong.

The instinct is to model the thing properly. Validate every field, give each field a column, write a migration. That instinct is right for data you own. For data you don’t, it welds your deploy cycle to somebody else’s roadmap, and you find out about it one field at a time: a migration for this, a controller branch for that, and a 422 for anything you didn’t anticipate, which is a row of data you will never see again.

So I want to walk through the alternative, which rests on one idea: you own the envelope, you don’t own the payload, and that raw payload is the only thing in the system you’re unwilling to lose. Everything derived from it is disposable by design.

Laravel 13 throughout:

laravel new ingest
php artisan install:api
composer require juststeveking/cloudevents

That last package needs PHP 8.4, which is a step above Laravel 13’s own 8.3 floor, worth knowing before you add it to something already running.

Schema on write for the envelope, schema on read for the payload

The trick isn’t to abandon structure, it’s to be precise about which structure is actually yours.

I use CloudEvents for the envelope, it’s a CNCF specification that describes an event in a transport-agnostic way, and it enforces the split I want, between a fixed set of context attributes that describe the event and a data block carrying whatever the producer wants to send.

{
    "specversion": "1.0",
    "id": "9c8b2e1a-1f4d-4f0f-9e3a-2b6a5f0d1c77",
    "source": "/billing/stripe",
    "type": "com.acme.subscription.renewed.v1",
    "subject": "cus_01HZY",
    "time": "2026-08-24T09:14:22Z",
    "datacontenttype": "application/json",
    "data": {
        "plan": "team",
        "amount": 4900,
        "currency": "gbp"
    }
}

The attributes are schema on write, I validate them, index them and route on them, and I will reject anything that does not have them. The data block is schema on read. I store it as it arrives and I don’t form an opinion about it until something needs one.

Two attributes do most of the work. id combined with source gives you an idempotency key for free, which you need because every real producer retries, and you will receive things twice. And type becomes the routing key for the whole system, which is why the version belongs in it. When the producer changes the shape of a payload, they emit v2, you write a second handler, and the two coexist for as long as they need to. No coordinated deploy, no flag day.

What if you can’t make producers speak CloudEvents? Often you can’t, because they’re Stripe or GitHub. Then you put a thin adapter at the edge that wraps their body in an envelope you generate. The important thing is that everything past the front door looks the same.

For the PHP side I use my own juststeveking/cloudevents, it’s a pretty small (and simple) package. You get a CloudEvent value object with the eight spec attributes, a SPEC_VERSION constant, and a toArray() that emits the wire format. It does no validation, which is the right call. Validation belongs to your Form Request, and the value object’s job is to stop the envelope from decaying into an associative array the moment it leaves the controller.

Validating the envelope and nothing else

Here’s the Form Request. What matters about it is the line that isn’t there:

final class IngestEventRequest extends FormRequest
{
    public function rules(): array
    {
        return [
            'specversion' => ['required', 'string', 'in:'.CloudEvent::SPEC_VERSION],
            'id' => ['required', 'string', 'max:255'],
            'source' => ['required', 'string', 'max:255'],
            'type' => ['required', 'string', 'max:255'],
            'subject' => ['nullable', 'string', 'max:255'],
            'time' => ['required', 'date'],
            'datacontenttype' => ['nullable', 'string', 'in:application/json'],
            'data' => ['required', 'array'],
        ];
    }

    public function payload(): CloudEvent
    {
        return new CloudEvent(
            id: $this->string('id')->toString(),
            source: $this->string('source')->toString(),
            type: $this->string('type')->toString(),
            data: $this->array('data'),
            dataContentType: $this->string('datacontenttype')->toString() ?: null,
            subject: $this->string('subject')->toString() ?: null,
            time: $this->date('time')->toImmutable(),
        );
    }
}

'data' => ['required', 'array'] is the entire flexibility story. I assert that it’s an array and then I stop asking questions. There is no data.plan, no data.amount, nothing that ties my release schedule to a decision made in somebody else’s sprint planning.

payload() returns a CloudEvent rather than a DTO I had to write, which is most of why the package is here at all.

One thing to watch. The package also ships a CloudEvent::make() that builds from an array, but it expects snake_case keys, data_content_type and data_schema, while the wire format that just arrived in your request body uses the spec’s flattened datacontenttype and dataschema. Feed one into the other and those two attributes silently come back null. make() is for arrays you built yourself; at the edge, construct it directly and skip the translation.

Note also that data is typed mixed, so the decoded array goes straight in. No json_encode on the way through, which keeps the payload queryable once it lands in a jsonb column.

One table, and the two indexes that matter

Schema::create('events', function (Blueprint $table): void {
    $table->ulid('id')->primary();
    $table->string('source');
    $table->string('event_id');
    $table->string('type')->index();
    $table->string('subject')->nullable()->index();
    $table->jsonb('data');
    $table->timestamp('occurred_at');
    $table->timestamp('processed_at')->nullable()->index();
    $table->timestamps();

    $table->unique(['source', 'event_id']);
});

The unique constraint on source and event_id is the deduplication strategy. An index, rather than a cache check or a where clause in PHP. Retries arrive concurrently, so any check-then-insert you write in PHP will eventually let a duplicate slip through the gap between the two statements. Let the database be the thing that says no.

processed_at as a nullable timestamp rather than a status enum is a choice I’ll come back to.

jsonb rather than json matters if you’re on Postgres, and you should be. It gives you operators and indexes over the payload, which is what keeps schema on read from being a polite name for a write-only table.

Accept and get out of the way

The ingest path has exactly one job: make the event durable and return. Everything else you do here couples your uptime to your interpretation logic, and interpretation logic is the part that changes constantly.

final readonly class IngestEvent
{
    public function handle(CloudEvent $event): Event
    {
        $record = Event::query()->firstOrCreate(
            attributes: [
                'source' => $event->source,
                'event_id' => $event->id,
            ],
            values: [
                'type' => $event->type,
                'subject' => $event->subject,
                'data' => $event->data,
                'occurred_at' => $event->time,
            ],
        );

        if ($record->wasRecentlyCreated) {
            ProcessEvent::dispatch($record->id);
        }

        return $record;
    }
}

firstOrCreate leans on the unique constraint. When two deliveries race, one insert wins and the other catches the violation and re-reads the winner. The wasRecentlyCreated check means a duplicate never dispatches a second job, so you have killed one class of double processing before the queue is even involved.

The controller has almost nothing left to do:

final readonly class StoreController
{
    public function __construct(
        private IngestEvent $action,
    ) {}

    public function __invoke(IngestEventRequest $request): JsonResponse
    {
        $event = $this->action->handle(
            payload: $request->payload(),
        );

        return new JsonDataResponse(
            data: ['id' => $event->id],
            status: Response::HTTP_ACCEPTED,
        );
    }
}

A 202 and a ULID. The producer learns that its data is safe and learns nothing about whether you understood it, which is about the right amount of information to share.

Where you are allowed to have opinions

This is the part I think people get wrong, and it’s the reason the whole design holds together.

You do still need to validate the payload. A field can be missing, an amount can arrive as a string, a plan can be a name you have never seen. You can’t write correct code against a payload you haven’t checked. So the real question is where to do it, and that follows from asking what happens when validation fails.

Validation failure at ingest is data loss. You return a 422, the producer logs a warning nobody reads, and the event is gone for good.

Validation failure during processing is a retry. The event is already on disk. You fix the processor, or you ask the producer what changed, and you run it again.

So the payload rules live in the processor:

final readonly class RecordRenewal implements Processor
{
    public function handle(Event $event): void
    {
        $data = Validator::make($event->data, [
            'plan' => ['required', 'string'],
            'amount' => ['required', 'integer'],
            'currency' => ['required', 'string', 'size:3'],
        ])->validate();

        $projection = MonthlyRevenue::query()->firstOrCreate(
            attributes: [
                'month' => $event->occurred_at->format('Y-m'),
                'currency' => $data['currency'],
            ],
            values: ['total' => 0],
        );

        $projection->increment('total', $data['amount']);
    }
}

Same strictness as a Form Request, moved somewhere recoverable. That single change is what lets you be relaxed at the door without being reckless about correctness.

Set tries low on the job while you are at it. A payload that fails validation will fail identically on every attempt, so retrying it is just noise in your logs. Either way the event stays on disk with processed_at still null, which is the only property that actually matters.

One thing that projection migration needs, which I have not shown: a unique index on month and currency. firstOrCreate is only safe under concurrency because it falls back to a select when the insert trips a constraint, and the select it retries with is built from the attributes you passed it. With no unique index, two workers happily create two rows. With a unique index on columns other than the ones you passed, the retry finds nothing and rethrows.

Resolving a handler from the type

Now the routing. How do you get from a type string to a class without a match statement that grows a new arm every fortnight?

interface Processor
{
    public function handle(Event $event): void;
}

final class ProcessorRegistry
{
    /** @var array<string, class-string<Processor>> */
    private array $processors = [];

    public function register(string $type, string $processor): void
    {
        $this->processors[$type] = $processor;
    }

    public function resolve(string $type): ?Processor
    {
        if (! isset($this->processors[$type])) {
            return null;
        }

        return app($this->processors[$type]);
    }
}

You could discover these by scanning for a PHP attribute, and it’d be nicer to write. I keep the map explicit anyway, because when something isn’t being processed at two in the morning I want one file that lists every type this server understands. Boring wins in the place you go to debug.

public function register(): void
{
    $this->app->singleton(ProcessorRegistry::class);
}

public function boot(ProcessorRegistry $registry): void
{
    $registry->register('com.acme.subscription.renewed.v1', RecordRenewal::class);
    $registry->register('com.acme.subscription.renewed.v2', RecordRenewalV2::class);
}

The job joins the two halves:

final class ProcessEvent implements ShouldQueue
{
    use Queueable;

    public function __construct(
        private readonly string $eventId,
    ) {}

    public function handle(ProcessorRegistry $registry): void
    {
        $event = Event::query()->findOrFail($this->eventId);

        if ($event->processed_at !== null) {
            return;
        }

        $processor = $registry->resolve($event->type);

        if ($processor === null) {
            return;
        }

        $processor->handle($event);

        $event->forceFill(['processed_at' => now()])->save();
    }
}

Look at what happens to a type you’ve never seen. It doesn’t throw. Nothing lands in failed jobs, nobody gets paged, and the row just sits there with a null processed_at.

That behaviour is worth arguing for, because the instinct is to treat an unrecognised type as an error when what you actually have is a backlog. A producer can start emitting something new this afternoon, entirely without telling you, and you can write the handler for it next month with every single event from the intervening weeks still sitting there waiting.

Two constraints come from running this on a queue. Your queue is at-least-once, so processors have to be idempotent, which the processed_at guard mostly handles. And you have no ordering guarantees, so processors want to be commutative where they can be. An atomic increment does not care what order it arrives in. If you find yourself writing something that genuinely needs sequence, that is a signal the state belongs inside the payload rather than in your accumulation of it.

Querying data you have not modelled yet

The obvious objection to storing payloads whole: how do you ever ask anything of them? On Postgres, more easily than you’d think.

Event::query()
    ->where('type', 'com.acme.subscription.renewed.v1')
    ->where('data->currency', 'gbp')
    ->whereBetween('occurred_at', [$from, $to])
    ->count();

That runs today, against events you ingested last year, for a question nobody had thought to ask when you wrote the schema. If a particular question becomes routine, you index it without touching the table structure:

DB::statement("
    CREATE INDEX events_renewal_currency_idx
    ON events ((data->>'currency'))
    WHERE type = 'com.acme.subscription.renewed.v1'
");

A partial expression index, scoped to one event type, added when the access pattern shows up rather than when you guessed it might. That’s the part that turns schema on read from a slogan into something you can actually operate.

The payoff: replay

Here’s where the discipline pays for itself.

Your projection tables are not data. They are a cache of a calculation over the events table. When the calculation turns out to be wrong, and it will be, because you wrote it against a payload you didn’t design, you don’t go back to the producer and beg them to resend six months of history. You throw the projection away and run the events through again.

final class ReplayEvents extends Command
{
    protected $signature = 'events:replay {--type=}';

    public function handle(): int
    {
        Event::query()
            ->when($this->option('type'), fn ($query, $type) => $query->where('type', $type))
            ->whereNull('processed_at')
            ->chunkById(500, function ($events): void {
                foreach ($events as $event) {
                    ProcessEvent::dispatch($event->id);
                }
            });

        return self::SUCCESS;
    }
}

Two things about that command matter. It uses chunkById rather than chunk, because the jobs it dispatches set processed_at and shrink the result set while you are still iterating over it, which is the case where offset paging starts skipping rows. And it adds no orderBy of its own, because chunkById only strips existing orders on the column it pages by. Add orderBy('occurred_at') and it survives, sorts ahead of the key order, and the cursor quietly starts missing records. Replay doesn’t need chronological order anyway. That was the whole point of keeping processors commutative.

This is where that nullable timestamp earns its keep. Rebuilding is two statements and a command:

MonthlyRevenue::query()->truncate();

Event::query()
    ->where('type', 'com.acme.subscription.renewed.v1')
    ->update(['processed_at' => null]);

php artisan events:replay --type=com.acme.subscription.renewed.v1

A status enum would have made that a state machine question. A nullable timestamp makes it a whereNull. Small decision, large difference in how the system feels to operate at the exact moment you’re least happy with it.

The same command drains the backlog of types you had not written handlers for. Register the processor, replay, and the history fills itself in behind you.

The seam at the far end

One more piece, kept short because it is a separate problem that deserves its own article.

When a processor writes derived state, something downstream usually wants to know. A cache needs busting, a partner wants a webhook, a warehouse wants the row. The trap is writing to the database and then publishing, because those aren’t atomic and one of them will fail on its own.

DB::transaction(function () use ($projection, $data): void {
    $projection->increment('total', $data['amount']);

    $message = new CloudEvent(
        id: (string) Str::ulid(),
        source: '/ingest/revenue',
        type: 'com.acme.revenue.updated.v1',
        data: $projection->only(['month', 'currency', 'total']),
        dataContentType: 'application/json',
        time: CarbonImmutable::now(),
    );

    OutboxMessage::query()->create([
        'type' => $message->type,
        'payload' => $message->toArray(),
    ]);
});

The state change and the intent to publish commit together or not at all, and a separate worker drains the outbox and retries as much as it likes. toArray() does the work here. It emits the flattened spec keys, so what lands in the outbox is already the shape a downstream consumer expects to receive.

None of that symmetry is accidental. At ingest you deduplicate on the producer’s identifier so that at-least-once delivery cannot corrupt you. At egress you write the message inside the transaction so that at-least-once publishing is possible at all. Same problem, opposite ends, and in both cases the answer is to make the durable write the only thing you rely on.

What you actually built

Count what has to change when a producer sends you something new tomorrow. A processor class, and a line in a service provider. Nothing on the ingest path, nothing at the edge, no controller branch and no deploy anybody has to coordinate. And if they started sending it before you were ready, which they will, you didn’t lose a thing in the meantime.

It all comes down to one refusal: you will not let your understanding of the data be a precondition for keeping it.

The outbox worker is the piece I skipped over here, and it gets interesting fast once you start thinking about ordering, poison messages and what to do when the consumer on the other end has been down for a day.