PHP code example of projectsaturnstudios / consumption-engine

    
        
<?php
require_once('vendor/autoload.php');

/* Start to develop here. Best regards https://php-download.com/ */

    

projectsaturnstudios / consumption-engine example snippets

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
];

'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'),
    ],
],

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

'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,
    ],
],

'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],
    ],
],

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
}

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

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 */, []);
        });
    }
}

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

bash
php artisan migrate

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

bash
# 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

text
consume-{task}

js
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 */ });