npm install @siglo/core @siglo/scheduler @siglo/postgres pg
npm install --save-dev @types/pg

CREATE TABLE IF NOT EXISTS siglo_wait_records (
    correlation_key VARCHAR(255) PRIMARY KEY,
    stack_id VARCHAR(255) NOT NULL,
    namespace VARCHAR(255),
    event_name VARCHAR(255),
    status VARCHAR(50) NOT NULL DEFAULT 'WAITING',
    continuation_step_index INT NOT NULL DEFAULT 0,
    version VARCHAR(50),
    data JSONB,
    resume_payload JSONB,
    locked_by VARCHAR(255),
    locked_at TIMESTAMPTZ,
    resumed_by VARCHAR(255),
    resumed_at TIMESTAMPTZ,
    expires_at TIMESTAMPTZ,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

-- Crucial composite index for high-throughput polling
CREATE INDEX IF NOT EXISTS idx_siglo_wait_records_due
    ON siglo_wait_records(status, expires_at);

WITH due_records AS (
    SELECT * FROM siglo_wait_records
    WHERE (status = 'WAITING' AND expires_at IS NOT NULL AND expires_at <= NOW())
       OR (status = 'RESUMING' AND resume_payload IS NOT NULL)
    ORDER BY updated_at ASC
    LIMIT $1
) SELECT * FROM due_records

import { Pool } from 'pg';
import { PostgresWaitPoller } from '@siglo/scheduler';
import { orderProcessingFlow, onboardingFlow } from './workflows';

const pgPool = new Pool({ connectionString: process.env.DATABASE_URL });

// 1. Maintain a registry of known workflow definitions
const workflowRegistry = new Map<string, any>([
    ['order-processing', orderProcessingFlow],
    ['onboarding', onboardingFlow],
]);

// 2. Initialize the poller daemon
const poller = new PostgresWaitPoller({
    client: pgPool,
    waitRecordsTable: 'siglo_wait_records',
    pollIntervalMs: 1000, // Poll every 1 second
    batchSize: 50, // Claim up to 50 due records per batch
    workerId: `worker-${process.env.HOSTNAME || process.pid}`,
    flowLookup: (workflowId, stackId) => {
        // Resolve workflow by ID, or fallback based on stack prefix
        if (workflowId && workflowRegistry.has(workflowId)) {
            return workflowRegistry.get(workflowId);
        }
        if (stackId?.startsWith('order:')) return orderProcessingFlow;
        return onboardingFlow;
    },
    onResumed: async (record, result) => {
        console.log(`[Scheduler] Successfully resumed stack ${record.stackId}`);
    },
    onFailed: async (record, error) => {
        console.error(`[Scheduler] Failed to resume stack ${record.stackId}:`, error.message);
    },
});

// 3. Ensure tables exist and start polling
await poller.ensureTable();
poller.start();

// 4. Handle graceful shutdown (Kubernetes SIGTERM)
process.on('SIGTERM', async () => {
    console.log('[Scheduler] Shutting down poller daemon...');
    await poller.stop();
    await pgPool.end();
    process.exit(0);
});

import { MemoryWaitPoller } from '@siglo/scheduler';
import { myFlow } from './flow';

const poller = new MemoryWaitPoller({
    pollIntervalMs: 500,
    flowLookup: () => myFlow,
    onResumed: rec => console.log('In-memory timer elapsed:', rec.correlationKey),
});

poller.start();

// In tests or shutdown:
await poller.stop();

import http from 'http';
import { createSchedulerWebhookHandler } from '@siglo/scheduler';
import { orderFlow } from './workflows';

const webhookHandler = createSchedulerWebhookHandler({
    // Expected in Authorization: Bearer <token> or X-Siglo-Token
    secretToken: process.env.SCHEDULER_SECRET_TOKEN,
    flowLookup: (workflowId, stackId) => orderFlow,
    onResumed: async (stackId, result) => {
        console.log(`Cloud dispatcher resumed workflow ${stackId}`);
    },
});

const server = http.createServer(webhookHandler);
server.listen(8080, () => {
    console.log('Scheduler webhook handler listening on port 8080');
});

{
    "stackId": "stack-order-12345",
    "workflowId": "order-processing",
    "payload": {
        "type": "timer:expired",
        "timestamp": 1730598000000
    }
}