const {getDB} = require('../db'); const { EPISODIC, getEnv, SERVICES, parseRow, formatEpisodeText, SUMMARIES, logger } = require('@nexusai/shared'); const semantic = require('../semantic'); const { extractAndStoreEntities } = require('../entities/extraction') // Common English function words that add noise to keyword matching. Kept short // on purpose — FTS5's BM25 ranking already down-weights frequent terms; this is // mainly to stop an all-stopword query ("how do I do this") from matching everything. const FTS_STOPWORDS = new Set([ 'the','a','an','and','or','but','if','then','of','to','in','on','at','for', 'with','is','are','was','were','be','been','being','do','does','did','how', 'what','why','when','where','who','which','this','that','these','those', 'i','you','it','we','they','me','my','your','can','could','would','should', 'will','about','as','by','from','so','just','get','got','have','has','had', ]); // Turn a free-text message into an FTS5 MATCH expression. The message is tokenized // into significant terms, each wrapped in quotes (so any FTS5-significant token in // the message — e.g. a literal "OR" or "*" — is treated as a search term, not an // operator), and the terms are OR-joined for recall. Returns null when nothing // usable remains, so the caller can skip keyword search rather than run a // degenerate query. // // This replaces the previous behaviour of quoting the ENTIRE message as one phrase, // which required the whole message to appear verbatim and made keyword recall ~nil. function buildFtsQuery(query) { const tokens = query .toLowerCase() .split(/[^\p{L}\p{N}]+/u) // split on any non-letter/number, unicode-aware .filter(t => t.length >= 2 && !FTS_STOPWORDS.has(t)); if (tokens.length === 0) return null; return tokens.map(t => `"${t}"`).join(' OR '); } // --Sessions -------------------------------------------------- // Creates a new session with the given external ID and optional metadata function createSession(externalId, metadata = null) { const db = getDB(); const stmt = db.prepare(` INSERT INTO sessions (external_id, metadata) VALUES (?, ?) `); const result = stmt.run(externalId, metadata ? JSON.stringify(metadata) : null); return getSession(result.lastInsertRowid); } // Retrieves a session by its external ID function getSession(id) { const db = getDB(); const stmt = db.prepare(`SELECT * FROM sessions WHERE id = ?`); return parseRow(stmt.get(id)); } function getSessions(limit = EPISODIC.DEFAULT_PAGE_SIZE, offset = EPISODIC.DEFAULT_OFFSET, projectId = null) { const db = getDB(); const stmt = projectId ? db.prepare(` SELECT * FROM sessions WHERE project_id = ? ORDER BY updated_at DESC LIMIT ? OFFSET ? `) : db.prepare(` SELECT * FROM sessions ORDER BY updated_at DESC LIMIT ? OFFSET ? `); return projectId ? stmt.all(projectId, limit, offset).map(parseRow) : stmt.all(limit, offset).map(parseRow); } // Retrieves a session by its external ID function getSessionByExternalId(externalId) { const db = getDB(); const stmt = db.prepare(`SELECT * FROM sessions WHERE external_id = ?`); return parseRow(stmt.get(externalId)); } // Updates the updated_at timestamp of a session to the current time function touchSession(id) { const db = getDB(); db.prepare(`UPDATE sessions SET updated_at = unixepoch() WHERE id = ?`).run(id); } // Deletes a session and all associated episodes, entities, relationships, and summaries function deleteSession(id) { const db = getDB(); db.prepare(`DELETE FROM sessions WHERE id = ?`).run(id); //SQLite cascade removed episode rows, clean up QDrant vectors too // Fire-and-forget - payload-filter delete, so it doesn't depend on the rows existing semantic.deleteEpisodesBySession(id) .catch(err => logger.error(`[Memory] QDrant cleanup failed for session ${id}:`, err.message)); } function updateSession(id, { name, projectId } = {}) { const db = getDB(); // Build update dynamically based on what was provided const updates = []; const values = []; if (name !== undefined) { updates.push('name = ?'); values.push(name ?? null); } if (projectId !== undefined) { updates.push('project_id = ?'); values.push(projectId ?? null); } if (updates.length === 0) return getSession(id); updates.push('updated_at = unixepoch()'); values.push(id); db.prepare(`UPDATE sessions SET ${updates.join(', ')} WHERE id = ?`).run(...values); return getSession(id); } function updateSessionByExternalId(externalId, fields) { const session = getSessionByExternalId(externalId); if (!session) throw new Error('Session not found'); return updateSession(session.id, fields); } function deleteSessionByExternalId(externalId) { const session = getSessionByExternalId(externalId); if(!session) throw new Error('Session not found'); deleteSession(session.id); } // --Episodes -------------------------------------------------- // Creates a new episode linked to a session, with user message, AI response, optional token count, and metadata async function createEpisode(sessionId, userMessage, aiResponse, tokenCount = null, projectId=null) { const db = getDB(); // Wrap insert + session touch in a transaction — both succeed or neither does const insert = db.transaction(() => { const stmt = db.prepare(` INSERT INTO episodes (session_id, user_message, ai_response, token_count) VALUES (?, ?, ?, ?) `); const result = stmt.run( sessionId, userMessage, aiResponse, tokenCount, ); touchSession(sessionId); return getEpisode(result.lastInsertRowid); }); const episode = insert(); //embed ascynchronously after SQLite completes, non-blocking. If embedding fail, the episode still saved. getEpisodeEmbedding(userMessage, aiResponse) .then(vector => semantic.upsertEpisode(episode.id, vector, { sessionId: episode.session_id, createdAt: episode.created_at })) .catch(err => logger.error(`Failed to embed episode ${episode.id}:`, err.message)); extractAndStoreEntities(userMessage, aiResponse, episode.id, projectId) .catch(err => logger.error(`Failed to extract entities for episode ${episode.id}:`, err.message)); return episode; } // Retrieves an episode by its ID function getEpisode(id) { const db = getDB(); const stmt = db.prepare(`SELECT * FROM episodes WHERE id = ?`); return parseRow(stmt.get(id)); } // Retrieves episodes for a given session, ordered by creation time descending, with pagination function getEpisodesBySession(sessionId, limit = EPISODIC.DEFAULT_PAGE_SIZE, offset = EPISODIC.DEFAULT_OFFSET) { const db = getDB(); const stmt = db.prepare(` SELECT * FROM episodes WHERE session_id = ? ORDER BY created_at DESC LIMIT ? OFFSET ? `); return stmt.all(sessionId, limit, offset).map(parseRow); } // Retrieves recent episodes across all sessions, ordered by creation time descending, with a limit function getRecentEpisodes(sessionId, limit = EPISODIC.DEFAULT_RECENT_LIMIT) { // Cross-session recent episodes — useful for recency-based retrieval const db = getDB(); const stmt = db.prepare(` SELECT * FROM episodes WHERE session_id = ? ORDER BY created_at DESC LIMIT ? `); return stmt.all(sessionId, limit).map(parseRow); } // Aggregate stats for a session — count, total tokens, and highest episode id. // Lets summarization check the token threshold without pulling every episode row. function getSessionEpisodeStats(sessionId) { const db = getDB(); return db.prepare(` SELECT COUNT(*) AS count, COALESCE(SUM(token_count), 0) AS totalTokens, COALESCE(MAX(id), 0) AS maxId FROM episodes WHERE session_id = ? `).get(sessionId); } // Episodes newer than a given id, in chronological order. Summarization uses // this to fetch only the un-summarized tail rather than the whole session. function getEpisodesSince(sessionId, afterId) { const db = getDB(); return db.prepare(` SELECT * FROM episodes WHERE session_id = ? AND id > ? ORDER BY id ASC `).all(sessionId, afterId).map(parseRow); } // Searches episodes using FTS5 full-text search, ordered by relevance, with a limit function searchEpisodes(query, limit = EPISODIC.DEFAULT_SEARCH_LIMIT, sessionIds = null) { const db = getDB(); const match = buildFtsQuery(query); if (!match) return []; // no usable keyword terms — let semantic retrieval carry this turn if (sessionIds && sessionIds.length > 0) { const ph = sessionIds.map(() => '?').join(','); return db.prepare(` SELECT e.* FROM episodes e JOIN episodes_fts fts ON e.id = fts.rowid WHERE episodes_fts MATCH ? AND e.session_id IN (${ph}) ORDER BY rank LIMIT ? `).all(match, ...sessionIds, limit).map(parseRow); } return db.prepare(` SELECT e.* FROM episodes e JOIN episodes_fts fts ON e.id = fts.rowid WHERE episodes_fts MATCH ? ORDER BY rank LIMIT ? `).all(match, limit).map(parseRow); } // Deletes an episode by its ID function deleteEpisode(id) { const db = getDB(); db.prepare(`DELETE FROM episodes WHERE id = ?`).run(id); } /******** Embedding Helper ********/ async function getEpisodeEmbedding(userMessage, aiResponse){ const url = getEnv('EMBEDDING_SERVICE_URL', SERVICES.EMBEDDING_URL); //Combine user message and AI response for embedding const text = formatEpisodeText(userMessage, aiResponse); const res = await fetch(`${url}/embed`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ text }), signal: AbortSignal.timeout(30_000), }) if (!res.ok) { throw new Error(`Embedding service error: ${res.status} ${res.statusText}`); } const data = await res.json(); return data.embedding; } function getEpisodesByProject(projectId, limit = SUMMARIES.MAX_PROJECT_EPISODE_LIMIT) { const db = getDB(); return db.prepare(` SELECT e.* FROM episodes e JOIN sessions s ON s.id = e.session_id WHERE s.project_id = ? ORDER BY e.created_at ASC LIMIT ? `).all(projectId, limit).map(parseRow); } module.exports = { createSession, getSession, getSessions, getSessionByExternalId, deleteSession, updateSession, updateSessionByExternalId, deleteSessionByExternalId, createEpisode, getEpisode, getEpisodesBySession, getRecentEpisodes, getSessionEpisodeStats, getEpisodesSince, searchEpisodes, deleteEpisode, getEpisodesByProject };