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\Data\BulkExportProbe;
use App\Data\CustomerData;
use App\Data\ExportRequest;
use App\Data\OrderData;
use App\Data\ProductData;
use App\Enums\SystemCapability;
use App\Exceptions\IntegrationRequestException;
use App\Models\SystemConnection;
use App\Services\AbstractIntegration;
use Closure;
use Illuminate\Support\Collection;

class AcmeIntegration extends AbstractIntegration
{
    public function __construct(private readonly AcmeApiClient $client) {}

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

        $customers = $this->client->getCustomers($connection->base_url);
        $batch = collect(array_map(AcmeTransformer::mapCustomer(...), $customers));

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

    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 AbstractIntegration::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 AbstractIntegration::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 AbstractIntegration::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.

<?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\SystemDriver;
use App\Services\AbstractTransformer;
use Illuminate\Support\Collection;

class AcmeTransformer extends AbstractTransformer
{
    /**
     * @param  array<string, mixed>  $data
     */
    public static function mapCustomer(array $data): CustomerData
    {
        return new CustomerData(
            sourceId: (string) $data['id'],
            sourceSystem: SystemDriver::Acme,
            email: strtolower($data['email']),
            createdAt: $data['date_created_gmt'],
            updatedAt: $data['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),
            newsletterConsent: isset($data['meta_data']['newsletter']) ? new ConsentData(
                subscribed: (bool) $data['meta_data']['newsletter'],
                confirmed: true,
                channel: ConsentChannel::Email,
            ) : null,
        );
    }

    /**
     * @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 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\SystemDriver;

/**
 * @param  array<string, mixed>  $data
 */
public static function mapOrder(array $data): OrderData
{
    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: $data['date_created_gmt'],
        updatedAt: $data['date_modified_gmt'],
        datePaid: $data['date_paid_gmt'] ?? null,
        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'] ?? []),
    );
}

5. Enable parallel fetching (optional)¶

By default, AbstractIntegration::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, CustomerData|OrderData|ProductData>): void  $onBatch
 */
#[\Override]
protected function performFetchBulkExportPageRange(SystemConnection $connection, ExportRequest $request, int $startPage, int $endPage, int $perPage, Closure $onBatch): void
{
    for ($page = $startPage; $page <= $endPage; $page++) {
        $batch = collect(array_map(AcmeTransformer::mapCustomer(...), $this->client->getCustomers($connection->base_url, $page, $perPage)));

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

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.

6. 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. AbstractIntegration::streamBulkExport()/requestBulkExport() pass it through to the caller unchanged.

7. Handle inbound webhooks (optional)¶

If the platform sends webhooks to Conflux, override processWebhook() in the integration. This method receives the SystemConnection and the raw payload array. The default implementation throws BadMethodCallException, so only implement it if the platform actually sends webhooks.

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 AbstractIntegration::handleWebhook(), which is sealed as final and wraps the call with the same error-handling guarantee as requestBulkExport() — any unexpected Throwable is caught and re-thrown as InvalidIntegrationResponseException.

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.

8. 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.

9. 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.