Worker Examples

1import { defineWorker } from "heliumts/server";
2import { Queue, Worker } from "bullmq";
3import { redis } from "../lib/redis";
4
5export const emailQueueConsumer = defineWorker(
6    async (ctx) => {
7        const worker = new Worker(
8            "email-queue",
9            async (job) => {
10                const { to, subject, body } = job.data;
11                await sendEmail(to, subject, body);
12            },
13            { connection: redis }
14        );
15
16        worker.on("completed", (job) => {
17            console.log(`Email job ${job.id} completed`);
18        });
19
20        worker.on("failed", (job, err) => {
21            console.error(`Email job ${job?.id} failed:`, err);
22        });
23
24        // Keep the worker running
25        await new Promise(() => {});
26    },
27    { name: "emailQueueConsumer" }
28);

1import { defineWorker } from "heliumts/server";
2
3export const dailyCleanup = defineWorker(
4    async (ctx) => {
5        while (true) {
6            const now = new Date();
7
8            // Run at midnight
9            if (now.getHours() === 0 && now.getMinutes() === 0) {
10                console.log("Running daily cleanup...");
11                await cleanupOldRecords();
12                await pruneExpiredSessions();
13                await generateDailyReport();
14            }
15
16            // Check every minute
17            await new Promise((resolve) => setTimeout(resolve, 60000));
18        }
19    },
20    { name: "dailyCleanup" }
21);

1import { defineWorker } from "heliumts/server";
2
3export const priceSync = defineWorker(
4    async (ctx) => {
5        const ws = new WebSocket("wss://api.exchange.com/prices");
6
7        ws.on("message", async (data) => {
8            const prices = JSON.parse(data.toString());
9            await updatePricesInDatabase(prices);
10            await notifySubscribers(prices);
11        });
12
13        ws.on("close", () => {
14            throw new Error("WebSocket connection closed");
15        });
16
17        // Keep the connection alive
18        await new Promise(() => {});
19    },
20    {
21        name: "priceSync",
22        autoRestart: true,
23        restartDelayMs: 5000,
24    }
25);

1import { defineWorker } from "heliumts/server";
2
3export const cacheWarmer = defineWorker(
4    async (ctx) => {
5        // Initial warm-up
6        console.log("Warming up cache...");
7        await warmupProductCache();
8        await warmupUserCache();
9        await warmupConfigCache();
10
11        // Periodic refresh
12        while (true) {
13            await new Promise((resolve) => setTimeout(resolve, 300000)); // Every 5 minutes
14            await refreshHotCache();
15        }
16    },
17    { name: "cacheWarmer" }
18);

1import { defineWorker } from "heliumts/server";
2import { SQSClient, ReceiveMessageCommand, DeleteMessageCommand } from "@aws-sdk/client-sqs";
3
4const sqs = new SQSClient({ region: "us-east-1" });
5const queueUrl = process.env.SQS_QUEUE_URL!;
6
7export const sqsConsumer = defineWorker(
8    async (ctx) => {
9        while (true) {
10            const { Messages } = await sqs.send(
11                new ReceiveMessageCommand({
12                    QueueUrl: queueUrl,
13                    MaxNumberOfMessages: 10,
14                    WaitTimeSeconds: 20, // Long polling
15                })
16            );
17
18            if (Messages) {
19                for (const message of Messages) {
20                    try {
21                        await processMessage(JSON.parse(message.Body!));
22                        await sqs.send(
23                            new DeleteMessageCommand({
24                                QueueUrl: queueUrl,
25                                ReceiptHandle: message.ReceiptHandle,
26                            })
27                        );
28                    } catch (error) {
29                        console.error("Failed to process message:", error);
30                    }
31                }
32            }
33        }
34    },
35    { name: "sqsConsumer" }
36);

1import { defineWorker } from "heliumts/server";
2import Redis from "ioredis";
3
4export const pubsubSubscriber = defineWorker(
5    async (ctx) => {
6        const subscriber = new Redis(process.env.REDIS_URL);
7
8        subscriber.subscribe("notifications", "updates");
9
10        subscriber.on("message", async (channel, message) => {
11            const data = JSON.parse(message);
12
13            switch (channel) {
14                case "notifications":
15                    await handleNotification(data);
16                    break;
17                case "updates":
18                    await handleUpdate(data);
19                    break;
20            }
21        });
22
23        // Keep the subscriber running
24        await new Promise(() => {});
25    },
26    { name: "pubsubSubscriber" }
27);