import { createClient } from "redis"; import { logger } from "./logger.js"; const REDIS_URL = process.env.REDIS_URL ?? "redis://localhost:6379"; export const redisPub = createClient({ url: REDIS_URL }); export const redisSub = redisPub.duplicate(); export const redisClient = redisPub.duplicate(); // general purpose (presence, push subs cache, etc.) let connected = false; export async function connectRedis() { if (connected) return; await Promise.all([redisPub.connect(), redisSub.connect(), redisClient.connect()]); connected = true; redisPub.on("error", (err) => logger.error({ err }, "Redis pub error")); redisSub.on("error", (err) => logger.error({ err }, "Redis sub error")); redisClient.on("error", (err) => logger.error({ err }, "Redis client error")); logger.info("Redis connected"); } // ── Presence helpers (so "online" status works across multiple instances) ── const ONLINE_KEY = "nexus:online"; // Redis SET of online emails export async function markOnline(email) { await redisClient.sAdd(ONLINE_KEY, email); } export async function markOffline(email) { await redisClient.sRem(ONLINE_KEY, email); } export async function getOnlineEmails() { return redisClient.sMembers(ONLINE_KEY); } // ── Pub/Sub for cross-instance broadcast ──────────────────────── const CHANNEL = "nexus:events"; export async function publishEvent(payload) { await redisPub.publish(CHANNEL, JSON.stringify(payload)); } export async function subscribeEvents(handler) { await redisSub.subscribe(CHANNEL, (message) => { try { handler(JSON.parse(message)); } catch (err) { logger.warn({ err }, "Failed to parse redis event"); } }); }