src/lib/redis.js

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");
    }
  });
}