use anyhow::{Context, Result};
use chrono::Utc;
use serde::{Deserialize, Serialize};
use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
use sqlx::{Row, SqlitePool};
use std::path::PathBuf;
use std::str::FromStr;
use crate::web::github::GitHubIssue;
#[derive(Clone)]
pub struct SessionDb {
pool: SqlitePool,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Session {
pub id: String,
pub project: String,
pub machine: String,
pub agent: String,
pub first_message: Option<String>,
pub display_name: Option<String>,
pub started_at: Option<String>,
pub ended_at: Option<String>,
pub message_count: i64,
pub user_message_count: i64,
pub parent_session_id: Option<String>,
pub relationship_type: String,
pub total_output_tokens: i64,
pub peak_context_tokens: i64,
pub deleted_at: Option<String>,
pub created_at: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SessionFileInfo {
pub file_path: Option<String>,
pub file_size: Option<i64>,
pub file_mtime: Option<i64>,
pub file_hash: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Message {
pub id: i64,
pub session_id: String,
pub ordinal: i64,
pub role: String,
pub content: String,
pub timestamp: Option<String>,
pub has_thinking: bool,
pub has_tool_use: bool,
pub content_length: i64,
pub is_system: bool,
pub model: String,
pub token_usage: String,
pub context_tokens: i64,
pub output_tokens: i64,
#[serde(skip_deserializing)]
pub tool_calls: Vec<ToolCall>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ToolCall {
pub tool_name: String,
pub category: String,
pub tool_use_id: Option<String>,
pub input_json: Option<String>,
pub skill_name: Option<String>,
pub result_content_length: Option<i64>,
pub result_content: Option<String>,
pub subagent_session_id: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MinimapEntry {
pub ordinal: i64,
pub role: String,
pub content_length: i64,
pub has_thinking: bool,
pub has_tool_use: bool,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SearchResult {
pub session_id: String,
pub project: String,
pub agent: String,
pub name: String,
pub ordinal: i64,
pub session_ended_at: Option<String>,
pub snippet: String,
pub rank: f64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Pin {
pub id: i64,
pub session_id: String,
pub message_id: i64,
pub ordinal: i64,
pub note: Option<String>,
pub created_at: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub content: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub role: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub session_project: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub session_agent: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub session_display_name: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub session_first_message: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PathNote {
pub id: String,
pub path: String,
pub content: String,
pub created_at: String,
pub updated_at: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Stats {
pub session_count: i64,
pub message_count: i64,
pub project_count: i64,
pub machine_count: i64,
pub earliest_session: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ProjectInfo {
pub name: String,
pub session_count: i64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AgentInfo {
pub name: String,
pub session_count: i64,
}
#[derive(Debug, Default, Clone)]
pub struct SessionFilter {
pub project: Option<String>,
pub exclude_project: Option<String>,
pub machine: Option<Vec<String>>,
pub agent: Option<Vec<String>>,
pub date: Option<String>,
pub date_from: Option<String>,
pub date_to: Option<String>,
pub active_since: Option<String>,
pub min_messages: Option<i64>,
pub max_messages: Option<i64>,
pub min_user_messages: Option<i64>,
pub include_one_shot: bool,
pub include_children: bool,
}
impl SessionDb {
pub async fn open() -> Result<Self> {
let db_path = Self::db_path()?;
if let Some(parent) = db_path.parent() {
std::fs::create_dir_all(parent)
.with_context(|| format!("failed to create {}", parent.display()))?;
}
let opts = SqliteConnectOptions::from_str(&format!("sqlite:{}", db_path.display()))?
.create_if_missing(true)
.journal_mode(sqlx::sqlite::SqliteJournalMode::Wal)
.busy_timeout(std::time::Duration::from_secs(5))
.foreign_keys(true)
.pragma("mmap_size", "268435456") .pragma("cache_size", "-64000") .synchronous(sqlx::sqlite::SqliteSynchronous::Normal);
let pool = SqlitePoolOptions::new()
.max_connections(5)
.connect_with(opts)
.await
.with_context(|| format!("failed to open {}", db_path.display()))?;
let db = Self { pool };
db.migrate().await?;
Ok(db)
}
fn db_path() -> Result<PathBuf> {
let cwd = std::env::current_dir().context("cannot determine current directory")?;
Ok(cwd.join(".wsb").join("sessions.db"))
}
async fn migrate(&self) -> Result<()> {
for sql_block in &[SCHEMA_SQL, FTS_SQL] {
for stmt in split_sql_script(sql_block) {
sqlx::query(&stmt)
.execute(&self.pool)
.await
.with_context(|| format!("migration failed at: {}", sql_preview(&stmt)))?;
}
}
self.migrate_github_cache().await?;
self.migrate_path_notes().await?;
Ok(())
}
async fn migrate_path_notes(&self) -> Result<()> {
sqlx::query(
r#"CREATE TABLE IF NOT EXISTS path_notes (
id TEXT PRIMARY KEY,
path TEXT NOT NULL,
content TEXT NOT NULL,
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
)"#,
)
.execute(&self.pool)
.await
.context("failed to create path_notes table")?;
sqlx::query(
"CREATE INDEX IF NOT EXISTS idx_path_notes_path ON path_notes(path)",
)
.execute(&self.pool)
.await
.context("failed to create idx_path_notes_path")?;
Ok(())
}
async fn migrate_github_cache(&self) -> Result<()> {
for stmt in [
r#"
CREATE TABLE IF NOT EXISTS github_issue_cache (
repo TEXT NOT NULL,
number INTEGER NOT NULL,
state TEXT NOT NULL,
payload_json TEXT NOT NULL,
fetched_at TEXT NOT NULL,
PRIMARY KEY (repo, number)
)
"#,
r#"
CREATE TABLE IF NOT EXISTS github_repo_cache_state (
repo TEXT PRIMARY KEY,
fetched_at TEXT NOT NULL,
last_error TEXT
)
"#,
] {
sqlx::query(stmt)
.execute(&self.pool)
.await
.with_context(|| format!("migration failed at: {}", sql_preview(stmt)))?;
}
Ok(())
}
pub fn pool(&self) -> &SqlitePool {
&self.pool
}
pub async fn cache_github_repo_issues(&self, repo: &str, issues: &[GitHubIssue]) -> Result<()> {
let mut tx = self.pool.begin().await?;
let fetched_at = Utc::now().to_rfc3339();
for issue in issues {
let payload_json = serde_json::to_string(issue)
.with_context(|| format!("failed to serialize GitHub issue {}#{}", repo, issue.number))?;
sqlx::query(
r#"
INSERT INTO github_issue_cache (repo, number, state, payload_json, fetched_at)
VALUES (?, ?, ?, ?, ?)
ON CONFLICT(repo, number) DO UPDATE SET
state = excluded.state,
payload_json = excluded.payload_json,
fetched_at = excluded.fetched_at
"#,
)
.bind(repo)
.bind(issue.number)
.bind(&issue.state)
.bind(payload_json)
.bind(&fetched_at)
.execute(&mut *tx)
.await?;
}
sqlx::query(
r#"
INSERT INTO github_repo_cache_state (repo, fetched_at, last_error)
VALUES (?, ?, NULL)
ON CONFLICT(repo) DO UPDATE SET
fetched_at = excluded.fetched_at,
last_error = NULL
"#,
)
.bind(repo)
.bind(&fetched_at)
.execute(&mut *tx)
.await?;
tx.commit().await?;
Ok(())
}
pub async fn cache_github_repo_failure(&self, repo: &str, error: &str) -> Result<()> {
sqlx::query(
r#"
INSERT INTO github_repo_cache_state (repo, fetched_at, last_error)
VALUES (?, ?, ?)
ON CONFLICT(repo) DO UPDATE SET
fetched_at = excluded.fetched_at,
last_error = excluded.last_error
"#,
)
.bind(repo)
.bind(Utc::now().to_rfc3339())
.bind(error)
.execute(&self.pool)
.await?;
Ok(())
}
pub async fn upsert_session(&self, s: &SessionInsert) -> Result<()> {
sqlx::query(
r#"INSERT INTO sessions (
id, project, machine, agent, first_message, started_at, ended_at,
message_count, user_message_count, file_path, file_size, file_mtime,
file_hash, parent_session_id, relationship_type,
total_output_tokens, peak_context_tokens
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(id) DO UPDATE SET
project = excluded.project,
machine = excluded.machine,
agent = excluded.agent,
first_message = COALESCE(excluded.first_message, sessions.first_message),
started_at = COALESCE(excluded.started_at, sessions.started_at),
ended_at = COALESCE(excluded.ended_at, sessions.ended_at),
message_count = excluded.message_count,
user_message_count = excluded.user_message_count,
file_path = excluded.file_path,
file_size = excluded.file_size,
file_mtime = excluded.file_mtime,
file_hash = COALESCE(excluded.file_hash, sessions.file_hash),
parent_session_id = COALESCE(excluded.parent_session_id, sessions.parent_session_id),
relationship_type = CASE
WHEN excluded.relationship_type != '' THEN excluded.relationship_type
ELSE sessions.relationship_type
END,
total_output_tokens = excluded.total_output_tokens,
peak_context_tokens = excluded.peak_context_tokens
"#,
)
.bind(&s.id)
.bind(&s.project)
.bind(&s.machine)
.bind(&s.agent)
.bind(&s.first_message)
.bind(&s.started_at)
.bind(&s.ended_at)
.bind(s.message_count)
.bind(s.user_message_count)
.bind(&s.file_path)
.bind(s.file_size)
.bind(s.file_mtime)
.bind(&s.file_hash)
.bind(&s.parent_session_id)
.bind(&s.relationship_type)
.bind(s.total_output_tokens)
.bind(s.peak_context_tokens)
.execute(&self.pool)
.await?;
Ok(())
}
pub async fn is_excluded(&self, id: &str) -> Result<bool> {
let row = sqlx::query("SELECT COUNT(*) as cnt FROM excluded_sessions WHERE id = ?")
.bind(id)
.fetch_one(&self.pool)
.await?;
Ok(row.get::<i64, _>("cnt") > 0)
}
pub async fn get_session_file_info(&self, id: &str) -> Result<Option<(i64, i64)>> {
let row = sqlx::query(
"SELECT file_size, file_mtime FROM sessions WHERE id = ? AND file_size IS NOT NULL",
)
.bind(id)
.fetch_optional(&self.pool)
.await?;
Ok(row.map(|r| (r.get("file_size"), r.get("file_mtime"))))
}
pub async fn get_file_info_by_path(&self, path: &str) -> Result<Option<(i64, i64)>> {
let row = sqlx::query(
"SELECT file_size, file_mtime FROM sessions WHERE file_path = ? AND file_size IS NOT NULL LIMIT 1",
)
.bind(path)
.fetch_optional(&self.pool)
.await?;
Ok(row.map(|r| (r.get("file_size"), r.get("file_mtime"))))
}
pub async fn count_sessions_for_path(&self, path: &str) -> Result<i64> {
let row = sqlx::query("SELECT COUNT(*) as cnt FROM sessions WHERE file_path = ?")
.bind(path)
.fetch_one(&self.pool)
.await?;
Ok(row.get("cnt"))
}
pub async fn max_ordinal(&self, session_id: &str) -> Result<Option<i64>> {
let row = sqlx::query("SELECT MAX(ordinal) as max_ord FROM messages WHERE session_id = ?")
.bind(session_id)
.fetch_one(&self.pool)
.await?;
Ok(row.get("max_ord"))
}
pub async fn insert_messages(&self, messages: &[MessageInsert]) -> Result<()> {
for msg in messages {
let result = sqlx::query(
r#"INSERT INTO messages (
session_id, ordinal, role, content, timestamp,
has_thinking, has_tool_use, content_length, is_system,
model, token_usage, context_tokens, output_tokens
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(session_id, ordinal) DO UPDATE SET
content = excluded.content,
timestamp = excluded.timestamp,
has_thinking = excluded.has_thinking,
has_tool_use = excluded.has_tool_use,
content_length = excluded.content_length,
model = excluded.model,
token_usage = excluded.token_usage,
context_tokens = excluded.context_tokens,
output_tokens = excluded.output_tokens
"#,
)
.bind(&msg.session_id)
.bind(msg.ordinal)
.bind(&msg.role)
.bind(&msg.content)
.bind(&msg.timestamp)
.bind(msg.has_thinking)
.bind(msg.has_tool_use)
.bind(msg.content_length)
.bind(msg.is_system)
.bind(&msg.model)
.bind(&msg.token_usage)
.bind(msg.context_tokens)
.bind(msg.output_tokens)
.execute(&self.pool)
.await?;
let message_id = if result.rows_affected() > 0 {
let row =
sqlx::query("SELECT id FROM messages WHERE session_id = ? AND ordinal = ?")
.bind(&msg.session_id)
.bind(msg.ordinal)
.fetch_one(&self.pool)
.await?;
row.get::<i64, _>("id")
} else {
continue;
};
for tc in &msg.tool_calls {
sqlx::query(
r#"INSERT OR IGNORE INTO tool_calls (
message_id, session_id, tool_name, category,
tool_use_id, input_json, skill_name,
result_content_length, result_content, subagent_session_id
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"#,
)
.bind(message_id)
.bind(&msg.session_id)
.bind(&tc.tool_name)
.bind(&tc.category)
.bind(&tc.tool_use_id)
.bind(&tc.input_json)
.bind(&tc.skill_name)
.bind(tc.result_content_length)
.bind(&tc.result_content)
.bind(&tc.subagent_session_id)
.execute(&self.pool)
.await?;
}
}
Ok(())
}
pub async fn update_session_incremental(
&self,
id: &str,
ended_at: Option<&str>,
message_count: i64,
user_message_count: i64,
file_size: i64,
file_mtime: i64,
total_output_tokens: i64,
peak_context_tokens: i64,
) -> Result<()> {
sqlx::query(
r#"UPDATE sessions SET
ended_at = COALESCE(?, ended_at),
message_count = ?,
user_message_count = ?,
file_size = ?,
file_mtime = ?,
total_output_tokens = ?,
peak_context_tokens = ?
WHERE id = ?"#,
)
.bind(ended_at)
.bind(message_count)
.bind(user_message_count)
.bind(file_size)
.bind(file_mtime)
.bind(total_output_tokens)
.bind(peak_context_tokens)
.bind(id)
.execute(&self.pool)
.await?;
Ok(())
}
pub async fn get_session_user_count(&self, session_id: &str) -> Result<i64> {
let row = sqlx::query("SELECT user_message_count FROM sessions WHERE id = ?")
.bind(session_id)
.fetch_optional(&self.pool)
.await?;
match row {
Some(r) => Ok(r.get("user_message_count")),
None => Ok(0),
}
}
pub async fn list_sessions(
&self,
filter: &SessionFilter,
limit: i64,
cursor: Option<(&str, &str)>, ) -> Result<Vec<Session>> {
let mut sql = String::from(
"SELECT id, project, machine, agent, first_message, display_name, \
started_at, ended_at, message_count, user_message_count, \
parent_session_id, relationship_type, total_output_tokens, \
peak_context_tokens, deleted_at, created_at \
FROM sessions WHERE deleted_at IS NULL",
);
let mut binds: Vec<String> = Vec::new();
if !filter.include_children {
sql.push_str(" AND (parent_session_id IS NULL OR parent_session_id = '')");
}
if !filter.include_one_shot {
sql.push_str(" AND user_message_count > 1");
}
if let Some(ref p) = filter.project {
sql.push_str(" AND project = ?");
binds.push(p.clone());
}
if let Some(ref ep) = filter.exclude_project {
sql.push_str(" AND project != ?");
binds.push(ep.clone());
}
if let Some(ref machines) = filter.machine {
let placeholders: Vec<&str> = machines.iter().map(|_| "?").collect();
sql.push_str(&format!(" AND machine IN ({})", placeholders.join(",")));
binds.extend(machines.iter().cloned());
}
if let Some(ref agents) = filter.agent {
let placeholders: Vec<&str> = agents.iter().map(|_| "?").collect();
sql.push_str(&format!(" AND agent IN ({})", placeholders.join(",")));
binds.extend(agents.iter().cloned());
}
if let Some(ref d) = filter.date {
sql.push_str(" AND date(ended_at) = ?");
binds.push(d.clone());
}
if let Some(ref df) = filter.date_from {
sql.push_str(" AND date(ended_at) >= ?");
binds.push(df.clone());
}
if let Some(ref dt) = filter.date_to {
sql.push_str(" AND date(ended_at) <= ?");
binds.push(dt.clone());
}
if let Some(ref since) = filter.active_since {
sql.push_str(" AND ended_at >= ?");
binds.push(since.clone());
}
if let Some(min) = filter.min_messages {
sql.push_str(&format!(" AND message_count >= {}", min));
}
if let Some(max) = filter.max_messages {
sql.push_str(&format!(" AND message_count <= {}", max));
}
if let Some(min) = filter.min_user_messages {
sql.push_str(&format!(" AND user_message_count >= {}", min));
}
if let Some((ended, id)) = cursor {
sql.push_str(" AND (ended_at, id) < (?, ?)");
binds.push(ended.to_string());
binds.push(id.to_string());
}
sql.push_str(" ORDER BY ended_at DESC, id DESC");
sql.push_str(&format!(" LIMIT {}", limit.min(500)));
let mut query = sqlx::query(&sql);
for b in &binds {
query = query.bind(b);
}
let rows = query.fetch_all(&self.pool).await?;
let sessions = rows
.iter()
.map(|r| Session {
id: r.get("id"),
project: r.get("project"),
machine: r.get("machine"),
agent: r.get("agent"),
first_message: r.get("first_message"),
display_name: r.get("display_name"),
started_at: r.get("started_at"),
ended_at: r.get("ended_at"),
message_count: r.get("message_count"),
user_message_count: r.get("user_message_count"),
parent_session_id: r.get("parent_session_id"),
relationship_type: r.get("relationship_type"),
total_output_tokens: r.get("total_output_tokens"),
peak_context_tokens: r.get("peak_context_tokens"),
deleted_at: r.get("deleted_at"),
created_at: r.get("created_at"),
})
.collect();
Ok(sessions)
}
pub async fn get_session(&self, id: &str) -> Result<Option<Session>> {
let row = sqlx::query(
"SELECT id, project, machine, agent, first_message, display_name, \
started_at, ended_at, message_count, user_message_count, \
parent_session_id, relationship_type, total_output_tokens, \
peak_context_tokens, deleted_at, created_at \
FROM sessions WHERE id = ? AND deleted_at IS NULL",
)
.bind(id)
.fetch_optional(&self.pool)
.await?;
Ok(row.map(|r| Session {
id: r.get("id"),
project: r.get("project"),
machine: r.get("machine"),
agent: r.get("agent"),
first_message: r.get("first_message"),
display_name: r.get("display_name"),
started_at: r.get("started_at"),
ended_at: r.get("ended_at"),
message_count: r.get("message_count"),
user_message_count: r.get("user_message_count"),
parent_session_id: r.get("parent_session_id"),
relationship_type: r.get("relationship_type"),
total_output_tokens: r.get("total_output_tokens"),
peak_context_tokens: r.get("peak_context_tokens"),
deleted_at: r.get("deleted_at"),
created_at: r.get("created_at"),
}))
}
pub async fn get_messages(
&self,
session_id: &str,
limit: i64,
from_ordinal: i64,
desc: bool,
) -> Result<Vec<Message>> {
let dir = if desc { "DESC" } else { "ASC" };
let cmp = if desc { "<=" } else { ">=" };
let sql = format!(
"SELECT id, session_id, ordinal, role, content, timestamp, \
has_thinking, has_tool_use, content_length, is_system, \
model, token_usage, context_tokens, output_tokens \
FROM messages WHERE session_id = ? AND ordinal {} ? \
ORDER BY ordinal {} LIMIT ?",
cmp, dir
);
let rows = sqlx::query(&sql)
.bind(session_id)
.bind(from_ordinal)
.bind(limit.min(1000))
.fetch_all(&self.pool)
.await?;
let mut messages: Vec<Message> = rows
.iter()
.map(|r| Message {
id: r.get("id"),
session_id: r.get("session_id"),
ordinal: r.get("ordinal"),
role: r.get("role"),
content: r.get("content"),
timestamp: r.get("timestamp"),
has_thinking: r.get::<bool, _>("has_thinking"),
has_tool_use: r.get::<bool, _>("has_tool_use"),
content_length: r.get("content_length"),
is_system: r.get::<bool, _>("is_system"),
model: r.get("model"),
token_usage: r.get("token_usage"),
context_tokens: r.get("context_tokens"),
output_tokens: r.get("output_tokens"),
tool_calls: Vec::new(),
})
.collect();
if !messages.is_empty() {
let msg_ids: Vec<i64> = messages.iter().map(|m| m.id).collect();
let placeholders: Vec<String> = msg_ids.iter().map(|id| id.to_string()).collect();
let tc_sql = format!(
"SELECT message_id, tool_name, category, tool_use_id, input_json, \
skill_name, result_content_length, result_content, subagent_session_id \
FROM tool_calls WHERE message_id IN ({})",
placeholders.join(",")
);
let tc_rows = sqlx::query(&tc_sql).fetch_all(&self.pool).await?;
let mut tc_map: std::collections::HashMap<i64, Vec<ToolCall>> =
std::collections::HashMap::new();
for r in &tc_rows {
let mid: i64 = r.get("message_id");
tc_map.entry(mid).or_default().push(ToolCall {
tool_name: r.get("tool_name"),
category: r.get("category"),
tool_use_id: r.get("tool_use_id"),
input_json: r.get("input_json"),
skill_name: r.get("skill_name"),
result_content_length: r.get("result_content_length"),
result_content: r.get("result_content"),
subagent_session_id: r.get("subagent_session_id"),
});
}
for msg in &mut messages {
if let Some(tcs) = tc_map.remove(&msg.id) {
msg.tool_calls = tcs;
}
}
}
Ok(messages)
}
pub async fn get_children(&self, parent_id: &str) -> Result<Vec<Session>> {
let rows = sqlx::query(
"SELECT id, project, machine, agent, first_message, display_name, \
started_at, ended_at, message_count, user_message_count, \
parent_session_id, relationship_type, total_output_tokens, \
peak_context_tokens, deleted_at, created_at \
FROM sessions WHERE parent_session_id = ? AND deleted_at IS NULL \
ORDER BY started_at ASC",
)
.bind(parent_id)
.fetch_all(&self.pool)
.await?;
Ok(rows
.iter()
.map(|r| Session {
id: r.get("id"),
project: r.get("project"),
machine: r.get("machine"),
agent: r.get("agent"),
first_message: r.get("first_message"),
display_name: r.get("display_name"),
started_at: r.get("started_at"),
ended_at: r.get("ended_at"),
message_count: r.get("message_count"),
user_message_count: r.get("user_message_count"),
parent_session_id: r.get("parent_session_id"),
relationship_type: r.get("relationship_type"),
total_output_tokens: r.get("total_output_tokens"),
peak_context_tokens: r.get("peak_context_tokens"),
deleted_at: r.get("deleted_at"),
created_at: r.get("created_at"),
})
.collect())
}
pub async fn get_minimap(
&self,
session_id: &str,
from: i64,
max: Option<i64>,
) -> Result<Vec<MinimapEntry>> {
let rows = sqlx::query(
"SELECT ordinal, role, content_length, has_thinking, has_tool_use \
FROM messages WHERE session_id = ? AND ordinal >= ? \
ORDER BY ordinal ASC",
)
.bind(session_id)
.bind(from)
.fetch_all(&self.pool)
.await?;
let mut entries: Vec<MinimapEntry> = rows
.iter()
.map(|r| MinimapEntry {
ordinal: r.get("ordinal"),
role: r.get("role"),
content_length: r.get("content_length"),
has_thinking: r.get("has_thinking"),
has_tool_use: r.get("has_tool_use"),
})
.collect();
if let Some(max_entries) = max {
if entries.len() as i64 > max_entries && max_entries > 0 {
let step = entries.len() as f64 / max_entries as f64;
let sampled: Vec<MinimapEntry> = (0..max_entries as usize)
.map(|i| entries[(i as f64 * step) as usize].clone())
.collect();
entries = sampled;
}
}
Ok(entries)
}
pub async fn search_session(&self, session_id: &str, query: &str) -> Result<Vec<i64>> {
let pattern = format!("%{}%", query);
let rows = sqlx::query(
"SELECT DISTINCT ordinal FROM ( \
SELECT ordinal FROM messages WHERE session_id = ? AND content LIKE ? \
UNION \
SELECT m.ordinal FROM tool_calls tc \
JOIN messages m ON m.id = tc.message_id \
WHERE tc.session_id = ? AND tc.result_content LIKE ? \
) ORDER BY ordinal ASC",
)
.bind(session_id)
.bind(&pattern)
.bind(session_id)
.bind(&pattern)
.fetch_all(&self.pool)
.await?;
Ok(rows.iter().map(|r| r.get("ordinal")).collect())
}
pub async fn global_search(
&self,
query: &str,
limit: i64,
offset: i64,
sort: &str,
project: Option<&str>,
) -> Result<Vec<SearchResult>> {
let fts_query = if query.contains(' ') && !query.starts_with('"') {
format!("\"{}\"", query)
} else {
query.to_string()
};
let order = if sort == "recency" {
"s.ended_at DESC"
} else {
"fts.rank"
};
let mut sql = format!(
"SELECT m.session_id, s.project, s.agent, \
COALESCE(s.display_name, s.first_message) as name, \
m.ordinal, s.ended_at as session_ended_at, \
snippet(messages_fts, 0, '<mark>', '</mark>', '...', 40) as snippet, \
fts.rank \
FROM messages_fts fts \
JOIN messages m ON m.id = fts.rowid \
JOIN sessions s ON s.id = m.session_id \
WHERE messages_fts MATCH ? AND s.deleted_at IS NULL"
);
let mut binds: Vec<String> = vec![fts_query];
if let Some(p) = project {
sql.push_str(" AND s.project = ?");
binds.push(p.to_string());
}
sql.push_str(&format!(
" ORDER BY {} LIMIT {} OFFSET {}",
order,
limit.min(500),
offset
));
let mut q = sqlx::query(&sql);
for b in &binds {
q = q.bind(b);
}
let rows = q.fetch_all(&self.pool).await?;
Ok(rows
.iter()
.map(|r| SearchResult {
session_id: r.get("session_id"),
project: r.get("project"),
agent: r.get("agent"),
name: r.get("name"),
ordinal: r.get("ordinal"),
session_ended_at: r.get("session_ended_at"),
snippet: r.get("snippet"),
rank: r.get("rank"),
})
.collect())
}
pub async fn rename_session(&self, id: &str, display_name: Option<&str>) -> Result<bool> {
let result =
sqlx::query("UPDATE sessions SET display_name = ?, local_modified_at = ? WHERE id = ?")
.bind(display_name)
.bind(Utc::now().to_rfc3339())
.bind(id)
.execute(&self.pool)
.await?;
Ok(result.rows_affected() > 0)
}
pub async fn soft_delete_session(&self, id: &str) -> Result<bool> {
let result =
sqlx::query("UPDATE sessions SET deleted_at = ? WHERE id = ? AND deleted_at IS NULL")
.bind(Utc::now().to_rfc3339())
.bind(id)
.execute(&self.pool)
.await?;
Ok(result.rows_affected() > 0)
}
pub async fn restore_session(&self, id: &str) -> Result<bool> {
let result = sqlx::query(
"UPDATE sessions SET deleted_at = NULL WHERE id = ? AND deleted_at IS NOT NULL",
)
.bind(id)
.execute(&self.pool)
.await?;
Ok(result.rows_affected() > 0)
}
pub async fn permanent_delete(&self, id: &str) -> Result<bool> {
let row = sqlx::query("SELECT deleted_at FROM sessions WHERE id = ?")
.bind(id)
.fetch_optional(&self.pool)
.await?;
match row {
Some(r) => {
let deleted_at: Option<String> = r.get("deleted_at");
if deleted_at.is_none() {
return Ok(false); }
}
None => return Ok(false),
}
sqlx::query("INSERT OR IGNORE INTO excluded_sessions (id) VALUES (?)")
.bind(id)
.execute(&self.pool)
.await?;
sqlx::query("DELETE FROM sessions WHERE id = ?")
.bind(id)
.execute(&self.pool)
.await?;
Ok(true)
}
pub async fn list_trash(&self) -> Result<Vec<Session>> {
let rows = sqlx::query(
"SELECT id, project, machine, agent, first_message, display_name, \
started_at, ended_at, message_count, user_message_count, \
parent_session_id, relationship_type, total_output_tokens, \
peak_context_tokens, deleted_at, created_at \
FROM sessions WHERE deleted_at IS NOT NULL \
ORDER BY deleted_at DESC LIMIT 500",
)
.fetch_all(&self.pool)
.await?;
Ok(rows
.iter()
.map(|r| Session {
id: r.get("id"),
project: r.get("project"),
machine: r.get("machine"),
agent: r.get("agent"),
first_message: r.get("first_message"),
display_name: r.get("display_name"),
started_at: r.get("started_at"),
ended_at: r.get("ended_at"),
message_count: r.get("message_count"),
user_message_count: r.get("user_message_count"),
parent_session_id: r.get("parent_session_id"),
relationship_type: r.get("relationship_type"),
total_output_tokens: r.get("total_output_tokens"),
peak_context_tokens: r.get("peak_context_tokens"),
deleted_at: r.get("deleted_at"),
created_at: r.get("created_at"),
})
.collect())
}
pub async fn empty_trash(&self) -> Result<i64> {
let rows = sqlx::query("SELECT id FROM sessions WHERE deleted_at IS NOT NULL")
.fetch_all(&self.pool)
.await?;
let count = rows.len() as i64;
for r in &rows {
let id: String = r.get("id");
sqlx::query("INSERT OR IGNORE INTO excluded_sessions (id) VALUES (?)")
.bind(&id)
.execute(&self.pool)
.await?;
}
sqlx::query("DELETE FROM sessions WHERE deleted_at IS NOT NULL")
.execute(&self.pool)
.await?;
Ok(count)
}
pub async fn list_starred(&self) -> Result<Vec<String>> {
let rows = sqlx::query("SELECT session_id FROM starred_sessions ORDER BY created_at DESC")
.fetch_all(&self.pool)
.await?;
Ok(rows.iter().map(|r| r.get("session_id")).collect())
}
pub async fn star_session(&self, id: &str) -> Result<()> {
sqlx::query("INSERT OR IGNORE INTO starred_sessions (session_id) VALUES (?)")
.bind(id)
.execute(&self.pool)
.await?;
Ok(())
}
pub async fn unstar_session(&self, id: &str) -> Result<()> {
sqlx::query("DELETE FROM starred_sessions WHERE session_id = ?")
.bind(id)
.execute(&self.pool)
.await?;
Ok(())
}
pub async fn bulk_star(&self, ids: &[String]) -> Result<()> {
for id in ids {
sqlx::query("INSERT OR IGNORE INTO starred_sessions (session_id) VALUES (?)")
.bind(id)
.execute(&self.pool)
.await?;
}
Ok(())
}
pub async fn is_starred(&self, id: &str) -> Result<bool> {
let row = sqlx::query("SELECT COUNT(*) as cnt FROM starred_sessions WHERE session_id = ?")
.bind(id)
.fetch_one(&self.pool)
.await?;
Ok(row.get::<i64, _>("cnt") > 0)
}
pub async fn list_all_pins(&self) -> Result<Vec<Pin>> {
let rows = sqlx::query(
"SELECT p.id, p.session_id, p.message_id, p.ordinal, p.note, p.created_at, \
m.content, m.role, \
s.project as session_project, s.agent as session_agent, \
s.display_name as session_display_name, s.first_message as session_first_message \
FROM pinned_messages p \
LEFT JOIN messages m ON m.id = p.message_id \
LEFT JOIN sessions s ON s.id = p.session_id \
ORDER BY p.created_at DESC LIMIT 500",
)
.fetch_all(&self.pool)
.await?;
Ok(rows
.iter()
.map(|r| Pin {
id: r.get("id"),
session_id: r.get("session_id"),
message_id: r.get("message_id"),
ordinal: r.get("ordinal"),
note: r.get("note"),
created_at: r.get("created_at"),
content: r.get("content"),
role: r.get("role"),
session_project: r.get("session_project"),
session_agent: r.get("session_agent"),
session_display_name: r.get("session_display_name"),
session_first_message: r.get("session_first_message"),
})
.collect())
}
pub async fn list_session_pins(&self, session_id: &str) -> Result<Vec<Pin>> {
let rows = sqlx::query(
"SELECT id, session_id, message_id, ordinal, note, created_at \
FROM pinned_messages WHERE session_id = ? ORDER BY ordinal ASC",
)
.bind(session_id)
.fetch_all(&self.pool)
.await?;
Ok(rows
.iter()
.map(|r| Pin {
id: r.get("id"),
session_id: r.get("session_id"),
message_id: r.get("message_id"),
ordinal: r.get("ordinal"),
note: r.get("note"),
created_at: r.get("created_at"),
content: None,
role: None,
session_project: None,
session_agent: None,
session_display_name: None,
session_first_message: None,
})
.collect())
}
pub async fn pin_message(
&self,
session_id: &str,
message_id: i64,
note: Option<&str>,
) -> Result<i64> {
let row = sqlx::query("SELECT ordinal FROM messages WHERE id = ? AND session_id = ?")
.bind(message_id)
.bind(session_id)
.fetch_optional(&self.pool)
.await?;
let ordinal: i64 = match row {
Some(r) => r.get("ordinal"),
None => anyhow::bail!("message not found in session"),
};
sqlx::query(
"INSERT INTO pinned_messages (session_id, message_id, ordinal, note) \
VALUES (?, ?, ?, ?) \
ON CONFLICT(session_id, message_id) DO UPDATE SET note = excluded.note",
)
.bind(session_id)
.bind(message_id)
.bind(ordinal)
.bind(note)
.execute(&self.pool)
.await?;
let pin_row =
sqlx::query("SELECT id FROM pinned_messages WHERE session_id = ? AND message_id = ?")
.bind(session_id)
.bind(message_id)
.fetch_one(&self.pool)
.await?;
Ok(pin_row.get("id"))
}
pub async fn unpin_message(&self, session_id: &str, message_id: i64) -> Result<()> {
sqlx::query("DELETE FROM pinned_messages WHERE session_id = ? AND message_id = ?")
.bind(session_id)
.bind(message_id)
.execute(&self.pool)
.await?;
Ok(())
}
pub async fn get_stats(&self, include_one_shot: bool) -> Result<Stats> {
let one_shot_filter = if include_one_shot {
""
} else {
" AND user_message_count > 1"
};
let sql = format!(
"SELECT \
(SELECT COUNT(*) FROM sessions WHERE deleted_at IS NULL{}) as session_count, \
(SELECT value FROM stats WHERE key = 'message_count') as message_count, \
(SELECT COUNT(DISTINCT project) FROM sessions WHERE deleted_at IS NULL{}) as project_count, \
(SELECT COUNT(DISTINCT machine) FROM sessions WHERE deleted_at IS NULL{}) as machine_count, \
(SELECT MIN(started_at) FROM sessions WHERE deleted_at IS NULL{}) as earliest_session",
one_shot_filter, one_shot_filter, one_shot_filter, one_shot_filter
);
let row = sqlx::query(&sql).fetch_one(&self.pool).await?;
Ok(Stats {
session_count: row.get("session_count"),
message_count: row.get("message_count"),
project_count: row.get("project_count"),
machine_count: row.get("machine_count"),
earliest_session: row.get("earliest_session"),
})
}
pub async fn get_overview_stats(&self) -> Result<std::collections::HashMap<String, i64>> {
let rows = sqlx::query("SELECT key, value FROM stats")
.fetch_all(&self.pool)
.await?;
let mut map = std::collections::HashMap::new();
for row in &rows {
let key: String = row.get("key");
let value: i64 = row.get("value");
map.insert(key, value);
}
Ok(map)
}
pub async fn set_stat(&self, key: &str, value: i64) -> Result<()> {
sqlx::query("INSERT OR REPLACE INTO stats (key, value) VALUES (?, ?)")
.bind(key)
.bind(value)
.execute(&self.pool)
.await?;
Ok(())
}
pub async fn list_projects(&self, include_one_shot: bool) -> Result<Vec<ProjectInfo>> {
let filter = if include_one_shot {
""
} else {
" AND user_message_count > 1"
};
let sql = format!(
"SELECT project as name, COUNT(*) as session_count \
FROM sessions WHERE deleted_at IS NULL{} \
GROUP BY project ORDER BY session_count DESC",
filter
);
let rows = sqlx::query(&sql).fetch_all(&self.pool).await?;
Ok(rows
.iter()
.map(|r| ProjectInfo {
name: r.get("name"),
session_count: r.get("session_count"),
})
.collect())
}
pub async fn list_machines(&self, include_one_shot: bool) -> Result<Vec<String>> {
let filter = if include_one_shot {
""
} else {
" AND user_message_count > 1"
};
let sql = format!(
"SELECT DISTINCT machine FROM sessions WHERE deleted_at IS NULL{} ORDER BY machine",
filter
);
let rows = sqlx::query(&sql).fetch_all(&self.pool).await?;
Ok(rows.iter().map(|r| r.get("machine")).collect())
}
pub async fn list_agents(&self, include_one_shot: bool) -> Result<Vec<AgentInfo>> {
let filter = if include_one_shot {
""
} else {
" AND user_message_count > 1"
};
let sql = format!(
"SELECT agent as name, COUNT(*) as session_count \
FROM sessions WHERE deleted_at IS NULL{} \
GROUP BY agent ORDER BY session_count DESC",
filter
);
let rows = sqlx::query(&sql).fetch_all(&self.pool).await?;
Ok(rows
.iter()
.map(|r| AgentInfo {
name: r.get("name"),
session_count: r.get("session_count"),
})
.collect())
}
pub async fn get_session_version(&self, id: &str) -> Result<Option<(i64, i64)>> {
let row = sqlx::query(
"SELECT message_count, COALESCE(file_mtime, 0) as file_mtime FROM sessions WHERE id = ?",
)
.bind(id)
.fetch_optional(&self.pool)
.await?;
Ok(row.map(|r| (r.get("message_count"), r.get("file_mtime"))))
}
pub async fn load_skip_cache(&self) -> Result<std::collections::HashMap<String, i64>> {
let rows = sqlx::query("SELECT file_path, file_mtime FROM skipped_files")
.fetch_all(&self.pool)
.await?;
let mut cache = std::collections::HashMap::new();
for r in &rows {
let path: String = r.get("file_path");
let mtime: i64 = r.get("file_mtime");
cache.insert(path, mtime);
}
Ok(cache)
}
pub async fn persist_skip_cache(
&self,
cache: &std::collections::HashMap<String, i64>,
) -> Result<()> {
sqlx::query("DELETE FROM skipped_files")
.execute(&self.pool)
.await?;
for (path, mtime) in cache {
sqlx::query("INSERT INTO skipped_files (file_path, file_mtime) VALUES (?, ?)")
.bind(path)
.bind(mtime)
.execute(&self.pool)
.await?;
}
Ok(())
}
pub async fn delete_skipped_file(&self, path: &str) -> Result<()> {
sqlx::query("DELETE FROM skipped_files WHERE file_path = ?")
.bind(path)
.execute(&self.pool)
.await?;
Ok(())
}
pub async fn link_subagent_sessions(&self) -> Result<()> {
sqlx::query(
"UPDATE sessions SET \
parent_session_id = ( \
SELECT tc.session_id FROM tool_calls tc \
WHERE tc.subagent_session_id = sessions.id \
LIMIT 1 \
), \
relationship_type = 'subagent' \
WHERE id IN ( \
SELECT DISTINCT tc.subagent_session_id FROM tool_calls tc \
WHERE tc.subagent_session_id IS NOT NULL \
) AND (parent_session_id IS NULL OR parent_session_id = '')",
)
.execute(&self.pool)
.await?;
Ok(())
}
pub async fn list_path_notes(&self, path: &str) -> Result<Vec<PathNote>> {
let rows = sqlx::query(
"SELECT id, path, content, created_at, updated_at \
FROM path_notes WHERE path = ? ORDER BY created_at ASC",
)
.bind(path)
.fetch_all(&self.pool)
.await?;
rows.into_iter()
.map(|r| {
Ok(PathNote {
id: r.get("id"),
path: r.get("path"),
content: r.get("content"),
created_at: r.get("created_at"),
updated_at: r.get("updated_at"),
})
})
.collect()
}
pub async fn create_path_note(&self, path: &str, content: &str) -> Result<PathNote> {
let id = uuid::Uuid::new_v4().to_string();
let now = Utc::now().format("%Y-%m-%dT%H:%M:%S%.3fZ").to_string();
sqlx::query(
"INSERT INTO path_notes (id, path, content, created_at, updated_at) \
VALUES (?, ?, ?, ?, ?)",
)
.bind(&id)
.bind(path)
.bind(content)
.bind(&now)
.bind(&now)
.execute(&self.pool)
.await?;
Ok(PathNote {
id,
path: path.to_string(),
content: content.to_string(),
created_at: now.clone(),
updated_at: now,
})
}
pub async fn update_path_note(&self, id: &str, content: &str) -> Result<Option<PathNote>> {
let now = Utc::now().format("%Y-%m-%dT%H:%M:%S%.3fZ").to_string();
let rows_affected = sqlx::query(
"UPDATE path_notes SET content = ?, updated_at = ? WHERE id = ?",
)
.bind(content)
.bind(&now)
.bind(id)
.execute(&self.pool)
.await?
.rows_affected();
if rows_affected == 0 {
return Ok(None);
}
let row = sqlx::query(
"SELECT id, path, content, created_at, updated_at FROM path_notes WHERE id = ?",
)
.bind(id)
.fetch_one(&self.pool)
.await?;
Ok(Some(PathNote {
id: row.get("id"),
path: row.get("path"),
content: row.get("content"),
created_at: row.get("created_at"),
updated_at: row.get("updated_at"),
}))
}
pub async fn delete_path_note(&self, id: &str) -> Result<bool> {
let rows_affected = sqlx::query("DELETE FROM path_notes WHERE id = ?")
.bind(id)
.execute(&self.pool)
.await?
.rows_affected();
Ok(rows_affected > 0)
}
}
#[derive(Debug, Clone)]
pub struct SessionInsert {
pub id: String,
pub project: String,
pub machine: String,
pub agent: String,
pub first_message: Option<String>,
pub started_at: Option<String>,
pub ended_at: Option<String>,
pub message_count: i64,
pub user_message_count: i64,
pub file_path: Option<String>,
pub file_size: Option<i64>,
pub file_mtime: Option<i64>,
pub file_hash: Option<String>,
pub parent_session_id: Option<String>,
pub relationship_type: String,
pub total_output_tokens: i64,
pub peak_context_tokens: i64,
}
#[derive(Debug, Clone)]
pub struct MessageInsert {
pub session_id: String,
pub ordinal: i64,
pub role: String,
pub content: String,
pub timestamp: Option<String>,
pub has_thinking: bool,
pub has_tool_use: bool,
pub content_length: i64,
pub is_system: bool,
pub model: String,
pub token_usage: String,
pub context_tokens: i64,
pub output_tokens: i64,
pub tool_calls: Vec<ToolCallInsert>,
}
#[derive(Debug, Clone)]
pub struct ToolCallInsert {
pub tool_name: String,
pub category: String,
pub tool_use_id: Option<String>,
pub input_json: Option<String>,
pub skill_name: Option<String>,
pub result_content_length: Option<i64>,
pub result_content: Option<String>,
pub subagent_session_id: Option<String>,
}
const SCHEMA_SQL: &str = r#"
CREATE TABLE IF NOT EXISTS sessions (
id TEXT PRIMARY KEY,
project TEXT NOT NULL,
machine TEXT NOT NULL DEFAULT 'local',
agent TEXT NOT NULL DEFAULT 'claude',
first_message TEXT,
display_name TEXT,
started_at TEXT,
ended_at TEXT,
message_count INTEGER NOT NULL DEFAULT 0,
user_message_count INTEGER NOT NULL DEFAULT 0,
file_path TEXT,
file_size INTEGER,
file_mtime INTEGER,
file_hash TEXT,
local_modified_at TEXT,
parent_session_id TEXT,
relationship_type TEXT NOT NULL DEFAULT '',
total_output_tokens INTEGER NOT NULL DEFAULT 0,
peak_context_tokens INTEGER NOT NULL DEFAULT 0,
deleted_at TEXT,
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
);
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY,
session_id TEXT NOT NULL REFERENCES sessions(id) ON DELETE CASCADE,
ordinal INTEGER NOT NULL,
role TEXT NOT NULL,
content TEXT NOT NULL,
timestamp TEXT,
has_thinking INTEGER NOT NULL DEFAULT 0,
has_tool_use INTEGER NOT NULL DEFAULT 0,
content_length INTEGER NOT NULL DEFAULT 0,
is_system INTEGER NOT NULL DEFAULT 0,
model TEXT NOT NULL DEFAULT '',
token_usage TEXT NOT NULL DEFAULT '',
context_tokens INTEGER NOT NULL DEFAULT 0,
output_tokens INTEGER NOT NULL DEFAULT 0,
UNIQUE(session_id, ordinal)
);
CREATE TABLE IF NOT EXISTS tool_calls (
id INTEGER PRIMARY KEY,
message_id INTEGER NOT NULL REFERENCES messages(id) ON DELETE CASCADE,
session_id TEXT NOT NULL REFERENCES sessions(id) ON DELETE CASCADE,
tool_name TEXT NOT NULL,
category TEXT NOT NULL,
tool_use_id TEXT,
input_json TEXT,
skill_name TEXT,
result_content_length INTEGER,
result_content TEXT,
subagent_session_id TEXT
);
CREATE TABLE IF NOT EXISTS stats (
key TEXT PRIMARY KEY,
value INTEGER NOT NULL DEFAULT 0
);
INSERT OR IGNORE INTO stats (key, value) VALUES ('session_count', 0);
INSERT OR IGNORE INTO stats (key, value) VALUES ('message_count', 0);
-- Stats triggers
CREATE TRIGGER IF NOT EXISTS sessions_insert_stats AFTER INSERT ON sessions BEGIN
UPDATE stats SET value = value + 1 WHERE key = 'session_count';
END;
CREATE TRIGGER IF NOT EXISTS sessions_delete_stats AFTER DELETE ON sessions BEGIN
UPDATE stats SET value = value - 1 WHERE key = 'session_count';
END;
CREATE TRIGGER IF NOT EXISTS messages_insert_stats AFTER INSERT ON messages BEGIN
UPDATE stats SET value = value + 1 WHERE key = 'message_count';
END;
CREATE TRIGGER IF NOT EXISTS messages_delete_stats AFTER DELETE ON messages BEGIN
UPDATE stats SET value = value - 1 WHERE key = 'message_count';
END;
-- Indexes
CREATE INDEX IF NOT EXISTS idx_sessions_ended ON sessions(ended_at DESC, id);
CREATE INDEX IF NOT EXISTS idx_sessions_project ON sessions(project);
CREATE INDEX IF NOT EXISTS idx_sessions_machine ON sessions(machine);
CREATE INDEX IF NOT EXISTS idx_sessions_agent ON sessions(agent);
CREATE INDEX IF NOT EXISTS idx_sessions_started ON sessions(started_at);
CREATE INDEX IF NOT EXISTS idx_sessions_message_count ON sessions(message_count);
CREATE INDEX IF NOT EXISTS idx_sessions_user_message_count ON sessions(user_message_count);
CREATE INDEX IF NOT EXISTS idx_sessions_parent ON sessions(parent_session_id)
WHERE parent_session_id IS NOT NULL;
CREATE INDEX IF NOT EXISTS idx_sessions_file_path ON sessions(file_path)
WHERE file_path IS NOT NULL;
CREATE INDEX IF NOT EXISTS idx_messages_session_ordinal ON messages(session_id, ordinal);
CREATE INDEX IF NOT EXISTS idx_messages_session_role ON messages(session_id, role);
CREATE INDEX IF NOT EXISTS idx_tool_calls_session ON tool_calls(session_id);
CREATE INDEX IF NOT EXISTS idx_tool_calls_category ON tool_calls(category);
CREATE INDEX IF NOT EXISTS idx_tool_calls_skill ON tool_calls(skill_name)
WHERE skill_name IS NOT NULL;
CREATE INDEX IF NOT EXISTS idx_tool_calls_subagent ON tool_calls(subagent_session_id)
WHERE subagent_session_id IS NOT NULL;
CREATE TABLE IF NOT EXISTS pinned_messages (
id INTEGER PRIMARY KEY,
session_id TEXT NOT NULL REFERENCES sessions(id) ON DELETE CASCADE,
message_id INTEGER NOT NULL REFERENCES messages(id) ON DELETE CASCADE,
ordinal INTEGER NOT NULL,
note TEXT,
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
UNIQUE(session_id, message_id)
);
CREATE INDEX IF NOT EXISTS idx_pinned_session ON pinned_messages(session_id);
CREATE INDEX IF NOT EXISTS idx_pinned_created ON pinned_messages(created_at DESC);
CREATE TABLE IF NOT EXISTS starred_sessions (
session_id TEXT PRIMARY KEY REFERENCES sessions(id) ON DELETE CASCADE,
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
);
CREATE TABLE IF NOT EXISTS excluded_sessions (
id TEXT PRIMARY KEY,
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
);
CREATE TABLE IF NOT EXISTS skipped_files (
file_path TEXT PRIMARY KEY,
file_mtime INTEGER NOT NULL
);
CREATE TABLE IF NOT EXISTS insights (
id INTEGER PRIMARY KEY,
type TEXT NOT NULL,
date_from TEXT NOT NULL,
date_to TEXT NOT NULL,
project TEXT,
agent TEXT NOT NULL,
model TEXT,
prompt TEXT,
content TEXT NOT NULL,
created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now'))
);
CREATE INDEX IF NOT EXISTS idx_insights_lookup ON insights(type, date_from, project);
CREATE INDEX IF NOT EXISTS idx_insights_created ON insights(created_at DESC);
"#;
const FTS_SQL: &str = r#"
CREATE VIRTUAL TABLE IF NOT EXISTS messages_fts USING fts5(
content,
content='messages',
content_rowid='id',
tokenize='porter unicode61'
);
CREATE TRIGGER IF NOT EXISTS messages_ai AFTER INSERT ON messages BEGIN
INSERT INTO messages_fts(rowid, content) VALUES (new.id, new.content);
END;
CREATE TRIGGER IF NOT EXISTS messages_ad AFTER DELETE ON messages BEGIN
INSERT INTO messages_fts(messages_fts, rowid, content)
VALUES('delete', old.id, old.content);
END;
CREATE TRIGGER IF NOT EXISTS messages_au AFTER UPDATE ON messages BEGIN
INSERT INTO messages_fts(messages_fts, rowid, content)
VALUES('delete', old.id, old.content);
INSERT INTO messages_fts(rowid, content) VALUES (new.id, new.content);
END;
"#;
fn split_sql_script(script: &str) -> Vec<String> {
let mut statements = Vec::new();
let mut current = String::new();
let mut in_trigger = false;
for line in script.lines() {
let trimmed = line.trim();
if trimmed.is_empty() || trimmed.starts_with("--") {
continue;
}
if !current.is_empty() {
current.push('\n');
}
current.push_str(trimmed);
let upper = trimmed.to_ascii_uppercase();
if upper.starts_with("CREATE TRIGGER") {
in_trigger = true;
}
let statement_complete = if in_trigger {
upper == "END;"
} else {
trimmed.ends_with(';')
};
if statement_complete {
statements.push(current.trim().to_string());
current.clear();
in_trigger = false;
}
}
if !current.trim().is_empty() {
statements.push(current.trim().to_string());
}
statements
}
fn sql_preview(stmt: &str) -> String {
let normalized = stmt.split_whitespace().collect::<Vec<_>>().join(" ");
if normalized.len() <= 120 {
normalized
} else {
format!("{}...", &normalized[..120])
}
}
#[cfg(test)]
mod tests {
use super::{split_sql_script, FTS_SQL, SCHEMA_SQL};
use sqlx::{
sqlite::{SqliteConnectOptions, SqlitePoolOptions},
Row,
};
use std::str::FromStr;
use tempfile::TempDir;
#[test]
fn split_sql_script_keeps_trigger_bodies_intact() {
let schema_statements = split_sql_script(SCHEMA_SQL);
let fts_statements = split_sql_script(FTS_SQL);
assert!(schema_statements.iter().any(|stmt| {
stmt.contains("CREATE TRIGGER IF NOT EXISTS sessions_insert_stats")
&& stmt.contains("UPDATE stats SET value = value + 1 WHERE key = 'session_count';")
&& stmt.ends_with("END;")
}));
assert!(fts_statements.iter().any(|stmt| {
stmt.contains("CREATE TRIGGER IF NOT EXISTS messages_au")
&& stmt.contains("VALUES('delete', old.id, old.content);")
&& stmt.contains("VALUES (new.id, new.content);")
&& stmt.ends_with("END;")
}));
}
#[tokio::test]
async fn migration_statements_execute_and_triggers_work() {
let temp_dir = TempDir::new().unwrap();
let db_path = temp_dir.path().join("sessions.db");
let connect_options =
SqliteConnectOptions::from_str(&format!("sqlite:{}", db_path.display()))
.unwrap()
.create_if_missing(true);
let pool = SqlitePoolOptions::new()
.max_connections(1)
.connect_with(connect_options)
.await
.unwrap();
for stmt in split_sql_script(SCHEMA_SQL) {
sqlx::query(&stmt).execute(&pool).await.unwrap();
}
for stmt in split_sql_script(FTS_SQL) {
sqlx::query(&stmt).execute(&pool).await.unwrap();
}
sqlx::query(
"INSERT INTO sessions (id, project, machine, agent, relationship_type) VALUES (?, ?, ?, ?, ?)",
)
.bind("session-1")
.bind("proj")
.bind("local")
.bind("claude")
.bind("")
.execute(&pool)
.await
.unwrap();
sqlx::query(
"INSERT INTO messages (id, session_id, ordinal, role, content) VALUES (?, ?, ?, ?, ?)",
)
.bind(1_i64)
.bind("session-1")
.bind(0_i64)
.bind("user")
.bind("hello world")
.execute(&pool)
.await
.unwrap();
let session_count: i64 = sqlx::query("SELECT value FROM stats WHERE key = 'session_count'")
.fetch_one(&pool)
.await
.unwrap()
.get(0);
let message_count: i64 = sqlx::query("SELECT value FROM stats WHERE key = 'message_count'")
.fetch_one(&pool)
.await
.unwrap()
.get(0);
let fts_count: i64 =
sqlx::query("SELECT count(*) FROM messages_fts WHERE messages_fts MATCH 'hello'")
.fetch_one(&pool)
.await
.unwrap()
.get(0);
assert_eq!(session_count, 1);
assert_eq!(message_count, 1);
assert_eq!(fts_count, 1);
}
}