Download the PHP package projectsaturnstudios/consumption-engine without Composer

Consumption Engine

Opinionated Laravel package for consuming cleaned CSV extracts into your domain via queued jobs, event sourcing, and realtime progress events.

Designed to run on job queues (Horizon recommended). Consumption work is dispatched to a processing queue; realtime broadcast events ride a dedicated broadcasts queue; projectors use your event-sourcing queue. Your Horizon supervisors must listen to all three.

Quick start

1. Install the package — use Option A or B under Install.
2. Publish config + migration, then migrate.
3. Define the source/audit disks in config/filesystems.php and point config/tasks.php at them.
4. Configure Spatie / PSS event sourcing + Media Library (stored events, media tables, projector workers) — see Event sourcing dependencies.
5. Register at least one task + ConsumptionJob in config/tasks.php.
6. Start Horizon workers on data-proc (required for jobs + CLI progress), broadcasts (Echo), and your projector queue (e.g. EVENT_PROJECTOR_QUEUE_NAME).
7. Run php artisan consume {task} or call ConsumptionEngine::executeTask().

php artisan consume listens for progress for 10 seconds of inactivity after the last update. Start the data-proc worker before running the command.

Requirements

• PHP ^8.3
• Laravel with queues + broadcasting configured
• Redis (queues and the CLI progress list side-channel)
• Horizon (or equivalent) supervising data-proc, broadcasts, and your projector queue
• projectsaturnstudios/pss-event-sourcing — wraps Spatie Laravel Event Sourcing and provides DataEvent / event_command()
• Spatie Laravel Event Sourcing (pulled in via pss-event-sourcing)
• Spatie Media Library — required; audit finish attaches the JSON audit file via Media Library
• Optional: Laravel Reverb / Echo (browser progress UI only; CLI progress uses Redis lists)

Install

Option A — Packagist / VCS require

If transitive projectsaturnstudios/* packages are private VCS deps, add Bitbucket/GitHub repositories entries and Composer auth in the host app before requiring this package (or use Option B).

Option B — path repository (local monorepo)

Add this to the host app composer.json first:

Then:

Adjust the path URL if your package does not live under projectsaturnstudios/ relative to the host app.

The service provider is auto-discovered.

Publish config & migration

Publish the tasks config:

Publish the audit_logs migration:

Then migrate:

Configuration

Published file: config/tasks.php.

Source & audit disks

Both folder listing (FileStorageRepo) and CSV reads (ConsumptionJob) use config('tasks.disk'). Audit JSON writes use config('tasks.audit_disk') under /audits/{task}/....

Define those disk names in the host app before running consume. Example config/filesystems.php entries:

CSVs are read from {disk root}/{folder}/…. Audits are written to {audit_disk}/audits/{task}/….

Registering consumption jobs

Each task key maps a slug to a folder and a job class that extends ConsumptionJob:

• slug (catalog-products) — passed to executeTask() / php artisan consume
• folder — path prefix on the source disk; FileStorageRepo lists CSVs here
• task — FQCN of your job

Filenames must end with YYYY-MM-DD.csv (for example cp_2024-05-30.csv). The date is parsed from that trailing segment.

Queues & Horizon

This package is queue-first:

Without a data-proc worker, jobs never run and consume hits the 10s inactivity timeout. Without a broadcasts worker, Echo broadcasts stall — the CLI Redis list side-channel still works because RPUSH happens in the data-proc worker.

Example Horizon supervisors

Add (or merge) supervisors like these in the host app config/horizon.php defaults array:

Also reference each supervisor key under environments (for example local / staging / production). Horizon defaults alone does not start workers — and a missing APP_ENV key under environments boots zero supervisors.

Then:

Starting a task

Artisan

FileFinder builds {folder}/{filename} from the task config, so --file is the basename (or relative name under the folder), not the full storage key.

consume validates, dispatches the job, then listens on a Redis list key for progress until TaskCompleted, TaskFailed, or 10 seconds of inactivity.

Programmatically

executeTask() only validates and dispatches. For progress monitoring, subscribe via Echo (Artisan).

executeTask() returns TASK_STARTED on success. Other common returns: INVALID_TASK, INVALID_FILENAME, TASK_ALREADY_COMPLETED, TASK_PREVIOUSLY_FAILED, NO_MORE_TASKS_AVAILABLE. (READY is an internal validation gate only — callers never receive it.)

Creating a consumption job

Pipeline (handle())

1. TaskInitializing
2. setupAuditLog() — load CSV, key by uuid, seed empty audit entries, fire CreateAuditLogRecord, then TaskInitialized
3. validateRecords(&$contents)
4. setRecordActions(&$contents)
5. updateRecords($contents) — optional; default no-op; runs before creates
6. consumeNewRecords($contents)
7. saveAuditLog() — SavingAuditLog, then TaskCompleted or TaskFailed

UUID requirement

Every CSV row must be keyable by uuid. Either include a uuid column, or override getContents() to inject one before validation.

Implement the abstract methods

What belongs in each abstract method

validateRecords and setRecordActions take the collection by reference — assign back when you replace it. consumeNewRecords / updateRecords do not.

audit_log contract

• Pre-seeded with an empty array per uuid during setupAuditLog()
• Typical field progression: status → action → optional updates → success / reason
• Prefer get() → merge → put() so you do not wipe earlier fields
• updateActionRecords / consumeActionRecords only process rows whose action matches, and they write success / reason for you
• Authors never call save(); the base job does that at the end

Helpers on the base class

• updateRecords / updateActionRecords / consumeActionRecords / processActionRecords
• mapRecordsWithProgress, broadcastTaskProgress
• fireEvent(...) for realtime events
• consumptionMeta() for domain command metadata
• getContents() — override when you need to inject uuids or reshape rows

Register the class under config/tasks.php as shown above.

Realtime / Echo events

Every progress event implements ShouldBroadcast and is published on a public channel:

Example: task catalog-products → channel consume-catalog-products.

The same string is also used as a Redis list key: events RPUSH JSON so php artisan consume can BLPOP without Echo. Echo uses the broadcast driver; the CLI uses the list. Same name, different mechanisms.

Host broadcasting setup (required for Echo only)

CLI progress does not need broadcasting. Browser UI does.

1. Set BROADCAST_CONNECTION=reverb (or pusher) in the host .env.
2. Configure Reverb/Pusher credentials (REVERB_APP_ID, REVERB_APP_KEY, REVERB_APP_SECRET, REVERB_HOST, …).
3. Ensure a Horizon/broadcasts worker is running so broadcast jobs are processed.
4. Initialize Laravel Echo in the frontend with the same app key/host, then subscribe as below.

Subscribe with Laravel Echo

Events define broadcastAs() as the short class basename. Listen with a leading dot:

Failure signaling: TaskFailed is emitted only when saveAuditLog() throws. Uncaught exceptions earlier in validateRecords / update / consume leave the CLI waiting until the 10s inactivity timeout (no terminal Redis event).

Dual payloads: Echo receives public event properties (pct, reason, warning, num_records, …). The Redis CLI payload params comes from toArray() and may omit those fields (for example pct / num_records). CLI listeners should rely on message + event class for terminal handling.

Broadcast jobs use the broadcasts queue — keep a Horizon worker on that queue for Echo.

Event sourcing dependencies

Domain audit lifecycle uses PSS Event Sourcing (projectsaturnstudios/pss-event-sourcing), which builds on Spatie Laravel Event Sourcing:

• Sourced events extend DataEvent (e.g. AuditLogStarted, AuditLogFinished)
• Commands are dispatched with event_command(...) (e.g. CreateAuditLogRecord, LogConsumption)
• ConsumptionJobProjector projects into AuditLogProjection (audit_logs table)

This package registers ConsumptionJobProjector automatically; it does not replace your host event-sourcing bootstrap.

Host checklist (minimum)

1. Require the packages in the host app:

2. Publish and run Spatie event-sourcing migrations / config (stored events table + config/event-sourcing.php) per Spatie’s docs.
3. Publish and migrate Spatie Media Library tables (required for audit finish attachment).
4. Ensure PSS helpers are available (event_command(), DataEvent) — normally via the PSS package service provider / autoload.

5. Set the projector queue, for example:

6. Run a worker that consumes that queue (see Queues & Horizon).
7. Publish and migrate this package’s audit_logs table (see Publish config & migration).

If event sourcing is missing, CreateAuditLogRecord / LogConsumption fail when a job starts or finishes. If Media Library is missing, audit finish (addMediaFromDisk) fails after the JSON audit is written.

Testing

Pest unit tests live in this package (tests/Unit), not in the host application.

License

MIT