jsdevart/laravel-managed-jobs

Laravel Managed Jobs

A Laravel package for managing background jobs with lifecycle tracking, real-time progress broadcasting, and file management.

What it does

You dispatch a job. The package:

• Creates a ManagedJob record that tracks its full lifecycle (PENDING → RUNNING → COMPLETED / FAILED / STOPPED)
• Broadcasts real-time progress events via WebSockets so your frontend can show a progress bar
• Stores files generated by the job with automatic expiration
• Fires lifecycle events (JobCompleted, JobFailed, etc.) your app can listen to

Requirements

Installation

composer require jsdevart/laravel-managed-jobs
php artisan migrate

Publish the config if you need to customise it:

php artisan vendor:publish --tag=managed-jobs-config

Minimal implementation

This section walks through wiring the package into your app — one decision (who owns a job) and three pieces of code (payload, job, dispatch).

1. Decide who owns a job

Nothing to implement. A job's owner is any Eloquent model — a User, a Team, a Tenant, a Service — stored polymorphically (owner_type / owner_id). You just pass the model to JobRunner::dispatch() (step 4) and read it back with $job->owner.

Optionally register a morph map so the database stores short, stable aliases instead of full class names (this also keeps broadcast channel names tidy — jobs.user.5 instead of jobs.app_models_user.5):

// AppServiceProvider::boot()
use Illuminate\Database\Eloquent\Relations\Relation;

Relation::enforceMorphMap([
    'user'    => \App\Models\User::class,
    'tenant'  => \App\Models\Tenant::class,
    'service' => \App\Models\Service::class,
]);

Upgrading from v1? The old JobOwner interface (getManagedJobOwnerId() / getManagedJobTenantId()) is no longer required — see UPGRADE.md.

2. Define the job's input

Implement JobPayload on any class that has a toArray() method.

use YourVendor\ManagedJobs\Contracts\JobPayload;

class GenerateReportPayload implements JobPayload
{
    public function __construct(
        public readonly string $dateFrom,
        public readonly string $dateTo,
    ) {}

    public function toArray(): array
    {
        return [
            'date_from' => $this->dateFrom,
            'date_to'   => $this->dateTo,
        ];
    }
}

Any class that already has a toArray() method — DTOs, Form Requests, Eloquent models — satisfies JobPayload without modification. Just add implements JobPayload.

3. Write the job

Extend BaseJob and implement handle().

use YourVendor\ManagedJobs\Jobs\BaseJob;

class GenerateReportJob extends BaseJob
{
    public function handle(): void
    {
        // Deserialize the stored payload back into your DTO
        ['date_from' => $from, 'date_to' => $to] = $this->jobExecution->payload;

        $rows  = Report::whereBetween('date', [$from, $to])->get();
        $total = $rows->count();

        foreach ($rows as $i => $row) {
            if ($this->isStopped()) {
                return; // user requested stop — exit cleanly
            }

            // ... your processing logic ...

            $this->updateProgress(
                percent: (int) (($i + 1) / $total * 100),
                message: "Processing row " . ($i + 1) . " of {$total}",
            );
        }
    }
}

4. Dispatch it

use YourVendor\ManagedJobs\Support\JobRunner;

$job = JobRunner::dispatch(
    job:     GenerateReportJob::class,
    payload: new GenerateReportPayload('2024-01-01', '2024-12-31'),
    owner:   $request->user(),
);

return response()->json(['job_id' => $job->job_id]);

That's it. The job record is created, the job is queued, and the lifecycle is tracked automatically.

Job API

Methods available inside handle():

failed(Throwable $e) is called automatically by Laravel when the job exhausts its retry attempts. It sets status = FAILED, stores the error message, and fires JobFailed.

Lifecycle

The middleware in BaseJob manages status transitions automatically:

PENDING  →  RUNNING  →  COMPLETED
                      ↘  FAILED     (can be retried)
                      ↘  STOPPED    (can be retried)

When a job completes, the package fires a JobCompleted event. When a job fails, it fires a JobFailed event. What happens next is entirely up to your app — listen to those events and react however you need.

HTTP endpoints

The package does not register routes. Add them yourself based on what your app needs:

// routes/api.php  or  routes/web.php
Route::middleware('auth')->prefix('jobs')->group(function () {

    // List the authenticated user's jobs
    Route::get('/', function (Request $request) {
        return ManagedJob::ownedBy($request->user())
            ->latest()
            ->paginate();
    });

    // Dispatch a new job
    Route::post('/', function (Request $request) {
        $validated = $request->validate([
            'date_from' => 'required|date',
            'date_to'   => 'required|date',
        ]);

        $job = JobRunner::dispatch(
            job:     GenerateReportJob::class,
            payload: new GenerateReportPayload($validated['date_from'], $validated['date_to']),
            owner:   $request->user(),
        );
        return response()->json(['job_id' => $job->job_id], 202);
    });

    // Stop a running/pending job
    Route::delete('/{jobId}', function (Request $request, string $jobId) {
        $job = ManagedJob::where('job_id', $jobId)
            ->ownedBy($request->user())
            ->firstOrFail();

        $job->update(['status' => JobStatusEnum::STOPPED]);
        event(new JobStopped($job));
    });

    // Retry a failed or stopped job
    Route::post('/{jobId}/retry', function (Request $request, string $jobId) {
        $job = ManagedJob::where('job_id', $jobId)
            ->ownedBy($request->user())
            ->firstOrFail();

        $job->update([
            'status'              => JobStatusEnum::PENDING,
            'progress_percentage' => 0,
            'progress_message'    => null,
            'failed_reason'       => null,
            'started_at'          => null,
            'finished_at'         => null,
        ]);

        DB::afterCommit(fn () => $job->type::dispatch($job));
    });

    // List non-expired files for a job
    Route::get('/{jobId}/files', function (Request $request, string $jobId) {
        $job = ManagedJob::where('job_id', $jobId)
            ->ownedBy($request->user())
            ->firstOrFail();

        return $job->files()->where('expires_at', '>', now())->get();
    });

    // Download a file
    Route::get('/{jobId}/files/{fileId}/download', function (Request $request, string $jobId, string $fileId) {
        $job  = ManagedJob::where('job_id', $jobId)
            ->ownedBy($request->user())
            ->firstOrFail();

        $file = $job->files()
            ->where('job_file_id', $fileId)
            ->where('expires_at', '>', now())
            ->firstOrFail();

        return Storage::download($file->path, $file->filename, [
            'Content-Type' => $file->mime_type,
        ]);
    });
});

In a real app you would extract this into a controller class. The inline closures above are for readability.

Real-time broadcasting

By default every event broadcasts on a single channel scoped to the job's owner:

• jobs.{ownerType}.{ownerId} — e.g. jobs.user.5, jobs.tenant.9, jobs.service.12

The owner type is part of the channel name, so owners of different types no longer collide even when their ids match (a user 7 job and a tenant 7 job resolve to jobs.user.7 and jobs.tenant.7). ownerType is the morph-map alias when you register one, otherwise a snake_case of the class name — so registering a morph map is recommended to give every type a distinct, stable alias (two classes sharing a basename would otherwise collapse to the same segment).

Need to broadcast somewhere else — a tenant-wide channel, a team channel, several channels at once? That is application policy, so the package hands it to you: see Custom channel policy below. To keep the flat v1-style name (jobs.{id}), set broadcasting.include_owner_type to false — only safe when every job is owned by a single owner type.

Frontend example (Laravel Echo):

Echo.channel(`jobs.user.${userId}`)
    .listen('.job.progress',  (e) => updateProgressBar(e.progress, e.progress_message))
    .listen('.job.completed', (e) => showDownloadButton(e.job_id))
    .listen('.job.failed',    (e) => showError(e.failed_reason));

File management

Register files produced by the job so users can download them later:

public function handle(): void
{
    // ... generate a CSV ...
    $path = "exports/{$this->jobExecution->getKey()}/report.csv";
    Storage::put($path, $csv);

    $this->addFile(
        path:      $path,
        filename:  'report.csv',
        mimeType:  'text/csv',
        sizeBytes: Storage::size($path),
        // expiresAt: Carbon instance — defaults to now() + config('managed-jobs.file_expiry_days')
    );
}

The managed-jobs:expire-files command deletes physical files whose expires_at has passed and soft-deletes their database records. It runs automatically every day at the configured time.

Run it manually:

php artisan managed-jobs:expire-files

Fault tolerance

Use saveState() to write a checkpoint after each unit of work. On retry, read it back with getState() to skip already-processed items:

public function handle(): void
{
    $lastId = $this->getState()['last_id'] ?? 0;

    Item::where('id', '>', $lastId)->lazyById()->each(function (Item $item) {
        if ($this->isStopped()) {
            return false;
        }

        // ... process ...

        $this->saveState(['last_id' => $item->id]);
    });
}

Configuration

Full reference after publishing with php artisan vendor:publish --tag=managed-jobs-config:

return [
    // Days before job-generated files expire (default: 3)
    'file_expiry_days' => 3,

    // Optional prefix for table names: 'bg_' → bg_managed_jobs, bg_managed_job_files
    // Must also be applied in your published migrations.
    'table_prefix' => '',

    // Broadcasting
    'broadcasting' => [
        'enabled'            => true,
        // Channel policy. Swap for your own JobChannelResolver to control
        // exactly which channels events broadcast on (tenant, team, service…).
        'resolver'           => \YourVendor\ManagedJobs\Support\DefaultJobChannelResolver::class,
        'channel_prefix'     => 'jobs',   // → jobs.{ownerType}.{ownerId}
        'include_owner_type' => true,     // false → v1-style jobs.{id} (single owner type only)
        'channel_type'       => 'public', // 'public' | 'private' | 'presence'
    ],

    // Queue settings applied to all managed jobs
    'queue' => [
        'connection' => null, // null = Laravel default
        'name'       => null, // null = connection default
    ],

    // Filesystem disk used for job file operations
    'storage' => [
        'disk' => null, // null = Laravel default
    ],

    // Scheduler for the expire-files command
    'schedule' => [
        'enabled'             => true,
        'expire_files_at'     => '22:00',
        'without_overlapping' => 5,    // minutes, or false to disable
        'on_one_server'       => true, // requires atomic-lock cache driver (Redis)
        'run_in_background'   => true,
    ],
];

Reacting to lifecycle events

The package fires a plain Laravel event at every status transition. Listen to them in your AppServiceProvider or EventServiceProvider and do whatever your app needs:

use YourVendor\ManagedJobs\Events\JobCompleted;
use YourVendor\ManagedJobs\Events\JobFailed;

// Send an email
Event::listen(JobCompleted::class, function (JobCompleted $event) {
    $event->jobRecord->owner?->notify(new YourJobCompletedNotification($event->jobRecord));
});

// Log the failure, alert on Slack, trigger a webhook — anything
Event::listen(JobFailed::class, function (JobFailed $event) {
    Log::error("Job failed: {$event->jobRecord->failed_reason}");
});

All five events (JobStarted, JobProgressUpdated, JobCompleted, JobStopped, JobFailed) expose $event->jobRecord — the ManagedJob model with full state.

Using private broadcast channels

Set channel_type to 'private' and define the authorization rule in routes/channels.php:

// config/managed-jobs.php
'broadcasting' => ['channel_type' => 'private'],

// routes/channels.php — matches the default jobs.{ownerType}.{ownerId} name
Broadcast::channel('jobs.user.{userId}', function ($user, $userId) {
    return (int) $user->id === (int) $userId;
});

Custom channel policy

Where events broadcast is application policy, so the package lets you own it. Implement JobChannelResolver and point the config at your class — this is the supported way to broadcast to a tenant, a team, several channels at once, or with any naming you like, without patching the package:

namespace App\ManagedJobs;

use Illuminate\Broadcasting\PrivateChannel;
use YourVendor\ManagedJobs\Contracts\JobChannelResolver;
use YourVendor\ManagedJobs\Models\ManagedJob;

class TenantChannelResolver implements JobChannelResolver
{
    public function channelsFor(ManagedJob $job): array
    {
        if (! config('managed-jobs.broadcasting.enabled', true)) {
            return [];
        }

        $channels = [new PrivateChannel("jobs.user.{$job->owner_id}")];

        // Broadcast to a tenant-wide channel too, derived from the owner.
        if ($tenantId = $job->owner?->tenant_id) {
            $channels[] = new PrivateChannel("jobs.tenant.{$tenantId}");
        }

        return $channels;
    }
}

// config/managed-jobs.php
'broadcasting' => [
    'resolver' => \App\ManagedJobs\TenantChannelResolver::class,
],

Authorize each channel your resolver emits in routes/channels.php.

Per-dispatch queue override

JobRunner::dispatch(
    job:        HeavyJob::class,
    payload:    $payload,
    owner:      $user,
    queue:      'heavy',       // overrides config queue.name for this dispatch only
    connection: 'sqs',         // overrides config queue.connection for this dispatch only
);

Database schema

managed_jobs

managed_job_files