diff --git a/packages/monitor/src/broadcaster.ts b/packages/monitor/src/broadcaster.ts index d9a321f..3b58e33 100644 --- a/packages/monitor/src/broadcaster.ts +++ b/packages/monitor/src/broadcaster.ts @@ -13,7 +13,11 @@ import { config } from './config.js' import { log } from './logger.js' import { isEncryptionEnabled, encrypt, maskApiKey } from './crypto.js' import { login, register, verifyToken } from './auth.js' -import { getAgentConfig, upsertAgentConfig, findUserById, upsertScheduleConfig, getScheduleConfig, getReportsForDate } from './db.js' +import { + getAgentConfig, upsertAgentConfig, findUserById, upsertScheduleConfig, getScheduleConfig, getReportsForDate, + appendAlertLogForAllUsers, getRecentAlerts, appendAnalysisLogForAllUsers, getRecentAnalyses, + getChatHistory, +} from './db.js' import type { JwtPayload } from './auth.js' import { handleOAuthRoute, getEnabledOAuthProviders } from './oauth.js' import { getUserWatchlistsWithItems, saveWatchlist, removeWatchlistItem, syncFromTastytrade, searchSymbols } from './watchlistService.js' @@ -54,7 +58,16 @@ export function onAlertForOrchestrator(handler: OrchestratorHandler): void { export function broadcastToAll(msg: WsMessage | Record): void { const typed = msg as Record if (typed.type === 'agent_analysis' && typed.data) { - pushAnalysis(typed.data as AgentAnalysis) + const analysis = typed.data as AgentAnalysis + pushAnalysis(analysis) + if (isMultiTenant()) { + try { + const analysisId = (analysis as Record).id as string ?? `analysis-${Date.now()}` + appendAnalysisLogForAllUsers(analysisId, JSON.stringify(analysis)) + } catch (err) { + log.warn(`Failed to persist analysis: ${(err as Error).message}`) + } + } } if (typed.type === 'agent_status' && typed.data) { agentStatus = typed.data as AgentStatus @@ -189,6 +202,14 @@ export function startBroadcaster(): void { onAlert((alert: OptionsAlert) => { pushAlert(alert) broadcast({ type: 'alert', data: alert }) + if (isMultiTenant()) { + try { + const alertId = (alert as Record).id as string ?? `alert-${Date.now()}` + appendAlertLogForAllUsers(alertId, JSON.stringify(alert)) + } catch (err) { + log.warn(`Failed to persist alert: ${(err as Error).message}`) + } + } if (orchestratorHandler) { Promise.resolve(orchestratorHandler(alert)).catch(err => { log.error('Orchestrator dispatch error:', err) @@ -255,9 +276,52 @@ function send(ws: WebSocket, msg: WsMessage | Record): void { function sendFullState(ws: WebSocket): void { send(ws, { type: 'snapshot', data: getAllSnapshots() }) send(ws, { type: 'account', data: getAccountContext() }) - if (recentAlerts.length > 0) send(ws, { type: 'alert_history', data: recentAlerts }) - if (recentAnalyses.length > 0) send(ws, { type: 'analysis_history', data: recentAnalyses }) if (agentStatus) send(ws, { type: 'agent_status', data: agentStatus }) + + if (isMultiTenant()) { + sendPerUserState(ws) + } else { + if (recentAlerts.length > 0) send(ws, { type: 'alert_history', data: recentAlerts }) + if (recentAnalyses.length > 0) send(ws, { type: 'analysis_history', data: recentAnalyses }) + } +} + +function sendPerUserState(ws: WebSocket): void { + const auth = clientAuth.get(ws) + if (!auth) return + + try { + const alertPayloads = getRecentAlerts(auth.userId) + if (alertPayloads.length > 0) { + const alerts = alertPayloads.map(p => JSON.parse(p)) + send(ws, { type: 'alert_history', data: alerts }) + } + + const analysisPayloads = getRecentAnalyses(auth.userId) + if (analysisPayloads.length > 0) { + const analyses = analysisPayloads.map(p => JSON.parse(p)) + send(ws, { type: 'analysis_history', data: analyses }) + } + + const chatRows = getChatHistory(auth.userId) + if (chatRows.length > 0) { + const messages = chatRows.map(r => ({ + id: `db-${r.id}`, + role: r.role, + content: r.content, + timestamp: r.created_at, + })) + send(ws, { type: 'chat_history', data: messages }) + } + + handleRequestAgentConfig(ws) + handleRequestWatchlist(ws) + handleRequestScheduleConfig(ws) + handleRequestBudgetStatus(ws) + handleRequestReports(ws, undefined) + } catch (err) { + log.warn(`Failed to send per-user state for ${auth.email}: ${(err as Error).message}`) + } } async function handleAuth(ws: WebSocket, msg: Record): Promise { @@ -643,7 +707,16 @@ const authenticatedWsRoutes: Record = { if (!data || typeof data !== 'object') return const d = data as Record log.info(`Agent analysis received for ${d.ticker ?? 'unknown'} (model: ${d.model ?? 'unknown'})`) - pushAnalysis(data as unknown as AgentAnalysis) + const analysis = data as unknown as AgentAnalysis + pushAnalysis(analysis) + if (isMultiTenant()) { + try { + const analysisId = d.id as string ?? `analysis-${Date.now()}` + appendAnalysisLogForAllUsers(analysisId, JSON.stringify(analysis)) + } catch (err) { + log.warn(`Failed to persist analysis: ${(err as Error).message}`) + } + } broadcastRaw(raw) }, @@ -654,6 +727,14 @@ const authenticatedWsRoutes: Record = { log.info(`Injected test alert for ${payload.trigger?.ticker ?? 'unknown'}`) pushAlert(payload) broadcast({ type: 'alert', data: payload }) + if (isMultiTenant()) { + try { + const alertId = (data as Record).id as string ?? `alert-${Date.now()}` + appendAlertLogForAllUsers(alertId, JSON.stringify(payload)) + } catch (err) { + log.warn(`Failed to persist alert: ${(err as Error).message}`) + } + } if (orchestratorHandler) { Promise.resolve(orchestratorHandler(payload)).catch(err => { log.error('Orchestrator dispatch error:', err) diff --git a/packages/monitor/src/chatHandler.ts b/packages/monitor/src/chatHandler.ts index 2f9f883..ce84e4b 100644 --- a/packages/monitor/src/chatHandler.ts +++ b/packages/monitor/src/chatHandler.ts @@ -4,8 +4,8 @@ import { getAllSnapshots } from './state.js' import { getAccountContext } from './account.js' import { getUserWatchlistsWithItems } from './watchlistService.js' import { checkBudget, trackUsage } from './budgetTracker.js' -import { getAgentConfig } from './db.js' -import { decrypt } from './crypto.js' +import { getAgentConfig, appendChatMessage, getChatHistory as getChatHistoryDb, clearChatHistoryDb } from './db.js' +import { decrypt, isEncryptionEnabled } from './crypto.js' import { log } from './logger.js' const MAX_HISTORY = 20 @@ -60,11 +60,20 @@ export async function handleChatMessage(userId: number, message: string): Promis } } - const history = conversations.get(userId) ?? [] + let history = conversations.get(userId) + if (!history && isEncryptionEnabled()) { + const rows = getChatHistoryDb(userId) + history = rows.map(r => ({ role: r.role, content: r.content })) + } + if (!history) history = [] history.push({ role: 'user', content: message }) if (history.length > MAX_HISTORY) history.splice(0, history.length - MAX_HISTORY) conversations.set(userId, history) + if (isEncryptionEnabled()) { + try { appendChatMessage(userId, 'user', message) } catch { /* best-effort */ } + } + try { const apiKey = decrypt(agentCfg.encrypted_api_key) const systemPrompt = buildChatSystemPrompt(userId) @@ -83,6 +92,10 @@ export async function handleChatMessage(userId: number, message: string): Promis if (history.length > MAX_HISTORY) history.splice(0, history.length - MAX_HISTORY) conversations.set(userId, history) + if (isEncryptionEnabled()) { + try { appendChatMessage(userId, 'assistant', result.text) } catch { /* best-effort */ } + } + log.info(`Chat: user ${userId} response (${result.outputTokens} tokens, $${costEstimate.toFixed(4)})`) return { @@ -95,10 +108,14 @@ export async function handleChatMessage(userId: number, message: string): Promis } catch (err) { const errMsg = (err as Error).message log.error(`Chat failed for user ${userId}: ${errMsg}`) + const errorMsg = `Error: ${errMsg}` + if (isEncryptionEnabled()) { + try { appendChatMessage(userId, 'assistant', errorMsg) } catch { /* best-effort */ } + } return { id: randomUUID(), role: 'assistant', - content: `Error: ${errMsg}`, + content: errorMsg, timestamp: new Date().toISOString(), costUsd: 0, } @@ -107,10 +124,25 @@ export async function handleChatMessage(userId: number, message: string): Promis export function clearChatHistory(userId: number): void { conversations.delete(userId) + if (isEncryptionEnabled()) { + try { clearChatHistoryDb(userId) } catch { /* best-effort */ } + } log.info(`Chat: cleared history for user ${userId}`) } -export function getChatHistory(userId: number): ChatMessage[] { +export function getChatHistoryForUser(userId: number): ChatMessage[] { + if (isEncryptionEnabled()) { + const rows = getChatHistoryDb(userId) + if (rows.length > 0 && !conversations.has(userId)) { + conversations.set(userId, rows.map(r => ({ role: r.role, content: r.content }))) + } + return rows.map(r => ({ + id: `db-${r.id}`, + role: r.role, + content: r.content, + timestamp: r.created_at, + })) + } const history = conversations.get(userId) ?? [] return history.map((h, i) => ({ id: `hist-${i}`, diff --git a/packages/monitor/src/db.ts b/packages/monitor/src/db.ts index a0e0a2e..32cc8a6 100644 --- a/packages/monitor/src/db.ts +++ b/packages/monitor/src/db.ts @@ -127,6 +127,39 @@ function migrateV2Tables(db: Database.Database): void { CREATE INDEX IF NOT EXISTS idx_analysis_reports_user ON analysis_reports(user_id, created_at); + + CREATE TABLE IF NOT EXISTS chat_history ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE, + role TEXT NOT NULL CHECK(role IN ('user','assistant')), + content TEXT NOT NULL, + created_at TEXT NOT NULL DEFAULT (datetime('now')) + ); + + CREATE INDEX IF NOT EXISTS idx_chat_history_user + ON chat_history(user_id, created_at); + + CREATE TABLE IF NOT EXISTS alerts_log ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE, + alert_id TEXT NOT NULL, + payload TEXT NOT NULL, + created_at TEXT NOT NULL DEFAULT (datetime('now')) + ); + + CREATE INDEX IF NOT EXISTS idx_alerts_log_user + ON alerts_log(user_id, created_at); + + CREATE TABLE IF NOT EXISTS analyses_log ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE, + analysis_id TEXT NOT NULL, + payload TEXT NOT NULL, + created_at TEXT NOT NULL DEFAULT (datetime('now')) + ); + + CREATE INDEX IF NOT EXISTS idx_analyses_log_user + ON analyses_log(user_id, created_at); `) } @@ -463,3 +496,77 @@ export function getLatestReport(userId: number): AnalysisReportRow | undefined { 'SELECT * FROM analysis_reports WHERE user_id = ? ORDER BY created_at DESC LIMIT 1' ).get(userId) as AnalysisReportRow | undefined } + +// ─── Chat History ───────────────────────────────────────────────── + +export interface ChatHistoryRow { + id: number + user_id: number + role: 'user' | 'assistant' + content: string + created_at: string +} + +export function appendChatMessage(userId: number, role: 'user' | 'assistant', content: string): ChatHistoryRow { + return getDb().prepare( + "INSERT INTO chat_history (user_id, role, content) VALUES (?, ?, ?) RETURNING *" + ).get(userId, role, content) as ChatHistoryRow +} + +export function getChatHistory(userId: number, limit = 50): ChatHistoryRow[] { + return getDb().prepare( + 'SELECT * FROM (SELECT * FROM chat_history WHERE user_id = ? ORDER BY created_at DESC LIMIT ?) ORDER BY created_at ASC' + ).all(userId, limit) as ChatHistoryRow[] +} + +export function clearChatHistoryDb(userId: number): void { + getDb().prepare('DELETE FROM chat_history WHERE user_id = ?').run(userId) +} + +// ─── Alerts Log ─────────────────────────────────────────────────── + +export function appendAlertLog(userId: number, alertId: string, payload: string): void { + getDb().prepare( + "INSERT INTO alerts_log (user_id, alert_id, payload) VALUES (?, ?, ?)" + ).run(userId, alertId, payload) +} + +export function getRecentAlerts(userId: number, limit = 200): string[] { + const rows = getDb().prepare( + 'SELECT payload FROM alerts_log WHERE user_id = ? ORDER BY created_at DESC LIMIT ?' + ).all(userId, limit) as { payload: string }[] + return rows.reverse().map(r => r.payload) +} + +export function appendAlertLogForAllUsers(alertId: string, payload: string): void { + const users = getDb().prepare('SELECT id FROM users').all() as { id: number }[] + const stmt = getDb().prepare("INSERT INTO alerts_log (user_id, alert_id, payload) VALUES (?, ?, ?)") + const tx = getDb().transaction(() => { + for (const u of users) stmt.run(u.id, alertId, payload) + }) + tx() +} + +// ─── Analyses Log ───────────────────────────────────────────────── + +export function appendAnalysisLog(userId: number, analysisId: string, payload: string): void { + getDb().prepare( + "INSERT INTO analyses_log (user_id, analysis_id, payload) VALUES (?, ?, ?)" + ).run(userId, analysisId, payload) +} + +export function getRecentAnalyses(userId: number, limit = 50): string[] { + const rows = getDb().prepare( + 'SELECT payload FROM analyses_log WHERE user_id = ? ORDER BY created_at DESC LIMIT ?' + ).all(userId, limit) as { payload: string }[] + return rows.reverse().map(r => r.payload) +} + +export function appendAnalysisLogForAllUsers(analysisId: string, payload: string): void { + const users = getDb().prepare('SELECT id FROM users').all() as { id: number }[] + const stmt = getDb().prepare("INSERT INTO analyses_log (user_id, analysis_id, payload) VALUES (?, ?, ?)") + const tx = getDb().transaction(() => { + for (const u of users) stmt.run(u.id, analysisId, payload) + }) + tx() +}