Implement an Outbound Writer¶

Use this guide when a system has to be able to receive data from Conflux. A writer is stage 5b of outbound delivery: it takes a projected record and puts it into a system Conflux does not own. It is the only part of the delivery pipeline that knows a target's API.

The examples below build a writer for a fictional AcmeCrm — substitute your own target. For a real one, see UserPlatformOutboundWriter, which creates and updates Positive User contacts.

Prerequisites¶

The target system exists as a System with its own SystemDriver case
It declares push_out for the domains it should receive, in its direction matrix (ADR-0004) — without it, routing resolves no route and the runner delivers nothing
Its update path is addressed by an identifier the target itself hands back, and it answers distinguishably when that identifier no longer exists

The last one is not a formality. It is the only channel through which Conflux ever learns that a record it created was deleted in the target. A target whose update silently creates instead, or that answers the same way for "not found" and "bad request", cannot be written to safely: Conflux would recreate records somebody deleted on purpose and never know.

1. Declare the verbs in the type¶

Extend App\Services\AbstractOutboundWriter, implement one capability interface per verb the target offers, and compose the matching trait for each:

final class AcmeCrmWriter extends AbstractOutboundWriter implements
    CanCreateRemoteRecords,
    CanUpdateRemoteRecords,
    CanDeleteRemoteRecords
{
    use CreatesRemoteRecords;
    use DeletesRemoteRecords;
    use UpdatesRemoteRecords;

    public function requestsPerMinute(): ?int
    {
        return 120;
    }
}

Implement an interface only when the target offers the path and you write the method below. This example omits CanDeactivateRemoteRecords because Acme CRM has no inactive state — so it also writes no performDeactivate(), and callers can see that from the type without asking.

The interface and its trait belong together, always. The interface declares the public verb; the trait supplies it, wrapped in the circuit breaker, the span and the error mapping, and leaves you the perform*() half. Implement the interface without the trait and you would have to write the public verb yourself — reaching the target with none of that. The architecture test a writer that declares a capability composes the trait that supplies it fails the build if you do.

requestsPerMinute() is the only fact about a target that is not tied to a verb. It defaults to null on the base class, so leave it out entirely unless your target publishes a rate — this example overrides it because Acme CRM does. Keep an override free of I/O; it is read when a sync rule is saved, before any delivery happens.

Everything else a target has to say about itself belongs to the verb it qualifies. A writer implementing CanIngestRemoteRecords also answers bulkIngestSuppressesSideEffects(), because whether a bulk path fires the notifications an ordinary write sets off is the fact that decides whether a first load of 50,000 records is survivable — and it is a question a writer without a bulk path is never asked.

2. Implement the verbs you declared¶

One perform*() per trait. The trait declares each abstract, so a missing one fails at compile time rather than at delivery time.

protected function performCreate(SystemConnection $connection, Projection $projection): string
{
    $response = $this->request($connection)
        ->post($this->endpoint($connection, $projection->domain), $this->payload($projection))
        ->throw();

    $remoteId = $response->json('id');

    if (! is_string($remoteId) || $remoteId === '') {
        throw new UnaddressableRemoteRecordException('Acme CRM created a record but returned no id for it.');
    }

    return $remoteId;
}

Note the return type: the identifier alone, not a RemoteReference. The trait pairs it with the projection's domain, so you cannot label a record with the wrong one — which matters because the ledger is keyed by domain, and a wrong one makes the next run create the record a second time.

Throw rather than invent an identifier — and throw the right thing. The identifier you return is the only pointer to what you just made, and it cannot be reconstructed: nothing can ask the target which of its records are ours. UnaddressableRemoteRecordException is the type for it, not OutboundWriteException, because the latter promises the write is safe to retry and this one is the opposite — the record already exists, so a retry makes a second one just as unreachable.

If you return an empty string instead of throwing, the trait raises the same exception for you. That backstop exists because the fault is silent otherwise, but your own throw carries the better message: you know what the response looked like.

The bulk path has the same backstop, and there it is worth avoiding rather than relying on: an empty identifier anywhere in the batch discards the whole result, good identifiers included. A record the target took without handing back an identifier was not usefully accepted — report it as an IngestRejection instead, and the rest of the batch survives.

protected function performUpdate(SystemConnection $connection, Projection $projection, RemoteReference $reference): void
{
    $response = $this->request($connection)->put(
        $this->recordUrl($connection, $reference),
        $this->payload($projection),
    );

    if ($response->status() === 404) {
        throw new RemoteRecordGoneException(sprintf('Acme CRM no longer holds record [%s].', $reference->remoteId));
    }

    $response->throw();
}

Map "not found" to RemoteRecordGoneException and nothing else to it. Mapping a 404 onto an ordinary failure costs the one signal that tells Conflux a record was deleted on purpose; mapping an ordinary failure onto it stops a record being delivered for good. Note the ->throw() comes after the 404 check — the other way round, the status is never reached.

performDelete() follows the same shape.

Refuse a connection in requireDeliverableConnection(), not inside a verb. If your target has a precondition that is about the connection rather than the record — a status that no longer entitles it to deliveries, credentials it cannot be addressed with — override that hook:

protected function requireDeliverableConnection(SystemConnection $connection): void
{
    $this->client->deliveryRequest($connection);
}

write() calls it before the breaker and the error mapping, so the exception reaches the caller with its own type. Checked inside a verb instead, it is thrown within the breaker's callback and comes out as a retryable OutboundWriteException with a breaker failure counted against the target — so one rotated token puts every record of the run on a retry schedule that cannot succeed.

Everything else the target needs stays inside the class:

/**
 * @return array<string, mixed>
 */
private function payload(Projection $projection): array
{
    /** @var array<string, mixed> $payload */
    $payload = $projection->record->toArray();

    return $payload;
}

private function request(SystemConnection $connection): PendingRequest
{
    /** @var array<string, mixed> $credentials */
    $credentials = $connection->credentials ?? [];
    $token = is_string($credentials['api_token'] ?? null) ? $credentials['api_token'] : '';

    return Http::withToken($token)->acceptJson();
}

private function endpoint(SystemConnection $connection, SystemCapability $domain): string
{
    return sprintf('%s/v1/%s', rtrim((string) $connection->base_url, '/'), $domain->value);
}

The payload() indirection is not decoration: toArray() is annotated array<mixed>, which PHPStan level 10 refuses to pass to the HTTP client's $data. The narrowing docblock is the same convention the existing integrations use for $credentials.

Once the mapping is more than a pass-through, it belongs in its own class, one per domain. A writer knows how a target is addressed, authenticated and how its answers read; what a customer, an order or a product is in that target is a different question, and it grows field by field. UserPlatformContactMapper is the worked example: one invokable call, no state, no I/O, named so that an …OrderMapper sits beside it rather than inside it. The writer then names each domain in a table — which mapper a record goes through, and which collection it is written to — so a new domain costs those two arms and its own mapper, and inherits nothing from the first domain about which verbs its API offers. Tests follow the same line: the mapper's own file asks it for a payload directly, and the writer's file proves only that it sends what came back.

3. Register the writer¶

Add one line to AppServiceProvider::registerOutboundWriters(), which boot() calls with the registry and which holds every writer registration in one place:

public function registerOutboundWriters(OutboundWriterRegistry $writers): void
{
    $writers->register(SystemDriver::WooCommerce, WooCommerceOutboundWriter::class);
    $writers->register(SystemDriver::PrestaShop, PrestaShopOutboundWriter::class);
    $writers->register(SystemDriver::UserPlatform, UserPlatformOutboundWriter::class);
    $writers->register(SystemDriver::AcmeCrm, AcmeCrmWriter::class);
}

The registry holds one writer per driver. A target that carries several domains, as Positive User carries customers and customer_consents, gets one writer that handles them all, and a writer that can express only part of a domain says so in accepts().

OutboundWriterRegistry is a sibling of IntegrationRegistry, not an extension of it (ADR-0020). A driver can appear in one, the other, or both. The registry resolves through the container, so a writer with no constructor of its own gets CircuitBreaker and TracerInterface injected; one with its own dependencies passes them up with parent::__construct().

4. What a writer must not do¶

Do not catch transport errors. AbstractOutboundWriter maps anything unexpected to OutboundWriteException, the one type the ledger records. A try/catch that swallows a failure turns it into a silent success, and the runner writes delivered for a call that never landed.
Do not retry, sleep, or throttle inside a call. requestsPerMinute paces the runner; a writer that throttles itself holds a worker rather than yielding it. Retries belong to the ledger, which counts attempts and schedules the next one.
Do not search the target for a record. There is no resolve verb. The ledger answers "is there a record for this source id" from a local row, exactly, without a network call (ADR-0015).
Do not decide create against update. The runner has already decided from the ledger before it calls you.
Do not interpret the delivery policy. Whether a tombstone means deactivate, delete or nothing is the rule's delete_policy, not the writer's call.
Do not add a verb you cannot perform. There is no flag to set to false and no method to leave throwing: if the target has no update path, leave CanUpdateRemoteRecords off, and a caller sees it in the type.
Do not add fields the DTO does not carry. The payload shape lives in the DTO layer; a field only one target wants is either defined in the canonical DTO or not sent at all.

5. Prove it with tests¶

Cover each declared verb against Http::fake(), never a live target. The two cases worth writing first are the ones that are load-bearing rather than obvious:

test('surfaces a missing remote record as gone rather than as a failure', function (): void {
    Http::fake(['*' => Http::response(status: 404)]);

    $writer = app(AcmeCrmWriter::class);

    expect(fn () => $writer->update($connection, $projection, $reference))
        ->toThrow(RemoteRecordGoneException::class);
});

test('refuses a create whose response carries no identifier', function (): void {
    Http::fake(['*' => Http::response(['name' => 'Ada'])]);

    $writer = app(AcmeCrmWriter::class);

    expect(fn (): RemoteReference => $writer->create($connection, $projection))
        ->toThrow(OutboundWriteException::class);
});

There is no test to write for "refuses a verb it does not support", and no exception for it either: that verb does not exist on the class, so nothing can call it. The check a caller performs is instanceof, and PHPStan verifies it statically.

The second one above is the test people skip. A create whose response is not parsed correctly leaves a record in the target that nothing can address, and the next run creates a second one — with no way to detect or repair it afterwards.