README

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

composer require projectsaturnstudios/consumption-engine

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:

{
  "repositories": [
    {
      "type": "path",
      "url": "projectsaturnstudios/*",
      "options": { "symlink": true }
    }
  ]
}

Then:

composer require projectsaturnstudios/consumption-engine:0.1.0

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:

php artisan vendor:publish --tag=consumption-engine

Publish the audit_logs migration:

php artisan vendor:publish --tag=consumption-engine.migrations.audit_logs
# or all package migrations:
php artisan vendor:publish --tag=consumption-engine.migrations.all

Then migrate:

php artisan migrate

Configuration

Published file: config/tasks.php.

Source & audit disks

return [
    // Filesystem disk that holds the cleaned CSV buffet (folder listing + reads)
    'disk' => env('CONSUMPTION_DISK', 'local'),

    // Filesystem disk where JSON audit payloads are written
    'audit_disk' => env('CONSUMPTION_AUDIT_DISK', 'local'),

    // ...task entries below
];

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:

'disks' => [
    's3-clean' => [
        'driver' => 's3',
        'key' => env('AWS_ACCESS_KEY_ID'),
        'secret' => env('AWS_SECRET_ACCESS_KEY'),
        'region' => env('AWS_DEFAULT_REGION'),
        'bucket' => env('AWS_CLEAN_BUCKET'),
    ],
    'local' => [
        'driver' => 'local',
        'root' => storage_path('app/private'),
    ],
],

CONSUMPTION_DISK=s3-clean
CONSUMPTION_AUDIT_DISK=local
AWS_CLEAN_BUCKET=your-cleaned-extracts-bucket

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:

'catalog-products' => [
    'name' => 'Consume Catalog Products',
    'folder' => 'cleaned/catalog-products',
    'task' => \App\Jobs\Consumption\CatalogProducts\ConsumeCatalogProductData::class,
],

• 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:

Concern	Default queue	Notes
Consumption jobs	data-proc	Dispatched by executeTask() / consume
Realtime broadcast events	broadcasts	ConsumptionRTEvent::$queue / broadcastQueue() — keep off the projector queue
Event-sourcing projectors	EVENT_PROJECTOR_QUEUE_NAME	Spatie / PSS projector workers (host-app env)

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:

'defaults' => [
    'event-supervisor' => [
        'connection' => 'redis',
        'queue' => [env('EVENT_PROJECTOR_QUEUE_NAME')],
        'balance' => 'auto',
        'maxProcesses' => 1,
        'memory' => 64,
        'tries' => 1,
        'timeout' => 60,
    ],
    'broadcast-supervisor' => [
        'connection' => 'redis',
        'queue' => ['broadcasts'],
        'balance' => 'auto',
        'maxProcesses' => 3,
        'memory' => 64,
        'tries' => 1,
        'timeout' => 30,
    ],
    'consumption-data' => [
        'connection' => 'redis',
        'queue' => ['data-proc'],
        'balance' => 'auto',
        'maxProcesses' => 3,
        'memory' => 128,
        'tries' => 1,
        'timeout' => 900,
    ],
],

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.

'environments' => [
    'local' => [
        'consumption-data' => ['maxProcesses' => 3],
        'broadcast-supervisor' => ['maxProcesses' => 3],
    ],
    'staging' => [
        'consumption-data' => ['maxProcesses' => 3],
        'broadcast-supervisor' => ['maxProcesses' => 2],
    ],
    'production' => [
        'consumption-data' => ['maxProcesses' => 3],
        'broadcast-supervisor' => ['maxProcesses' => 3],
    ],
],

Then:

php artisan horizon
# or restart after config changes:
php artisan horizon:terminate

Starting a task

Artisan

# oldest unconsumed file for the task
php artisan consume catalog-products

# specific file — basename only (folder comes from config/tasks.php)
php artisan consume catalog-products --file=cp_2024-05-30.csv

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

use ProjectSaturnStudios\ConsumptionEngine\ConsumptionEngine;
use ProjectSaturnStudios\ConsumptionEngine\Enums\TaskExecutionStatus;

$filename = null; // input: null or basename under the task folder
$status = app(ConsumptionEngine::class)->executeTask('catalog-products', $filename);
// on success, $filename is filled with the full storage path by reference

if ($status === TaskExecutionStatus::TASK_STARTED) {
    // job is on data-proc
}

executeTask() only validates and dispatches. For progress monitoring, subscribe via Echo (Realtime / Echo events) or BLPOP the Redis list consume-{task} as php artisan consume does — including the 10 seconds of inactivity exit and terminal handling for TaskCompleted / TaskFailed (see 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.

protected function getContents(): ?array
{
    return array_map(function (array $row) {
        $row['uuid'] = /* deterministic id for this row */;
        return $row;
    }, parent::getContents());
}

Implement the abstract methods

namespace App\Jobs\Consumption\Example;

use Illuminate\Support\Collection;
use ProjectSaturnStudios\ConsumptionEngine\Jobs\ConsumptionJob;
use ProjectSaturnStudios\ConsumptionEngine\Events\Realtime\TaskDataValidationStarted;
use ProjectSaturnStudios\ConsumptionEngine\Events\Realtime\TaskWarning;

class ConsumeExampleData extends ConsumptionJob
{
    protected function validateRecords(Collection &$raw_contents_sorted): void
    {
        $this->fireEvent(new TaskDataValidationStarted(
            $this->channel, $this->filename, $this->task, count($raw_contents_sorted)
        ));

        $results = collect();

        foreach ($raw_contents_sorted as $uuid => $content) {
            try {
                // Validate / build DTO...
                $results->put($uuid, $content);
                $this->audit_log->put($uuid, ['status' => 'valid']);
            } catch (\Throwable $e) {
                $this->audit_log->put($uuid, [
                    'status' => 'invalid',
                    'error' => $e->getMessage(),
                ]);
                $this->fireEvent(new TaskWarning(
                    $this->channel, $this->filename, $this->task, "Invalid: {$e->getMessage()}"
                ));
            }
        }

        // Replace with processable rows only (invalid rows stay in audit_log)
        $raw_contents_sorted = $results;
    }

    protected function setRecordActions(Collection &$raw_processable_contents): void
    {
        // Load existing domain state yourself, then decide create / update / none.
        $raw_processable_contents = $this->mapRecordsWithProgress(
            $raw_processable_contents,
            function ($uuid, $content, Collection $results): void {
                $entry = $this->audit_log->get($uuid) ?? [];
                $entry['action'] = 'create'; // or update / none
                $this->audit_log->put($uuid, $entry);
                $results->put($uuid, $content);
            }
        );
    }

    protected function consumeNewRecords(Collection $raw_processable_contents): void
    {
        $this->consumeActionRecords($raw_processable_contents, 'create', function ($dto): void {
            event_command(/* your create command */, []);
        });
    }

    protected function updateRecords(Collection $raw_processable_contents): void
    {
        $this->updateActionRecords($raw_processable_contents, 'update', function ($dto): void {
            event_command(/* your update command */, []);
        });
    }
}

What belongs in each abstract method

Method	Responsibility
validateRecords	Schema/DTO validation; keep invalid rows out of the returned collection; seed audit status. Whatever you put in the collection (arrays or DTOs) is what later phases receive.
setRecordActions	Diff against existing projections (your code); set action (create / update / none / …). Rows with none stay in the audit log but are skipped by update/create helpers.
updateRecords	Optional. Apply update actions (usually via updateActionRecords(..., 'update', ...)). Runs before creates.
consumeNewRecords	Apply create actions (usually via event_command(...) + your domain packages)

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:

consume-{task}

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:

Echo.channel('consume-catalog-products')
  .listen('.TaskInitializing', (e) => { /* ... */ })
  .listen('.TaskInitialized', (e) => { /* ... */ })
  .listen('.TaskDataValidationStarted', (e) => { /* e.num_records */ })
  .listen('.TaskProgress', (e) => { /* e.pct */ })
  .listen('.TaskUpdateConsumeStarted', (e) => { /* e.num_records */ })
  .listen('.TaskRecordConsumeStarted', (e) => { /* e.num_records */ })
  .listen('.SavingAuditLog', (e) => { /* ... */ })
  .listen('.TaskCompleted', (e) => { /* ... */ })
  .listen('.TaskFailed', (e) => { /* e.reason */ })
  .listen('.TaskWarning', (e) => { /* e.warning */ });

Event	When	Fired by
TaskInitializing	Job handle starts	Base job (automatic)
TaskInitialized	Audit log started / contents loaded	Base job (automatic)
TaskDataValidationStarted	Validation phase begins	Your job via fireEvent()
TaskProgress	Throttled percent progress	Base helpers (automatic when used)
TaskUpdateConsumeStarted	Update phase begins	updateActionRecords()
TaskRecordConsumeStarted	Create/consume phase begins	consumeActionRecords()
SavingAuditLog	Before audit JSON + finish command	Base job (automatic)
TaskCompleted	Happy path finished	Base job after successful audit save
TaskFailed	Exception during audit save	Base job catch in saveAuditLog() only
TaskWarning	Non-fatal row issues	Your job via fireEvent()

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:

   composer require spatie/laravel-event-sourcing projectsaturnstudios/pss-event-sourcing spatie/laravel-medialibrary

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:

   EVENT_PROJECTOR_QUEUE_NAME=app-events

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.

cd vendor/projectsaturnstudios/consumption-engine
# or your path-repo checkout of the package
composer install
composer test

License

MIT