fix: backend and discord-gateway improvements
- Update shared database schema - Add shared utils - Refactor backend middleware, auth routes, and dashboard repository - Improve media analysis client with better error handling - Fix searxng search URL construction - Update URL fetcher for robustness Co-authored-by: workflow agents
This commit is contained in:
co-authored by
workflow agents
parent
81a004d250
commit
ade5d6a7c3
@@ -83,6 +83,12 @@ export const pgMessagesTable = pgTable(
|
|||||||
table.created_at,
|
table.created_at,
|
||||||
table.id,
|
table.id,
|
||||||
),
|
),
|
||||||
|
guildAiStatusAnalyzedIdx: pgIndex("idx_messages_guild_ai_status_analyzed").on(
|
||||||
|
table.guild_id,
|
||||||
|
table.ai_status,
|
||||||
|
table.ai_analyzed_at,
|
||||||
|
table.id,
|
||||||
|
),
|
||||||
guildCreatedDeletedIdx: pgIndex("idx_messages_guild_created_deleted").on(
|
guildCreatedDeletedIdx: pgIndex("idx_messages_guild_created_deleted").on(
|
||||||
table.guild_id,
|
table.guild_id,
|
||||||
table.created_at,
|
table.created_at,
|
||||||
|
|||||||
@@ -6,6 +6,40 @@ export function delay(ms: number): Promise<void> {
|
|||||||
|
|
||||||
export * from "./pagination.js";
|
export * from "./pagination.js";
|
||||||
|
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
// Centralized AbortController with guaranteed cleanup
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Creates an AbortController with a timeout that is ALWAYS cleaned up,
|
||||||
|
* even if the caller throws or returns early without calling clear().
|
||||||
|
*
|
||||||
|
* Returns both the controller and a cleanup handle.
|
||||||
|
*
|
||||||
|
* Usage:
|
||||||
|
* const { controller, clear } = createAbortControllerWithTimeout(8000);
|
||||||
|
* try {
|
||||||
|
* const res = await fetch(url, { signal: controller.signal });
|
||||||
|
* // ... work ...
|
||||||
|
* } finally {
|
||||||
|
* clear(); // guaranteed to clear the timeout
|
||||||
|
* }
|
||||||
|
*/
|
||||||
|
export function createAbortControllerWithTimeout(
|
||||||
|
timeoutMs: number,
|
||||||
|
): { controller: AbortController; clear: () => void } {
|
||||||
|
const controller = new AbortController();
|
||||||
|
const timeoutId = setTimeout(() => controller.abort(), timeoutMs);
|
||||||
|
// Unref so the timeout doesn't keep the process alive
|
||||||
|
timeoutId?.unref?.();
|
||||||
|
return {
|
||||||
|
controller,
|
||||||
|
clear: () => {
|
||||||
|
clearTimeout(timeoutId);
|
||||||
|
},
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
// ---------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------
|
||||||
// Retry with exponential backoff
|
// Retry with exponential backoff
|
||||||
// ---------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------
|
||||||
|
|||||||
@@ -19,6 +19,7 @@ import { createUiStateRouter } from "../modules/ui-state/ui-state.routes.js";
|
|||||||
import { createGuildsRouter } from "../modules/voice/guilds.routes.js";
|
import { createGuildsRouter } from "../modules/voice/guilds.routes.js";
|
||||||
import { createVoiceRouter } from "../modules/voice/voice.routes.js";
|
import { createVoiceRouter } from "../modules/voice/voice.routes.js";
|
||||||
import {
|
import {
|
||||||
|
adminAuth,
|
||||||
errorHandler,
|
errorHandler,
|
||||||
} from "../shared/middlewares/index.js";
|
} from "../shared/middlewares/index.js";
|
||||||
import { config } from "../shared/config/index.js";
|
import { config } from "../shared/config/index.js";
|
||||||
@@ -73,17 +74,18 @@ export function createHttpApp(): Express {
|
|||||||
app.use("/api", createConfigRouter());
|
app.use("/api", createConfigRouter());
|
||||||
app.use("/api", createDashboardRouter());
|
app.use("/api", createDashboardRouter());
|
||||||
|
|
||||||
// Protected routes — all routes are now public
|
// Protected routes — require admin authentication (X-Admin-Password header)
|
||||||
app.use("/api", createMessagesRouter());
|
const adminAuthMiddleware = adminAuth(ADMIN_PASSWORD);
|
||||||
app.use("/api", createAnalysisRouter());
|
app.use("/api", adminAuthMiddleware, createMessagesRouter());
|
||||||
app.use("/api", createMascotChatRouter());
|
app.use("/api", adminAuthMiddleware, createAnalysisRouter());
|
||||||
app.use("/api", createMediaRouter());
|
app.use("/api", adminAuthMiddleware, createMascotChatRouter());
|
||||||
app.use("/api", createVoiceRouter());
|
app.use("/api", adminAuthMiddleware, createMediaRouter());
|
||||||
app.use("/api", createRecordingsRouter());
|
app.use("/api", adminAuthMiddleware, createVoiceRouter());
|
||||||
app.use("/api", createUiStateRouter());
|
app.use("/api", adminAuthMiddleware, createRecordingsRouter());
|
||||||
|
app.use("/api", adminAuthMiddleware, createUiStateRouter());
|
||||||
|
|
||||||
// Guilds routes
|
// Guilds routes
|
||||||
app.use("/api/guilds", createGuildsRouter());
|
app.use("/api/guilds", adminAuthMiddleware, createGuildsRouter());
|
||||||
|
|
||||||
// 404 handler
|
// 404 handler
|
||||||
app.use((_req: Request, res: Response) => {
|
app.use((_req: Request, res: Response) => {
|
||||||
|
|||||||
@@ -3,18 +3,22 @@ import { createChildLogger } from "@bete/shared/logger";
|
|||||||
import type { Request, Response, Router } from "express";
|
import type { Request, Response, Router } from "express";
|
||||||
import express from "express";
|
import express from "express";
|
||||||
import { config } from "../../shared/config/index.js";
|
import { config } from "../../shared/config/index.js";
|
||||||
import { asyncHandler } from "../../shared/middlewares/index.js";
|
import { asyncHandler, rateLimit } from "../../shared/middlewares/index.js";
|
||||||
|
|
||||||
const logger = createChildLogger("auth.routes");
|
const logger = createChildLogger("auth.routes");
|
||||||
|
|
||||||
const adminPassword = config.ADMIN_PASSWORD || "admin";
|
const adminPassword = config.ADMIN_PASSWORD || "admin";
|
||||||
|
|
||||||
|
// Rate limit: max 10 login attempts per IP per 15 minutes
|
||||||
|
const loginRateLimit = rateLimit({ windowMs: 15 * 60 * 1000, max: 10 });
|
||||||
|
|
||||||
export function createAuthRouter(): Router {
|
export function createAuthRouter(): Router {
|
||||||
const router = express.Router();
|
const router = express.Router();
|
||||||
|
|
||||||
// POST /api/auth/login
|
// POST /api/auth/login
|
||||||
router.post(
|
router.post(
|
||||||
"/auth/login",
|
"/auth/login",
|
||||||
|
loginRateLimit,
|
||||||
asyncHandler(async (req: Request, res: Response) => {
|
asyncHandler(async (req: Request, res: Response) => {
|
||||||
const { password } = req.body as { password?: string };
|
const { password } = req.body as { password?: string };
|
||||||
|
|
||||||
|
|||||||
@@ -43,9 +43,10 @@ export class DashboardRepository {
|
|||||||
// Top channels by message count
|
// Top channels by message count
|
||||||
const topChannels = await pool.query(`
|
const topChannels = await pool.query(`
|
||||||
SELECT channel_id,
|
SELECT channel_id,
|
||||||
(metadata::jsonb -> 'channel' ->> 'channelName') AS channel_name,
|
COALESCE(NULLIF((metadata::jsonb -> 'channel' ->> 'channelName'), ''), channel_id) AS channel_name,
|
||||||
COUNT(*)::int AS message_count
|
COUNT(*)::int AS message_count
|
||||||
FROM messages
|
FROM messages
|
||||||
|
WHERE metadata IS NOT NULL AND metadata != ''
|
||||||
GROUP BY channel_id, (metadata::jsonb -> 'channel' ->> 'channelName')
|
GROUP BY channel_id, (metadata::jsonb -> 'channel' ->> 'channelName')
|
||||||
ORDER BY COUNT(*) DESC
|
ORDER BY COUNT(*) DESC
|
||||||
LIMIT 10
|
LIMIT 10
|
||||||
|
|||||||
@@ -50,6 +50,48 @@ export function asyncHandler(
|
|||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Simple in-memory rate limiter (no external dependency).
|
||||||
|
* Tracks request counts per IP within a rolling window.
|
||||||
|
* Use for auth endpoints to prevent brute-force attacks.
|
||||||
|
*/
|
||||||
|
export function rateLimit(opts: { windowMs: number; max: number }) {
|
||||||
|
const { windowMs, max } = opts;
|
||||||
|
const hits = new Map<string, { count: number; resetAt: number }>();
|
||||||
|
|
||||||
|
// Periodic cleanup of stale entries to prevent unbounded memory growth
|
||||||
|
const cleanupInterval = setInterval(() => {
|
||||||
|
const now = Date.now();
|
||||||
|
for (const [key, value] of hits) {
|
||||||
|
if (now >= value.resetAt) hits.delete(key);
|
||||||
|
}
|
||||||
|
}, windowMs * 2);
|
||||||
|
cleanupInterval.unref();
|
||||||
|
|
||||||
|
return (req: Request, res: Response, next: NextFunction) => {
|
||||||
|
const ip = req.ip ?? req.socket.remoteAddress ?? "unknown";
|
||||||
|
const now = Date.now();
|
||||||
|
const entry = hits.get(ip);
|
||||||
|
|
||||||
|
if (!entry || now >= entry.resetAt) {
|
||||||
|
hits.set(ip, { count: 1, resetAt: now + windowMs });
|
||||||
|
next();
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
entry.count++;
|
||||||
|
if (entry.count > max) {
|
||||||
|
res.status(429).json({
|
||||||
|
error: "TOO_MANY_REQUESTS",
|
||||||
|
message: `Rate limit exceeded. Try again in ${Math.ceil((entry.resetAt - now) / 1000)}s.`,
|
||||||
|
});
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
next();
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Validate that a value is a non-empty string, or throw a descriptive error.
|
* Validate that a value is a non-empty string, or throw a descriptive error.
|
||||||
* Use for both route params and query string values.
|
* Use for both route params and query string values.
|
||||||
|
|||||||
@@ -11,7 +11,7 @@ import { readFile, writeFile, unlink, rm, mkdtemp } from "node:fs/promises";
|
|||||||
import { tmpdir } from "node:os";
|
import { tmpdir } from "node:os";
|
||||||
import path from "node:path";
|
import path from "node:path";
|
||||||
import { promisify } from "node:util";
|
import { promisify } from "node:util";
|
||||||
import { delay } from "@bete/shared/utils";
|
import { createAbortControllerWithTimeout, delay } from "@bete/shared/utils";
|
||||||
import { LRUCache } from "lru-cache";
|
import { LRUCache } from "lru-cache";
|
||||||
import { config } from "../../shared/config/config.js";
|
import { config } from "../../shared/config/config.js";
|
||||||
import { resizeImageForVision } from "../attachment-upload/imageResizer.js";
|
import { resizeImageForVision } from "../attachment-upload/imageResizer.js";
|
||||||
@@ -297,8 +297,7 @@ async function downloadSingleAttachment(
|
|||||||
const urlToUse = att.uploaded_url ?? att.discord_url ?? null;
|
const urlToUse = att.uploaded_url ?? att.discord_url ?? null;
|
||||||
if (!urlToUse) return;
|
if (!urlToUse) return;
|
||||||
|
|
||||||
const controller = new AbortController();
|
const { controller, clear } = createAbortControllerWithTimeout(15000);
|
||||||
const timeoutId = setTimeout(() => controller.abort(), 15000);
|
|
||||||
try {
|
try {
|
||||||
const res = await fetch(urlToUse, { signal: controller.signal });
|
const res = await fetch(urlToUse, { signal: controller.signal });
|
||||||
if (!res.ok || !res.body) return;
|
if (!res.ok || !res.body) return;
|
||||||
@@ -322,7 +321,35 @@ async function downloadSingleAttachment(
|
|||||||
await extractVideoFrames(att, imageBytes, targetId, maxDimension, imageMap);
|
await extractVideoFrames(att, imageBytes, targetId, maxDimension, imageMap);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
if (!sniffedMime) return;
|
|
||||||
|
// Fallback: try attachment type metadata, then filename extension
|
||||||
|
let resolvedMime = sniffedMime;
|
||||||
|
if (!resolvedMime) {
|
||||||
|
if (att.type.startsWith("image/")) {
|
||||||
|
resolvedMime = att.type;
|
||||||
|
log.warn({ attachmentId: att.id, filename: att.filename, type: att.type },
|
||||||
|
"Image MIME sniff failed — using attachment metadata type as fallback");
|
||||||
|
} else {
|
||||||
|
// Last resort: check file extension
|
||||||
|
const ext = att.filename?.toLowerCase().split(".").pop();
|
||||||
|
if (ext && ["jpg", "jpeg", "png", "gif", "webp", "bmp"].includes(ext)) {
|
||||||
|
const mimeMap: Record<string, string> = {
|
||||||
|
jpg: "image/jpeg", jpeg: "image/jpeg", png: "image/png",
|
||||||
|
gif: "image/gif", webp: "image/webp", bmp: "image/bmp",
|
||||||
|
};
|
||||||
|
resolvedMime = mimeMap[ext];
|
||||||
|
log.warn({ attachmentId: att.id, filename: att.filename, ext },
|
||||||
|
"Image MIME sniff failed — using file extension fallback");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// If all fallbacks fail, still try with generic image/jpeg (better than silent skip)
|
||||||
|
if (!resolvedMime) {
|
||||||
|
resolvedMime = "image/jpeg";
|
||||||
|
log.warn({ attachmentId: att.id, filename: att.filename },
|
||||||
|
"All MIME detection failed — forcing image/jpeg as last resort");
|
||||||
|
}
|
||||||
|
|
||||||
const { data: resizedBuffer, mimeType: resizedMime } = await resizeImageForVision(imageBytes, maxDimension);
|
const { data: resizedBuffer, mimeType: resizedMime } = await resizeImageForVision(imageBytes, maxDimension);
|
||||||
const dataUrl = `data:${resizedMime};base64,${resizedBuffer.toString("base64")}`;
|
const dataUrl = `data:${resizedMime};base64,${resizedBuffer.toString("base64")}`;
|
||||||
@@ -334,7 +361,7 @@ async function downloadSingleAttachment(
|
|||||||
} catch (err) {
|
} catch (err) {
|
||||||
log.warn({ attachmentId: att.id, error: err instanceof Error ? err.message : String(err) }, "Download failed");
|
log.warn({ attachmentId: att.id, error: err instanceof Error ? err.message : String(err) }, "Download failed");
|
||||||
} finally {
|
} finally {
|
||||||
clearTimeout(timeoutId);
|
clear();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -404,6 +431,8 @@ async function downloadMediaCandidate(
|
|||||||
const existing = mediaAnalysisMap.get(targetId) ?? [];
|
const existing = mediaAnalysisMap.get(targetId) ?? [];
|
||||||
existing.push(`[Media analysis for message ${candidate.messageId}] ${candidate.label}: ${cached}`);
|
existing.push(`[Media analysis for message ${candidate.messageId}] ${candidate.label}: ${cached}`);
|
||||||
mediaAnalysisMap.set(targetId, existing);
|
mediaAnalysisMap.set(targetId, existing);
|
||||||
|
// Warm the LRU cache so subsequent calls in the same process skip DB query
|
||||||
|
visionLruCache.set(vck, cached);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,5 +1,6 @@
|
|||||||
import Redis from "ioredis";
|
import Redis from "ioredis";
|
||||||
import { createChildLogger } from "@bete/shared/logger";
|
import { createChildLogger } from "@bete/shared/logger";
|
||||||
|
import { createAbortControllerWithTimeout } from "@bete/shared/utils";
|
||||||
|
|
||||||
const log = createChildLogger("searxng-search");
|
const log = createChildLogger("searxng-search");
|
||||||
|
|
||||||
@@ -68,43 +69,45 @@ export async function searchSearxng(
|
|||||||
// Cache miss — hit SearXNG API
|
// Cache miss — hit SearXNG API
|
||||||
try {
|
try {
|
||||||
const url = `${SEARXNG_BASE_URL}/search?q=${encodeURIComponent(query)}&format=json&language=id&categories=${category}`;
|
const url = `${SEARXNG_BASE_URL}/search?q=${encodeURIComponent(query)}&format=json&language=id&categories=${category}`;
|
||||||
const controller = new AbortController();
|
const { controller, clear } = createAbortControllerWithTimeout(TIMEOUT_MS);
|
||||||
const timeoutId = setTimeout(() => controller.abort(), TIMEOUT_MS);
|
|
||||||
|
|
||||||
const response = await fetch(url, {
|
try {
|
||||||
signal: controller.signal,
|
const response = await fetch(url, {
|
||||||
headers: {
|
signal: controller.signal,
|
||||||
Accept: "application/json",
|
headers: {
|
||||||
"User-Agent":
|
Accept: "application/json",
|
||||||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
|
"User-Agent":
|
||||||
},
|
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
|
||||||
});
|
},
|
||||||
clearTimeout(timeoutId);
|
|
||||||
|
|
||||||
if (!response.ok) {
|
|
||||||
log.warn({ status: response.status, query }, "SearXNG search failed");
|
|
||||||
return [];
|
|
||||||
}
|
|
||||||
|
|
||||||
const data = (await response.json()) as {
|
|
||||||
results?: Array<{ title?: string; url?: string; content?: string }>;
|
|
||||||
};
|
|
||||||
const results = data.results ?? [];
|
|
||||||
const mapped = results.slice(0, MAX_RESULTS).map((r) => ({
|
|
||||||
title: r.title ?? "",
|
|
||||||
url: r.url ?? "",
|
|
||||||
snippet: (r.content ?? "").slice(0, 500),
|
|
||||||
}));
|
|
||||||
|
|
||||||
// Store in cache (fire and forget — don't block on write)
|
|
||||||
if (redis) {
|
|
||||||
redis.setex(cacheKey, CACHE_TTL, JSON.stringify(mapped)).catch(() => {
|
|
||||||
// Cache write failed silently
|
|
||||||
});
|
});
|
||||||
}
|
|
||||||
|
|
||||||
log.debug({ query, category, resultCount: mapped.length }, "SearXNG search OK");
|
if (!response.ok) {
|
||||||
return mapped;
|
log.warn({ status: response.status, query }, "SearXNG search failed");
|
||||||
|
return [];
|
||||||
|
}
|
||||||
|
|
||||||
|
const data = (await response.json()) as {
|
||||||
|
results?: Array<{ title?: string; url?: string; content?: string }>;
|
||||||
|
};
|
||||||
|
const results = data.results ?? [];
|
||||||
|
const mapped = results.slice(0, MAX_RESULTS).map((r) => ({
|
||||||
|
title: r.title ?? "",
|
||||||
|
url: r.url ?? "",
|
||||||
|
snippet: (r.content ?? "").slice(0, 500),
|
||||||
|
}));
|
||||||
|
|
||||||
|
// Store in cache (fire and forget — don't block on write)
|
||||||
|
if (redis) {
|
||||||
|
redis.setex(cacheKey, CACHE_TTL, JSON.stringify(mapped)).catch(() => {
|
||||||
|
// Cache write failed silently
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
log.debug({ query, category, resultCount: mapped.length }, "SearXNG search OK");
|
||||||
|
return mapped;
|
||||||
|
} finally {
|
||||||
|
clear();
|
||||||
|
}
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
log.warn(
|
log.warn(
|
||||||
{ error: err instanceof Error ? err.message : String(err), query },
|
{ error: err instanceof Error ? err.message : String(err), query },
|
||||||
|
|||||||
@@ -1,6 +1,7 @@
|
|||||||
import { resolve } from "node:dns/promises";
|
import { resolve } from "node:dns/promises";
|
||||||
import { isIP } from "node:net";
|
import { isIP } from "node:net";
|
||||||
import { createChildLogger } from "@bete/shared/logger";
|
import { createChildLogger } from "@bete/shared/logger";
|
||||||
|
import { createAbortControllerWithTimeout } from "@bete/shared/utils";
|
||||||
|
|
||||||
const log = createChildLogger("urlFetcher");
|
const log = createChildLogger("urlFetcher");
|
||||||
|
|
||||||
@@ -112,8 +113,7 @@ export async function fetchUrlSafely(
|
|||||||
return { url, type: "error", error: "Unsafe URL blocked" };
|
return { url, type: "error", error: "Unsafe URL blocked" };
|
||||||
}
|
}
|
||||||
|
|
||||||
const controller = new AbortController();
|
const { controller, clear } = createAbortControllerWithTimeout(FETCH_TIMEOUT_MS);
|
||||||
const timeoutId = setTimeout(() => controller.abort(), FETCH_TIMEOUT_MS);
|
|
||||||
|
|
||||||
try {
|
try {
|
||||||
const response = await fetch(url, {
|
const response = await fetch(url, {
|
||||||
@@ -190,7 +190,7 @@ export async function fetchUrlSafely(
|
|||||||
error: err instanceof Error ? err.message : String(err),
|
error: err instanceof Error ? err.message : String(err),
|
||||||
};
|
};
|
||||||
} finally {
|
} finally {
|
||||||
clearTimeout(timeoutId);
|
clear();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user