PHP API

use Queuety\Queuety;

use Queuety\Connection;
use Queuety\Queuety;

$conn = new Connection(
    host: '127.0.0.1:3306',
    dbname: 'wordpress',
    user: 'root',
    password: 'secret',
    prefix: 'wp_',
);

Queuety::init( $conn );

$conn = new Connection(
    host: '127.0.0.1:3306',
    dbname: 'wordpress',
    user: 'root',
    password: 'secret',
    prefix: 'wp_',
    table_prefix: 'themequeue_',
);

Queuety::init( $conn );
Queuety::ensure_schema();

// Handler name + payload (original API)
Queuety::dispatch( 'send_email', [ 'to' => 'user@example.com' ] );

// Contracts\Job instance (v0.6.0+)
Queuety::dispatch( new SendEmailJob( to: 'user@example.com' ) );

// With options
Queuety::dispatch( 'send_email', [ 'to' => 'user@example.com' ] )
    ->on_queue( 'emails' )
    ->with_priority( Priority::High )
    ->delay( 300 )
    ->max_attempts( 5 )
    ->rate_limit( 10, 60 )
    ->unique()
    ->after( $other_job_id );

// Get the job ID
$job_id = Queuety::dispatch( 'send_email', $payload )->id();

// With a Job instance
Queuety::dispatch_sync( new SendEmailJob( to: 'user@example.com', subject: 'Hi', body: 'Hello' ) );

// With a handler name and payload
Queuety::dispatch_sync( 'send_email', [ 'to' => 'user@example.com' ] );

$job_ids = Queuety::batch( [
    [ 'handler' => 'send_email', 'payload' => [ 'to' => 'a@example.com' ], 'queue' => 'emails' ],
    [ 'handler' => 'send_email', 'payload' => [ 'to' => 'b@example.com' ], 'priority' => 2 ],
] );

$batch = Queuety::create_batch( [
    new ImportUsersJob( $chunk_1 ),
    new ImportUsersJob( $chunk_2 ),
] )
    ->name( 'Import users' )
    ->then( ImportCompleteHandler::class )
    ->catch( ImportFailedHandler::class )
    ->finally( ImportCleanupHandler::class )
    ->allow_failures()
    ->on_queue( 'imports' )
    ->dispatch();

$batch = Queuety::find_batch( $batch_id );

if ( null !== $batch ) {
    echo $batch->progress() . '% complete';
}

Queuety::run_worker( 'default' );
Queuety::run_worker( 'critical,default', true );

Queuety::run_worker_pool( 4, 'providers' );

Queuety::run_auto_scaling_worker_pool( 2, 6, 'providers' );

$first_job_id = Queuety::chain( [
    new FetchDataJob( $url ),
    new ProcessDataJob(),
    new NotifyCompleteJob(),
] )
    ->on_queue( 'pipeline' )
    ->catch( ChainFailedHandler::class )
    ->dispatch();

Queuety::retry( 42 );

$count = Queuety::retry_buried();

$count = Queuety::purge();     // uses QUEUETY_RETENTION_DAYS
$count = Queuety::purge( 30 ); // purge jobs older than 30 days

$stats = Queuety::stats();          // all queues
$stats = Queuety::stats( 'emails' ); // specific queue
// ['pending' => 12, 'processing' => 3, 'completed' => 150, 'failed' => 2, 'buried' => 1]

$jobs = Queuety::buried();
$jobs = Queuety::buried( 'emails' );

Queuety::pause( 'emails' );

Queuety::resume( 'emails' );

if ( Queuety::is_paused( 'emails' ) ) {
    // Queue is paused
}

$workflow_id = Queuety::workflow( 'generate_report' )
    ->then( FetchDataHandler::class )
    ->then( CallLLMHandler::class )
    ->then( FormatOutputHandler::class )
    ->dispatch( [ 'user_id' => 42 ] );

$state = Queuety::workflow_status( $workflow_id );
echo $state->name;          // 'generate_report'
echo $state->status->value; // 'running'
echo $state->current_step;  // 1
echo $state->total_steps;   // 3
print_r( $state->state );   // accumulated public state array
echo $state->wait_type;     // 'signal' or 'workflow' while blocked
echo $state->wait_mode;     // 'all', 'any', or 'quorum'
echo $state->definition_version; // 'research.v2'
echo $state->definition_hash;    // '4c7a...'
echo $state->idempotency_key;    // 'brief:42:research'
echo $state->artifact_count;     // 2
print_r( $state->budget );       // configured workflow budget + current usage
print_r( $state->wait_details ); // matched/remaining blockers, result key, correlation info
print_r( $state->artifact_keys ); // ['draft', 'citations']

Queuety::retry_workflow( $workflow_id );

Queuety::pause_workflow( $workflow_id );

Queuety::resume_workflow( $workflow_id );

Queuety::cancel_workflow( $workflow_id );

Queuety::rewind_workflow( $workflow_id, to_step: 1 );

$forked_id = Queuety::fork_workflow( $workflow_id );

$data = Queuety::export_workflow( $workflow_id );

$new_id = Queuety::replay_workflow( $data );

Queuety::replace_wp_cron();

Queuety::restore_wp_cron();

Queuety::signal( $workflow_id, 'approved' );

// With data merged into workflow state
Queuety::signal( $workflow_id, 'approved', [
    'approved_by' => 'admin@example.com',
] );

Queuety::approve_workflow( $workflow_id, [
    'reviewer' => 'editor@example.com',
] );

Queuety::reject_workflow( $workflow_id, [
    'reason' => 'needs citations',
] );

Queuety::submit_workflow_input( $workflow_id, [
    'notes' => 'Ship after legal review',
] );

Queuety::put_artifact(
    $workflow_id,
    'research_brief',
    [ 'summary' => 'done' ],
    'json',
    metadata: [ 'source' => 'agent' ],
);

Queuety::put_current_artifact(
    'draft',
    [ 'status' => 'ready' ],
    'json',
);

$artifact = Queuety::workflow_artifact( $workflow_id, 'draft' );

$artifacts = Queuety::workflow_artifacts( $workflow_id );

Queuety::delete_workflow_artifact( $workflow_id, 'draft' );

use Queuety\Enums\StateMachineStatus;

$machine_id = Queuety::machine( 'agent_session' )
    ->state( 'awaiting_user' )
    ->on( 'user_message', 'planning' )
    ->state( 'planning' )
    ->action( PlanSessionAction::class )
    ->on( 'planned', 'completed' )
    ->state( 'completed', StateMachineStatus::Completed )
    ->dispatch( [ 'thread_id' => 42 ] );

$machine_id = Queuety::dispatch_state_machine_definition(
    $definition,
    [ 'thread_id' => 42 ],
    [ 'idempotency_key' => 'thread:42' ],
);

$status = Queuety::machine_status( $machine_id );

echo $status->name;               // 'agent_session'
echo $status->status->value;      // 'waiting_event'
echo $status->current_state;      // 'awaiting_review'
print_r( $status->state );        // public machine state
print_r( $status->available_events ); // valid incoming events in the current state

Queuety::machine_event(
    $machine_id,
    'approve',
    [
        'approved' => true,
        'reviewer' => 'editor@example.com',
    ]
);

$machines = Queuety::list_machines();
$waiting  = Queuety::list_machines( status: 'waiting_event' );

$timeline = Queuety::machine_timeline( $machine_id );
$page_two = Queuety::machine_timeline( $machine_id, 100, 100 );

$trace = Queuety::machine_trace( $machine_id );

$state = Queuety::machine_state_at( $machine_id, $event_id );

$builder = Queuety::define_workflow( 'onboarding' )
    ->then( CreateAccountHandler::class )
    ->then( SendWelcomeEmailHandler::class );

Queuety::register_workflow_template( $builder );

$workflow_id = Queuety::run_workflow( 'onboarding', [ 'email' => 'user@example.com' ] );

$workflow_id = Queuety::run_workflow(
    'onboarding',
    [ 'email' => 'user@example.com' ],
    [ 'idempotency_key' => 'onboarding:user@example.com' ],
);

$workflow_id = Queuety::dispatch_workflow_definition(
    $definition,
    [ 'email' => 'user@example.com' ],
    [ 'idempotency_key' => 'onboarding:user@example.com' ],
);

Queuety::on_action(
    'save_post',
    workflow: 'content_review',
    map: static fn ( int $post_id, object $post, bool $update ): array => [
        'post_id' => $post_id,
        'post_type' => $post->post_type,
        'update' => $update,
    ],
    when: static fn ( int $post_id, object $post ): bool => 'post' === $post->post_type,
    idempotency_key: static fn ( int $post_id ): string => "save_post:{$post_id}",
);

$registry = Queuety::workflow_templates();
$template = $registry->get( 'onboarding' ); // WorkflowTemplate or null

$count = Queuety::load_workflows( __DIR__ . '/workflows/' );
$count = Queuety::load_workflows( __DIR__ . '/workflows/', recursive: true );

$template = Queuety::load_workflow_file( __DIR__ . '/workflows/onboard-user.php' );

Queuety::schedule( 'cleanup_handler' )
    ->every( '1 hour' )
    ->on_queue( 'maintenance' );

Queuety::schedule( 'nightly_report' )
    ->cron( '0 3 * * *' )
    ->on_queue( 'reports' );

$scheduler = Queuety::scheduler();
$schedules = $scheduler->list();
$scheduler->remove( 'handler_name' );
$count = $scheduler->tick(); // manually trigger

Queuety::register( 'send_email', SendEmailHandler::class );

$count = Queuety::discover_handlers( __DIR__ . '/handlers', 'MyPlugin\\Handlers' );

$registry = Queuety::registry();

$logger = Queuety::logger();
$logs   = $logger->for_job( 42 );
$logs   = $logger->for_workflow( 7 );
$logs   = $logger->for_handler( 'send_email', 50 );
$logs   = $logger->for_event( LogEvent::Failed, 50 );
$logs   = $logger->since( new \DateTimeImmutable( '-1 hour' ), 100 );
$count  = $logger->purge( 30 );

$stats = Queuety::metrics()->handler_stats( 60 ); // last 60 minutes

$notifier = Queuety::webhook_notifier();
$id       = $notifier->register( 'job.buried', 'https://hooks.slack.com/...' );
$webhooks = $notifier->list();
$notifier->remove( $id );

$limiter = Queuety::rate_limiter();
$limiter->register( 'call_openai', 60, 60 );

$cache = Queuety::cache();
$cache->set( 'my_key', 'my_value', 60 );
$value = $cache->get( 'my_key' );

Queuety::set_cache( new RedisCache() );
Queuety::init( $connection );

define( 'QUEUETY_CACHE_TTL', 10 );

$log = Queuety::machine_events();

$log = Queuety::workflow_events();

$events = Queuety::workflow_timeline( $workflow_id );
$next   = Queuety::workflow_timeline( $workflow_id, 100, 100 );

$trace = Queuety::workflow_trace( $workflow_id );

$state = Queuety::workflow_state_at( $workflow_id, 2 );

$fake = Queuety::fake();
SendEmailJob::dispatch( 'user@example.com', 'Hello', 'Hi!' );
$fake->assert_pushed( SendEmailJob::class );

Queuety::reset();

interface Handler {
    public function handle( array $payload ): void;
    public function config(): array;
}

interface Step {
    public function handle( array $state ): array;
    public function config(): array;
}

namespace Queuety\Contracts;

interface ForEachHandler {
    public function handle_item( array $state, mixed $item, int $index ): array;
    public function config(): array;
}

namespace Queuety\Contracts;

interface ForEachReducer {
    public function reduce( array $state, array $for_each ): array;
}

namespace Queuety\Contracts;

interface Compensation {
    public function handle( array $state ): void;
}

namespace Queuety\Contracts;

interface RepeatCondition {
    public function matches( array $state ): bool;
}

namespace Queuety\Contracts;

interface StateAction {
    public function handle( array $state, ?string $event = null, array $event_payload = [], array $payload = [] ): array|string;
}

namespace Queuety\Contracts;

interface StateGuard {
    public function allows( array $state, array $event_payload, string $event, array $payload = [] ): bool;
}

namespace Queuety\Contracts;

interface Job {
    public function handle(): void;
}

namespace Queuety\Contracts;

interface Cache {
    public function get( string $key ): mixed;
    public function set( string $key, mixed $value, int $ttl = 0 ): void;
    public function delete( string $key ): void;
    public function has( string $key ): bool;
    public function add( string $key, mixed $value, int $ttl = 0 ): bool;
    public function flush(): void;
}

namespace Queuety\Contracts;

interface StreamingStep {
    public function stream( array $state, array $existing_chunks = [] ): \Generator;
    public function on_complete( array $chunks, array $state ): array;
    public function config(): array;
}

namespace Queuety\Contracts;

interface Middleware {
    public function handle( object $job, \Closure $next ): void;
}

use Queuety\Contracts\Job;
use Queuety\Dispatchable;

class SendEmailJob implements Job {
    use Dispatchable;
    // ...
}

// Dispatches a new SendEmailJob instance
SendEmailJob::dispatch( 'user@example.com', 'Subject', 'Body' );

// Conditional dispatch
SendEmailJob::dispatch_if( $user->wants_email, $user->email, 'Hello', 'Hi!' );
SendEmailJob::dispatch_unless( $maintenance_mode, $user->email, 'Hello', 'Hi!' );

// Chain from a job class
FetchDataJob::with_chain( [
    new ProcessDataJob(),
    new NotifyCompleteJob(),
] )->dispatch();

class ImportProductsJob implements Job {
    use Dispatchable;

    public int $tries = 5;
    public int $timeout = 120;
    public array $backoff = [ 10, 30, 60, 120 ];

    // ...
}

public function failed( \Throwable $exception ): void {
    error_log( "Job failed: {$exception->getMessage()}" );
}

enum Priority: int {
    case Low    = 0;
    case Normal = 1;
    case High   = 2;
    case Urgent = 3;
}

enum JobStatus: string {
    case Pending    = 'pending';
    case Processing = 'processing';
    case Completed  = 'completed';
    case Failed     = 'failed';
    case Buried     = 'buried';
}

enum WorkflowStatus: string {
    case Running        = 'running';
    case Completed      = 'completed';
    case Failed         = 'failed';
    case Paused         = 'paused';
    case WaitingForSignal  = 'waiting_for_signal';
    case WaitingForWorkflows = 'waiting_for_workflows';
    case Cancelled      = 'cancelled';
}

enum StateMachineStatus: string {
    case Running = 'running';
    case WaitingEvent = 'waiting_event';
    case Completed = 'completed';
    case Failed = 'failed';
    case Paused = 'paused';
    case Cancelled = 'cancelled';
}

enum BackoffStrategy: string {
    case Fixed       = 'fixed';
    case Linear      = 'linear';
    case Exponential = 'exponential';
}

enum ForEachMode: string {
    case All = 'all';
    case FirstSuccess = 'first_success';
    case Quorum = 'quorum';
}

enum WaitMode: string {
    case All    = 'all';
    case Any    = 'any';
    case Quorum = 'quorum';
}

enum LogEvent: string {
    case Started                  = 'started';
    case Completed                = 'completed';
    case Failed                   = 'failed';
    case Buried                   = 'buried';
    case Retried                  = 'retried';
    case WorkflowStarted          = 'workflow_started';
    case WorkflowCompleted        = 'workflow_completed';
    case WorkflowFailed           = 'workflow_failed';
    case WorkflowPaused           = 'workflow_paused';
    case WorkflowResumed          = 'workflow_resumed';
    case WorkflowCancelled        = 'workflow_cancelled';
    case WorkflowRewound          = 'workflow_rewound';
    case WorkflowForked           = 'workflow_forked';
    case WorkflowDeadlineExceeded = 'workflow_deadline_exceeded';
    case Debug                    = 'debug';
}

enum OverlapPolicy: string {
    case Allow  = 'allow';
    case Skip   = 'skip';
    case Buffer = 'buffer';
}

use Queuety\Attributes\QueuetyHandler;

#[QueuetyHandler( name: 'send_email', queue: 'emails', max_attempts: 5 )]
class SendEmailHandler implements Handler {
    // ...
}

$plan = Queuety::resolve_cli_command(
    [ 'workflow', 'export' ],
    [ '42' ],
    [ 'output' => '/tmp/workflow.json' ]
);

echo $plan['callable']; // Queuety\Queuety::export_workflow_to_file
print_r( $plan['arguments'] );