npm install @siglo/core @siglo/bullmq bullmq ioredis

function createBullMQProcessor<TInput = any, TOutput = any, TJob = any>(
    workflow: Flow<TInput, TOutput>,
    options?: SigloBullMQOptions<TJob, TInput, TOutput>,
): (job: any) => Promise<any>;

import { Worker, Job } from 'bullmq';
import { Step, flow } from '@siglo/core';
import { createBullMQProcessor } from '@siglo/bullmq';
import { Redis } from 'ioredis';

const connection = new Redis(process.env.REDIS_URL || 'redis://localhost:6379', {
    maxRetriesPerRequest: null,
});

// 1. Define workflow steps
interface VideoJobData {
    videoId: string;
    sourceBucket: string;
    resolutions: ('720p' | '1080p' | '4k')[];
}

const downloadSource = Step.create('downloadSource', async (param: VideoJobData) => {
    console.log(`Downloading video ${param.videoId} from ${param.sourceBucket}`);
    return { localPath: `/tmp/${param.videoId}.mp4` };
});

const transcode = Step.create('transcode', async (param: any) => {
    console.log(`Transcoding ${param.localPath} into ${param.resolutions.join(', ')}`);
    return {
        transcodedFiles: param.resolutions.map((r: string) => `/tmp/${param.videoId}_${r}.mp4`),
        durationSec: 124,
    };
});

const uploadCdn = Step.create('uploadCdn', async (param: any) => {
    console.log(`Uploading ${param.transcodedFiles.length} files to CDN`);
    return {
        cdnUrls: param.transcodedFiles.map((f: string) => `https://cdn.example.com/${param.videoId}`),
        status: 'READY',
    };
});

const transcodeFlow = flow(downloadSource, transcode, uploadCdn);

// 2. Create BullMQ worker with processor adapter
const worker = new Worker<VideoJobData>(
    'video-processing-queue',
    createBullMQProcessor(transcodeFlow, {
        timeout: 120000, // 2-minute deadline
        input: (job: Job<VideoJobData>) => ({
            videoId: job.data.videoId,
            sourceBucket: job.data.sourceBucket,
            resolutions: job.data.resolutions,
        }),
        output: (result, job) => ({
            jobId: job.id,
            videoId: job.data.videoId,
            cdnUrls: result.cdnUrls,
            durationSec: result.durationSec,
        }),
    }),
    { connection, concurrency: 5 },
);

worker.on('completed', (job, returnvalue) => {
    console.log(`Job ${job.id} completed. Transcoded video ready:`, returnvalue.cdnUrls);
});

worker.on('failed', (job, err) => {
    console.error(`Job ${job?.id} failed: ${err.message}`);
});

import { Worker } from 'bullmq';
import { Step, flow } from '@siglo/core';
import { createBullMQProcessor } from '@siglo/bullmq';
import { RedisIdempotencyStore, RedisStorageAdapter } from '@siglo/redis';
import { Redis } from 'ioredis';

const redis = new Redis('redis://localhost:6379');

// Configure global storage for checkpoints
Step.useStorage(new RedisStorageAdapter({ client: redis }));

// Configure distributed idempotency store
const idempotencyStore = new RedisIdempotencyStore({
    client: redis,
    keyPrefix: 'siglo:idemp:bullmq:',
    defaultTtlMs: 3600000, // 1 hour
});

const billingWorkflow = flow(
    Step.create('chargeInvoice', async ({ invoiceId, amount }: any) => {
        // Charging payment gateway
        return { chargeId: `ch_${invoiceId}`, settled: true };
    }),
);

const worker = new Worker(
    'billing-queue',
    createBullMQProcessor(billingWorkflow, {
        idempotencyStore,
        idempotencyTtlMs: 3600000,
        timeout: 10000,
        namespace: job => job.data.tenantId, // Multi-tenant namespace
    }),
    { connection: redis, concurrency: 10 },
);

async function shutdown() {
    console.log('Shutting down BullMQ worker...');
    await worker.close();
    await redis.quit();
    process.exit(0);
}

process.on('SIGTERM', shutdown);
process.on('SIGINT', shutdown);