Aime/app/db.js

import mysql from 'mysql2/promise';

export const pool = mysql.createPool({
  host: 'localhost', user: 'callrec', password: 'callrec123',
  database: 'callrec', connectionLimit: 5, waitForConnections: true
});

// Idempotent migration: the LLM analysis columns (auto title lives in `title`).
// MariaDB supports ADD COLUMN IF NOT EXISTS.
export async function ensureSchema() {
  await pool.query(`ALTER TABLE recordings
    ADD COLUMN IF NOT EXISTS summary TEXT NULL,
    ADD COLUMN IF NOT EXISTS todos TEXT NULL,
    ADD COLUMN IF NOT EXISTS participants TEXT NULL,
    ADD COLUMN IF NOT EXISTS transcript_clean TEXT NULL,
    ADD COLUMN IF NOT EXISTS wav_file VARCHAR(255) NULL,
    ADD COLUMN IF NOT EXISTS speaker_names TEXT NULL,
    ADD COLUMN IF NOT EXISTS speaker_labels TEXT NULL,
    ADD COLUMN IF NOT EXISTS analysis_status VARCHAR(16) NOT NULL DEFAULT 'pending',
    ADD COLUMN IF NOT EXISTS analysis_started_at BIGINT NULL,
    ADD COLUMN IF NOT EXISTS analysis_ms INT NULL`);
  await pool.query(`ALTER TABLE words ADD COLUMN IF NOT EXISTS speaker VARCHAR(32) NULL`);
  await pool.query(`CREATE TABLE IF NOT EXISTS voiceprints (
    id INT AUTO_INCREMENT PRIMARY KEY,
    name VARCHAR(128) NOT NULL,
    dim INT NOT NULL,
    vec MEDIUMTEXT NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)`);
  // n = number of samples averaged into this voiceprint (running centroid).
  await pool.query(`ALTER TABLE voiceprints ADD COLUMN IF NOT EXISTS n INT NOT NULL DEFAULT 1`);
  // Embedding model the vector came from: prints of different models are NOT
  // comparable (cosine across models is noise), so identification only ever
  // uses the rows of the model currently loaded. NULL = pre-2026-07-27 TitaNet.
  await pool.query(`ALTER TABLE voiceprints ADD COLUMN IF NOT EXISTS model VARCHAR(64) NULL`);
  await pool.query(`UPDATE voiceprints SET model='nemo_titanet_small' WHERE model IS NULL`);
  // Domain glossary: known proper nouns / terms the ASR mis-hears (e.g. "Aime"
  // heard as "Amy"). Fed into the LLM correction step so soundalikes get mapped
  // to the correct spelling. `soundalikes` = optional comma-list of mis-hearings.
  await pool.query(`CREATE TABLE IF NOT EXISTS glossary (
    id INT AUTO_INCREMENT PRIMARY KEY,
    term VARCHAR(190) NOT NULL UNIQUE,
    soundalikes TEXT NULL,
    enabled TINYINT NOT NULL DEFAULT 1,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)`);
  // Live speaker markers: names typed DURING the recording ("who is talking
  // right now"). t_ms = offset from recording start. After diarization each
  // marker is resolved to the acoustic cluster that was speaking around t_ms
  // (`speaker`), which names that cluster — ground truth instead of guessing.
  await pool.query(`CREATE TABLE IF NOT EXISTS markers (
    id INT AUTO_INCREMENT PRIMARY KEY,
    recording_id INT NOT NULL,
    t_ms INT NOT NULL,
    name VARCHAR(128) NOT NULL,
    speaker VARCHAR(32) NULL,
    conf VARCHAR(8) NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    INDEX rec_idx (recording_id))`);
  // Trash instead of hard delete: `deleted_at` set = in the bin, rows and audio
  // files stay untouched until purgeExpired() passes TRASH_DAYS. Every listing
  // filters on it, so a binned recording is invisible everywhere else.
  // `title_locked` marks a title the USER typed — a later re-analysis must not
  // overwrite it with a fresh LLM guess (see updateAnalysis).
  await pool.query(`ALTER TABLE recordings
    ADD COLUMN IF NOT EXISTS deleted_at TIMESTAMP NULL DEFAULT NULL,
    ADD COLUMN IF NOT EXISTS title_locked TINYINT(1) NOT NULL DEFAULT 0,
    ADD COLUMN IF NOT EXISTS doc_vec MEDIUMTEXT NULL`);
  // Search index. One row per passage (a few sentences of ONE speaker, see
  // search.js/buildPassages), carrying the time window so a hit can name the
  // exact minute and the player can jump there. `vec` = optional text embedding
  // for semantic ranking; NULL when the embedding model is not installed, in
  // which case search degrades to full-text only.
  await pool.query(`CREATE TABLE IF NOT EXISTS passages (
    id INT AUTO_INCREMENT PRIMARY KEY,
    recording_id INT NOT NULL,
    idx INT NOT NULL,
    speaker VARCHAR(32) NULL,
    speaker_name VARCHAR(128) NULL,
    t_start_ms INT NOT NULL,
    t_end_ms INT NOT NULL,
    text TEXT NOT NULL,
    vec MEDIUMTEXT NULL,
    INDEX rec_idx (recording_id),
    FULLTEXT KEY ft_text (text)) ENGINE=InnoDB`);
}

// --- live speaker markers ---------------------------------------------------
export async function addMarkers(recId, markers) {
  const list = (markers || []).filter(m => m && m.name);
  if (!list.length) return;
  await pool.query('INSERT INTO markers (recording_id, t_ms, name) VALUES ?',
    [list.map(m => [recId, Math.max(0, Math.round(m.t_ms || 0)), String(m.name).slice(0, 128)])]);
}
export async function listMarkers(recId) {
  const [rows] = await pool.query(
    'SELECT id, t_ms, name, speaker, conf FROM markers WHERE recording_id=? ORDER BY t_ms', [recId]);
  return rows;
}
export async function setMarkerSpeaker(id, speaker, conf) {
  await pool.execute('UPDATE markers SET speaker=?, conf=? WHERE id=?', [speaker || null, conf || null, id]);
}
export async function deleteMarker(id) { await pool.execute('DELETE FROM markers WHERE id=?', [id]); }
export async function addMarker(recId, tMs, name) {
  const [res] = await pool.execute('INSERT INTO markers (recording_id, t_ms, name) VALUES (?,?,?)',
    [recId, Math.max(0, Math.round(tMs || 0)), String(name).slice(0, 128)]);
  return res.insertId;
}

export async function listGlossary() {
  const [rows] = await pool.query('SELECT id, term, soundalikes, enabled, created_at FROM glossary ORDER BY term');
  return rows.map(r => ({ ...r, enabled: !!r.enabled }));
}
export async function addGlossary(term, soundalikes) {
  await pool.execute(
    'INSERT INTO glossary (term, soundalikes) VALUES (?,?) ON DUPLICATE KEY UPDATE soundalikes=VALUES(soundalikes), enabled=1',
    [term, soundalikes || null]);
  const [rows] = await pool.execute('SELECT id, term, soundalikes, enabled FROM glossary WHERE term=?', [term]);
  return rows[0] ? { ...rows[0], enabled: !!rows[0].enabled } : null;
}
export async function deleteGlossary(id) {
  await pool.execute('DELETE FROM glossary WHERE id=?', [id]);
}
export async function updateGlossary(id, term, soundalikes) {
  await pool.execute('UPDATE glossary SET term=?, soundalikes=? WHERE id=?', [term, soundalikes || null, id]);
  const [rows] = await pool.execute('SELECT id, term, soundalikes, enabled FROM glossary WHERE id=?', [id]);
  return rows[0] ? { ...rows[0], enabled: !!rows[0].enabled } : null;
}

export async function addVoiceprint(name, vec, model) {
  const [res] = await pool.execute(
    'INSERT INTO voiceprints (name, dim, vec, model) VALUES (?,?,?,?)',
    [name, vec.length, JSON.stringify(vec), model || null]);
  return res.insertId;
}

// Add a sample to a named voiceprint, keeping ONE row per name as a running
// centroid (mean of all samples). This is how the app "learns who is who": every
// time a speaker is named in a recording, their voice is folded into the centroid,
// so future recordings match them more reliably. New name -> fresh row.
export async function upsertVoiceprint(name, vec, model) {
  const [rows] = await pool.query(
    'SELECT id, vec, n FROM voiceprints WHERE name=? AND (model=? OR (model IS NULL AND ? IS NULL)) ORDER BY id LIMIT 1',
    [name, model || null, model || null]);
  if (!rows.length) {
    const [res] = await pool.execute('INSERT INTO voiceprints (name, dim, vec, n, model) VALUES (?,?,?,1,?)',
      [name, vec.length, JSON.stringify(vec), model || null]);
    return res.insertId;
  }
  const old = rows[0];
  const on = old.n || 1;
  const ov = JSON.parse(old.vec);
  const mean = vec.map((x, i) => ((ov[i] || 0) * on + x) / (on + 1));
  await pool.execute('UPDATE voiceprints SET vec=?, n=?, dim=? WHERE id=?',
    [JSON.stringify(mean), on + 1, mean.length, old.id]);
  return old.id;
}
// model given → only the prints of that embedding model (what identification
// must use); omitted → all rows (GUI listing).
export async function listVoiceprints(model) {
  const [rows] = model
    ? await pool.query('SELECT id,name,dim,vec,model,created_at FROM voiceprints WHERE model=? ORDER BY name', [model])
    : await pool.query('SELECT id,name,dim,vec,model,created_at FROM voiceprints ORDER BY name');
  return rows.map(r => ({ ...r, vec: JSON.parse(r.vec) }));
}
export async function deleteVoiceprint(id) { await pool.query('DELETE FROM voiceprints WHERE id=?', [id]); }

export async function addRecording(r) {
  const [res] = await pool.execute(
    'INSERT INTO recordings (title, mp3_file, wav_file, device, duration_ms, transcript, language, status, analysis_status) VALUES (?,?,?,?,?,?,?,?,?)',
    [r.title, r.mp3_file, r.wav_file || null, r.device, r.duration_ms, r.transcript, r.language, r.status || 'done', 'pending']);
  return res.insertId;
}

export async function addWords(recId, words) {
  if (!words || !words.length) return;
  const values = words.map((w, i) => [recId, w.word, w.t_start_ms, w.t_end_ms, i, w.speaker || null]);
  await pool.query('INSERT INTO words (recording_id, word, t_start_ms, t_end_ms, idx, speaker) VALUES ?', [values]);
}

// Replace a recording's words wholesale (used when re-diarizing: the text is
// unchanged, only the speaker labels differ). Words are re-indexed by array order.
export async function replaceWords(recId, words) {
  await pool.query('DELETE FROM words WHERE recording_id=?', [recId]);
  await addWords(recId, words);
}

// Store the transcript + words after async (thermal-guarded) transcription and
// flip the row out of 'transcribing'. The wav + row were saved before whisper
// ran, so this just fills them in.
export async function setTranscript(id, transcript, words) {
  await pool.execute('UPDATE recordings SET transcript=?, status=? WHERE id=?', [transcript || '', 'done', id]);
  await replaceWords(id, words || []);
}

// Every in-flight job dies with the service: systemd SIGKILLs the whisper,
// spk_embed and rkllm children on stop, and nothing resumes them. The row
// however still reads 'pending' with a start stamp, so the GUI shows
// "KI-Analyse läuft…" and a timer counting up forever on work that no longer
// exists (observed: 2 h 34 min on a 29-minute recording after a deploy
// restart). Called once at boot — at that moment no job CAN legitimately be
// running — so the retry button appears instead of a phantom counter.
export async function failStaleJobs() {
  const [a] = await pool.query(
    "UPDATE recordings SET analysis_status='error' WHERE analysis_status='pending' AND analysis_started_at IS NOT NULL");
  const [t] = await pool.query(
    "UPDATE recordings SET status='error' WHERE status='transcribing'");
  return { analyses: a.affectedRows, transcriptions: t.affectedRows };
}

// Stamp the wall-clock start of a processing run (transcribe+diar+LLM). The
// GUI shows a live counter from here; updateAnalysis('done') freezes the total
// into analysis_ms. Called at every trigger (record-stop, retranscribe, reanalyze).
export async function markAnalysisStart(id) {
  await pool.execute('UPDATE recordings SET analysis_started_at=?, analysis_ms=NULL WHERE id=?', [Date.now(), id]);
}

// Update just the coarse recording status ('transcribing' | 'done' | 'error').
export async function setStatus(id, status) {
  await pool.execute('UPDATE recordings SET status=? WHERE id=?', [status, id]);
}

// Per-recording map { rawSpeaker -> displayName }. Manual edits (setSpeakerName)
// are authoritative; acoustic identification only fills blanks (mergeSpeakerNames).
async function getSpeakerNames(id) {
  const [rows] = await pool.query('SELECT speaker_names FROM recordings WHERE id=?', [id]);
  try { return JSON.parse((rows[0] && rows[0].speaker_names) || '{}') || {}; } catch { return {}; }
}
export async function setSpeakerName(id, speaker, name) {
  const m = await getSpeakerNames(id);
  if (name) m[speaker] = name; else delete m[speaker];   // empty name clears back to "Sprecher N"
  await pool.execute('UPDATE recordings SET speaker_names=? WHERE id=?', [JSON.stringify(m), id]);
}
// The label EVERY speaker carried in the text the LLM produced ({speaker_07:
// "Sprecher 5"}). Stored with the analysis so a later rename can swap the name
// into the finished summary/todos instead of re-running the 20-minute analysis.
export async function setSpeakerLabels(id, labels) {
  await pool.execute('UPDATE recordings SET speaker_labels=? WHERE id=?', [JSON.stringify(labels || {}), id]);
}
export async function getSpeakerLabels(id) {
  const [rows] = await pool.query('SELECT speaker_labels FROM recordings WHERE id=?', [id]);
  try { return JSON.parse((rows[0] && rows[0].speaker_labels) || '{}') || {}; } catch { return {}; }
}

// Seed names from acoustic identification WITHOUT clobbering existing (manual) ones.
export async function mergeSpeakerNames(id, map) {
  if (!map || !Object.keys(map).length) return;
  const m = await getSpeakerNames(id);
  let changed = false;
  for (const k of Object.keys(map)) if (!m[k] && map[k]) { m[k] = map[k]; changed = true; }
  if (changed) await pool.execute('UPDATE recordings SET speaker_names=? WHERE id=?', [JSON.stringify(m), id]);
}

// Store the LLM result. `todos` is an array → persisted as JSON text.
// A user-typed title (title_locked=1) survives every re-analysis: without the
// CASE the LLM would silently rename the recording back on the next run.
export async function updateAnalysis(id, { title, summary, todos, participants, transcriptClean, status }) {
  const st = status || 'done';
  await pool.execute(
    `UPDATE recordings SET title=CASE WHEN title_locked=1 THEN title ELSE COALESCE(?,title) END,
       summary=?, todos=?, participants=?, transcript_clean=?, analysis_status=?,
       analysis_ms=CASE WHEN ?='done' AND analysis_started_at IS NOT NULL THEN ?-analysis_started_at ELSE analysis_ms END
     WHERE id=?`,
    [title || null, summary || null, todos ? JSON.stringify(todos) : null,
     participants ? JSON.stringify(participants) : null, transcriptClean || null, st,
     st, Date.now(), id]);
}

// --- title rename ------------------------------------------------------------
// Empty/blank title unlocks again: the next analysis is then free to propose one.
export async function renameRecording(id, title) {
  const t = (title || '').trim().slice(0, 255);
  await pool.execute('UPDATE recordings SET title=?, title_locked=? WHERE id=?', [t || null, t ? 1 : 0, id]);
  const [rows] = await pool.query('SELECT id,title,title_locked FROM recordings WHERE id=?', [id]);
  return rows[0] || null;
}

function parseJsonCols(row) {
  for (const c of ['todos', 'participants']) {
    if (row && typeof row[c] === 'string') {
      try { row[c] = JSON.parse(row[c]); } catch { row[c] = []; }
    }
  }
  return row;
}
const parseTodos = parseJsonCols;

const LIST_COLS = 'id,title,mp3_file,device,duration_ms,language,status,summary,todos,participants,analysis_status,analysis_started_at,analysis_ms,created_at';

export async function listRecordings() {
  const [rows] = await pool.query(
    `SELECT ${LIST_COLS} FROM recordings WHERE deleted_at IS NULL ORDER BY id DESC`);
  return rows.map(parseTodos);
}

export async function getRecording(id) {
  const [recRows] = await pool.query('SELECT * FROM recordings WHERE id=?', [id]);
  const rec = recRows[0];
  if (!rec) return null;
  const [words] = await pool.query(
    'SELECT word,t_start_ms,t_end_ms,idx,speaker FROM words WHERE recording_id=? ORDER BY idx', [id]);
  let speaker_names = {};
  try { speaker_names = JSON.parse(rec.speaker_names || '{}') || {}; } catch {}
  const markers = await listMarkers(id).catch(() => []);
  return { ...parseTodos(rec), speaker_names, words, markers };
}

// --- trash -------------------------------------------------------------------
// Deleting only stamps `deleted_at`; nothing is destroyed. Everything (words,
// markers, passages, wav/mp3) stays put so a restore is complete, not partial.
export const TRASH_DAYS = +(process.env.AIME_TRASH_DAYS || 30);

export async function deleteRecording(id) {
  await pool.execute('UPDATE recordings SET deleted_at=CURRENT_TIMESTAMP WHERE id=? AND deleted_at IS NULL', [id]);
}

export async function restoreRecording(id) {
  await pool.execute('UPDATE recordings SET deleted_at=NULL WHERE id=?', [id]);
}

// Binned recordings, newest first, with the days left before auto-purge.
export async function listTrash() {
  const [rows] = await pool.query(
    `SELECT ${LIST_COLS}, deleted_at,
       GREATEST(0, ? - TIMESTAMPDIFF(DAY, deleted_at, CURRENT_TIMESTAMP)) AS days_left
     FROM recordings WHERE deleted_at IS NOT NULL ORDER BY deleted_at DESC`, [TRASH_DAYS]);
  return rows.map(parseTodos);
}

// Hard delete. Returns the audio file names so the caller can unlink them —
// db.js deliberately does no filesystem work. Note this also clears the `words`
// rows, which the pre-trash delete leaked (they outlived their recording).
export async function purgeRecording(id) {
  const [rows] = await pool.query('SELECT mp3_file, wav_file FROM recordings WHERE id=?', [id]);
  const files = rows[0] ? [rows[0].mp3_file, rows[0].wav_file].filter(Boolean) : [];
  for (const t of ['markers', 'words', 'passages'])
    await pool.query(`DELETE FROM ${t} WHERE recording_id=?`, [id]).catch(() => {});
  await pool.query('DELETE FROM recordings WHERE id=?', [id]);
  return files;
}

// Ids whose retention window has run out. The caller purges them one by one so
// a single bad row cannot abort the whole sweep.
export async function expiredTrashIds() {
  const [rows] = await pool.query(
    'SELECT id FROM recordings WHERE deleted_at IS NOT NULL AND deleted_at < (CURRENT_TIMESTAMP - INTERVAL ? DAY)',
    [TRASH_DAYS]);
  return rows.map(r => r.id);
}