PHP code example of dealnews / inngest-php-sdk

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

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

    

dealnews / inngest-php-sdk example snippets

use DealNews\Inngest\Client\Inngest;

$client = new Inngest('my-app');

use DealNews\Inngest\Function\InngestFunction;
use DealNews\Inngest\Function\TriggerEvent;

$function = new InngestFunction(
    id: 'hello-world',
    handler: function ($ctx) {
        $name = $ctx->getEvent()->getData()['name'] ?? 'World';
        return ['message' => "Hello, {$name}!"];
    },
    triggers: [
        new TriggerEvent('demo/hello')
    ],
    name: 'Hello World Function'
);

$client->registerFunction($function);

use DealNews\Inngest\Http\ServeHandler;

$handler = new ServeHandler($client, '/api/inngest');

// In your framework (e.g., Laravel, Symfony):
$response = $handler->handle(
    method: $_SERVER['REQUEST_METHOD'],
    path: $_SERVER['REQUEST_URI'],
    headers: getallheaders(),
    body: file_get_contents('php://input'),
    query: $_GET
);

http_response_code($response['status']);
foreach ($response['headers'] as $key => $value) {
    header("{$key}: {$value}");
}
echo $response['body'];

use DealNews\Inngest\Event\Event;

$client->send(new Event(
    name: 'demo/hello',
    data: ['name' => 'PHP Developer']
));

use DealNews\Inngest\Function\InngestFunction;
use DealNews\Inngest\Function\TriggerEvent;

$function = new InngestFunction(
    id: 'process-order',
    handler: function ($ctx) {
        $step = $ctx->getStep();
        
        // Each step is individually retriable
        $order = $step->run('fetch-order', function () use ($ctx) {
            return fetchOrder($ctx->getEvent()->getData()['order_id']);
        });
        
        $step->run('charge-customer', function () use ($order) {
            return chargeCustomer($order);
        });
        
        // Sleep for a duration
        $step->sleep('wait-for-fulfillment', '1h');
        
        $step->run('send-confirmation', function () use ($order) {
            return sendConfirmationEmail($order);
        });
        
        return ['status' => 'complete'];
    },
    triggers: [new TriggerEvent('order/created')]
);

$payment = $step->waitForEvent(
    id: 'wait-for-payment',
    event: 'payment/completed',
    timeout: '1h',
    if: 'event.data.order_id == async.data.order_id'
);

$result = $step->invoke(
    id: 'call-function',
    function_id: 'my-app-other-function',
    payload: ['data' => ['foo' => 'bar']]
);

$step->sendEvent('notify-team', new Event(
    name: 'order/fulfilled',
    data: ['order_id' => $order['id']]
));

$response = $step->fetch(
    id: 'get-user',
    url: 'https://api.example.com/users/123',
    method: 'GET',
    headers: ['Authorization' => 'Bearer ' . $token]
);

use DealNews\Inngest\Config\Config;

$config = new Config(
    event_key: 'your-event-key',
    signing_key: 'signkey-prod-xxxxx',
    is_dev: false
);

$client = new Inngest('my-app', $config);

use DealNews\Inngest\Error\NonRetriableError;

$function = new InngestFunction(
    id: 'validate-data',
    handler: function ($ctx) {
        if (!isValid($ctx->getEvent()->getData())) {
            throw new NonRetriableError('Invalid data');
        }
        return ['status' => 'ok'];
    },
    triggers: [new TriggerEvent('data/received')]
);

use DealNews\Inngest\Error\RetryAfterError;

throw new RetryAfterError('Rate limited', retry_after: 60); // Retry after 60 seconds

use DealNews\Inngest\Function\TriggerCron;

$function = new InngestFunction(
    id: 'daily-report',
    handler: function ($ctx) {
        generateDailyReport();
        return ['status' => 'complete'];
    },
    triggers: [
        new TriggerCron('0 0 * * *') // Every day at midnight
    ]
);

use DealNews\Inngest\Function\Concurrency;

$function = new InngestFunction(
    id: 'process-data',
    handler: function ($ctx) {
        // Function logic
    },
    triggers: [new TriggerEvent('data/process')],
    concurrency: [
        new Concurrency(limit: 10) // Max 10 concurrent steps
    ]
);

// Limit to 2 concurrent runs per user
$function = new InngestFunction(
    id: 'user-task',
    handler: function ($ctx) {
        // Process user-specific task
    },
    triggers: [new TriggerEvent('user/task')],
    concurrency: [
        new Concurrency(
            limit: 2,
            key: 'event.data.user_id' // Group by user ID
        )
    ]
);

// Set both regional and account-wide limits
$function = new InngestFunction(
    id: 'process-orders',
    handler: function ($ctx) {
        // Process order
    },
    triggers: [new TriggerEvent('order/created')],
    concurrency: [
        // Limit per region
        new Concurrency(
            limit: 5,
            key: 'event.data.region',
            scope: 'fn' // Per function (default)
        ),
        // Overall account limit
        new Concurrency(
            limit: 100,
            scope: 'account' // Across all environments
        )
    ]
);

use DealNews\Inngest\Function\Priority;

$function = new InngestFunction(
    id: 'process-task',
    handler: function ($ctx) {
        // Function logic
    },
    triggers: [new TriggerEvent('task/process')],
    priority: new Priority(
        run: 'event.data.priority' // Use priority from event
    )
);

// Prioritize enterprise customers
$function = new InngestFunction(
    id: 'ai-generate-summary',
    handler: function ($ctx) {
        // Generate AI summary
    },
    triggers: [new TriggerEvent('ai/summary.requested')],
    priority: new Priority(
        // Enterprise accounts run up to 120 seconds ahead
        run: 'event.data.account_type == "enterprise" ? 120 : 0'
    )
);

// Delay free tier users
$function = new InngestFunction(
    id: 'process-report',
    handler: function ($ctx) {
        // Generate report
    },
    triggers: [new TriggerEvent('report/generate')],
    priority: new Priority(
        // Free plan users delayed by 60 seconds
        run: 'event.data.plan == "free" ? -60 : 0'
    )
);

use DealNews\Inngest\Function\Debounce;

$function = new InngestFunction(
    id: 'process-user-input',
    handler: function ($ctx) {
        $text = $ctx->getEvent()->getData()['text'];
        // Process final input after user stops typing
        return saveUserInput($text);
    },
    triggers: [new TriggerEvent('user/input')],
    debounce: new Debounce(
        period: '30s' // Wait 30 seconds after last event
    )
);

// Separate debounce window for each user
$function = new InngestFunction(
    id: 'sync-user-data',
    handler: function ($ctx) {
        $user_id = $ctx->getEvent()->getData()['user_id'];
        // Sync data once updates stop for this user
        return syncUserData($user_id);
    },
    triggers: [new TriggerEvent('user/updated')],
    debounce: new Debounce(
        period: '5m',
        key: 'event.data.user_id' // Each user has own debounce
    )
);

// Process webhooks, but force execution after maximum wait
$function = new InngestFunction(
    id: 'process-webhook',
    handler: function ($ctx) {
        $data = $ctx->getEvent()->getData();
        // Process either when events stop OR timeout reached
        return processWebhook($data);
    },
    triggers: [new TriggerEvent('webhook/received')],
    debounce: new Debounce(
        period: '1m',    // Wait 1 minute after last event
        timeout: '10m'   // Force run after 10 minutes max
    )
);

// Debounce per customer and region combination
$function = new InngestFunction(
    id: 'aggregate-metrics',
    handler: function ($ctx) {
        $data = $ctx->getEvent()->getData();
        return aggregateMetrics($data['customer_id'], $data['region']);
    },
    triggers: [new TriggerEvent('metrics/collected')],
    debounce: new Debounce(
        period: '2m',
        key: 'event.data.customer_id + "-" + event.data.region'
    )
);

use DealNews\Inngest\Function\Singleton;

$function = new InngestFunction(
    id: 'data-sync',
    handler: function ($ctx) {
        // Sync data with third-party API
        // Only one sync can run at a time
        return syncDataWithAPI();
    },
    triggers: [new TriggerEvent('sync/start')],
    singleton: new Singleton(
        mode: 'skip' // Skip new runs if one is executing
    )
);

// Each user has their own singleton rule
$function = new InngestFunction(
    id: 'process-user-data',
    handler: function ($ctx) {
        $user_id = $ctx->getEvent()->getData()['user_id'];
        // Process user data (only one run per user at a time)
        return processUserData($user_id);
    },
    triggers: [new TriggerEvent('user/data.updated')],
    singleton: new Singleton(
        mode: 'skip',
        key: 'event.data.user_id' // Separate singleton per user
    )
);

// Always process the latest event
$function = new InngestFunction(
    id: 'sync-latest-profile',
    handler: function ($ctx) {
        $user_id = $ctx->getEvent()->getData()['user_id'];
        // Cancel old sync and start new one with latest data
        return syncUserProfile($user_id);
    },
    triggers: [new TriggerEvent('profile/updated')],
    singleton: new Singleton(
        mode: 'cancel', // Cancel existing run, start new one
        key: 'event.data.user_id'
    )
);

// Singleton per customer and region combination
$function = new InngestFunction(
    id: 'generate-report',
    handler: function ($ctx) {
        $data = $ctx->getEvent()->getData();
        return generateReport($data['customer_id'], $data['region']);
    },
    triggers: [new TriggerEvent('report/generate')],
    singleton: new Singleton(
        mode: 'skip',
        key: 'event.data.customer_id + "-" + event.data.region'
    )
);

use DealNews\Inngest\Function\RateLimit;

$function = new InngestFunction(
    id: 'api-request',
    handler: function ($ctx) {
        // Call external API
        // Maximum 100 calls per hour
        return callExternalAPI($ctx->getEvent()->getData());
    },
    triggers: [new TriggerEvent('api/request')],
    rate_limit: new RateLimit(
        limit: 100,
        period: '1h'
    )
);

// Each user has their own rate limit
$function = new InngestFunction(
    id: 'user-api-request',
    handler: function ($ctx) {
        $user_id = $ctx->getEvent()->getData()['user_id'];
        // Limit: 10 requests per 30 minutes per user
        return processUserRequest($user_id);
    },
    triggers: [new TriggerEvent('user/api.request')],
    rate_limit: new RateLimit(
        limit: 10,
        period: '30m',
        key: 'event.data.user_id'
    )
);

// Complex key expression
$function = new InngestFunction(
    id: 'process-webhook',
    handler: function ($ctx) {
        $data = $ctx->getEvent()->getData();
        return processWebhook($data);
    },
    triggers: [new TriggerEvent('webhook/received')],
    rate_limit: new RateLimit(
        limit: 50,
        period: '1h',
        key: 'event.data.customer_id + "-" + event.data.region'
    )
);

use DealNews\Inngest\Function\Throttle;

$function = new InngestFunction(
    id: 'process-job',
    handler: function ($ctx) {
        // Process job at controlled rate
        // Maximum 10 jobs per hour (excess are queued)
        return processJob($ctx->getEvent()->getData());
    },
    triggers: [new TriggerEvent('job/created')],
    throttle: new Throttle(
        limit: 10,
        period: '1h'
    )
);

// Allow bursts above base limit
$function = new InngestFunction(
    id: 'api-sync',
    handler: function ($ctx) {
        // Base: 100/hour, can burst up to 110/hour
        return syncWithAPI($ctx->getEvent()->getData());
    },
    triggers: [new TriggerEvent('sync/request')],
    throttle: new Throttle(
        limit: 100,
        period: '1h',
        burst: 10  // Allow 10 extra runs during spikes
    )
);

// Each user has their own throttle
$function = new InngestFunction(
    id: 'user-report',
    handler: function ($ctx) {
        $user_id = $ctx->getEvent()->getData()['user_id'];
        // Limit: 5 reports per 30 minutes per user
        return generateReport($user_id);
    },
    triggers: [new TriggerEvent('report/generate')],
    throttle: new Throttle(
        limit: 5,
        period: '30m',
        key: 'event.data.user_id'
    )
);

// Throttle per customer and region
$function = new InngestFunction(
    id: 'aggregate-data',
    handler: function ($ctx) {
        $data = $ctx->getEvent()->getData();
        return aggregateData($data);
    },
    triggers: [new TriggerEvent('data/received')],
    throttle: new Throttle(
        limit: 100,
        period: '1h',
        burst: 20,
        key: 'event.data.customer_id + "-" + event.data.region'
    )
);

use DealNews\Inngest\Function\Checkpoint;

$function = new InngestFunction(
    id: 'bulk-processor',
    handler: function ($ctx) {
        $step = $ctx->getStep();
        // Steps execute without yielding after each one
        $a = $step->run('step-a', fn() => computeA());
        $b = $step->run('step-b', fn() => computeB());
        return ['a' => $a, 'b' => $b];
    },
    triggers: [new TriggerEvent('bulk/process')],
    checkpointing: new Checkpoint()
);

Checkpoint::safe()        // Checkpoint after every step (most durable, default)
Checkpoint::performant()  // Buffer up to 1000 steps (fastest, least durable)
Checkpoint::blended()     // Buffer 3 steps or 3 seconds, whichever comes first

new Checkpoint(
    bufferedSteps: 10,      // Steps to buffer before flushing
    maxInterval: '5s',      // Flush at least every 5 seconds
    maxRuntime: '30s'       // Maximum total execution time per invocation
)

use DealNews\Inngest\Middleware\AbstractMiddleware;
use DealNews\Inngest\Function\FunctionContext;

class LoggingMiddleware extends AbstractMiddleware
{
    public function beforeExecution(FunctionContext $ctx): void
    {
        error_log('Executing: ' . $ctx->getEvent()->getName());
    }

    public function afterExecution(FunctionContext $ctx): void
    {
        error_log('Completed: ' . $ctx->getEvent()->getName());
    }
}

$client = new Inngest('my-app');
$client->addMiddleware(new LoggingMiddleware());

bash
composer