stack/packages/queue/src/queue.ts

import Redis from 'ioredis';

const DEFAULT_VALKEY_URL = 'redis://localhost:6380';

export interface QueueConfig {
  url?: string;
}

export interface QueueHandle {
  redis: Redis;
  close: () => Promise<void>;
}

export interface TaskPayload {
  id: string;
  type: string;
  data: Record<string, unknown>;
  createdAt: string;
}

export function createQueue(config?: QueueConfig): QueueHandle {
  const url = config?.url ?? process.env['VALKEY_URL'] ?? DEFAULT_VALKEY_URL;
  const redis = new Redis(url, { maxRetriesPerRequest: 3 });

  return {
    redis,
    close: async () => {
      await redis.quit();
    },
  };
}

export function createQueueClient(handle: QueueHandle) {
  const { redis } = handle;

  return {
    async enqueue(queueName: string, payload: TaskPayload): Promise<void> {
      await redis.lpush(queueName, JSON.stringify(payload));
    },

    async dequeue(queueName: string): Promise<TaskPayload | null> {
      const item = await redis.rpop(queueName);
      if (!item) return null;
      return JSON.parse(item) as TaskPayload;
    },

    async length(queueName: string): Promise<number> {
      return redis.llen(queueName);
    },

    async publish(channel: string, message: string): Promise<void> {
      await redis.publish(channel, message);
    },

    subscribe(channel: string, handler: (message: string) => void): () => void {
      const sub = redis.duplicate();
      sub.subscribe(channel).catch(() => {});
      sub.on('message', (_ch: string, msg: string) => handler(msg));
      return () => {
        sub.unsubscribe(channel).catch(() => {});
        sub.disconnect();
      };
    },
  };
}

export type QueueClient = ReturnType<typeof createQueueClient>;