Jobpipeline Laravel Package

Getting Started

Minimal Steps

1. Install the Package:

   composer require stancl/jobpipeline
   

   Ensure compatibility with your Laravel version (supports 10+).

2. Define a Job Pipeline: Create a pipeline for a series of jobs (e.g., in a service provider or event listener):

   use Stancl\JobPipeline\JobPipeline;
   use App\Jobs\ProcessOrder;
   use App\Jobs\SendNotification;
   use App\Jobs\UpdateInventory;
   
   $pipeline = JobPipeline::make([
       ProcessOrder::class,
       SendNotification::class,
       UpdateInventory::class,
   ]);
   

3. Map Event Data to Jobs: Attach a closure to extract data from an event and pass it to jobs:

   $pipeline->send(function ($event) {
       return $event->order; // Passes order data to each job
   });
   

4. Register as a Listener: Bind the pipeline to an event in EventServiceProvider or dynamically:

   // In EventServiceProvider.php
   protected $listen = [
       'order.placed' => [
           JobPipeline::make([...])->send(...)->toListener(),
       ],
   ];
   
   // Or dynamically:
   Event::listen('order.placed', $pipeline->toListener());
   

5. Trigger the Pipeline: Fire the event to execute the pipeline:

   event(new OrderPlaced($order));
   

First Use Case: Tenant Onboarding

Convert a multi-step tenant provisioning workflow (DB creation → migrations → seeding) into a reusable pipeline:

// In EventServiceProvider.php
protected $listen = [
    'tenant.created' => [
        JobPipeline::make([
            CreateDatabase::class,
            MigrateDatabase::class,
            SeedDatabase::class,
        ])->send(fn ($event) => $event->tenant)
         ->shouldBeQueued('tenants')
         ->toListener(),
    ],
];

Implementation Patterns

1. Event-Driven Workflows

• Pattern: Use pipelines for sequential, dependent jobs triggered by events. Example: Order processing (ProcessPayment → ShipOrder → NotifyCustomer).

  Event::listen('order.paid', JobPipeline::make([
      ShipOrder::class,
      NotifyCustomer::class,
  ])->send(fn ($event) => $event->order)->toListener());
  

• Integration Tip: Combine with Laravel’s dispatch() for hybrid sync/async flows:

  // Sync step + async pipeline
  ProcessOrder::dispatch($order);
  event(new OrderProcessed($order)); // Triggers pipeline
  

2. Conditional Pipeline Execution

• Pattern: Use return false in a job to cancel subsequent jobs in the pipeline. Example: Skip seeding if the database already exists:

  // In SeedDatabase job
  public function handle() {
      if (Schema::hasTable('users')) {
          return false; // Stops pipeline
      }
      // Seed logic...
  }
  

• Workflow: Split pipelines into logical groups (e.g., db-pipeline, seed-pipeline) for granular control:

  // EventServiceProvider.php
  $listen['tenant.created'] = [
      JobPipeline::make([CreateDatabase::class, MigrateDatabase::class])
          ->send(fn ($event) => $event->tenant)
          ->toListener(),
      JobPipeline::make([SeedDatabase::class])
          ->send(fn ($event) => $event->tenant)
          ->toListener(),
  ];
  

3. Dynamic Pipeline Configuration

• Pattern: Reuse pipelines with dynamic job lists or data. Example: Load jobs from a config or database:

  $jobs = config('pipelines.order_workflow.jobs');
  JobPipeline::make($jobs)->send(fn ($event) => $event->data)->toListener();
  

• Integration Tip: Use dependency injection to pass context (e.g., tenant ID, user ID):

  $pipeline = JobPipeline::make([...])
      ->send(fn ($event) => [
          'tenantId' => $event->tenantId,
          'userId' => auth()->id(),
      ]);
  

4. Queue Management

• Pattern: Default queued execution for long-running tasks:

  // Set globally (once)
  \Stancl\JobPipeline\JobPipeline::$shouldBeQueuedByDefault = true;
  
  // Override per pipeline
  JobPipeline::make([...])->shouldBeQueued('high-priority');
  

• Best Practices:

  • Use named queues (shouldBeQueued('queue-name')) for prioritization.
  • Monitor queues with Laravel Horizon or Supervisor.
  • Avoid blocking queues (e.g., sync) for user-facing flows.

5. Testing Pipelines

• Pattern: Mock pipelines in tests using JobPipeline::fake() (if supported) or manually:

  public function test_pipeline_execution() {
      $pipeline = JobPipeline::make([FakeJob::class])
          ->send(fn () => ['data' => 'test']);
  
      // Mock the job
      FakeJob::fake()->shouldReceive('handle')->once();
  
      event(new TestEvent());
  }
  

• Tip: Test failure scenarios by throwing exceptions in jobs and verifying pipeline cancellation.

Gotchas and Tips

Pitfalls

1. Pipeline Cancellation:

   • Returning false from a job stops all subsequent jobs in the same pipeline. If you need partial execution, split into multiple pipelines.
   • Debugging Tip: Log the pipeline state in jobs to trace where it fails:

     public function handle() {
         logger()->debug('Pipeline step: ' . static::class);
         if ($this->shouldFail) return false;
     }
     

2. Data Serialization:

   • Job payloads must be serializable (e.g., arrays, JSON-able objects). Avoid passing non-serializable objects (e.g., closures, resources) directly.
   • Fix: Use IDs or lazy-loading:

     ->send(fn ($event) => $event->user->id) // Pass ID instead of User model
     

3. Queue Deadlocks:

   • If jobs in a pipeline fail silently, the entire pipeline may hang. Use shouldQueue() with a timeout or retry logic:

     // In JobPipeline.php (custom extension)
     public function withRetry(int $attempts = 3) { ... }
     

4. Event Listener Registration:

   • Pipelines registered via Event::listen() won’t auto-wire like EventServiceProvider. Ensure closures are bound to the correct event class:

     // Wrong: Uses closure class name, not event class
     Event::listen(TestEvent::class, $pipeline->toListener());
     
     // Correct: Uses event class
     

Debugging Tips

1. Log Pipeline Execution: Add a middleware or job decorator to log pipeline steps:

   // In AppServiceProvider
   JobPipeline::make([...])->tap(function ($pipeline) {
       $pipeline->onStart(fn () => logger()->info('Pipeline started'));
       $pipeline->onJob(fn ($job) => logger()->debug("Running {$job}"));
   });
   

2. Inspect Queued Jobs: Use queue:work or Horizon to verify jobs are dispatched:

   php artisan queue:work --queue=tenants
   

3. Handle Exceptions: Wrap pipeline execution in a try-catch to log failures:

   try {
       event(new TenantCreated($tenant));
   } catch (\Throwable $e) {
       logger()->error("Pipeline failed: " . $e->getMessage());
   }
   

Extension Points

1. Custom Pipeline Behavior: Extend JobPipeline to add methods (e.g., withTimeout(), withRetry()):

   // app/Extensions/JobPipeline.php
   namespace App\Extensions;
   
   use Stancl\JobPipeline\JobPipeline;
   
   class EnhancedPipeline extends JobPipeline {
       public function withTimeout(int $seconds) {
           $this->timeout = $seconds;
           return $this;
       }
   }
   

2. Dynamic Job Loading: Load jobs dynamically from a database or API:

   $jobs = JobConfig::where('pipeline', 'tenant_onboarding')->pluck('job_class');
   JobPipeline::make($jobs)->send(...)->toListener();
   

3. Pipeline Events: Listen to pipeline lifecycle events (e.g., pipeline.starting, job.failed):

   Event::listen('pipeline.starting', fn ($pipeline) => logger()->info('Pipeline started'));
   

Configuration Quirks

1. Default Queue Behavior:
   • If `$shouldBeQueued