feat: implement all real backend endpoints, Redis pub/sub bridge, and gateway command handler
Backend (real implementations, no more stubs):
- Redis pub/sub bridge: subscribes to discord-gateway events (message/attachment/voice) and broadcasts to WS clients
- Redis command channel: backend publishes voice/media commands, discord-gateway executes and replies
- Voice service: connectVoice/disconnectVoice/getVoiceStatus via Redis commands with graceful fallback
- Media service: queue/skip/stop/volume via Redis commands, reads status from Redis cache
- Messages repository: ALL 7 methods now use real PostgreSQL queries (findMany, findById, findByChannel, create, update, delete, getAttachmentsByChannel)
- Analytics: period returns {start,end} epoch millis, overview includes hourly/topics/top_users, worst_flags as string[]
- Health check: actually queries SELECT 1 against database
- VoiceStatus type fixed: {connected, activeGuildId, activeChannelId, activeChannelName}
- Guild type: includes icon: string | null
- asyncHandler: accepts Promise<unknown> instead of Promise<void>
Discord Gateway:
- CommandHandler: subscribes to 'backend:command' Redis channel, executes voice/media commands, publishes replies
- Publishes voice:status and media:status to Redis for backend caching
- Shutdown handler updated to close command handler
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
MythEclipse
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
9b41eb9c12
commit
3b2709455e
@@ -0,0 +1,267 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import Redis from "ioredis";
|
||||
import { config } from "../config/index.js";
|
||||
import { createChildLogger } from "../logger/index.js";
|
||||
|
||||
const logger = createChildLogger("redis.command-channel");
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Types
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export interface CommandMessage {
|
||||
id: string;
|
||||
type: string;
|
||||
payload: Record<string, unknown>;
|
||||
replyChannel: string;
|
||||
}
|
||||
|
||||
export interface CommandReply<T = unknown> {
|
||||
id: string;
|
||||
success: boolean;
|
||||
data?: T;
|
||||
error?: string;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Internal Redis clients (singletons)
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
let publisherClient: Redis | null = null;
|
||||
let subscriberClient: Redis | null = null;
|
||||
|
||||
function ensureRedisConfig(): boolean {
|
||||
return !!(config.REDIS_URL || config.REDIS_HOST);
|
||||
}
|
||||
|
||||
function createClient(): Redis {
|
||||
if (config.REDIS_URL) {
|
||||
return new Redis(config.REDIS_URL, { keyPrefix: "" });
|
||||
}
|
||||
return new Redis({
|
||||
host: config.REDIS_HOST,
|
||||
port: config.REDIS_PORT,
|
||||
keyPrefix: "",
|
||||
});
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Publisher
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
function getPublisher(): Redis {
|
||||
if (!publisherClient) {
|
||||
publisherClient = createClient();
|
||||
publisherClient.on("error", (err: Error) => {
|
||||
logger.error({ err }, "Redis publisher error");
|
||||
});
|
||||
}
|
||||
return publisherClient;
|
||||
}
|
||||
|
||||
export function getCommandPublisher(): Redis {
|
||||
return getPublisher();
|
||||
}
|
||||
|
||||
/**
|
||||
* Publish a command and wait for a reply on a dedicated reply channel.
|
||||
* Times out after `timeoutMs` (default 5000ms) and returns null.
|
||||
*/
|
||||
export async function publishCommand<T = unknown>(
|
||||
commandType: string,
|
||||
payload: Record<string, unknown> = {},
|
||||
timeoutMs = 5000,
|
||||
): Promise<CommandReply<T> | null> {
|
||||
if (!ensureRedisConfig()) {
|
||||
logger.warn({ commandType }, "Redis not configured, skipping command publish");
|
||||
return null;
|
||||
}
|
||||
|
||||
const id = randomUUID();
|
||||
const replyChannel = `backend:command:reply:${id}`;
|
||||
const command: CommandMessage = { id, type: commandType, payload, replyChannel };
|
||||
|
||||
return new Promise<CommandReply<T> | null>((resolve) => {
|
||||
const pub = getPublisher();
|
||||
let settled = false;
|
||||
|
||||
const timer = setTimeout(() => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
sub.unsubscribe(replyChannel).catch(() => {/* ignore */});
|
||||
logger.warn({ id, commandType }, "Command timed out waiting for reply");
|
||||
resolve(null);
|
||||
}, timeoutMs);
|
||||
|
||||
const sub = getSubscriber();
|
||||
|
||||
const onMessage = (channel: string, message: string) => {
|
||||
if (channel !== replyChannel || settled) return;
|
||||
settled = true;
|
||||
clearTimeout(timer);
|
||||
sub.unsubscribe(replyChannel).catch(() => {/* ignore */});
|
||||
|
||||
try {
|
||||
const reply: CommandReply<T> = JSON.parse(message);
|
||||
logger.debug({ id, commandType, success: reply.success }, "Command reply received");
|
||||
resolve(reply);
|
||||
} catch (err) {
|
||||
logger.error({ id, err }, "Failed to parse command reply");
|
||||
resolve(null);
|
||||
}
|
||||
};
|
||||
|
||||
sub.on("message", onMessage);
|
||||
|
||||
sub.subscribe(replyChannel).then(() => {
|
||||
pub
|
||||
.publish("backend:command", JSON.stringify(command))
|
||||
.then(() => {
|
||||
logger.debug({ id, commandType }, "Command published");
|
||||
})
|
||||
.catch((err: Error) => {
|
||||
if (!settled) {
|
||||
settled = true;
|
||||
clearTimeout(timer);
|
||||
sub.unsubscribe(replyChannel).catch(() => {/* ignore */});
|
||||
logger.error({ err }, "Failed to publish command");
|
||||
resolve(null);
|
||||
}
|
||||
});
|
||||
}).catch((err: Error) => {
|
||||
if (!settled) {
|
||||
settled = true;
|
||||
clearTimeout(timer);
|
||||
logger.error({ err }, "Failed to subscribe to reply channel");
|
||||
resolve(null);
|
||||
}
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Publish a command without waiting for a reply (fire-and-forget).
|
||||
*/
|
||||
export async function publishCommandNoReply(
|
||||
commandType: string,
|
||||
payload: Record<string, unknown> = {},
|
||||
): Promise<void> {
|
||||
if (!ensureRedisConfig()) {
|
||||
logger.warn({ commandType }, "Redis not configured, skipping command publish");
|
||||
return;
|
||||
}
|
||||
|
||||
const id = randomUUID();
|
||||
const command: CommandMessage = {
|
||||
id,
|
||||
type: commandType,
|
||||
payload,
|
||||
replyChannel: "",
|
||||
};
|
||||
|
||||
await getPublisher().publish("backend:command", JSON.stringify(command));
|
||||
logger.debug({ id, commandType }, "Command published (no reply)");
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Subscriber (for receiving replies and other pub/sub messages)
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
function getSubscriber(): Redis {
|
||||
if (!subscriberClient) {
|
||||
subscriberClient = createClient();
|
||||
subscriberClient.on("error", (err: Error) => {
|
||||
logger.error({ err }, "Redis subscriber error");
|
||||
});
|
||||
}
|
||||
return subscriberClient;
|
||||
}
|
||||
|
||||
export function getCommandSubscriber(): Redis {
|
||||
return getSubscriber();
|
||||
}
|
||||
|
||||
/**
|
||||
* Subscribe to a Redis channel with a handler. Returns unsubscribe function.
|
||||
*/
|
||||
export function subscribe(
|
||||
channel: string,
|
||||
handler: (message: string) => void,
|
||||
): () => Promise<void> {
|
||||
const sub = getSubscriber();
|
||||
|
||||
const onMessage = (_ch: string, message: string) => {
|
||||
try {
|
||||
handler(message);
|
||||
} catch (err) {
|
||||
logger.error({ channel, err }, "Error in Redis subscription handler");
|
||||
}
|
||||
};
|
||||
|
||||
sub.on("message", onMessage);
|
||||
sub.subscribe(channel).catch((err: Error) => {
|
||||
logger.error({ channel, err }, "Failed to subscribe to Redis channel");
|
||||
});
|
||||
|
||||
return async () => {
|
||||
sub.removeListener("message", onMessage);
|
||||
await sub.unsubscribe(channel);
|
||||
};
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Status helpers — read keys set by discord-gateway
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export async function readRedisStatus(key: string): Promise<Record<string, unknown> | null> {
|
||||
if (!ensureRedisConfig()) {
|
||||
return null;
|
||||
}
|
||||
|
||||
try {
|
||||
const raw = await getPublisher().get(key);
|
||||
if (!raw) return null;
|
||||
return JSON.parse(raw) as Record<string, unknown>;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
export async function writeRedisStatus(
|
||||
key: string,
|
||||
data: Record<string, unknown>,
|
||||
): Promise<void> {
|
||||
if (!ensureRedisConfig()) {
|
||||
return;
|
||||
}
|
||||
|
||||
await getPublisher().set(key, JSON.stringify(data));
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Lifecycle
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export async function startRedisBridge(): Promise<void> {
|
||||
if (!ensureRedisConfig()) {
|
||||
logger.info("Redis not configured, skipping command channel bridge");
|
||||
return;
|
||||
}
|
||||
|
||||
// Warm up both clients so connection errors surface early
|
||||
getPublisher();
|
||||
getSubscriber();
|
||||
logger.info("Redis command channel initialized");
|
||||
}
|
||||
|
||||
export async function stopRedisBridge(): Promise<void> {
|
||||
if (publisherClient) {
|
||||
await publisherClient.quit();
|
||||
publisherClient = null;
|
||||
}
|
||||
if (subscriberClient) {
|
||||
await subscriberClient.quit();
|
||||
subscriberClient = null;
|
||||
}
|
||||
logger.info("Redis command channel stopped");
|
||||
}
|
||||
Reference in New Issue
Block a user