Etl Core Bundle Laravel Package

Getting Started

This is a Symfony bundle, not a Laravel package. There is no Laravel service provider, no config/etl.php, and no Eloquent integration. In a Laravel codebase, treat it as a Symfony/Doctrine subsystem you either run as a separate Symfony service or embed behind an adapter layer. If you are not already running Symfony DI, Messenger, Workflow, and Doctrine ORM 3.6, the integration cost is significant.

Minimal path:

1. composer require boutdecode/etl-core-bundle
2. Register BoutDeCode\ETLCoreBundle\BoutDeCodeETLCoreBundle in config/bundles.php (Symfony) or build an equivalent bootstrap in Laravel.
3. Import the bundle config:

   imports:
       - { resource: "@BoutDeCodeETLCoreBundle/Resources/config/config.yaml" }
   

4. Create the five application entities by extending the abstract classes:
   • AbstractWorkflow
   • AbstractStep
   • AbstractPipeline
   • AbstractStepHistory
   • AbstractPipelineHistory
5. Add Doctrine mappings, $id properties/getters, constructors, and collection initialization, then run migrations.
6. Create a Workflow row whose stepConfiguration lists step codes in order.
7. Create a Pipeline from that workflow, put the run input in Pipeline::$input, dispatch it.
8. Implement at least one step with #[AsExecutableStep(code: '...')] and extend AbstractExtractorStep, AbstractTransformerStep, or AbstractLoaderStep.

Where to look first:

• README “Concepts” section: Workflow vs Pipeline, and config Step vs executable service.
• README entity examples: they are the fastest way to get the persistence layer working.
• ETL/Domain/Model/ for the step base classes and Context usage.
• Run/Domain/Middleware/ if you need logging, metrics, validation, or retry behavior around steps.
• CQS/Application/Operation/ if you want command/query dispatch rather than calling the runner directly.

First useful use case: a CSV import pipeline.

CsvFileExtractStep
  -> FilterTransformStep / DataMappingTransformStep
  -> DatabaseLoadStep

That covers the common day-to-day job: pull a file, normalize/filter rows, persist the result, and retain PipelineHistory / StepHistory for audit and debugging.

Implementation Patterns

Treat Workflow as seed data, not runtime state. Create workflows through fixtures, an admin command, or a small application service. Day to day, pipelines are created from existing workflow definitions. Keep workflow step configuration stable and override only the per-run values on the Pipeline.

Pass run-specific data through Pipeline::$input and Context. For example, store the uploaded CSV path or API parameters in Pipeline::$input. Steps should read from Context::getInput() and write to Context::setResult(). Do not pass data through service constructor state; steps are resolved and cloned at runtime.

Use the built-in steps before writing custom ones. CsvFileExtractStep, ApiExtractStep, XmlFileExtractStep, FilterTransformStep, DataMappingTransformStep, JsonFileLoadStep, and DatabaseLoadStep cover many ingestion jobs. Custom steps should be reserved for domain transformations, validation, enrichment, or proprietary destinations.

Model custom steps around one responsibility. Extend the appropriate abstract class and declare a globally unique code:

#[AsExecutableStep(
    code: 'app.import.normalize_customer',
    configurationDescription: [
        'trim' => 'Trim string values before mapping',
    ],
)]
final class NormalizeCustomerStep extends AbstractTransformerStep
{
    public function process(Context $context): Context
    {
        $rows = $context->getInput();
        // transform...
        return $context->setResult($rows);
    }
}

Use getConfigurationValue() for per-step options rather than hard-coding behavior. That is what makes a workflow reusable across pipelines.

Use middleware for cross-cutting concerns, not business logic. Implement Run\Domain\Middleware\Middleware, tag it as boutdecode_etl_core.pipeline_middleware or boutdecode_etl_core.step_middleware, and control ordering with priority. Built-in priorities range from -100 to 100; higher runs first. Good candidates: structured logging, timing/metrics, audit enrichment, input validation, and failure notification.

Use CQS buses for orchestration boundaries. Commands modify state and return void; queries read state and return data. Implement AsyncCommand or AsyncQuery only when the operation should go through Messenger asynchronously. Handlers implement CommandHandler or QueryHandler; do not add #[AsMessageHandler], because the bundle autoconfigures handlers.

Keep Laravel and Symfony boundaries explicit. A pragmatic Laravel pattern:

• Run the ETL engine in a Symfony service/container (or a separately deployed worker).
• From Laravel, dispatch a queued job that calls the ETL command bus or an HTTP/CLI boundary.
• Persist pipeline status and history in the Doctrine-managed database.
• Surface status to Laravel through a read model or query endpoint rather than coupling Eloquent models directly to the bundle entities.

Do not try to make the bundle’s Doctrine entities double as Eloquent models. That path creates more problems than it solves.

Use history aggressively. PipelineHistory and StepHistory are your operational record. Persist enough input/result detail to answer: which step failed, what data did it receive, and what did it produce? Avoid storing huge payloads in JSON columns without a retention policy.

Schedule recurring pipelines through the host application. The bundle provides a PipelineScheduler integration for Symfony Scheduler. In Laravel, the equivalent is a scheduled job or console command that creates and dispatches a pipeline. Keep scheduling outside the step logic.

Test at the seam, not through the whole framework. Unit-test custom steps by constructing a Context and asserting getResult(). Mock repositories, buses, and external clients. The bundle’s own quality bar is ECS, PHPStan, and PHPUnit; adopt the same for your custom steps and middleware.

Gotchas and Tips

• No entities are shipped. You must create and map all five entities yourself. The abstract classes hold properties, but you still need #[ORM\Column], relations, IDs, constructors, and collection initialization in the child classes. Missing collection initialization is a common first-run failure.
• Inherited property mapping is fiddly. You are mapping protected properties declared in a parent class from the child entity. Keep mappings consistent with the README examples and verify with doctrine:schema:validate before writing migrations.
• Doctrine ORM 3.6 is required. Laravel projects typically use Eloquent. If you are not prepared to run Doctrine alongside Eloquent, this bundle is probably the wrong tool for that codebase.
• Workflow vs Pipeline naming causes confusion. A Workflow is a reusable template. A Pipeline is a single execution instance with input, status, and timestamps. Most “why didn’t my workflow run?” issues are actually “I never created and dispatched a pipeline.”
• Step config and executable step are different objects. Step stores code, order, and configuration. The executable service is resolved by StepResolver using the code. A typo in Step::$code means no executable service is found.
• Step codes must be globally unique. The #[AsExecutableStep(code: ...)] value is the machine identifier. Use namespaced codes such as app.import.normalize_customer to avoid collisions.
• Do not assume step services retain state between runs. The resolver clones the executable step and applies configuration. Store run data in Context, not in service properties.
• Bus service IDs are inconsistent across the docs. The README configuration table lists boutdecode_etl_core.command.bus and boutdecode_etl_core.query.bus; the internal agent documentation refers to command.bus, query.bus, and event.bus. Verify the actual container IDs in your environment before wiring custom dispatchers.
• You must configure Messenger transports yourself. Async commands and queries require a transport and routing in the host application. The bundle provides buses, not your infrastructure.
• The bundle exposes no configurable keys under boutdecode_etl_core:. Service IDs, tags, and bus names are fixed constants. If you need different behavior, use middleware, custom steps, or override services explicitly rather than looking for a config switch.
• Middleware ordering is priority-based. If logging runs after the step it is meant to log, check the priority. Higher priority runs first. Register custom middleware explicitly with its tag and priority; do not rely solely on autowiring.
• The pipeline_lifecycle workflow is loaded by the bundle. If your application already defines a workflow with that name, or if you customize the marking store, inspect Resources/config/packages/workflow.yaml and resolve conflicts deliberately.
• Large inputs can exhaust memory. Context carries data in memory. For large CSV/XML files or API result sets, design steps to stream, chunk, or hand off file paths rather than fully materializing huge arrays.
• JSON columns plus mixed values need care. History entities store input and result as JSON. Avoid non-serializable objects, resources, and circular references. Add retention or truncation for large payloads.
• Failed pipelines need explicit operational handling. Status transitions are managed by the Symfony Workflow state machine, but retries, alerting, and dead-letter handling are application concerns. Add middleware or host-level monitoring around failed pipelines.
• Do not add #[AsMessageHandler] to command/query handlers. The bundle’s build() method autoconfigures them. Adding the attribute can cause duplicate registration or unexpected behavior.
• Respect the domain boundaries. ETL/, Run/, and CQS/ communicate through interfaces in Core/. If you find yourself importing infrastructure classes into a step or domain service, you are creating coupling that will make upgrades painful.
• Maturity risk is real. The package has very low adoption, a small star count, and a 2026 release date. Pin the version, wrap the engine behind your own application interfaces, and keep custom steps small enough to replace if the bundle stalls. The extension points—custom ExecutableStep, middleware, StepResolver, runner, and CQS buses—are your escape hatches.
• Debugging sequence: inspect Pipeline::$status, then the latest PipelineHistory, then the failing StepHistory. Confirm the pipeline’s step order, confirm each Step::$code resolves to a tagged service, and check Messenger logs if dispatch was async. If a step produced no history, the failure likely occurred before or during step resolution rather than inside process().