Implement a System Integration¶

Use this guide when adding support for a new e-commerce system. An integration translates system-native API responses into Conflux's unified DTOs so the rest of the system stays system-agnostic.

Prerequisites¶

The target system driver has a case in App\Enums\SystemDriver
A SystemConnection record for the target system exists (or will be created in tests via factory)

1. Create the integration class¶

Extend AbstractIntegration and implement its three abstract methods: performVerification() (see step 2), fetchBulkExportBatches(), and provideTransformerClass() (see step 3). Place the class in app/Integrations/<PlatformName>/ by convention — see App\Integrations\WooCommerce\WooCommerceIntegration for a real implementation.

<?php

declare(strict_types=1);

namespace App\Integrations\Acme;

use App\Contracts\DomainRecord;
use App\Data\BulkExportProbe;
use App\Data\CustomerData;
use App\Data\ExportRequest;
use App\Enums\SystemCapability;
use App\Enums\SystemDriver;
use App\Exceptions\IntegrationRequestException;
use App\Models\SystemConnection;
use App\Services\AbstractIntegration;
use App\Services\SourceDateTimes;
use Closure;
use Illuminate\Support\Collection;

class AcmeIntegration extends AbstractIntegration
{
    /** Acme sends `2024-03-01T09:05:00`, always in UTC. */
    public const string SOURCE_TIMEZONE = 'UTC';

    public const string DATETIME_FORMAT = 'Y-m-d\TH:i:s';

    public const string DATE_FORMAT = 'Y-m-d';

    public function __construct(private readonly AcmeApiClient $client) {}

    /** The datetime parse context for one export run. */
    public function sourceDateTimes(): SourceDateTimes
    {
        return new SourceDateTimes(SystemDriver::Acme, self::DATETIME_FORMAT, self::DATE_FORMAT, self::SOURCE_TIMEZONE);
    }

    /**
     * @param  Closure(Collection<int, DomainRecord>): void  $onBatch
     */
    protected function fetchBulkExportBatches(SystemConnection $connection, ExportRequest $request, Closure $onBatch): void
    {
        if ($request->capability !== SystemCapability::Customers) {
            return;
        }

        $source = $this->sourceDateTimes();

        try {
            $customers = $this->client->getCustomers($connection->base_url);
            $batch = collect($customers)->flatMap(
                fn (array $customer): array => AcmeTransformer::mapCustomerRecords($customer, $source),
            );

            if ($batch->isNotEmpty()) {
                $onBatch($batch);
            }
        } finally {
            $source->flush();
        }
    }

    protected function provideTransformerClass(): string
    {
        return AcmeTransformer::class;
    }
}

fetchBulkExportBatches() must call $onBatch once per non-empty page — page size and pagination strategy are entirely up to the integration. It is called by the sealed ExportsInBulk::streamBulkExport()/requestBulkExport(), so it never needs to build up the full collection itself; see Dispatch asynchronously for how batches flow into the persistence pipeline.

2. Implement verification¶

performVerification() is called via VerifiesConnections::verify() whenever a SystemConnection's credentials need to be checked against the live system — during both the plugin-initiated and tenant-initiated registration handshakes. See Connection Flow for where this fits in the request lifecycle.

Return VerificationResult::success($externalUserId) on success. On failure you can either return VerificationResult::failure($message) explicitly, or just throw — verify() catches any Throwable and converts it into a failed result, so a plain $response->throw() on a non-2xx HTTP response is usually enough.

use App\Data\VerificationResult;
use Illuminate\Support\Facades\Http;

protected function performVerification(SystemConnection $connection): VerificationResult
{
    $response = $this->client->getAccount($connection->base_url, $connection->credentials['api_key']);

    // Non-2xx → RequestException, caught by VerifiesConnections::verify()
    // and turned into a failed VerificationResult.
    $response->throw();

    return VerificationResult::success((string) $response->json('account.id'));
}

See it in practice

App\Integrations\PrestaShop\PrestaShopIntegration::performVerification() is a real implementation of this pattern against the PrestaShop Webservice API — see the Integration Contract reference.

3. Create the transformer and map customers¶

provideTransformerClass() (see step 1) must return a class-string<TransformerContract>. Create a class extending AbstractTransformer alongside the integration, in the same namespace — AbstractTransformer already implements the generic persistence logic (transform(), store(), and the storeCustomer()/storeOrder()/storeProduct() updateOrCreate() calls), so the subclass typically adds nothing there.

Instead, the transformer is where field-mapping from the platform's raw API response to Conflux's DTOs lives, as public static methods that the integration calls directly from fetchBulkExportBatches() (see App\Integrations\WooCommerce\WooCommerceTransformer::mapCustomer() for a real implementation). Pass all available fields; omit (leave null) anything the platform does not provide.

A payload that carries dependent data, such as consent, is mapped to [CustomerData, ConsentData…] by mapCustomerRecords(): the customer first, then one ConsentData per channel the platform reports. CustomerData has no consents, and AbstractTransformer::store() routes each record by type. See Transformer Contract.

<?php

declare(strict_types=1);

namespace App\Integrations\Acme;

use App\Data\AddressData;
use App\Data\ConsentData;
use App\Data\CustomAttributeData;
use App\Data\CustomerData;
use App\Enums\ConsentChannel;
use App\Enums\SystemCapability;
use App\Enums\SystemDriver;
use App\Services\AbstractTransformer;
use App\Services\RecordDateTimes;
use App\Services\SourceDateTimes;
use Carbon\CarbonImmutable;
use Illuminate\Support\Collection;

class AcmeTransformer extends AbstractTransformer
{
    /**
     * The records one customer payload carries: the customer, then its consents.
     *
     * @param  array<string, mixed>  $data
     * @return list<CustomerData|ConsentData>
     */
    public static function mapCustomerRecords(array $data, SourceDateTimes $source): array
    {
        $customer = self::mapCustomer($data, $source);
        $at = $source->forRecord(SystemCapability::Customers, $customer->sourceId);

        return [$customer, ...self::mapConsents($data, $customer->sourceId, $at)->all()];
    }

    /**
     * @param  array<string, mixed>  $data
     */
    public static function mapCustomer(array $data, SourceDateTimes $source): CustomerData
    {
        $sourceId = (string) $data['id'];
        $at = $source->forRecord(SystemCapability::Customers, $sourceId);

        return new CustomerData(
            sourceId: $sourceId,
            sourceSystem: SystemDriver::Acme,
            email: strtolower($data['email']),
            createdAt: $at->instant($data['date_created_gmt'] ?? null, 'date_created_gmt'),
            updatedAt: $at->instant($data['date_modified_gmt'] ?? null, 'date_modified_gmt'),
            customAttributes: self::flattenMeta($data['meta_data'] ?? []),
            firstName: $data['first_name'] ?: null,
            lastName: $data['last_name'] ?: null,
            phone: $data['billing']['phone'] ?: null,
            addressBilling: self::mapAddress($data['billing']),
            orderCount: $data['orders_count'],
            totalSpent: (int) round((float) $data['total_spent'] * 100),
        );
    }

    /**
     * Emit one ConsentData per channel the platform reports. Channels the platform has no
     * signal for are simply omitted — never emit a "withdrawn" consent to mean "unknown".
     *
     * `consentUpdatedAt` is required, because a two-way conflict between two mirrors is settled
     * by it. It is parsed through the same `RecordDateTimes` as every other datetime, so
     * a value that does not match the declared format is reported once and comes back null — and a
     * consent with no orderable timestamp is omitted rather than written with a guess. Acme has no
     * per-field consent timestamp, so `date_modified_gmt` stands in for one — see "Map datetimes"
     * below for what that costs.
     *
     * Emit `confirmed: null` unless the platform really reports a confirmation: a single opt-in
     * flag does not say the address was confirmed.
     *
     * @param  array<string, mixed>  $data
     * @return Collection<int, ConsentData>
     */
    private static function mapConsents(array $data, string $customerSourceId, RecordDateTimes $at): Collection
    {
        $consents = collect();
        $updatedAt = $at->instant($data['date_modified_gmt'] ?? null, 'date_modified_gmt');

        if (! $updatedAt instanceof CarbonImmutable) {
            return $consents;
        }

        if (isset($data['meta_data']['newsletter'])) {
            $consents->push(new ConsentData(
                sourceId: ConsentData::sourceIdFor($customerSourceId, ConsentChannel::Email),
                sourceSystem: SystemDriver::Acme,
                parentSourceId: $customerSourceId,
                channel: ConsentChannel::Email,
                consentUpdatedAt: $updatedAt,
                subscribed: (bool) $data['meta_data']['newsletter'],
                confirmed: null,
            ));
        }

        if (isset($data['meta_data']['sms_optin'])) {
            $consents->push(new ConsentData(
                sourceId: ConsentData::sourceIdFor($customerSourceId, ConsentChannel::Sms),
                sourceSystem: SystemDriver::Acme,
                parentSourceId: $customerSourceId,
                channel: ConsentChannel::Sms,
                consentUpdatedAt: $updatedAt,
                subscribed: (bool) $data['meta_data']['sms_optin'],
                confirmed: null,
            ));
        }

        return $consents;
    }

    /**
     * @param  array<string, mixed>  $data
     */
    private static function mapAddress(array $data): ?AddressData
    {
        if (empty($data['address_1']) && empty($data['city'])) {
            return null;
        }

        return new AddressData(
            firstName: $data['first_name'] ?: null,
            lastName: $data['last_name'] ?: null,
            company: $data['company'] ?: null,
            address1: $data['address_1'] ?: null,
            address2: $data['address_2'] ?: null,
            city: $data['city'] ?: null,
            postalCode: $data['postcode'] ?: null,
            country: $data['country'] ?: null,
            phone: $data['phone'] ?: null,
            email: $data['email'] ?: null,
        );
    }

    /**
     * @param  array<array{key: string, value: mixed}>  $meta
     * @return Collection<int, CustomAttributeData>
     */
    private static function flattenMeta(array $meta): Collection
    {
        $attributes = collect();

        foreach ($meta as $entry) {
            if (is_scalar($entry['value'])) {
                $attributes->push(new CustomAttributeData(key: $entry['key'], value: $entry['value']));
            }
        }

        return $attributes;
    }
}

Monetary values

All amounts (totalSpent, unitPrice, total, etc.) are integers in minor currency units. Multiply floats from the API by 100 and round: (int) round($float * 100).

4. Map datetimes¶

Six rules govern every datetime that crosses an integration boundary. They are decided in ADR-0018, which carries the reasoning; this section is what you do about them.

A datetime Conflux stores in a column of its own is parsed at the integration boundary and stored as an instant in UTC.
A calendar date is stored as a date and never converted, because it means the same day in every zone.
The source zone is established per system connection, never assumed. Where the zone is a fact about the platform, it is a constant in your integration. Where it varies per connection, it lives in system_connections.source_timezone. Where it cannot be established at all, rule 4 applies.
A failed parse is an event, never a silent default.
What your integration passes through without storing it in a column of its own is not covered by rule 1, and must not be presented as though it were.
A datetime you read and Conflux does not store is recorded as discarded rather than quietly disappearing.

In the order you will meet them:

Declare DATETIME_FORMAT and DATE_FORMAT as constants on your integration. Write down what the platform actually sends, separator and all: WooCommerce sends 2024-03-01T09:05:00, so its DATETIME_FORMAT is 'Y-m-d\TH:i:s', while PrestaShop sends 2024-03-01 10:05:00 and declares 'Y-m-d H:i:s'. Declare both constants even if you map no calendar date today, so the parse context needs no special case. A format that mentions fewer fields than the value carries is not a shortcut either: the parse context prefixes both formats with !, so what your format leaves out is reset rather than taken from the sync run's wall clock, and a platform sending 2024-03-01 as a datetime lands at midnight instead of at whatever time the run happened to start.

Declare SOURCE_TIMEZONE when the zone is a platform fact. WooCommerce reads the _gmt variant of every field, and those strings carry no offset but are always UTC, so SOURCE_TIMEZONE is 'UTC'. Note what such a constant means: it is the zone of the strings you read, not the zone the shop is in. If the zone instead varies per connection, resolve it and write it to system_connections.source_timezone, the way PrestaShopIntegration::resolveSourceTimezone() reads PS_TIMEZONE. Resolve it on every run, including a full sync, because the full sync is the run that writes the historical values for the first time. A failed lookup falls back to the value already stored on the connection, never to the application timezone.

Build the parse context once per run and hand it to the mappers. sourceDateTimes() returns a SourceDateTimes carrying the facts that hold for the whole run: the driver, both formats, and the zone. Pass a null zone through rather than substituting one; every instant parsed against it then becomes a recorded failure, which is what rule 3 requires.

Narrow it per record with forRecord(). That binds the domain and the sourceId a failure is reported against, so each call site names only the raw value and the field it came from.

Use instant() for a point in time and calendarDate() for a calendar date. A calendar date takes no zone, because it means the same day everywhere, so it parses even on a connection whose zone could not be established.

$at = $source->forRecord(SystemCapability::Customers, $sourceId);

createdAt: $at->instant($data['date_add'] ?? null, 'date_add'),
dateOfBirth: $at->calendarDate($data['birthday'] ?? null, 'birthday'),

Reach for instantForComparison() only where the value is never stored. A client-side delta filter parses a source datetime to decide whether a record belongs in the batch, and PrestaShopIntegration::wasModifiedAfter() is its one caller today. It takes no field, domain or source id and records no failure, because the mapper reads the same field of the same record moments later and reports it there; reporting it twice would double every count in the aggregated line. It returns null for anything it cannot read, an unestablished zone included, and a filter must treat that null as "not known" rather than as "older than the cursor". Keep the record, so a shop whose zone cannot be resolved looks broken in the log rather than empty in the batch.

Call flush() in a finally. Failures are aggregated over the run and written as one log line per domain and field, so a systematically broken field produces one line rather than one per record. The finally is what makes a run that throws still report what it already found.

$source = $this->sourceDateTimes();

try {
    // fetch pages, map them, hand each batch to $onBatch
} finally {
    $source->flush();
}

Never call Carbon::parse() on a source datetime string. It reads an undeclared format by heuristic and the heuristic turns on the separator: Carbon::parse('04.12.1985') is 4 December and Carbon::parse('04/12/1985') is 12 April. A wrong date that parses cleanly is worse than a parse failure, because nothing surfaces it.

An absent value is not a failure

A field the platform did not send, an empty string, and the MySQL zero dates 0000-00-00 and 0000-00-00 00:00:00 all yield null without recording a failure. They are how a source spells "no value", not a value it sent incorrectly. Anything else that does not match your declared format is a failure.

Which failure policy applies depends on what the datetime means, and today only the first case has any call site:

A datetime that describes the record (createdAt, updatedAt, datePaid, dateOfBirth) is nulled on failure. The record is then less precise but still usable, and skipping it would let one broken field stop a whole shop from syncing.
A datetime that is the evidence of something (the consent timestamps, where confirmed_at is the proof of a double opt-in) fails the record instead. Nulling it silently would turn a confirmed recipient into one with no proof. Nothing writes this kind today.
A datetime the platform's state is ordered by — only ConsentData::$consentUpdatedAt today — drops the value it belongs to. It is the required field a two-way conflict between two consent mirrors is settled by, so it can be neither nulled nor guessed at, and a consent without one is omitted the same way a channel the platform reports no signal for is. The record itself still lands, minus that channel.

Picking a consentUpdatedAt when the platform has no per-field timestamp

Neither WooCommerce nor PrestaShop records when a newsletter flag last changed — only the customer record's own date_modified/date_upd. Using that as a proxy is the only option available, and it comes with a real failure mode: if a plugin flips the consent flag without bumping the record's timestamp, a later conflict is judged on a stale instant without a trace. Prefer a timestamp scoped to the consent field itself wherever a platform offers one.

Finally, rules 5 and 6 are about what you do not store. A datetime you pass through into customAttributes stays a raw string and must not be described as a UTC instant, because you do not know an unknown field's format. A datetime you read and deliberately drop belongs in a named constant with the reason in its docblock, the way WooCommerceTransformer::DROPPED_ORDER_KEYS holds date_completed, rather than being discarded by the silent interaction of a mapped-keys list and a mapper that never reads it.

5. Map orders¶

Add an mapOrder() static method to the same transformer, following the same pattern:

use App\Data\LineItemData;
use App\Data\OrderData;
use App\Enums\SystemCapability;
use App\Enums\SystemDriver;

/**
 * @param  array<string, mixed>  $data
 */
public static function mapOrder(array $data, SourceDateTimes $source): OrderData
{
    $at = $source->forRecord(SystemCapability::Orders, (string) $data['id']);

    return new OrderData(
        sourceId: (string) $data['id'],
        sourceSystem: SystemDriver::Acme,
        sourceOrderNumber: $data['number'],
        customerSourceId: $data['customer_id'] ? (string) $data['customer_id'] : null,
        status: $data['status'],
        currency: $data['currency'],
        total: (int) round((float) $data['total'] * 100),
        totalTax: (int) round((float) $data['total_tax'] * 100),
        totalShipping: (int) round((float) $data['shipping_total'] * 100),
        createdAt: $at->instant($data['date_created_gmt'] ?? null, 'date_created_gmt'),
        updatedAt: $at->instant($data['date_modified_gmt'] ?? null, 'date_modified_gmt'),
        datePaid: $at->instant($data['date_paid_gmt'] ?? null, 'date_paid_gmt'),
        addressBilling: self::mapAddress($data['billing']),
        addressShipping: self::mapAddress($data['shipping']),
        lineItems: array_map(fn (array $li): LineItemData => new LineItemData(
            sourceId: (string) $li['id'],
            productSourceId: (string) $li['product_id'],
            name: $li['name'],
            quantity: (int) $li['quantity'],
            unitPrice: (int) round((float) $li['price'] * 100),
            total: (int) round((float) $li['total'] * 100),
            sku: $li['sku'] ?: null,
            currency: $data['currency'],
        ), $data['line_items']),
        customAttributes: self::flattenMeta($data['meta_data'] ?? []),
    );
}

6. Enable parallel fetching (optional)¶

By default, ExportsInBulk::performProbeBulkExport() reports a single logical page, so BulkSyncJob always falls back to one fetchBulkExportBatches() call. If the platform's API exposes a total page count up front, override performProbeBulkExport() and performFetchBulkExportPageRange() to let BulkSyncJob fan out parallel FetchBulkPageRangeJobs instead:

use App\Data\BulkExportProbe;

#[\Override]
protected function performProbeBulkExport(SystemConnection $connection, ExportRequest $request): BulkExportProbe
{
    $response = $this->client->getCustomers($connection->base_url, page: 1, perPage: self::PER_PAGE);

    return new BulkExportProbe(totalPages: $response->totalPages(), perPage: self::PER_PAGE);
}

/**
 * @param  Closure(Collection<int, DomainRecord>): void  $onBatch
 */
#[\Override]
protected function performFetchBulkExportPageRange(SystemConnection $connection, ExportRequest $request, int $startPage, int $endPage, int $perPage, Closure $onBatch): void
{
    $source = $this->sourceDateTimes();

    try {
        for ($page = $startPage; $page <= $endPage; $page++) {
            $batch = collect($this->client->getCustomers($connection->base_url, $page, $perPage))->flatMap(
                fn (array $customer): array => AcmeTransformer::mapCustomerRecords($customer, $source),
            );

            if ($batch->isNotEmpty()) {
                $onBatch($batch);
            }
        }
    } finally {
        $source->flush();
    }
}

The parse context is built per page-range job rather than shared across them, because each job is a separate process. One job's aggregated failure log therefore covers only its own pages, which is what makes the finally matter on this path too.

See App\Integrations\WooCommerce\WooCommerceIntegration for a real implementation — it derives the page count from the X-WP-TotalPages response header. If the platform's API exposes no such header, see App\Integrations\PrestaShop\PrestaShopIntegration — its probe lists all record ids in a single unpaginated request and derives the page count as ceil(count($ids) / $perPage) instead.

7. Handle errors¶

Throw IntegrationRequestException for expected failures (auth errors, rate limits, missing resources). AbstractIntegration catches everything else and re-throws it as InvalidIntegrationResponseException, so the caller always receives a typed exception.

use App\Exceptions\IntegrationRequestException;

protected function fetchBulkExportBatches(SystemConnection $connection, ExportRequest $request, Closure $onBatch): void
{
    $response = $this->client->getCustomers($connection->base_url);

    if ($response->unauthorized()) {
        throw new IntegrationRequestException('Acme API credentials are invalid or expired.');
    }

    if ($response->failed()) {
        throw new IntegrationRequestException("Acme API returned HTTP {$response->status()}.");
    }

    // ...
}

Do not catch IntegrationRequestException inside the integration — let it propagate. ExportsInBulk::streamBulkExport()/requestBulkExport() pass it through to the caller unchanged.

8. Handle inbound webhooks (optional)¶

If the platform sends webhooks to Conflux, add use HandlesWebhooks and implements CanHandleWebhooks to the integration and implement the trait's processWebhook() hook. It receives the SystemConnection and the raw payload array. Skip all three for a platform that sends no webhooks — there is nothing to stub out.

use App\Exceptions\IntegrationRequestException;
use App\Models\SystemConnection;

protected function processWebhook(SystemConnection $connection, array $payload): void
{
    $topic = $payload['topic'] ?? null;

    match ($topic) {
        'customer/created', 'customer/updated' => $this->handleCustomerWebhook($connection, $payload),
        'order/created', 'order/completed'     => $this->handleOrderWebhook($connection, $payload),
        default => throw new IntegrationRequestException("Unsupported webhook topic: {$topic}"),
    };
}

The processWebhook() method is called by HandlesWebhooks::handleWebhook(), which is sealed as final on the trait and wraps the call with the same error-handling guarantee as requestBulkExport() — any unexpected Throwable is caught and re-thrown as InvalidIntegrationResponseException. The implements CanHandleWebhooks declaration is what ProcessWebhookJob checks before calling it; without it the job logs a warning and discards the webhook.

Signature validation is handled by the controller

The webhook endpoint validates the HMAC-SHA256 signature before the job is dispatched. Integrations do not need to validate signatures themselves. See the Webhook Receiver reference for the full endpoint specification.

9. Register the integration¶

Before the integration can be used, it must be registered in IntegrationRegistry. By convention, both existing integrations (WooCommerceIntegration, PrestaShopIntegration) are registered directly in App\Providers\AppServiceProvider::boot() — there is no per-integration service provider:

public function boot(): void
{
    $registry = $this->app->make(IntegrationRegistry::class);
    $registry->register(SystemDriver::WooCommerce, WooCommerceIntegration::class);
    $registry->register(SystemDriver::PrestaShop, PrestaShopIntegration::class);
    $registry->register(SystemDriver::Acme, AcmeIntegration::class); // add the new driver here
}

Warning

An integration that is not registered cannot be resolved. Any queued BulkSyncJob dispatched for its driver will throw a RuntimeException at execution time.

10. Dispatch asynchronously¶

The integration is never called directly from a controller or a request lifecycle. Conflux is async-first — every bulk export runs inside a queued job on the rabbitmq connection so it does not block the web process and survives transient API failures.

The admin panel's Bulk Sync action already dispatches App\Jobs\BulkSyncJob — once an integration is registered, admin-triggered syncs work without any additional wiring. BulkSyncJob calls probeBulkExport() first: if it reports more than one page, it fans out parallel App\Jobs\FetchBulkPageRangeJobs via fetchBulkExportPageRange(); otherwise it falls back to a single streamBulkExport() call. Either way, each batch handed to $onBatch is persisted by the transformer returned from transformerClass().

// ✅ queued — returns immediately, job runs in the background
BulkSyncJob::dispatch($connection, [SystemCapability::Customers]);

// ❌ synchronous — blocks the process, do not do this
$integration->requestBulkExport($connection, $request);

The queue:work supervisor process (rabbitmq connection) picks up the job automatically. Failed jobs land in the failed_jobs table (database-uuid driver) and can be retried via php artisan queue:retry.