PHPackages projectsaturnstudios/consumption-engine - PHPackages - PHPackages [Skip to content](#main-content)[PHPackages](/)[Directory](/)[Categories](/categories)[Trending](/trending)[Leaderboard](/leaderboard)[Changelog](/changelog)[Analyze](/analyze)[Collections](/collections)[Log in](/login)[Sign up](/register) 1. [Directory](/) 2. / 3. [Utility & Helpers](/categories/utility) 4. / 5. projectsaturnstudios/consumption-engine ActiveLibrary[Utility & Helpers](/categories/utility) projectsaturnstudios/consumption-engine ======================================= Opinionated implementation on mass data consumption. For Laravel. 0.1.1(1mo ago)027↓80%MITPHPPHP ^8.3 Since Jul 10Pushed 1mo agoCompare [ Source](https://github.com/projectsaturnstudios/consumption-engine)[ Packagist](https://packagist.org/packages/projectsaturnstudios/consumption-engine)[ RSS](/packages/projectsaturnstudios-consumption-engine/feed)WikiDiscussions main Synced 1w ago READMEChangelogDependencies (20)Versions (4)Used By (0) Consumption Engine ================== [](#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 ----------- [](#quick-start) 1. Install the package — use **Option A or B** under [Install](#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](#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 ------------ [](#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](https://bitbucket.org/projectsaturnstudios) — 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 ------- [](#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-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 ------------- [](#configuration) Published file: `config/tasks.php`. ### Source & audit disks [](#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 [](#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 -------------------- [](#queues--horizon) This package is **queue-first**: ConcernDefault queueNotesConsumption jobs`data-proc`Dispatched by `executeTask()` / `consume`Realtime broadcast events`broadcasts``ConsumptionRTEvent::$queue` / `broadcastQueue()` — keep off the projector queueEvent-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 [](#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 --------------- [](#starting-a-task) ### Artisan [](#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 [](#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](#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](#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 -------------------------- [](#creating-a-consumption-job) ### Pipeline (`handle()`) [](#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 [](#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 [](#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 [](#what-belongs-in-each-abstract-method) MethodResponsibility`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 [](#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 [](#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 ---------------------- [](#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) [](#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 [](#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 */ }); ``` EventWhenFired by`TaskInitializing`Job handle startsBase job (automatic)`TaskInitialized`Audit log started / contents loadedBase job (automatic)`TaskDataValidationStarted`Validation phase beginsYour job via `fireEvent()``TaskProgress`Throttled percent progressBase helpers (automatic when used)`TaskUpdateConsumeStarted`Update phase begins`updateActionRecords()``TaskRecordConsumeStarted`Create/consume phase begins`consumeActionRecords()``SavingAuditLog`Before audit JSON + finish commandBase job (automatic)`TaskCompleted`Happy path finishedBase job after successful audit save`TaskFailed`Exception during audit saveBase job catch in `saveAuditLog()` only`TaskWarning`Non-fatal row issuesYour 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 --------------------------- [](#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) [](#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](#queues--horizon)). 7. Publish and migrate this package’s `audit_logs` table (see [Publish config & migration](#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 ------- [](#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 ------- [](#license) MIT ### Health Score 39 — LowBetter than 84% of packages Maintenance93 Actively maintained with recent releases Popularity10 Limited adoption so far Community6 Small or concentrated contributor base Maturity41 Maturing project, gaining track record Bus Factor1 Top contributor holds 100% of commits — single point of failure How is this calculated?**Maintenance (25%)** — Last commit recency, latest release date, and issue-to-star ratio. Uses a 2-year decay window. **Popularity (30%)** — Total and monthly downloads, GitHub stars, and forks. Logarithmic scaling prevents top-heavy scores. **Community (15%)** — Contributors, dependents, forks, watchers, and maintainers. Measures real ecosystem engagement. **Maturity (30%)** — Project age, version count, PHP version support, and release stability. ### Release Activity Cadence Every ~6 days Total 3 Last Release 35d ago ### Community Maintainers ![](https://avatars.githubusercontent.com/u/10563160?v=4)[Angel Gonzalez](/maintainers/projectsaturnstudios)[@projectsaturnstudios](https://github.com/projectsaturnstudios) --- Top Contributors [![projectsaturnstudios](https://avatars.githubusercontent.com/u/10563160?v=4)](https://github.com/projectsaturnstudios "projectsaturnstudios (3 commits)") ### Code Quality TestsPest ### Embed Badge ![Health badge](/badges/projectsaturnstudios-consumption-engine/health.svg) ``` [![Health](https://phpackages.com/badges/projectsaturnstudios-consumption-engine/health.svg)](https://phpackages.com/packages/projectsaturnstudios-consumption-engine) ``` ### Alternatives [filament/spatie-laravel-media-library-plugin Filament support for `spatie/laravel-medialibrary`. 1826.5M245](/packages/filament-spatie-laravel-media-library-plugin)[ebess/advanced-nova-media-library Laravel Nova tools for managing the Spatie media library. 6143.6M22](/packages/ebess-advanced-nova-media-library)[nasirkhan/laravel-starter A CMS like modular Laravel starter project. 1.4k2.7k](/packages/nasirkhan-laravel-starter)[tapp/filament-form-builder User facing form builder using Filament components 133.5k6](/packages/tapp-filament-form-builder) PHPackages © 2026 [Directory](/)[Categories](/categories)[Trending](/trending)[Changelog](/changelog)[Analyze](/analyze)