diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index 4ac1b33..372b536 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -58,6 +58,8 @@ export default function App() { const selectedVoiceChannel = uiState.selectedVoiceChannel || ""; const selectedTextGuild = uiState.selectedTextGuild || uiState.selectedGuild || ""; const selectedTextChannel = uiState.selectedTextChannel || ""; + const selectedAnalyticsGuild = uiState.selectedAnalyticsGuild || uiState.selectedGuild || ""; + const selectedAnalyticsChannel = uiState.selectedAnalyticsChannel || ""; const handleIncomingPcm = useCallback((data: ArrayBuffer) => { const headerView = new DataView(data, 0, 4); @@ -163,6 +165,7 @@ export default function App() { useEffect(() => { if (selectedVoiceGuild) voice.loadVoiceChannels(selectedVoiceGuild).catch(() => undefined); }, [selectedVoiceGuild]); useEffect(() => { if (selectedTextGuild) voice.loadTextTargets(selectedTextGuild).catch(() => undefined); }, [selectedTextGuild]); + useEffect(() => { if (selectedAnalyticsGuild) voice.loadTextTargets(selectedAnalyticsGuild).catch(() => undefined); }, [selectedAnalyticsGuild]); useEffect(() => { if (selectedTextChannel) messages.fetchMessages(selectedTextChannel).catch(() => undefined); }, [selectedTextChannel]); const toggleListening = useCallback(async () => { @@ -250,10 +253,10 @@ export default function App() { patchUIState({ selectedTextGuild: guildId, selectedTextChannel: "" })} - onChannelChange={(channelId) => patchUIState({ selectedTextChannel: channelId })} + selectedGuild={selectedAnalyticsGuild} + selectedChannel={selectedAnalyticsChannel} + onGuildChange={(guildId) => patchUIState({ selectedAnalyticsGuild: guildId, selectedAnalyticsChannel: "" })} + onChannelChange={(channelId) => patchUIState({ selectedAnalyticsChannel: channelId })} /> diff --git a/frontend/src/types/ui.ts b/frontend/src/types/ui.ts index 82f5705..9804c12 100644 --- a/frontend/src/types/ui.ts +++ b/frontend/src/types/ui.ts @@ -6,6 +6,8 @@ export interface UIState { selectedVoiceChannel?: string; selectedTextGuild?: string; selectedTextChannel?: string; + selectedAnalyticsGuild?: string; + selectedAnalyticsChannel?: string; activeTab?: DashboardTab; isListening?: boolean; isStreaming?: boolean; diff --git a/scripts/fix-missing-tables.sql b/scripts/fix-missing-tables.sql new file mode 100644 index 0000000..ebb998f --- /dev/null +++ b/scripts/fix-missing-tables.sql @@ -0,0 +1,153 @@ +-- Fix: missing messages and attachments tables on VPS +-- Run: PGPASSWORD=hunterz psql -h 100.108.1.124 -U asephs -d hub -f scripts/fix-missing-tables.sql + +BEGIN; + +-- 1. Create messages table if not exists +CREATE TABLE IF NOT EXISTS "messages" ( + "id" text PRIMARY KEY NOT NULL, + "guild_id" text NOT NULL, + "channel_id" text NOT NULL, + "thread_id" text, + "user_id" text NOT NULL, + "username" text NOT NULL, + "avatar_url" text, + "content" text NOT NULL, + "edited_content" text, + "created_at" bigint NOT NULL, + "edited_at" bigint, + "deleted_at" bigint, + "type" text DEFAULT 'text' NOT NULL, + "metadata" text, + "ai_status" text DEFAULT 'pending' NOT NULL, + "ai_moderation_flags" text, + "ai_moderation_score" real, + "ai_moderation_raw" text, + "ai_analysis" text, + "ai_analyzed_at" bigint, + "ai_error" text +); + +-- 2. Create attachments table if not exists +CREATE TABLE IF NOT EXISTS "attachments" ( + "id" text PRIMARY KEY NOT NULL, + "message_id" text NOT NULL, + "guild_id" text NOT NULL, + "channel_id" text NOT NULL, + "thread_id" text, + "user_id" text NOT NULL, + "filename" text NOT NULL, + "size" integer NOT NULL, + "type" text NOT NULL, + "discord_url" text NOT NULL, + "uploaded_url" text, + "upload_status" text DEFAULT 'pending' NOT NULL, + "upload_error" text, + "created_at" bigint NOT NULL, + "uploaded_at" bigint +); + +-- 3. Foreign key +ALTER TABLE "attachments" DROP CONSTRAINT IF EXISTS "fk_attachments_message_id"; +ALTER TABLE "attachments" ADD CONSTRAINT "fk_attachments_message_id" + FOREIGN KEY ("message_id") REFERENCES "public"."messages"("id") + ON DELETE cascade ON UPDATE no action; + +-- 4. Indexes from migration 0000 +CREATE INDEX IF NOT EXISTS "idx_attachments_channel" ON "attachments" USING btree ("channel_id"); +CREATE INDEX IF NOT EXISTS "idx_attachments_message" ON "attachments" USING btree ("message_id"); +CREATE INDEX IF NOT EXISTS "idx_attachments_status" ON "attachments" USING btree ("upload_status"); +CREATE INDEX IF NOT EXISTS "idx_messages_channel" ON "messages" USING btree ("channel_id"); +CREATE INDEX IF NOT EXISTS "idx_messages_user" ON "messages" USING btree ("user_id"); +CREATE INDEX IF NOT EXISTS "idx_messages_created" ON "messages" USING btree ("created_at"); +CREATE INDEX IF NOT EXISTS "idx_messages_thread" ON "messages" USING btree ("thread_id"); + +-- 5. Indexes from migration 0001 +CREATE INDEX IF NOT EXISTS "idx_attachments_channel_created" ON "attachments" ("channel_id","created_at","id"); +CREATE INDEX IF NOT EXISTS "idx_attachments_thread_created" ON "attachments" ("thread_id","created_at","id"); +CREATE INDEX IF NOT EXISTS "idx_messages_channel_created" ON "messages" ("channel_id","created_at","id"); +CREATE INDEX IF NOT EXISTS "idx_messages_thread_created" ON "messages" ("thread_id","created_at","id"); +CREATE INDEX IF NOT EXISTS "idx_messages_ai_status_created" ON "messages" ("ai_status","created_at","id"); +CREATE INDEX IF NOT EXISTS "idx_messages_guild_ai_status_created" ON "messages" ("guild_id","ai_status","created_at","id"); + +-- 6. Columns from migration 0003 +ALTER TABLE "messages" ADD COLUMN IF NOT EXISTS "ai_categories" text; +ALTER TABLE "messages" ADD COLUMN IF NOT EXISTS "ai_severity" text; +ALTER TABLE "messages" ADD COLUMN IF NOT EXISTS "ai_confidence" real; +ALTER TABLE "messages" ADD COLUMN IF NOT EXISTS "ai_recommended_action" text; +ALTER TABLE "messages" ADD COLUMN IF NOT EXISTS "ai_policy_version" text; +ALTER TABLE "messages" ADD COLUMN IF NOT EXISTS "ai_evidence" text; + +-- 7. message_reviews, moderation_actions, retention_policies (from 0003 — should already exist, but IF NOT EXISTS safe) +CREATE TABLE IF NOT EXISTS "message_reviews" ( + "id" text PRIMARY KEY NOT NULL, + "message_id" text NOT NULL, + "guild_id" text NOT NULL, + "channel_id" text NOT NULL, + "reviewer_id" text, + "status" text DEFAULT 'pending' NOT NULL, + "notes" text, + "created_at" bigint NOT NULL, + "reviewed_at" bigint +); +CREATE INDEX IF NOT EXISTS "idx_message_reviews_message_id" ON "message_reviews" USING btree ("message_id"); +CREATE INDEX IF NOT EXISTS "idx_message_reviews_guild_status" ON "message_reviews" USING btree ("guild_id", "status", "created_at"); + +CREATE TABLE IF NOT EXISTS "moderation_actions" ( + "id" text PRIMARY KEY NOT NULL, + "message_id" text, + "user_id" text, + "guild_id" text NOT NULL, + "action_type" text NOT NULL, + "reason" text, + "executed_by" text, + "status" text DEFAULT 'pending' NOT NULL, + "error" text, + "created_at" bigint NOT NULL, + "executed_at" bigint +); +CREATE INDEX IF NOT EXISTS "idx_moderation_actions_message_id" ON "moderation_actions" USING btree ("message_id"); +CREATE INDEX IF NOT EXISTS "idx_moderation_actions_user_id" ON "moderation_actions" USING btree ("user_id"); +CREATE INDEX IF NOT EXISTS "idx_moderation_actions_status" ON "moderation_actions" USING btree ("status"); +CREATE INDEX IF NOT EXISTS "idx_moderation_actions_guild_status" ON "moderation_actions" USING btree ("guild_id", "status", "created_at"); + +CREATE TABLE IF NOT EXISTS "retention_policies" ( + "id" text PRIMARY KEY NOT NULL, + "guild_id" text NOT NULL, + "channel_id" text, + "retention_days" integer DEFAULT 90 NOT NULL, + "apply_to_media" boolean DEFAULT true NOT NULL, + "apply_to_voice" boolean DEFAULT true NOT NULL, + "enabled" boolean DEFAULT true NOT NULL, + "created_at" bigint NOT NULL, + "updated_at" bigint NOT NULL +); +CREATE INDEX IF NOT EXISTS "idx_retention_policies_guild_id" ON "retention_policies" USING btree ("guild_id"); +CREATE INDEX IF NOT EXISTS "idx_retention_policies_enabled" ON "retention_policies" USING btree ("enabled"); + +-- 8. Indexes from migration 0005 +CREATE INDEX IF NOT EXISTS "idx_messages_guild_created_deleted" ON "messages" USING btree ("guild_id","created_at","deleted_at","id"); +CREATE INDEX IF NOT EXISTS "idx_messages_channel_ai_status_created" ON "messages" USING btree ("channel_id","ai_status","created_at","id"); +CREATE INDEX IF NOT EXISTS "idx_messages_thread_ai_status_created" ON "messages" USING btree ("thread_id","ai_status","created_at","id"); + +-- 9. Drizzle migration tracking table (so future drizzle-kit migrate works) +CREATE TABLE IF NOT EXISTS "__drizzle_migrations" ( + "id" serial PRIMARY KEY, + "hash" text NOT NULL, + "created_at" bigint NOT NULL +); + +-- Seed migration tracking so drizzle-kit doesn't try to re-apply +DO $$ +BEGIN + IF NOT EXISTS (SELECT 1 FROM "__drizzle_migrations" WHERE hash = '0000_rare_kitty_pryde') THEN + INSERT INTO "__drizzle_migrations" (hash, created_at) VALUES + ('0000_rare_kitty_pryde', 1778750697764), + ('0001_curious_zodiak', 1778764447718), + ('0002_dark_omega_flight', 1779109619461), + ('0003_ai_moderation_review_guardrails', 1780079000000), + ('0005_optimize-message-index', 1780218363790); + END IF; +END $$; + +COMMIT; diff --git a/src/muxer-queue.ts b/src/muxer-queue.ts index f60026e..c1d13e1 100644 --- a/src/muxer-queue.ts +++ b/src/muxer-queue.ts @@ -7,9 +7,18 @@ const logger = createChildLogger("muxer-queue"); // ── Redis client (lazy singleton) ────────────────────────────────────────── let redis: Redis | null = null; +let redisReady = false; +let redisAttempted = false; -function getRedis(): Redis { - if (redis !== null) return redis; +/** + * Try to create a Redis connection. Does NOT block if unreachable — + * sets lazyConnect + logs a single warning, then lets callers fall back + * to in-memory KV. + */ +function tryGetRedis(): Redis | null { + if (redis !== null) return redisReady ? redis : null; + if (redisAttempted) return null; + redisAttempted = true; redis = new Redis(config.REDIS_URL, { maxRetriesPerRequest: 3, @@ -17,20 +26,45 @@ function getRedis(): Redis { if (times > 5) return null; // stop retrying return Math.min(times * 200, 2000); }, - lazyConnect: false, + lazyConnect: true, // don't block startup on Redis }); redis.on("error", (err) => { - logger.error({ err }, "Redis connection error"); + logger.warn({ err }, "Redis connection failed — using in-memory fallback"); }); redis.on("connect", () => { logger.info({ url: config.REDIS_URL }, "Redis connected"); + redisReady = true; + }); + + redis.on("reconnecting", () => { + redisReady = false; }); return redis; } +/** + * Attempt a Redis operation. Returns the result on success, or null if + * Redis is unavailable (caller should fall back). + */ +async function redisOp(fn: (r: Redis) => Promise): Promise { + const r = tryGetRedis(); + if (!r) return null; + try { + return await fn(r); + } catch { + // Redis failed mid-operation — mark as not ready so next call falls back + redisReady = false; + return null; + } +} + +// ── In-memory KV fallback (volatile, survives only process lifetime) ────── + +const memKV = new Map(); + // ── Types ────────────────────────────────────────────────────────────────── export interface MuxerJobData { @@ -61,33 +95,39 @@ export async function getPersistedValue( key: string, fallback: T, ): Promise { - try { - const r = getRedis(); - const raw = await r.get(`${KV_PREFIX}${key}`); - if (raw === null) return fallback; - return JSON.parse(raw) as T; - } catch (error) { - logger.error( - { key, error: error instanceof Error ? error.message : String(error) }, - "Failed to get persisted value", - ); - return fallback; + const raw = await redisOp((r) => r.get(`${KV_PREFIX}${key}`)); + if (raw !== null && raw !== undefined) { + try { + return JSON.parse(raw) as T; + } catch { + logger.warn({ key }, "Failed to parse persisted value from Redis"); + } } + // Fallback to in-memory KV, then default + const memVal = memKV.get(`${KV_PREFIX}${key}`); + if (memVal !== undefined) { + try { + return JSON.parse(memVal) as T; + } catch { + // ignore corrupted in-memory data + } + } + return fallback; } export async function setPersistedValue( key: string, value: unknown, ): Promise { - try { - const r = getRedis(); - await r.set(`${KV_PREFIX}${key}`, JSON.stringify(value)); - } catch (error) { - logger.error( - { key, error: error instanceof Error ? error.message : String(error) }, - "Failed to set persisted value", + const serialized = JSON.stringify(value); + // Always store in in-memory KV so it works even without Redis + memKV.set(`${KV_PREFIX}${key}`, serialized); + const saved = await redisOp((r) => r.set(`${KV_PREFIX}${key}`, serialized)); + if (!saved) { + logger.verbose( + { key }, + "Persisted value stored in-memory only (Redis unavailable)", ); - throw error; } } @@ -115,8 +155,7 @@ function queueKey(status: string): string { } export async function enqueueMuxerJob(data: MuxerJobData): Promise { - try { - const r = getRedis(); + const result = await redisOp(async (r) => { const jobId = `${data.userId}-${data.sessionId}`; const now = Date.now(); @@ -142,16 +181,14 @@ export async function enqueueMuxerJob(data: MuxerJobData): Promise { ); return jobId; - } catch (error) { - logger.error( - { - userId: data.userId, - error: error instanceof Error ? error.message : String(error), - }, - "Failed to enqueue muxer job", + }); + if (!result) { + logger.warn( + { userId: data.userId }, + "Failed to enqueue muxer job (Redis unavailable)", ); - throw error; } + return result ?? ""; } export async function getPendingJobs(): Promise< @@ -166,9 +203,7 @@ export async function getPendingJobs(): Promise< error?: string; }> > { - try { - const r = getRedis(); - + const jobs = await redisOp(async (r) => { // Get up to 10 pending job IDs from the left (oldest first) const jobIds = await r.lrange(QUEUE_PENDING, 0, 9); if (jobIds.length === 0) return []; @@ -181,7 +216,7 @@ export async function getPendingJobs(): Promise< const results = await pipeline.exec(); if (!results) return []; - const jobs: Array<{ + const out: Array<{ id: string; data: string; status: "pending" | "processing" | "completed" | "failed"; @@ -197,7 +232,7 @@ export async function getPendingJobs(): Promise< if (err || !fields) continue; const raw = fields as Record; - jobs.push({ + out.push({ id: jobIds[i], data: raw.data || "", status: (raw.status as "pending") || "pending", @@ -209,14 +244,12 @@ export async function getPendingJobs(): Promise< }); } - return jobs; - } catch (error) { - logger.error( - { error: error instanceof Error ? error.message : String(error) }, - "Failed to get pending jobs", - ); - return []; + return out; + }); + if (!jobs) { + logger.verbose("No pending jobs (Redis unavailable)"); } + return jobs ?? []; } export async function updateJobStatus( @@ -224,8 +257,7 @@ export async function updateJobStatus( status: "processing" | "completed" | "failed", error?: string, ): Promise { - try { - const r = getRedis(); + const ok = await redisOp(async (r) => { const jobKey = `${JOB_PREFIX}${jobId}`; const now = Date.now(); @@ -256,21 +288,17 @@ export async function updateJobStatus( await pipeline.exec(); logger.info({ jobId, status, error }, "Job status updated"); - } catch (err) { - logger.error( - { - jobId, - error: err instanceof Error ? err.message : String(err), - }, - "Failed to update job status", + }); + if (!ok) { + logger.verbose( + { jobId, status }, + "Job status update skipped (Redis unavailable)", ); - throw err; } } export async function retryFailedJob(jobId: string): Promise { - try { - const r = getRedis(); + const result = await redisOp(async (r) => { const jobKey = `${JOB_PREFIX}${jobId}`; const [attemptsStr, maxAttemptsStr] = await r.hmget( @@ -299,20 +327,17 @@ export async function retryFailedJob(jobId: string): Promise { logger.info({ jobId, attempt: attempts + 1 }, "Job retried"); return true; - } catch (err) { - logger.error( - { jobId, error: err instanceof Error ? err.message : String(err) }, - "Failed to retry job", - ); - return false; + }); + if (!result) { + logger.verbose({ jobId }, "Job retry skipped (Redis unavailable)"); } + return result ?? false; } export async function cleanupCompletedJobs( olderThanMs: number = 24 * 60 * 60 * 1000, ): Promise { - try { - const r = getRedis(); + const deletedCount = await redisOp(async (r) => { const cutoffTime = Date.now() - olderThanMs; const jobIds = await r.lrange(QUEUE_COMPLETED, 0, -1); @@ -325,7 +350,7 @@ export async function cleanupCompletedJobs( const results = await pipeline.exec(); if (!results) return 0; - let deletedCount = 0; + let count = 0; const deletePipeline = r.pipeline(); for (let i = 0; i < jobIds.length; i++) { @@ -336,23 +361,21 @@ export async function cleanupCompletedJobs( if (updatedAt < cutoffTime) { deletePipeline.del(`${JOB_PREFIX}${jobIds[i]}`); deletePipeline.lrem(QUEUE_COMPLETED, 0, jobIds[i]); - deletedCount++; + count++; } } - if (deletedCount > 0) { + if (count > 0) { await deletePipeline.exec(); } - logger.info({ deletedCount }, "Cleaned up completed jobs"); - return deletedCount; - } catch (err) { - logger.error( - { error: err instanceof Error ? err.message : String(err) }, - "Failed to clean up completed jobs", - ); - return 0; + logger.info({ deletedCount: count }, "Cleaned up completed jobs"); + return count; + }); + if (deletedCount === null) { + logger.verbose("Cleanup skipped (Redis unavailable)"); } + return deletedCount ?? 0; } export async function getJobStats(): Promise<{ @@ -361,23 +384,16 @@ export async function getJobStats(): Promise<{ completed: number; failed: number; }> { - try { - const r = getRedis(); + const stats = await redisOp(async (r) => { const [pending, processing, completed, failed] = await Promise.all([ r.llen(QUEUE_PENDING), r.llen(QUEUE_PROCESSING), r.llen(QUEUE_COMPLETED), r.llen(QUEUE_FAILED), ]); - return { pending, processing, completed, failed }; - } catch (err) { - logger.error( - { error: err instanceof Error ? err.message : String(err) }, - "Failed to get job stats", - ); - return { pending: 0, processing: 0, completed: 0, failed: 0 }; - } + }); + return stats ?? { pending: 0, processing: 0, completed: 0, failed: 0 }; } export async function closeQueue(): Promise {