Module import command framework

Core provides reusable infrastructure for module-owned bulk import commands. It deliberately does not register a runnable core:import command: the destination domain must remain visible in each concrete command name, such as cms:import or erp:import.

Core components

The abstract command is excluded from automatic module command discovery because only instantiable console commands are registered.

Concrete module commands

A module command extends AbstractImportCommand, declares $name and $description, and injects module-aware resolver/discovery collaborators:

final class ImportCommand extends AbstractImportCommand
{
    protected $name = 'example:import';

    protected $description = 'Import example records <fg=green>(Modules\\Example)</fg=green>';
}

Do not declare $signature. Core defines the shared options through getOptions():

• --importer=: concrete importer FQCN;
• --bootstrap=: optional external Composer autoloader;
• --arg=*: repeatable importer constructor argument in key=value form;
• --dry-run: roll back writes on the connection declared by the importer, or the default connection;
• --limit=: non-negative import limit passed to the importer;
• --no-search: import without search indexing: no document is written and no embedding is generated;
• --index-batch=: distinct records indexed per bulk search flush (default scout.chunk.searchable, 500).

Module marker interfaces should extend Modules\Core\Import\Contracts\BulkImporterInterface. Configure the module resolver and discovery with that marker so a command rejects importers targeting another module before execution.

Domain boundary

Core does not know source schemas, destination entities, import ordering, DTOs, upserters, accounting rules, inventory rules, or conflict policies. A module owns its destination pipeline and an external plugin owns source access and source-specific mapping.

Importers must call module services for protected domain mutations. They must not bypass posting, numbering, locking, inventory, accounting, audit, or authorization rules through raw writes.

Dry-run guarantee

BulkImportRunner opens a transaction on the connection returned by an optional ConnectionAwareBulkImporterInterface, falling back to the current default connection, and restores its previous transaction nesting level. This rolls back database writes made on that connection only.

It does not roll back other connections, files, object storage, queued work, HTTP calls, or other external side effects. Importers receive dryRun as a named constructor parameter and are responsible for suppressing non-transactional effects. The command disables search indexing when dry-run is active.

Search indexing during an import

The command runs the importer inside Modules\Core\Search\DeferredSearchIndexing::run(). While it runs, a searchable() call made by Scout's observer (or by the AI single-locale re-embed listener) does not index anything: it records the model class and its scout key. A searchable model saved inside a transaction is also recorded at save time, so an interruption between the commit and Scout's after-commit callbacks loses none of the models that transaction wrote. Several saves of the same record collapse into one entry, and only keys are kept in memory.

The recorded records are indexed through the bulk path every --index-batch distinct records and once at the end. A flush reloads them by key through the model's Scout import query (makeAllSearchableQuery()), skips rows deleted in the meantime or no longer searchable, then calls makeSearchableInBulk(): one ModelsRequireIndexing pre-process pass (embeddings for every stale locale in one batched call) and adaptive engine writes.

Where that work runs depends on the queue. When scout.queue is on and the default queue connection is not sync, a flush only queues one Search\Jobs\IndexDeferredSearchChunkJob per chunk of keys on the scout.queue.queue queue (indexing), and a worker reloads and indexes the chunk there: the import does not wait for embeddings or engine writes, and a failing document fails that job (retried with the scout.queue settings) instead of the import. Otherwise the flush indexes each chunk in the import process.

The command prints one line per flushed chunk: Search: queued 500 Content record(s) for bulk indexing, or Search: indexed 500 Content record(s) in 8400ms when the chunk is indexed in place. The count includes every searchable model the import saved (categories, tags, contributors, locations), so a flush can come sooner than --index-batch imported contents.

• A threshold flush never runs inside an open transaction: it waits for the commit, and a rollback leaves the keys pending for the next flush.
• When the importer throws, the records already committed are flushed before the exception propagates.
• Records become searchable in blocks, not one by one.
• Ctrl+C (SIGINT), when records are waiting to be indexed, asks what to do: index them and quit (the default; a flush already in progress is finished), quit at once, or resume the import. Indexing and quitting rolls back the record being imported at that moment and exits with status 130. SIGTERM, or a run started with --no-interaction or without a terminal, indexes and quits without asking (status 143 for SIGTERM). A second Ctrl+C while indexing quits at once. Import\Support\ImportInterruptHandler implements this; it needs the pcntl extension.
• Quitting at once, kill -9, or a fatal error (out of memory) leave up to --index-batch minus one records unindexed; recover them with scout:import {Model} (and ai:embeddings:repair when vectors are enabled).
• With --no-search or --dry-run the recorded calls are dropped instead. Removals (unsearchable()) are never deferred.

Other bulk code can use the same service: app(DeferredSearchIndexing::class)->run($callback, $batchSize, discard: false, onFlush: null); onFlush receives the model class, the record count, whether the chunk was queued, and the milliseconds it took. A nested run() joins the outer one. interrupt() stops a run by throwing Search\Exceptions\DeferredRunInterruptedException, at once or right after the flush in progress; the run indexes what it recorded while the exception propagates. The exception deliberately extends Exception, not RuntimeException, so code wrapping runtime failures into row errors does not swallow it.

Imports versus synchronization

This framework executes bounded, operator-triggered imports. Continuous external-system synchronization is a separate layer that may reuse importer pipelines but also requires explicit cursors, remote identity, idempotency, conflict policy, direction, retries, observability, and scheduling. Those concerns must not be added implicitly to AbstractImportCommand.

External record identities

RecordOriginRegistry is the source-neutral persistence boundary for imported record identities. An identity is the tuple (referable_type, source_key, external_id); adapters should qualify source_key by source instance or group when the upstream identifier is not globally unique. A malformed url from the source is stored as null rather than failing the record: RecordOrigin::isValidUrl() is the same rule the model applies to manual attributions.

Adapters compute a lowercase SHA-256 fingerprint from normalized source fields and may supply the source modification timestamp as evidence. The registry reports Missing, Unchanged, or Changed, stores provenance on the referable model's database connection, and never decides whether the destination record may be changed. Destination modules own that policy. Source timestamps never replace destination created_at or updated_at values.

Interactive bulk import (upload → map → preview → run)

Distinct from the CLI framework above, Core also provides an interactive, entity-agnostic bulk import: a user uploads a file, maps its columns to a target entity's fields through dropdowns with a spreadsheet-like preview, and runs the import as a queued job. It is surfaced by the SPA (/app/crud/imports) and a Filament monitoring resource.

An entity importer validates the mapped row, upserts idempotently (typically via RecordOriginRegistry), and returns Created/Updated/Skipped or raises RowImportException for a per-row failure. Registered importers today:

A module adds an importable entity by implementing EntityImporterInterface and registering it from its provider's boot; the framework never touches an arbitrary table.

Relations by natural key

Foreign keys are never expressed as internal ids in a source file — a row references a related record by the same human-readable natural key an operator would type (a slug, name or code). A to-one relation resolves that key inline (e.g. cms.category.parent, erp.item.company). A to-many or cross-entity relation is declared as an ImportRelationField and resolved through RelationValueResolver, which splits a multi-value cell ("news, politics"), de-duplicates the tokens, and applies the field's OnMissingRelation policy to any token that matches nothing:

• create — provision the related record from the token (tags, cheap folksonomy);
• error — fail the row with a per-field message (categories, contributors: curated, so import them first);
• skip — silently drop the unmatched token.

cms.content is the reference implementation: it attaches tags, categories and contributors in the same row without the file ever carrying an id.

In-app notifications

When a queued import ends, SendImportFinishedNotification (listening on ImportSessionCompleted / ImportSessionFailed) sends an ImportFinishedNotification to the user who launched it — a Laravel database notification, the same store two surfaces read:

• the SPA notification tray (NotificationBell, mounted once in the shared AppShell), and
• Filament's native database-notifications bell (->databaseNotifications()).

The notification payload is UI-locale-agnostic — a machine type (import.finished) plus a meta bag (per-outcome counts) the SPA localizes itself, with English title/body as a backoffice fallback — and carries the module scope (the imported entity's module) and a semantic action target ({ target: 'import_session', id, view }) that each surface resolves to its own route. The notifications table adds one derived column, module_name (mirrored from data->scope by the Notification model on save), so a module-scoped tray filters at the database level.

GET /app/notifications (optionally ?scope=), .../unread-count, POST .../{id}/read and .../read-all back the SPA tray, each bound to the authenticated user. Both bells refresh by polling (SPA: unread-count on a ~30s interval, list on open; Filament: 30s) — real-time broadcast is a later, additive upgrade.