Transformer Contract¶

Reference for the pluggable transformation and storage pipeline introduced in CFX-46. This covers the interface every platform transformer must implement, the result value objects, and the unified Eloquent models.

Interface: TransformerContract¶

Namespace: App\Contracts\TransformerContract

public function transform(Collection $data, SystemConnection $connection, string $syncBatchId): TransformationResult;

Accepts a Collection<int, DomainRecord> (the output of an integration's requestBulkExport()) and a SystemConnection. Persists each item to the unified Eloquent models. A collection may hold records of more than one domain: a mapper emits a parent followed by its dependents (see Dependent records). Returns a TransformationResult summarising successes and failures.

The method never throws — any exception raised while storing an individual item is captured and included in the result.

Abstract Base Class: AbstractTransformer¶

Namespace: App\Services\AbstractTransformer

Implements TransformerContract. Concrete transformers extend this class and may override store() to intercept the dispatch.

transform() is declared final. It iterates over the data collection, calls store() for each item, and catches any Throwable per item — meaning a failure on one record does not abort the rest of the batch. The result carries the full failure list so callers can log or retry specific items.

protected function store(DomainRecord $item, SystemConnection $connection, string $syncBatchId): StoreOutcome;

Inheriting it is enough — a minimal subclass with no overrides is sufficient; a concrete transformer only needs to override it to handle a platform-specific mapping need.

Dependent records¶

A mapper that reads a payload with dependent data emits [parent, ...dependents]. mapCustomerRecords() on the WooCommerce and PrestaShop transformers returns the CustomerData followed by one ConsentData per channel the payload reports. The parent comes first, because storing a consent links the customer row of the same connection.

store() routes each record by its type: CustomerData, OrderData and ProductData go to their own store*() method, and ConsentData goes to ConsentRecorder::write(SystemConnection, ConsentData): StoreOutcome. See ADR-0041.

• The signal moves only on a value change. A consent's content_changed_at moves when a value field changes (subscribed, confirmed, the three timestamps, source, ip_address) or a trashed row is revived. consent_updated_at moving alone does not count, because WooCommerce reports the customer's modification time there and every customer edit would otherwise move the signal. Linking customer_id does not count either.
• Counters count roots. TransformationResult and the sync batch counters count records of the batch's own domain. A consent's inserted, updated or not-touched outcome is not counted, because it arrives with its customer, and PersistBulkDataJob adds root records only to total_source_items. A dependent's failure is counted, so it stays visible.
• A channel the source did not mention is not touched, and null never overwrites a known value, except that a withdrawal clears the confirmation of the opt-in it ends.

Result Value Objects¶

TransformationResult¶

Namespace: App\Data\TransformationResult

TransformationFailure¶

Namespace: App\Data\TransformationFailure

Unified Eloquent Models¶

These models are the persistent representation of normalised platform data inside Conflux. All use HasUuids and SoftDeletes.

Every datetime column below that holds a value the source sent is a UTC instant, and every calendar date is a date. The cast tells you which side a column belongs to: a column that fills a typed DTO field is immutable_datetime or immutable_date, while Conflux's own bookkeeping columns (content_changed_at, created_at, updated_at) stay mutable Carbon. See ADR-0018.

Deleting and restoring the row of a mirrored domain¶

Customer, Order and Product run every query through App\Models\Builders\MirroredDomainRecordBuilder, because a soft deletion has to carry the reason it happened and the two columns must move in one statement. RecordProjector refuses a row that names a cause without having been deleted, and a sequence writing the two separately can be interrupted between them.

AbstractTransformer un-trashes a returning record with untrash() rather than restore(). That is deliberate: restore() saves immediately, and deleted_at has to still be dirty when stampIfChanged() reads getDirty() a few lines later. A record that only came back would otherwise move no content_changed_at and be reported as NotTouched, which makes the resurrection invisible to outbound delivery.

LineItem carries neither column and uses neither method: a child row inherits its parent's cause by having none of its own. CustomerConsent is a mirror and carries both. Disconnect, cleanup, restore and purge handle consent by system_connection_id with a cause, and a customer deleted through the deletion log soft-deletes its consents with the same cause (source_event). See ADR-0019.

Customer¶

Namespace: App\Models\Customer
Table: customers

Unique constraint: (system_connection_id, source_id)

Relations: systemConnection() → BelongsTo<SystemConnection>, consents() → HasMany<CustomerConsent>, orders() → HasMany<Order>

CustomerConsent¶

Namespace: App\Models\CustomerConsent Table: customer_consents

A mirror of the dependent customer_consents domain: one connection's view of one customer's consent on one channel. Its parent is named by parent_source_id within the same connection, not by a foreign key. See ADR-0041.

Unique constraint: (system_connection_id, source_id) — one row per channel per parent per connection.

A migration set confirmed to null for the WooCommerce and PrestaShop rows that existed: neither platform can say an address was confirmed, and the old false and true were guesses.

Relations: customer() → BelongsTo<Customer>, systemConnection() → BelongsTo<SystemConnection>

Order¶

Namespace: App\Models\Order
Table: orders

Unique constraint: (system_connection_id, source_id)

Relations: systemConnection() → BelongsTo<SystemConnection>, customer() → BelongsTo<Customer>, lineItems() → HasMany<LineItem>

LineItem¶

Namespace: App\Models\LineItem Table: line_items

Unique constraint: (order_id, source_id)

Relations: order() → BelongsTo<Order>, product() → BelongsTo<Product>

product_id is resolved at import time by looking up Product with (system_connection_id, product_source_id). If the product has not been synced yet, product_id remains null until the order is re-imported after the product sync.

Product¶

Namespace: App\Models\Product
Table: products

Unique constraint: (system_connection_id, source_id)

Relations: systemConnection() → BelongsTo<SystemConnection>

Default store() Implementation¶

AbstractTransformer provides a concrete store() method that handles all three entity types via a match expression. It uses updateOrCreate keyed on (system_connection_id, source_id) so that re-running a bulk export updates existing records rather than creating duplicates.

Platform-specific transformers extend AbstractTransformer and override store() only when they need behaviour that differs from the default. Until that need arises, a minimal subclass with no overrides is sufficient.

TransformerRegistry¶

Namespace: App\Services\TransformerRegistry

A singleton that maps each SystemDriver to a concrete transformer class. It mirrors IntegrationRegistry and is the only mechanism by which Conflux locates the correct transformer at runtime.

public function register(SystemDriver $driver, string $transformer): void
public function resolve(SystemDriver $driver): AbstractTransformer

• register() — associates a SystemDriver case with a class-string<AbstractTransformer>. Called once per platform in AppServiceProvider::boot().
• resolve() — returns an instance of the registered transformer, resolved via the service container (app($class)). Throws RuntimeException if no transformer has been registered for the given driver.

The registry is bound as a singleton in AppServiceProvider::register():

$this->app->singleton(TransformerRegistry::class);

PersistBulkDataJob receives the registry via constructor injection and calls resolve() at job execution time to obtain the transformer for the connection's system driver.

PersistBulkDataJob¶

Namespace: App\Jobs\PersistBulkDataJob
Queue connection: rabbitmq

The job that bridges the bulk-fetch stage and the persistence stage. BulkSyncJob dispatches it after each successful requestBulkExport() call when the result is non-empty.

public function __construct(
    public readonly Collection $data,           // Collection<int, DomainRecord>
    public readonly SystemConnection $systemConnection,
) {}

On execution, it resolves the correct transformer from TransformerRegistry using the system driver of the associated SystemConnection, then calls transform(). If the connection has no associated system, the job exits silently.

The job runs on the rabbitmq connection, which places it on a different queue from BulkSyncJob (which runs on the default redis-backed queue). This separates the fetch stage from the persistence stage and allows them to scale independently.