researchChat/lib/db.js

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 Flash Preise (USD pro 1M Tokens, Peak) – obere Schätzung.
// Andere Modelle (z.B. OpenRouter) legen eigene Preise in der Registry an.
const PRICES = { inputCacheHit: 0.014, inputCacheMiss: 0.44, output: 1.32 };

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');
}

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`),
  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);
  },
  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),
};