memind/session-snapshot.mjs

/**
 * Session snapshot cache service.
 *
 * Stores a DB-side copy of each session's userVisible messages so that
 * repeated opens of historical chats skip the Goose round-trip entirely.
 *
 * Rollback: set SESSION_SNAPSHOT_CACHE_ENABLED=0 in env → all reads/writes
 * are no-ops; behaviour falls back to the existing Goose path unchanged.
 *
 * The table (h5_session_snapshots) is additive-only and has no FK deps on
 * conversation content, so it can be dropped/truncated safely at any time.
 */

export function createSessionSnapshotService(pool) {
  function isEnabled() {
    return process.env.SESSION_SNAPSHOT_CACHE_ENABLED !== '0';
  }

  /**
   * Persist a snapshot for a session.
   * @param {string} sessionId
   * @param {string} userId
   * @param {object} session  – raw Goose session object (includes message_count, updated_at, etc.)
   * @param {Array}  messages – already-normalized userVisible Message[]
   */
  async function save(sessionId, userId, session, messages) {
    if (!isEnabled() || !pool) return;
    try {
      const now = Date.now();
      const sessionMeta = {
        name: session.name ?? '',
        working_dir: session.working_dir ?? '',
        created_at_str: session.created_at ?? '',
        updated_at_str: session.updated_at ?? '',
        user_set_name: session.user_set_name ? 1 : 0,
        recipe_json: session.recipe != null ? JSON.stringify(session.recipe) : null,
      };
      await pool.query(
        `INSERT INTO h5_session_snapshots
           (agent_session_id, user_id, name, working_dir,
            created_at_str, updated_at_str, user_set_name, recipe_json,
            synced_msg_count, source_updated_at,
            messages_json, synced_at)
         VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
         ON DUPLICATE KEY UPDATE
           name = VALUES(name),
           working_dir = VALUES(working_dir),
           created_at_str = VALUES(created_at_str),
           updated_at_str = VALUES(updated_at_str),
           user_set_name = VALUES(user_set_name),
           recipe_json = VALUES(recipe_json),
           synced_msg_count = VALUES(synced_msg_count),
           source_updated_at = VALUES(source_updated_at),
           messages_json = VALUES(messages_json),
           synced_at = VALUES(synced_at)`,
        [
          sessionId,
          userId,
          sessionMeta.name,
          sessionMeta.working_dir,
          sessionMeta.created_at_str,
          sessionMeta.updated_at_str,
          sessionMeta.user_set_name,
          sessionMeta.recipe_json,
          session.message_count ?? 0,
          session.updated_at ?? '',
          JSON.stringify(messages),
          now,
        ],
      );
    } catch (err) {
      // Non-fatal: log and continue; caller falls back to Goose.
      console.warn('[snapshot] save failed:', err instanceof Error ? err.message : err);
    }
  }

  /**
   * Read a snapshot. Returns null if not found or cache is disabled.
   * The returned object reconstructs what `loadSessionDetail` returns.
   */
  async function get(sessionId) {
    if (!isEnabled() || !pool) return null;
    try {
      const [rows] = await pool.query(
        `SELECT agent_session_id, user_id, name, working_dir,
                created_at_str, updated_at_str, user_set_name, recipe_json,
                synced_msg_count, source_updated_at, messages_json, synced_at
         FROM h5_session_snapshots
         WHERE agent_session_id = ?
         LIMIT 1`,
        [sessionId],
      );
      if (!rows.length) return null;
      const row = rows[0];
      return {
        session: {
          id: sessionId,
          name: row.name,
          working_dir: row.working_dir,
          message_count: row.synced_msg_count,
          created_at: row.created_at_str || undefined,
          updated_at: row.updated_at_str || undefined,
          user_set_name: row.user_set_name === 1,
          recipe: row.recipe_json ? JSON.parse(row.recipe_json) : null,
          conversation: null,
        },
        messages: JSON.parse(row.messages_json),
        meta: {
          synced_msg_count: row.synced_msg_count,
          source_updated_at: row.source_updated_at,
          synced_at: row.synced_at,
        },
      };
    } catch (err) {
      console.warn('[snapshot] get failed:', err instanceof Error ? err.message : err);
      return null;
    }
  }

  /**
   * Delete a snapshot (called when the session itself is deleted).
   */
  async function remove(sessionId) {
    if (!pool) return;
    try {
      await pool.query(
        `DELETE FROM h5_session_snapshots WHERE agent_session_id = ?`,
        [sessionId],
      );
    } catch (err) {
      console.warn('[snapshot] remove failed:', err instanceof Error ? err.message : err);
    }
  }

  /**
   * Fetch full session from Goose and write/update the snapshot.
   * Fire-and-forget safe: errors are swallowed.
   */
  async function refresh(sessionId, userId, apiFetchFn) {
    if (!isEnabled() || !pool) return;
    try {
      const res = await apiFetchFn(`/sessions/${encodeURIComponent(sessionId)}`, {
        method: 'GET',
      });
      if (!res.ok) return;
      const gooseSession = await res.json();
      // Store raw userVisible messages — same as write-through path.
      const messages = (gooseSession.conversation ?? [])
        .filter((m) => m.metadata?.userVisible);
      await save(sessionId, userId, gooseSession, messages);
    } catch (err) {
      console.warn('[snapshot] refresh failed:', sessionId, err instanceof Error ? err.message : err);
    }
  }

  return { save, get, remove, refresh, isEnabled };
}