const path = require('path'); const { DatabaseSync } = require('node:sqlite'); const crypto = require('crypto'); const DATA_DIR = path.join(__dirname, '..', 'data'); const DB_PATH = path.join(DATA_DIR, 'research.db'); const fs = require('fs'); fs.mkdirSync(DATA_DIR, { recursive: true }); const db = new DatabaseSync(DB_PATH); db.exec(` PRAGMA journal_mode = WAL; CREATE TABLE IF NOT EXISTS sessions ( id TEXT PRIMARY KEY, title TEXT DEFAULT 'Neue Recherche', owner_name TEXT NOT NULL, share_token TEXT UNIQUE, status TEXT DEFAULT 'idle', dashboard TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, role TEXT NOT NULL, content TEXT, tool_name TEXT, tool_call_id TEXT, tool_calls TEXT, created_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_messages_session ON messages(session_id); CREATE TABLE IF NOT EXISTS push_subscriptions ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_name TEXT NOT NULL, endpoint TEXT UNIQUE NOT NULL, keys TEXT NOT NULL, created_at TEXT NOT NULL ); `); const now = () => new Date().toISOString(); const uid = () => crypto.randomBytes(12).toString('hex'); const shareToken = () => crypto.randomBytes(16).toString('hex'); // DeepSeek V4.1 Flash Preise (USD pro 1M Tokens, Peak) – obere Schätzung. const PRICES = { inputCacheHit: 0.006, inputCacheMiss: 0.3, output: 1.2 }; function computeCost(u, prices = PRICES) { return ( ((u.cacheHitTokens || 0) * prices.inputCacheHit + (u.cacheMissTokens || 0) * prices.inputCacheMiss + (u.completionTokens || 0) * prices.output) / 1e6 ); } // Migration: Usage-Spalten ergänzen (falls Tabelle älter ist) { const cols = db.prepare(`PRAGMA table_info(sessions)`).all().map((c) => c.name); const add = (name, def) => { if (!cols.includes(name)) db.exec(`ALTER TABLE sessions ADD COLUMN ${name} ${def}`); }; add('prompt_tokens', 'INTEGER DEFAULT 0'); add('completion_tokens', 'INTEGER DEFAULT 0'); add('cache_hit_tokens', 'INTEGER DEFAULT 0'); add('cache_miss_tokens', 'INTEGER DEFAULT 0'); add('cost_estimate', 'REAL DEFAULT 0'); add('model', 'TEXT'); } // Migration: Messages-Spalten ergänzen (Reasoning-Roundtrip + Bild-Upload) { const cols = db.prepare(`PRAGMA table_info(messages)`).all().map((c) => c.name); const add = (name, def) => { if (!cols.includes(name)) db.exec(`ALTER TABLE messages ADD COLUMN ${name} ${def}`); }; add('reasoning', 'TEXT'); add('image', 'TEXT'); } // Migration: Legacy-/OpenRouter-Modell-IDs auf den aktuellen Namen umstellen db.prepare( `UPDATE sessions SET model = 'deepseek-flash' WHERE model IN ('deepseek-v4-flash', 'deepseek-v4-flash-vision-exp', 'z-ai/glm-5.3-flash')` ).run(); const stmts = { createSession: db.prepare( `INSERT INTO sessions (id, title, owner_name, share_token, status, model, created_at, updated_at) VALUES (?, ?, ?, ?, 'idle', ?, ?, ?)` ), getSession: db.prepare(`SELECT * FROM sessions WHERE id = ?`), getSessionByShare: db.prepare(`SELECT * FROM sessions WHERE share_token = ?`), listSessions: db.prepare(`SELECT * FROM sessions ORDER BY updated_at DESC`), updateSession: db.prepare( `UPDATE sessions SET title = ?, status = ?, dashboard = ?, updated_at = ? WHERE id = ?` ), setSessionStatus: db.prepare( `UPDATE sessions SET status = ?, updated_at = ? WHERE id = ?` ), setSessionTitle: db.prepare(`UPDATE sessions SET title = ?, updated_at = ? WHERE id = ?`), setSessionDashboard: db.prepare(`UPDATE sessions SET dashboard = ?, updated_at = ? WHERE id = ?`), setSessionModel: db.prepare(`UPDATE sessions SET model = ?, updated_at = ? WHERE id = ?`), addUsage: db.prepare( `UPDATE sessions SET prompt_tokens = prompt_tokens + ?, completion_tokens = completion_tokens + ?, cache_hit_tokens = cache_hit_tokens + ?, cache_miss_tokens = cache_miss_tokens + ?, cost_estimate = cost_estimate + ? WHERE id = ?` ), deleteSession: db.prepare(`DELETE FROM sessions WHERE id = ?`), insertMessage: db.prepare( `INSERT INTO messages (session_id, role, content, tool_name, tool_call_id, tool_calls, reasoning, image, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)` ), listMessages: db.prepare(`SELECT * FROM messages WHERE session_id = ? ORDER BY id ASC`), listStaleSessions: db.prepare(`SELECT id FROM sessions WHERE status IN ('running', 'queued')`), lastMessage: db.prepare(`SELECT * FROM messages WHERE session_id = ? ORDER BY id DESC LIMIT 1`), updateMessageContent: db.prepare(`UPDATE messages SET content = ? WHERE id = ?`), deleteMessages: db.prepare(`DELETE FROM messages WHERE session_id = ?`), insertPushSub: db.prepare( `INSERT INTO push_subscriptions (user_name, endpoint, keys, created_at) VALUES (?, ?, ?, ?)` ), deletePushSub: db.prepare(`DELETE FROM push_subscriptions WHERE endpoint = ?`), listPushSubs: db.prepare(`SELECT * FROM push_subscriptions`), }; function createSession(ownerName, title = 'Neue Recherche', model = null) { const id = uid(); stmts.createSession.run(id, title, ownerName, shareToken(), model, now(), now()); return stmts.getSession.get(id); } function createMessage(sessionId, { role, content, toolName, toolCallId, toolCalls, reasoning, image }) { stmts.insertMessage.run( sessionId, role, content ?? null, toolName ?? null, toolCallId ?? null, toolCalls ? JSON.stringify(toolCalls) : null, reasoning ?? null, image ?? null, now() ); return stmts.lastMessage.get(sessionId); } function getSessionMessages(sessionId) { return stmts.listMessages.all(sessionId).map((m) => ({ ...m, tool_calls: m.tool_calls ? JSON.parse(m.tool_calls) : null, })); } const API_HISTORY_MAX = 60; /** * Bereinigt den Verlauf für die API: entfernt unvollständige Tool-Gruppen * (assistant-Tool-Call ohne passende tool-Antwort, z.B. nach einem abgebrochenen * Job) und kürzt auf die letzten API_HISTORY_MAX Nachrichten, ohne jemals mitten * in einer Tool-Gruppe zu schneiden. Verwaiste tool-Nachrichten lehnt die * DeepSeek-API sonst mit 400 ab ("Messages with role 'tool' must be a response * to a preceding message with 'tool_calls'"). */ function sanitizeHistory(out, max = API_HISTORY_MAX) { const resolved = new Set(); const withCalls = []; const owner = new Array(out.length).fill(-1); let cur = -1; for (let i = 0; i < out.length; i++) { const m = out[i]; if (m.role === 'assistant' && m.tool_calls) { withCalls.push(i); cur = i; } else if (m.role === 'tool') { owner[i] = cur; if (m.tool_call_id) resolved.add(m.tool_call_id); } } const incomplete = new Set( withCalls.filter((i) => (out[i].tool_calls || []).some((tc) => !resolved.has(tc.id))) ); const kept = []; for (let i = 0; i < out.length; i++) { if (incomplete.has(i)) continue; if (out[i].role === 'tool' && incomplete.has(owner[i])) continue; kept.push(out[i]); } if (kept.length <= max) return kept; let start = kept.length - max; while (start < kept.length && kept[start].role === 'tool') start++; return kept.slice(start); } /** * Baut aus dem Chat-Verlauf das Messages-Array für die Modell-API. * @param {object} opts { provider, vision } – steuert Bild-Content (Vision-Modelle) * und den reasoning_content-Roundtrip (Pflicht bei DeepSeek mit tools). */ function toApiMessages(sessionId, opts = {}) { const { provider = 'deepseek', vision = false } = opts; const msgs = getSessionMessages(sessionId); const out = []; for (const m of msgs) { if (m.role === 'user') { const msg = { role: 'user' }; if (m.image && vision) { msg.content = [ { type: 'text', text: m.content || '' }, { type: 'image_url', image_url: { url: m.image } }, ]; } else { msg.content = m.content; } out.push(msg); } else if (m.role === 'assistant') { const msg = { role: 'assistant' }; if (m.tool_calls) msg.tool_calls = m.tool_calls; if (m.content) msg.content = m.content; else if (!m.tool_calls) msg.content = ''; // DeepSeek verlangt bei tools-Requests den Roundtrip des Reasoning-Texts if (provider === 'deepseek' && m.reasoning) msg.reasoning_content = m.reasoning; out.push(msg); } else if (m.role === 'tool') { out.push({ role: 'tool', tool_call_id: m.tool_call_id, content: m.content ?? '' }); } } return sanitizeHistory(out); } module.exports = { db, createSession, createMessage, getSession: stmts.getSession.get.bind(stmts.getSession), getSessionByShare: stmts.getSessionByShare.get.bind(stmts.getSessionByShare), listSessions: stmts.listSessions.all.bind(stmts.listSessions), setSessionStatus: (id, status) => stmts.setSessionStatus.run(status, now(), id), setSessionTitle: (id, title) => stmts.setSessionTitle.run(title, now(), id), setSessionDashboard: (id, dashboard) => stmts.setSessionDashboard.run(dashboard, now(), id), setSessionModel: (id, model) => stmts.setSessionModel.run(model, now(), id), setSessionUsage: (id, u, prices) => stmts.addUsage.run( u.promptTokens || 0, u.completionTokens || 0, u.cacheHitTokens || 0, u.cacheMissTokens || 0, computeCost(u, prices), id ), computeCost, deleteSession: (id) => { stmts.deleteMessages.run(id); stmts.deleteSession.run(id); }, /** * Setzt nach einem Server-Neustart alle Sessions zurück, die im alten * Prozess noch als 'running'/'queued' festhingen (Jobs existieren im * Speicher nicht mehr). Gibt die Anzahl zurück. */ markStaleJobs() { const stale = stmts.listStaleSessions.all(); for (const s of stale) { stmts.setSessionStatus.run('done', now(), s.id); stmts.insertMessage.run( s.id, 'assistant', '⛔ Recherche wurde durch einen Server-Neustart abgebrochen.', null, null, null, null, null, now() ); } return stale.length; }, getSessionMessages, toApiMessages, upsertPushSub(userName, endpoint, keys) { try { stmts.insertPushSub.run(userName, endpoint, JSON.stringify(keys), now()); } catch { stmts.deletePushSub.run(endpoint); stmts.insertPushSub.run(userName, endpoint, JSON.stringify(keys), now()); } }, removePushSub: stmts.deletePushSub.run.bind(stmts.deletePushSub), listPushSubs: stmts.listPushSubs.all.bind(stmts.listPushSubs), };