import Redis from "ioredis";
import { config, log } from "./config";

let redis: Redis | null = null;

export function getRedis(): Redis {
  if (!redis) {
    redis = new Redis(config.redisUrl, { maxRetriesPerRequest: 3, lazyConnect: false, enableOfflineQueue: true });
    redis.on("error", (e) => log.warn({ err: e.message }, "redis error"));
  }
  return redis;
}

export const STREAM_EVENTS = "ws:events";
export const CHANNEL_LIVE = "ws:live";
export const STREAM_CHANGES = "ws:changes";

/**
 * Publish a compact event payload to the stream (durable, replayable by stream id) and the pub/sub
 * channel (realtime). The stream id is embedded as `sid` so clients can resume with `since`.
 */
export async function publishEvent(payload: Record<string, unknown>): Promise<string | null> {
  const r = getRedis();
  try {
    const sid = await r.xadd(STREAM_EVENTS, "MAXLEN", "~", "20000", "*", "event", JSON.stringify(payload));
    const json = JSON.stringify({ ...payload, sid });
    // keep the stream entry consistent with what subscribers saw
    await r.publish(CHANNEL_LIVE, json);
    return sid ?? null;
  } catch (e) {
    log.warn({ err: (e as Error).message }, "publish failed");
    return null;
  }
}

/** Engine heartbeat for the ops dashboard (`/api/v1/admin/ops`, `/api/ready`). */
export async function publishEngineStatus(status: Record<string, unknown>): Promise<void> {
  try {
    await getRedis().set("ws:engine:status", JSON.stringify({ ...status, at: new Date().toISOString(), pid: process.pid }), "EX", 30);
  } catch {
    // best effort
  }
}

export async function publishChange(payload: Record<string, unknown>): Promise<void> {
  try {
    await getRedis().xadd(STREAM_CHANGES, "MAXLEN", "~", "5000", "*", "change", JSON.stringify(payload));
  } catch {
    // best effort
  }
}

export async function closeRedis(): Promise<void> {
  if (redis) await redis.quit().catch(() => undefined);
  redis = null;
}