Spooled PHP SDK

Official PHP SDK for Spooled Cloud - a modern job queue service for distributed applications.

Live Demo (SpriteForge) • Documentation

Features

• Full REST API support - Jobs, queues, workers, schedules, workflows, webhooks, and more
• Worker runtime - Process jobs with lease renewal, lease fencing, events, and graceful shutdown
• Realtime events - SSE and WebSocket clients for live job/queue events
• Optional gRPC transport - High-performance binary protocol when extensions are available
• Retry & circuit breaker - Built-in resilience with exponential backoff and circuit breaker
• Type-safe - Full PHP 8.2+ type hints and readonly DTOs
• PSR-compliant - Works with any PSR-compatible HTTP client and logger
• Framework agnostic - Use with Laravel, Symfony, or vanilla PHP
• Webhook ingestion - Validate and process GitHub, Stripe, and custom webhooks
• Dead Letter Queue (DLQ) - Manage and retry failed jobs
• Billing integration - Stripe-powered subscription management
• Automatic case conversion - Between camelCase (PHP) and snake_case (API)

Requirements

• PHP 8.2 or higher
• Composer
• ext-json (usually bundled)

Optional Extensions

• ext-pcntl + ext-posix - Required by SpooledWorker at startup for automatic lease renewal while synchronous handlers run (Unix-like CLI environments); REST/gRPC/realtime clients do not require them
• ext-grpc + ext-protobuf - For gRPC transport support
• WebSocket library (e.g., ratchet/pawl) - For WebSocket realtime support

Installation

composer require spooled-cloud/spooled

With gRPC Support

# Install PHP extensions first
pecl install grpc protobuf

# Then require gRPC packages
composer require grpc/grpc google/protobuf

Quick Start

<?php

use Spooled\SpooledClient;
use Spooled\Config\ClientOptions;

// Create client with API key
$client = new SpooledClient(
    new ClientOptions(apiKey: 'sp_live_your_api_key')
);

// Create a job
$job = $client->jobs->create([
    'queue' => 'email-notifications',
    'payload' => [
        'to' => 'user@example.com',
        'subject' => 'Welcome!',
        'body' => 'Thanks for signing up.',
    ],
    'priority' => 5,
    'maxRetries' => 3,
]);

echo "Created job: {$job->id}\n";

// Get job status
$status = $client->jobs->get($job->id);
echo "Status: {$status->status}\n";

Documentation

Guide	Description
Getting Started	Installation, setup, and first job
Configuration	All configuration options, retry, and circuit breaker
Workers	SpooledWorker runtime, concurrency, and graceful shutdown
Workflows	DAG workflows with job dependencies
gRPC	High-performance gRPC transport
Resources	Complete API reference for all resources
Publishing	Publishing to Packagist

Examples

See the examples/ directory for runnable code:

Example	Description
basic-usage.php	Basic SDK usage
worker-example.php	Processing jobs with SpooledWorker
workflow-example.php	Complex workflows with dependencies
scheduled-jobs.php	Cron schedules
grpc-example.php	High-performance gRPC transport
realtime-example.php	SSE/WebSocket event streaming
webhook-ingestion-example.php	Webhook validation and ingestion
error-handling.php	Error handling patterns and retry logic

Real-world examples (beginner friendly)

If you want 5 copy/paste “real life” setups (Stripe → jobs, GitHub Actions → jobs, cron schedules, CSV import, website signup), see:

• https://github.com/spooled-cloud/spooled-backend/blob/main/docs/guides/real-world-examples.md

Core Concepts

Jobs

Jobs are units of work with payloads, priorities, and retry policies:

<?php

$job = $client->jobs->create([
    'queue' => 'my-queue',
    'payload' => ['data' => 'value'],
    'priority' => 5,                     // -100 to 100
    'maxRetries' => 3,
    'timeoutSeconds' => 300,
    'scheduledFor' => '2024-01-15T10:00:00Z',
    'idempotencyKey' => 'unique-key',
]);

// Get job status
$job = $client->jobs->get($job->id);

// List jobs
$jobs = $client->jobs->list([
    'queue' => 'my-queue',
    'status' => 'pending',
    'tag' => 'billing',  // Optional: filter by a single tag
]);

// Cancel a job
$client->jobs->cancel($job->id);

// Boost priority
$client->jobs->boostPriority($job->id, 10);

// Bulk enqueue
$result = $client->jobs->bulkEnqueue([
    ['queue' => 'my-queue', 'payload' => ['item' => 1]],
    ['queue' => 'my-queue', 'payload' => ['item' => 2]],
]);

Workers

Process jobs with the built-in worker runtime. SpooledWorker uses short-lived pcntl renewal processes so blocking synchronous handlers keep their leases without requiring an event loop; install ext-pcntl and ext-posix in the CLI environment.

<?php

use Spooled\SpooledClient;
use Spooled\Config\ClientOptions;
use Spooled\Worker\SpooledWorker;
use Spooled\Worker\JobContext;

$client = new SpooledClient(
    new ClientOptions(apiKey: 'sp_live_your_api_key')
);

$worker = new SpooledWorker($client, [
    'queueName' => 'my-queue',
    'concurrency' => 10,
    // Optional, but recommended: a stable id (1-128 chars, [A-Za-z0-9._-])
    // makes registration an upsert, so restarts reuse this worker's row
    // instead of leaving the old one against the plan worker cap.
    'workerId' => 'my-queue-worker-01',
]);

$worker->process(function (JobContext $ctx): array {
    echo "Processing job {$ctx->jobId}\n";
    
    // Access payload
    $data = $ctx->get('data');
    
    // Check if shutting down
    if ($ctx->isShuttingDown()) {
        throw new \RuntimeException('Worker shutting down');
    }
    
    // Return result (job completed on success)
    return ['success' => true];
});

// Handle shutdown gracefully
pcntl_signal(SIGTERM, fn() => $worker->stop());
pcntl_signal(SIGINT, fn() => $worker->stop());

// Start processing (blocking)
$worker->start();

Workflows (DAGs)

Orchestrate multiple jobs with dependencies:

<?php

$workflow = $client->workflows->create([
    'name' => 'ETL Pipeline',
    'jobs' => [
        ['key' => 'extract', 'queue' => 'etl', 'payload' => ['step' => 'extract']],
        ['key' => 'transform', 'queue' => 'etl', 'payload' => ['step' => 'transform'], 'dependsOn' => ['extract']],
        ['key' => 'load', 'queue' => 'etl', 'payload' => ['step' => 'load'], 'dependsOn' => ['transform']],
    ],
]);

// Get workflow status
$workflow = $client->workflows->get($workflow->id);
echo "Progress: {$workflow->completedJobs}/{$workflow->totalJobs}\n";

// List workflow jobs
$jobs = $client->workflows->jobs->list($workflow->id);

// Cancel workflow
$client->workflows->cancel($workflow->id);

Schedules

Run jobs on a cron schedule:

<?php

$schedule = $client->schedules->create([
    'name' => 'Daily Report',
    'queue' => 'reports',
    'schedule' => '0 9 * * *',           // 5-field cron
    'payload' => ['type' => 'daily'],
    'timezone' => 'America/New_York',
]);

// List schedules
$schedules = $client->schedules->list();

// Pause/resume
$client->schedules->pause($schedule->id);
$client->schedules->resume($schedule->id);

// Trigger immediately
$job = $client->schedules->trigger($schedule->id);

// Get execution history
$runs = $client->schedules->history($schedule->id);

// Delete schedule
$client->schedules->delete($schedule->id);

Realtime Events

Subscribe to real-time job events via SSE or WebSocket:

<?php

use Spooled\SpooledClient;
use Spooled\Config\ClientOptions;

$client = new SpooledClient(new ClientOptions(
    apiKey: 'sp_live_your_api_key',
));

// Get unified realtime client (auto-selects best transport)
$realtime = $client->realtime();

// Check available transport
if ($realtime->isWebSocketAvailable()) {
    echo "Using WebSocket\n";
} else {
    echo "Using SSE\n";
}

// Subscribe to queue events
$realtime->subscribeToQueue('my-queue', function (array $event): void {
    echo "Event: {$event['type']} - Job: {$event['data']['jobId']}\n";
});

// Or subscribe to specific job
$realtime->subscribeToJob($jobId, function (array $event): void {
    echo "Job event: {$event['type']}\n";
});

// Stop when done
$realtime->stop();
$client->close();

Direct SSE Client

<?php

use Spooled\Realtime\SseClient;

$sse = new SseClient(
    baseUrl: 'https://api.spooled.cloud',
    apiKey: 'sp_live_your_api_key',
);

// Register event handlers
$sse->on('job.created', fn($e) => print("New job!\n"));
$sse->on('job.completed', fn($e) => print("Job done!\n"));

// Subscribe to all events
$sse->subscribe(function (array $event): void {
    echo "[{$event['type']}] " . json_encode($event['data']) . "\n";
});

// Start listening (blocking)
$sse->listen();

Organization Management

Manage your organization and track usage:

<?php

// Get current usage and limits
$usage = $client->organizations->getUsage();
echo "Plan: {$usage->plan}\n";

// Generate a unique slug for a new organization
$slug = $client->organizations->generateSlug('My Company');

// Check if a slug is available
$result = $client->organizations->checkSlug('my-company');
echo $result['available'] ? 'Available' : 'Taken';

// List organizations
$orgs = $client->organizations->list();

// Webhook token management
$token = $client->organizations->getWebhookToken();
$client->organizations->regenerateWebhookToken();
$client->organizations->clearWebhookToken();

Webhooks

Configure outgoing webhooks for job events:

<?php

// Create webhook
$webhook = $client->webhooks->create([
    'name' => 'My Webhook',
    'url' => 'https://your-app.com/webhooks/spooled',
    'events' => ['job.completed', 'job.failed'],
    'secret' => 'hmac-secret',
]);

// Test webhook
$client->webhooks->test($webhook->id);

// Get delivery history (retained for the plan's window, see below)
$deliveries = $client->webhooks->getDeliveries($webhook->id);

// Retry a failed delivery
$client->webhooks->retryDelivery($webhook->id, $deliveryId);

// Delete webhook
$client->webhooks->delete($webhook->id);

On update, secret is three-state. Omit the key to keep the current secret, pass a string to replace it, and pass null to clear it. Clearing means deliveries go out unsigned, with no X-Spooled-Signature header, and a receiver that verifies signatures will reject every one of them. Send only the keys you mean to change; serialising a whole webhook back with null defaults now wipes the secret.

$client->webhooks->update($webhook->id, ['secret' => null]); // stops signing

After 20 consecutive failed deliveries a webhook is disabled automatically: enabled becomes false and lastStatus becomes "auto_disabled", and it receives no events until it is enabled again. failureCount counts consecutive failed deliveries rather than individual retry attempts, and any success resets it to 0.

if ($webhook->lastStatus === 'auto_disabled') {
    // Fix the endpoint first. Re-enabling is charged against the plan webhook
    // cap, so this can throw RateLimitError with errorCode "QUOTA_EXCEEDED".
    $client->webhooks->enable($webhook->id);
}

Delivery history is retained, not permanent: rows are removed once past the plan's history retention window (free 1 day, starter 7, pro 30, enterprise 90), and only the newest 100 deliveries per webhook are readable. Copy anything you need to keep into your own store; once a row is swept it can no longer be retried.

Dead Letter Queue (DLQ)

Manage jobs that have exhausted all retries:

<?php

// List DLQ jobs
$dlqJobs = $client->jobs->dlq->list(['limit' => 100]);

// Retry specific jobs from DLQ
$result = $client->jobs->dlq->retry([
    'jobIds' => ['job-1', 'job-2'],
]);

// Retry jobs by queue
$result = $client->jobs->dlq->retry([
    'queue' => 'my-queue',
    'limit' => 50,
]);

// Purge DLQ (requires confirmation)
$result = $client->jobs->dlq->purge([
    'queue' => 'my-queue',
    'confirm' => true,
]);

API Key Management

Manage API keys programmatically:

<?php

// Create a new API key
$apiKey = $client->apiKeys->create([
    'name' => 'Production Worker',
]);
echo "Save this key: {$apiKey->key}\n"; // Only shown once!

// List all API keys
$keys = $client->apiKeys->list();

// Update key
$client->apiKeys->update($keyId, ['name' => 'Updated Name']);

// Revoke a key
$client->apiKeys->delete($keyId);

$key->lastUsedAt is coarse. The API writes it at most once per key per five minutes, not once per request, so it can be up to five minutes behind. Use it to answer "has this key been used lately", not "did that request just happen", and give rotation tooling a margin wider than the write interval before treating a key as unused.

Billing & Subscriptions

Manage billing via Stripe integration:

<?php

// Get billing status
$status = $client->billing->getStatus();
echo "Plan: {$status->planTier}\n";

// Create customer portal session
$portal = $client->billing->createPortal([
    'returnUrl' => 'https://your-app.com/settings',
]);

// Redirect user to: $portal->url

Webhook Ingestion

Validate and process incoming webhooks from GitHub, Stripe, or custom sources:

<?php

// Ingest custom webhook (creates a job)
$result = $client->ingest->custom($orgId, [
    'queueName' => 'webhooks',
    'eventType' => 'user.created',
    'payload' => ['userId' => '123', 'email' => 'user@example.com'],
]);

echo "Created job: {$result['jobId']}\n";

// Ingest GitHub webhook (with raw body for signature)
$result = $client->ingest->github(
    orgId: $orgId,
    body: file_get_contents('php://input'),
    githubEvent: $_SERVER['HTTP_X_GITHUB_EVENT'] ?? 'push',
    signature: $_SERVER['HTTP_X_HUB_SIGNATURE_256'] ?? null,
    secret: 'your-github-secret',  // SDK computes signature if not provided
);

// Ingest Stripe webhook
$result = $client->ingest->stripe(
    orgId: $orgId,
    body: file_get_contents('php://input'),
    signature: $_SERVER['HTTP_STRIPE_SIGNATURE'] ?? null,
    secret: 'whsec_...',
);

Signature Validation Helpers

<?php

// Generate signatures (for testing)
$githubSig = $client->ingest->generateGitHubSignature($payload, $secret);
$stripeSig = $client->ingest->generateStripeSignature($payload, $secret);

// Validate signatures manually
$valid = $client->ingest->validateGitHubSignature($payload, $signature, $secret);
$valid = $client->ingest->validateStripeSignature($payload, $signature, $secret, tolerance: 300);

// Example webhook endpoint handler
$payload = file_get_contents('php://input');
$signature = $_SERVER['HTTP_X_HUB_SIGNATURE_256'] ?? '';

if (!$client->ingest->validateGitHubSignature($payload, $signature, $secret)) {
    http_response_code(401);
    exit('Invalid signature');
}

// Process webhook...

Configuration

<?php

use Spooled\SpooledClient;
use Spooled\Config\ClientOptions;
use Spooled\Config\RetryConfig;
use Spooled\Config\CircuitBreakerConfig;

$client = new SpooledClient(new ClientOptions(
    // Authentication (one of these)
    apiKey: 'sp_live_...',
    // accessToken: 'jwt-token',
    // adminKey: 'admin-key',
    
    // URLs
    baseUrl: 'https://api.spooled.cloud',
    // grpcAddress: 'grpc.spooled.cloud:443',
    
    // Timeouts (seconds)
    connectTimeout: 10.0,
    requestTimeout: 30.0,
    
    // Retry configuration
    retry: new RetryConfig(
        maxRetries: 3,
        baseDelay: 1.0,      // seconds
        maxDelay: 30.0,      // seconds
        factor: 2.0,
        jitter: 0.1,
    ),
    
    // Circuit breaker
    circuitBreaker: new CircuitBreakerConfig(
        enabled: true,
        failureThreshold: 5,
        successThreshold: 2,
        timeout: 30.0,       // seconds
    ),
    
    // Custom headers
    headers: [
        'X-Custom-Header' => 'value',
    ],
    
    // PSR-3 logger
    logger: $myLogger,
));

Error Handling

All errors extend SpooledError with specific subclasses:

<?php

use Spooled\Errors\SpooledError;
use Spooled\Errors\AuthenticationError;
use Spooled\Errors\NotFoundError;
use Spooled\Errors\RateLimitError;
use Spooled\Errors\ValidationError;

try {
    $job = $client->jobs->get('non-existent-id');
} catch (NotFoundError $e) {
    echo "Job not found: {$e->getMessage()}\n";
} catch (RateLimitError $e) {
    echo "Rate limited. Retry after: {$e->getRetryAfterSeconds()} seconds\n";
} catch (ValidationError $e) {
    echo "Validation failed: {$e->getMessage()}\n";
    foreach ($e->getFieldErrors() as $field => $errors) {
        echo "  {$field}: " . implode(', ', $errors) . "\n";
    }
} catch (AuthenticationError $e) {
    echo "Invalid API key\n";
} catch (SpooledError $e) {
    echo "Error [{$e->statusCode}]: {$e->getMessage()}\n";
    echo "Request ID: {$e->requestId}\n";
}

gRPC Transport (Optional)

For high-throughput workers, use the gRPC API:

<?php

use Spooled\SpooledClient;
use Spooled\Config\ClientOptions;

// Requires ext-grpc and ext-protobuf
$client = new SpooledClient(new ClientOptions(
    apiKey: 'sp_live_your_api_key',
    grpcAddress: 'grpc.spooled.cloud:443',
));

// Get gRPC client (lazy-initialized)
$grpc = $client->grpc();

// Wait for connection
$grpc->waitForReady();

// Enqueue via gRPC (higher throughput than REST)
$result = $grpc->queue->enqueue([
    'queueName' => 'fast-jobs',
    'payload' => ['data' => 'value'],
    'priority' => 5,
]);

echo "Created job: {$result['jobId']}\n";

// Get queue stats
$stats = $grpc->queue->getStats(['queueName' => 'fast-jobs']);

// For manual dequeue loops, preserve the returned leaseId and echo it on
// complete(), fail(), and renewLease(). renewLease uses extensionSecs.

// Register worker
$worker = $grpc->workers->register([
    'queueName' => 'fast-jobs',
    'hostname' => gethostname(),
    'concurrency' => 10,
]);

// Send heartbeat
$grpc->workers->heartbeat([
    'workerId' => $worker['workerId'],
    'currentJobs' => 0,
]);

// Deregister when done
$grpc->workers->deregister(['workerId' => $worker['workerId']]);

// Clean up connections
$client->close();

Standalone gRPC Client

<?php

use Spooled\Grpc\SpooledGrpcClient;
use Spooled\Grpc\GrpcOptions;

$grpc = new SpooledGrpcClient(new GrpcOptions(
    address: 'grpc.spooled.cloud:443',
    apiKey: 'sp_live_your_api_key',
    secure: true,
));

$grpc->waitForReady();
// ... use $grpc->queue and $grpc->workers
$grpc->close();

Plan Limits

All operations automatically enforce tier-based limits:

Tier	Active Jobs	Daily Jobs	Queues	Workers	Webhooks
Free	10	1,000	5	3	2
Starter	100	100,000	25	25	10
Enterprise	Unlimited	Unlimited	Unlimited	Unlimited	Unlimited

When a plan quota or limit is exceeded, you'll receive an HTTP 429 (a RateLimitError) with errorCode "QUOTA_EXCEEDED" and details describing the resource, current, limit, and plan:

<?php

use Spooled\Errors\RateLimitError;

try {
    $client->jobs->create([/* ... */]);
} catch (RateLimitError $e) {
    if ($e->errorCode === 'QUOTA_EXCEEDED') {
        echo "Plan quota exceeded: {$e->getMessage()}\n";
    } else {
        // Per-second rate limiting also returns 429
        echo "Rate limited. Retry after: {$e->getRetryAfterSeconds()} seconds\n";
    }
}

Version Source

Release 1.1.0 is defined once in Spooled\Version::VERSION. The default HTTP User-Agent and default SpooledWorker registration version derive from that constant. Application metadata may override the worker version explicitly.

Development

# Install dependencies
composer install

# Run tests
composer test

# Run with coverage
XDEBUG_MODE=coverage composer test:coverage

# Static analysis
composer analyse

# Code formatting
composer format

# All CI checks
composer ci

Testing Scripts

The SDK includes parity test scripts that match the Node.js and Python SDK test suites:

# Run local tests (requires running backend)
API_KEY=sp_test_... BASE_URL=http://localhost:8080 composer scripts:test-local

# Run production tests (safe subset)
API_KEY=sp_live_... composer scripts:test-production

# Interactive verification
composer scripts:verify-production

License

Apache License 2.0 - see LICENSE for details.