use crate::persist::global::models::{
AgentInfo, DBMessage, DBSession, DBToolCall, DailyUsage, MessageInsert, MinimapEntry,
ModelUsage, Pin, ProjectInfo, ProjectUsage, SearchResult, SessionFilter, SessionInsert, Stats,
UsageFilter, UsageSummary,
};
use crate::persist::global::service::DatabaseService;
use crate::prelude::*;
use crate::web::github::GitHubIssue;
use crate::web::pricing::ModelPricing;
use chrono::Utc;
use r2d2::PooledConnection;
use r2d2_sqlite::SqliteConnectionManager;
use rusqlite::{params, Row};
#[derive(Debug, Clone)]
pub struct SessionRepo {
service: Arc<DatabaseService>,
}
impl SessionRepo {
pub fn new(service: Arc<DatabaseService>) -> Self {
Self { service }
}
pub fn get_connection(&self) -> anyhow::Result<PooledConnection<SqliteConnectionManager>> {
self.service.get_connection()
}
fn map_session_row(row: &Row) -> rusqlite::Result<DBSession> {
Ok(DBSession {
id: row.get("id")?,
project: row.get("project")?,
machine: row.get("machine")?,
agent: row.get("agent")?,
first_message: row.get("first_message")?,
display_name: row.get("display_name")?,
started_at: row.get("started_at")?,
ended_at: row.get("ended_at")?,
message_count: row.get("message_count")?,
user_message_count: row.get("user_message_count")?,
file_path: row.get("file_path")?,
file_size: row.get("file_size")?,
file_mtime: row.get("file_mtime")?,
file_hash: row.get("file_hash")?,
parent_session_id: row.get("parent_session_id")?,
relationship_type: row.get("relationship_type")?,
total_output_tokens: row.get("total_output_tokens")?,
peak_context_tokens: row.get("peak_context_tokens")?,
has_total_output_tokens: row.get::<&str, i32>("has_total_output_tokens")? != 0,
has_peak_context_tokens: row.get::<&str, i32>("has_peak_context_tokens")? != 0,
is_automated: row.get::<&str, i32>("is_automated")? != 0,
cwd: row.get("cwd")?,
git_branch: row.get("git_branch")?,
source_session_id: row.get("source_session_id")?,
source_version: row.get("source_version")?,
parser_malformed_lines: row.get("parser_malformed_lines")?,
is_truncated: row.get::<&str, i32>("is_truncated")? != 0,
deleted_at: row.get("deleted_at")?,
created_at: row.get("created_at")?,
})
}
fn map_message_row(row: &Row) -> rusqlite::Result<DBMessage> {
Ok(DBMessage {
id: row.get("id")?,
session_id: row.get("session_id")?,
ordinal: row.get("ordinal")?,
role: row.get("role")?,
content: row.get("content")?,
timestamp: row.get("timestamp")?,
has_thinking: row.get::<&str, i32>("has_thinking")? != 0,
has_tool_use: row.get::<&str, i32>("has_tool_use")? != 0,
content_length: row.get("content_length")?,
is_system: row.get::<&str, i32>("is_system")? != 0,
model: row.get("model")?,
token_usage: row.get("token_usage")?,
context_tokens: row.get("context_tokens")?,
output_tokens: row.get("output_tokens")?,
has_context_tokens: row.get::<&str, i32>("has_context_tokens")? != 0,
has_output_tokens: row.get::<&str, i32>("has_output_tokens")? != 0,
claude_message_id: row.get("claude_message_id")?,
claude_request_id: row.get("claude_request_id")?,
source_type: row.get("source_type")?,
source_subtype: row.get("source_subtype")?,
source_uuid: row.get("source_uuid")?,
source_parent_uuid: row.get("source_parent_uuid")?,
is_sidechain: row.get::<&str, i32>("is_sidechain")? != 0,
is_compact_boundary: row.get::<&str, i32>("is_compact_boundary")? != 0,
tool_calls: Vec::new(),
})
}
pub fn upsert_session(&self, s: &SessionInsert) -> anyhow::Result<()> {
let conn = self.get_connection()?;
conn.execute(
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,
has_total_output_tokens, has_peak_context_tokens,
is_automated, cwd, git_branch,
source_session_id, source_version,
parser_malformed_lines, is_truncated
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18, ?19, ?20, ?21, ?22, ?23, ?24, ?25, ?26)
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,
has_total_output_tokens = excluded.has_total_output_tokens,
has_peak_context_tokens = excluded.has_peak_context_tokens,
is_automated = excluded.is_automated,
cwd = CASE WHEN excluded.cwd != '' THEN excluded.cwd ELSE sessions.cwd END,
git_branch = CASE WHEN excluded.git_branch != '' THEN excluded.git_branch ELSE sessions.git_branch END,
source_session_id = CASE WHEN excluded.source_session_id != '' THEN excluded.source_session_id ELSE sessions.source_session_id END,
source_version = CASE WHEN excluded.source_version != '' THEN excluded.source_version ELSE sessions.source_version END,
parser_malformed_lines = excluded.parser_malformed_lines,
is_truncated = excluded.is_truncated"#,
params![
s.id, s.project, s.machine, s.agent, s.first_message, s.started_at, s.ended_at,
s.message_count, s.user_message_count, s.file_path, s.file_size, s.file_mtime,
s.file_hash, s.parent_session_id, s.relationship_type,
s.total_output_tokens, s.peak_context_tokens,
s.has_total_output_tokens as i32, s.has_peak_context_tokens as i32,
s.is_automated as i32, s.cwd, s.git_branch,
s.source_session_id, s.source_version,
s.parser_malformed_lines, s.is_truncated as i32,
],
)?;
Ok(())
}
pub fn get_session(&self, id: &str) -> anyhow::Result<Option<DBSession>> {
let conn = self.get_connection()?;
let mut stmt = conn.prepare(
"SELECT id, project, machine, agent, first_message, display_name, \
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, \
has_total_output_tokens, has_peak_context_tokens, \
is_automated, cwd, git_branch, \
source_session_id, source_version, \
parser_malformed_lines, is_truncated, \
deleted_at, created_at \
FROM sessions WHERE id = ?1 AND deleted_at IS NULL",
)?;
let mut rows = stmt.query(params![id])?;
if let Some(row) = rows.next()? {
Ok(Some(Self::map_session_row(row)?))
} else {
Ok(None)
}
}
pub fn is_excluded(&self, id: &str) -> anyhow::Result<bool> {
let conn = self.get_connection()?;
let count: i64 = conn.query_row(
"SELECT COUNT(*) FROM excluded_sessions WHERE id = ?1",
params![id],
|row| row.get(0),
)?;
Ok(count > 0)
}
pub fn get_session_file_info(&self, id: &str) -> anyhow::Result<Option<(i64, i64)>> {
let conn = self.get_connection()?;
let result = conn.query_row(
"SELECT file_size, file_mtime FROM sessions WHERE id = ?1 AND file_size IS NOT NULL",
params![id],
|row| Ok((row.get::<_, i64>(0)?, row.get::<_, i64>(1)?)),
);
match result {
Ok(info) => Ok(Some(info)),
Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
Err(e) => Err(e.into()),
}
}
pub fn get_file_info_by_path(&self, path: &str) -> anyhow::Result<Option<(i64, i64)>> {
let conn = self.get_connection()?;
let result = conn.query_row(
"SELECT file_size, file_mtime FROM sessions WHERE file_path = ?1 AND file_size IS NOT NULL LIMIT 1",
params![path],
|row| Ok((row.get::<_, i64>(0)?, row.get::<_, i64>(1)?)),
);
match result {
Ok(info) => Ok(Some(info)),
Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
Err(e) => Err(e.into()),
}
}
pub fn count_sessions_for_path(&self, path: &str) -> anyhow::Result<i64> {
let conn = self.get_connection()?;
let count: i64 = conn.query_row(
"SELECT COUNT(*) FROM sessions WHERE file_path = ?1",
params![path],
|row| row.get(0),
)?;
Ok(count)
}
pub fn max_ordinal(&self, session_id: &str) -> anyhow::Result<Option<i64>> {
let conn = self.get_connection()?;
let result: Option<i64> = conn
.query_row(
"SELECT MAX(ordinal) FROM messages WHERE session_id = ?1",
params![session_id],
|row| row.get(0),
)
.ok();
Ok(result)
}
pub fn insert_messages(&self, messages: &[MessageInsert]) -> anyhow::Result<()> {
let conn = self.get_connection()?;
for msg in messages {
conn.execute(
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,
has_context_tokens, has_output_tokens,
claude_message_id, claude_request_id,
source_type, source_subtype,
source_uuid, source_parent_uuid,
is_sidechain, is_compact_boundary
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18, ?19, ?20, ?21, ?22, ?23)
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,
has_context_tokens = excluded.has_context_tokens,
has_output_tokens = excluded.has_output_tokens,
claude_message_id = CASE WHEN excluded.claude_message_id != '' THEN excluded.claude_message_id ELSE messages.claude_message_id END,
claude_request_id = CASE WHEN excluded.claude_request_id != '' THEN excluded.claude_request_id ELSE messages.claude_request_id END,
source_type = CASE WHEN excluded.source_type != '' THEN excluded.source_type ELSE messages.source_type END,
source_subtype = CASE WHEN excluded.source_subtype != '' THEN excluded.source_subtype ELSE messages.source_subtype END,
source_uuid = CASE WHEN excluded.source_uuid != '' THEN excluded.source_uuid ELSE messages.source_uuid END,
source_parent_uuid = CASE WHEN excluded.source_parent_uuid != '' THEN excluded.source_parent_uuid ELSE messages.source_parent_uuid END,
is_sidechain = excluded.is_sidechain,
is_compact_boundary = excluded.is_compact_boundary"#,
params![
msg.session_id, msg.ordinal, msg.role, msg.content, msg.timestamp,
msg.has_thinking as i32, msg.has_tool_use as i32, msg.content_length, msg.is_system as i32,
msg.model, msg.token_usage, msg.context_tokens, msg.output_tokens,
msg.has_context_tokens as i32, msg.has_output_tokens as i32,
msg.claude_message_id, msg.claude_request_id,
msg.source_type, msg.source_subtype,
msg.source_uuid, msg.source_parent_uuid,
msg.is_sidechain as i32, msg.is_compact_boundary as i32,
],
)?;
let message_id: i64 = conn.query_row(
"SELECT id FROM messages WHERE session_id = ?1 AND ordinal = ?2",
params![msg.session_id, msg.ordinal],
|row| row.get(0),
)?;
for tc in &msg.tool_calls {
conn.execute(
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 (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)"#,
params![
message_id,
msg.session_id,
tc.tool_name,
tc.category,
tc.tool_use_id,
tc.input_json,
tc.skill_name,
tc.result_content_length,
tc.result_content,
tc.subagent_session_id,
],
)?;
}
}
Ok(())
}
pub 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,
) -> anyhow::Result<()> {
let conn = self.get_connection()?;
conn.execute(
r#"UPDATE sessions SET
ended_at = COALESCE(?1, ended_at),
message_count = ?2,
user_message_count = ?3,
file_size = ?4,
file_mtime = ?5,
total_output_tokens = ?6,
peak_context_tokens = ?7
WHERE id = ?8"#,
params![
ended_at,
message_count,
user_message_count,
file_size,
file_mtime,
total_output_tokens,
peak_context_tokens,
id
],
)?;
Ok(())
}
pub fn get_session_user_count(&self, session_id: &str) -> anyhow::Result<i64> {
let conn = self.get_connection()?;
let result: Option<i64> = conn
.query_row(
"SELECT user_message_count FROM sessions WHERE id = ?1",
params![session_id],
|row| row.get(0),
)
.ok();
Ok(result.unwrap_or(0))
}
pub fn list_sessions(
&self,
filter: &SessionFilter,
limit: i64,
cursor: Option<(String, String)>,
) -> anyhow::Result<Vec<DBSession>> {
let conn = self.get_connection()?;
let mut sql = String::from(
"SELECT id, project, machine, agent, first_message, display_name, \
started_at, ended_at, message_count, user_message_count, \
file_path, file_size, file_mtime, file_hash, local_modified_at, \
parent_session_id, relationship_type, total_output_tokens, \
peak_context_tokens, has_total_output_tokens, has_peak_context_tokens, \
is_automated, tool_failure_signal_count, tool_retry_count, \
edit_churn_count, consecutive_failure_max, \
outcome, outcome_confidence, ended_with_role, final_failure_streak, \
signals_pending_since, compaction_count, mid_task_compaction_count, \
context_pressure_max, health_score, health_grade, \
has_tool_calls, has_context_data, data_version, \
cwd, git_branch, source_session_id, source_version, \
parser_malformed_lines, is_truncated, deleted_at, created_at \
FROM sessions WHERE deleted_at IS NULL",
);
let mut binds: Vec<Box<dyn rusqlite::ToSql>> = 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(Box::new(p.clone()));
}
if let Some(ref ep) = filter.exclude_project {
sql.push_str(" AND project != ?");
binds.push(Box::new(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(",")));
for m in machines {
binds.push(Box::new(m.clone()));
}
}
if let Some(ref agents) = filter.agent {
let placeholders: Vec<&str> = agents.iter().map(|_| "?").collect();
sql.push_str(&format!(" AND agent IN ({})", placeholders.join(",")));
for a in agents {
binds.push(Box::new(a.clone()));
}
}
if let Some(ref d) = filter.date {
sql.push_str(" AND date(ended_at) = ?");
binds.push(Box::new(d.clone()));
}
if let Some(ref df) = filter.date_from {
sql.push_str(" AND date(ended_at) >= ?");
binds.push(Box::new(df.clone()));
}
if let Some(ref dt) = filter.date_to {
sql.push_str(" AND date(ended_at) <= ?");
binds.push(Box::new(dt.clone()));
}
if let Some(ref since) = filter.active_since {
sql.push_str(" AND ended_at >= ?");
binds.push(Box::new(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(Box::new(ended.to_string()));
binds.push(Box::new(id.to_string()));
}
sql.push_str(" ORDER BY ended_at DESC, id DESC");
sql.push_str(&format!(" LIMIT {}", limit.min(500)));
let mut stmt = conn.prepare(&sql)?;
let params_refs: Vec<&dyn rusqlite::ToSql> = binds.iter().map(|b| b.as_ref()).collect();
let mut rows = stmt.query(params_refs.as_slice())?;
let mut sessions = Vec::new();
while let Some(row) = rows.next()? {
let mut session = Self::map_session_row(row)?;
let mut full_stmt = conn.prepare(
"SELECT file_path, file_size, file_mtime, file_hash, \
has_total_output_tokens, has_peak_context_tokens, is_automated, \
git_branch, source_session_id, source_version, parser_malformed_lines, is_truncated \
FROM sessions WHERE id = ?1",
)?;
if let Some(full_row) = full_stmt.query(params![session.id])?.next()? {
session.file_path = full_row.get("file_path").ok();
session.file_size = full_row.get("file_size").ok();
session.file_mtime = full_row.get("file_mtime").ok();
session.file_hash = full_row.get("file_hash").ok();
session.has_total_output_tokens = full_row
.get::<&str, i32>("has_total_output_tokens")
.unwrap_or(0)
!= 0;
session.has_peak_context_tokens = full_row
.get::<&str, i32>("has_peak_context_tokens")
.unwrap_or(0)
!= 0;
session.is_automated = full_row.get::<&str, i32>("is_automated").unwrap_or(0) != 0;
session.git_branch = full_row.get("git_branch").unwrap_or_default();
session.source_session_id = full_row.get("source_session_id").unwrap_or_default();
session.source_version = full_row.get("source_version").unwrap_or_default();
session.parser_malformed_lines =
full_row.get("parser_malformed_lines").unwrap_or(0);
session.is_truncated = full_row.get::<_, i32>("is_truncated").unwrap_or(0) != 0;
}
sessions.push(session);
}
Ok(sessions)
}
pub fn get_children(&self, parent_id: &str) -> anyhow::Result<Vec<DBSession>> {
let conn = self.get_connection()?;
let mut stmt = conn.prepare(
"SELECT id, project, machine, agent, first_message, display_name, \
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, \
has_total_output_tokens, has_peak_context_tokens, \
is_automated, cwd, git_branch, \
source_session_id, source_version, \
parser_malformed_lines, is_truncated, \
deleted_at, created_at \
FROM sessions WHERE parent_session_id = ?1 AND deleted_at IS NULL \
ORDER BY started_at ASC",
)?;
let mut rows = stmt.query(params![parent_id])?;
let mut sessions = Vec::new();
while let Some(row) = rows.next()? {
sessions.push(Self::map_session_row(row)?);
}
Ok(sessions)
}
pub fn get_messages(
&self,
session_id: &str,
limit: i64,
from_ordinal: i64,
desc: bool,
) -> anyhow::Result<Vec<DBMessage>> {
let conn = self.get_connection()?;
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, \
has_context_tokens, has_output_tokens, \
claude_message_id, claude_request_id, \
source_type, source_subtype, \
source_uuid, source_parent_uuid, \
is_sidechain, is_compact_boundary \
FROM messages WHERE session_id = ?1 AND ordinal {} ?2 \
ORDER BY ordinal {} LIMIT ?3",
cmp, dir
);
let mut stmt = conn.prepare(&sql)?;
let mut rows = stmt.query(params![session_id, from_ordinal, limit.min(1000)])?;
let mut messages = Vec::new();
while let Some(row) = rows.next()? {
messages.push(Self::map_message_row(row)?);
}
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 mut tc_stmt = conn.prepare(&tc_sql)?;
let mut tc_rows = tc_stmt.query([])?;
let mut tc_map: HashMap<i64, Vec<DBToolCall>> = HashMap::new();
while let Some(r) = tc_rows.next()? {
let mid: i64 = r.get("message_id")?;
tc_map.entry(mid).or_default().push(DBToolCall {
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 fn get_minimap(
&self,
session_id: &str,
from: i64,
max: Option<i64>,
) -> anyhow::Result<Vec<MinimapEntry>> {
let conn = self.get_connection()?;
let mut stmt = conn.prepare(
"SELECT ordinal, role, content_length, has_thinking, has_tool_use \
FROM messages WHERE session_id = ?1 AND ordinal >= ?2 \
ORDER BY ordinal ASC",
)?;
let mut rows = stmt.query(params![session_id, from])?;
let mut entries = Vec::new();
while let Some(row) = rows.next()? {
entries.push(MinimapEntry {
ordinal: row.get("ordinal")?,
role: row.get("role")?,
content_length: row.get("content_length")?,
has_thinking: row.get::<_, i32>("has_thinking")? != 0,
has_tool_use: row.get::<_, i32>("has_tool_use")? != 0,
});
}
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 fn search_session(&self, session_id: &str, query: &str) -> anyhow::Result<Vec<i64>> {
let conn = self.get_connection()?;
let pattern = format!("%{}%", query);
let mut stmt = conn.prepare(
"SELECT DISTINCT ordinal FROM ( \
SELECT ordinal FROM messages WHERE session_id = ?1 AND content LIKE ?2 \
UNION \
SELECT m.ordinal FROM tool_calls tc \
JOIN messages m ON m.id = tc.message_id \
WHERE tc.session_id = ?1 AND tc.result_content LIKE ?2 \
) ORDER BY ordinal ASC",
)?;
let mut rows = stmt.query(params![session_id, &pattern])?;
let mut results = Vec::new();
while let Some(row) = rows.next()? {
results.push(row.get("ordinal")?);
}
Ok(results)
}
pub fn global_search(
&self,
query: &str,
limit: i64,
offset: i64,
sort: &str,
project: Option<&str>,
) -> anyhow::Result<Vec<SearchResult>> {
let conn = self.get_connection()?;
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 ?1 AND s.deleted_at IS NULL"
);
if let Some(p) = project {
sql.push_str(" AND s.project = ?2");
let _stmt = conn.prepare(&sql)?;
let full_sql = format!(
"{} ORDER BY {} LIMIT {} OFFSET {}",
sql,
order,
limit.min(500),
offset
);
let mut s = conn.prepare(&full_sql)?;
let mut rows = s.query(params![fts_query, p])?;
let mut results = Vec::new();
while let Some(row) = rows.next()? {
results.push(SearchResult {
session_id: row.get("session_id")?,
project: row.get("project")?,
agent: row.get("agent")?,
name: row.get("name")?,
ordinal: row.get("ordinal")?,
session_ended_at: row.get("session_ended_at")?,
snippet: row.get("snippet")?,
rank: row.get("rank")?,
});
}
return Ok(results);
}
let full_sql = format!(
"{} ORDER BY {} LIMIT {} OFFSET {}",
sql,
order,
limit.min(500),
offset
);
let mut stmt = conn.prepare(&full_sql)?;
let mut rows = stmt.query(params![fts_query])?;
let mut results = Vec::new();
while let Some(row) = rows.next()? {
results.push(SearchResult {
session_id: row.get("session_id")?,
project: row.get("project")?,
agent: row.get("agent")?,
name: row.get("name")?,
ordinal: row.get("ordinal")?,
session_ended_at: row.get("session_ended_at")?,
snippet: row.get("snippet")?,
rank: row.get("rank")?,
});
}
Ok(results)
}
pub fn rename_session(&self, id: &str, display_name: Option<&str>) -> anyhow::Result<bool> {
let conn = self.get_connection()?;
let affected = conn.execute(
"UPDATE sessions SET display_name = ?1, local_modified_at = ?2 WHERE id = ?3",
params![display_name, Utc::now().to_rfc3339(), id],
)?;
Ok(affected > 0)
}
pub fn soft_delete_session(&self, id: &str) -> anyhow::Result<bool> {
let conn = self.get_connection()?;
let affected = conn.execute(
"UPDATE sessions SET deleted_at = ?1 WHERE id = ?2 AND deleted_at IS NULL",
params![Utc::now().to_rfc3339(), id],
)?;
Ok(affected > 0)
}
pub fn restore_session(&self, id: &str) -> anyhow::Result<bool> {
let conn = self.get_connection()?;
let affected = conn.execute(
"UPDATE sessions SET deleted_at = NULL WHERE id = ?1 AND deleted_at IS NOT NULL",
params![id],
)?;
Ok(affected > 0)
}
pub fn permanent_delete(&self, id: &str) -> anyhow::Result<bool> {
let conn = self.get_connection()?;
let deleted_at: Option<String> = conn
.query_row(
"SELECT deleted_at FROM sessions WHERE id = ?1",
params![id],
|row| row.get(0),
)
.ok();
if deleted_at.is_none() {
return Ok(false);
}
conn.execute(
"INSERT OR IGNORE INTO excluded_sessions (id) VALUES (?1)",
params![id],
)?;
conn.execute("DELETE FROM sessions WHERE id = ?1", params![id])?;
Ok(true)
}
pub fn list_trash(&self) -> anyhow::Result<Vec<DBSession>> {
let conn = self.get_connection()?;
let mut stmt = conn.prepare(
"SELECT id, project, machine, agent, first_message, display_name, \
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, \
has_total_output_tokens, has_peak_context_tokens, \
is_automated, cwd, git_branch, \
source_session_id, source_version, \
parser_malformed_lines, is_truncated, \
deleted_at, created_at \
FROM sessions WHERE deleted_at IS NOT NULL \
ORDER BY deleted_at DESC LIMIT 500",
)?;
let mut rows = stmt.query([])?;
let mut sessions = Vec::new();
while let Some(row) = rows.next()? {
sessions.push(Self::map_session_row(row)?);
}
Ok(sessions)
}
pub fn empty_trash(&self) -> anyhow::Result<i64> {
let conn = self.get_connection()?;
let mut stmt = conn.prepare("SELECT id FROM sessions WHERE deleted_at IS NOT NULL")?;
let mut rows = stmt.query([])?;
let mut count = 0i64;
while let Some(row) = rows.next()? {
let id: String = row.get(0)?;
conn.execute(
"INSERT OR IGNORE INTO excluded_sessions (id) VALUES (?1)",
params![id],
)?;
count += 1;
}
conn.execute("DELETE FROM sessions WHERE deleted_at IS NOT NULL", [])?;
Ok(count)
}
pub fn list_starred(&self) -> anyhow::Result<Vec<String>> {
let conn = self.get_connection()?;
let mut stmt =
conn.prepare("SELECT session_id FROM starred_sessions ORDER BY created_at DESC")?;
let mut rows = stmt.query([])?;
let mut ids = Vec::new();
while let Some(row) = rows.next()? {
ids.push(row.get("session_id")?);
}
Ok(ids)
}
pub fn star_session(&self, id: &str) -> anyhow::Result<()> {
let conn = self.get_connection()?;
conn.execute(
"INSERT OR IGNORE INTO starred_sessions (session_id) VALUES (?1)",
params![id],
)?;
Ok(())
}
pub fn unstar_session(&self, id: &str) -> anyhow::Result<()> {
let conn = self.get_connection()?;
conn.execute(
"DELETE FROM starred_sessions WHERE session_id = ?1",
params![id],
)?;
Ok(())
}
pub fn bulk_star(&self, ids: &[String]) -> anyhow::Result<()> {
let conn = self.get_connection()?;
for id in ids {
conn.execute(
"INSERT OR IGNORE INTO starred_sessions (session_id) VALUES (?1)",
params![id],
)?;
}
Ok(())
}
pub fn is_starred(&self, id: &str) -> anyhow::Result<bool> {
let conn = self.get_connection()?;
let count: i64 = conn.query_row(
"SELECT COUNT(*) FROM starred_sessions WHERE session_id = ?1",
params![id],
|row| row.get(0),
)?;
Ok(count > 0)
}
pub fn list_all_pins(&self) -> anyhow::Result<Vec<Pin>> {
let conn = self.get_connection()?;
let mut stmt = conn.prepare(
"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",
)?;
let mut rows = stmt.query([])?;
let mut pins = Vec::new();
while let Some(row) = rows.next()? {
pins.push(Pin {
id: row.get("id")?,
session_id: row.get("session_id")?,
message_id: row.get("message_id")?,
ordinal: row.get("ordinal")?,
note: row.get("note")?,
created_at: row.get("created_at")?,
content: row.get("content").ok(),
role: row.get("role").ok(),
session_project: row.get("session_project").ok(),
session_agent: row.get("session_agent").ok(),
session_display_name: row.get("session_display_name").ok(),
session_first_message: row.get("session_first_message").ok(),
});
}
Ok(pins)
}
pub fn list_session_pins(&self, session_id: &str) -> anyhow::Result<Vec<Pin>> {
let conn = self.get_connection()?;
let mut stmt = conn.prepare(
"SELECT id, session_id, message_id, ordinal, note, created_at \
FROM pinned_messages WHERE session_id = ?1 ORDER BY ordinal ASC",
)?;
let mut rows = stmt.query(params![session_id])?;
let mut pins = Vec::new();
while let Some(row) = rows.next()? {
pins.push(Pin {
id: row.get("id")?,
session_id: row.get("session_id")?,
message_id: row.get("message_id")?,
ordinal: row.get("ordinal")?,
note: row.get("note")?,
created_at: row.get("created_at")?,
content: None,
role: None,
session_project: None,
session_agent: None,
session_display_name: None,
session_first_message: None,
});
}
Ok(pins)
}
pub fn pin_message(
&self,
session_id: &str,
message_id: i64,
note: Option<&str>,
) -> anyhow::Result<i64> {
let conn = self.get_connection()?;
let ordinal: i64 = conn.query_row(
"SELECT ordinal FROM messages WHERE id = ?1 AND session_id = ?2",
params![message_id, session_id],
|row| row.get(0),
)?;
conn.execute(
"INSERT INTO pinned_messages (session_id, message_id, ordinal, note) \
VALUES (?1, ?2, ?3, ?4) \
ON CONFLICT(session_id, message_id) DO UPDATE SET note = excluded.note",
params![session_id, message_id, ordinal, note],
)?;
let pin_id: i64 = conn.query_row(
"SELECT id FROM pinned_messages WHERE session_id = ?1 AND message_id = ?2",
params![session_id, message_id],
|row| row.get(0),
)?;
Ok(pin_id)
}
pub fn unpin_message(&self, session_id: &str, message_id: i64) -> anyhow::Result<()> {
let conn = self.get_connection()?;
conn.execute(
"DELETE FROM pinned_messages WHERE session_id = ?1 AND message_id = ?2",
params![session_id, message_id],
)?;
Ok(())
}
pub fn get_stats(&self, include_one_shot: bool) -> anyhow::Result<Stats> {
let conn = self.get_connection()?;
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 = conn.query_row(&sql, [], |row| {
Ok(Stats {
session_count: row.get(0)?,
message_count: row.get::<_, Option<i64>>(1)?.unwrap_or(0),
project_count: row.get(2)?,
machine_count: row.get(3)?,
earliest_session: row.get(4)?,
})
})?;
Ok(row)
}
pub fn get_overview_stats(&self) -> anyhow::Result<HashMap<String, i64>> {
let conn = self.get_connection()?;
let mut stmt = conn.prepare("SELECT key, value FROM stats")?;
let mut rows = stmt.query([])?;
let mut map = HashMap::new();
while let Some(row) = rows.next()? {
let key: String = row.get("key")?;
let value: i64 = row.get("value")?;
map.insert(key, value);
}
Ok(map)
}
pub fn set_stat(&self, key: &str, value: i64) -> anyhow::Result<()> {
let conn = self.get_connection()?;
conn.execute(
"INSERT OR REPLACE INTO stats (key, value) VALUES (?1, ?2)",
params![key, value],
)?;
Ok(())
}
pub fn list_projects(&self, include_one_shot: bool) -> anyhow::Result<Vec<ProjectInfo>> {
let conn = self.get_connection()?;
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 mut stmt = conn.prepare(&sql)?;
let mut rows = stmt.query([])?;
let mut projects = Vec::new();
while let Some(row) = rows.next()? {
projects.push(ProjectInfo {
name: row.get("name")?,
session_count: row.get("session_count")?,
});
}
Ok(projects)
}
pub fn list_machines(&self, include_one_shot: bool) -> anyhow::Result<Vec<String>> {
let conn = self.get_connection()?;
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 mut stmt = conn.prepare(&sql)?;
let mut rows = stmt.query([])?;
let mut machines = Vec::new();
while let Some(row) = rows.next()? {
machines.push(row.get("machine")?);
}
Ok(machines)
}
pub fn list_agents(&self, include_one_shot: bool) -> anyhow::Result<Vec<AgentInfo>> {
let conn = self.get_connection()?;
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 mut stmt = conn.prepare(&sql)?;
let mut rows = stmt.query([])?;
let mut agents = Vec::new();
while let Some(row) = rows.next()? {
agents.push(AgentInfo {
name: row.get("name")?,
session_count: row.get("session_count")?,
});
}
Ok(agents)
}
pub fn get_session_version(&self, id: &str) -> anyhow::Result<Option<(i64, i64)>> {
let conn = self.get_connection()?;
let result = conn.query_row(
"SELECT message_count, COALESCE(file_mtime, 0) as file_mtime FROM sessions WHERE id = ?1",
params![id],
|row| Ok((row.get::<_, i64>(0)?, row.get::<_, i64>(1)?)),
);
match result {
Ok(info) => Ok(Some(info)),
Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
Err(e) => Err(e.into()),
}
}
pub fn load_skip_cache(&self) -> anyhow::Result<HashMap<String, i64>> {
let conn = self.get_connection()?;
let mut stmt = conn.prepare("SELECT file_path, file_mtime FROM skipped_files")?;
let mut rows = stmt.query([])?;
let mut cache = HashMap::new();
while let Some(row) = rows.next()? {
let path: String = row.get("file_path")?;
let mtime: i64 = row.get("file_mtime")?;
cache.insert(path, mtime);
}
Ok(cache)
}
pub fn persist_skip_cache(&self, cache: &HashMap<String, i64>) -> anyhow::Result<()> {
let conn = self.get_connection()?;
conn.execute("DELETE FROM skipped_files", [])?;
for (path, mtime) in cache {
conn.execute(
"INSERT INTO skipped_files (file_path, file_mtime) VALUES (?1, ?2)",
params![path, mtime],
)?;
}
Ok(())
}
pub fn delete_skipped_file(&self, path: &str) -> anyhow::Result<()> {
let conn = self.get_connection()?;
conn.execute(
"DELETE FROM skipped_files WHERE file_path = ?1",
params![path],
)?;
Ok(())
}
pub fn link_subagent_sessions(&self) -> anyhow::Result<()> {
let conn = self.get_connection()?;
conn.execute(
"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 = '')",
[],
)?;
Ok(())
}
pub fn get_model_pricing(&self) -> anyhow::Result<Vec<ModelPricing>> {
let conn = self.get_connection()?;
let mut stmt = conn.prepare(
"SELECT model_pattern, input_per_mtok, output_per_mtok, \
cache_creation_per_mtok, cache_read_per_mtok \
FROM model_pricing",
)?;
let mut rows = stmt.query([])?;
let mut pricing = Vec::new();
while let Some(row) = rows.next()? {
pricing.push(ModelPricing {
model_pattern: row.get("model_pattern")?,
input_per_mtok: row.get("input_per_mtok")?,
output_per_mtok: row.get("output_per_mtok")?,
cache_creation_per_mtok: row.get("cache_creation_per_mtok")?,
cache_read_per_mtok: row.get("cache_read_per_mtok")?,
});
}
Ok(pricing)
}
pub fn upsert_model_pricing(&self, pricing: &ModelPricing) -> anyhow::Result<()> {
let conn = self.get_connection()?;
conn.execute(
r#"INSERT INTO model_pricing (
model_pattern, input_per_mtok, output_per_mtok,
cache_creation_per_mtok, cache_read_per_mtok
) VALUES (?1, ?2, ?3, ?4, ?5)
ON CONFLICT(model_pattern) DO UPDATE SET
input_per_mtok = excluded.input_per_mtok,
output_per_mtok = excluded.output_per_mtok,
cache_creation_per_mtok = excluded.cache_creation_per_mtok,
cache_read_per_mtok = excluded.cache_read_per_mtok,
updated_at = strftime('%Y-%m-%dT%H:%M:%fZ','now')"#,
params![
pricing.model_pattern,
pricing.input_per_mtok,
pricing.output_per_mtok,
pricing.cache_creation_per_mtok,
pricing.cache_read_per_mtok,
],
)?;
Ok(())
}
pub fn sync_fallback_pricing(&self) -> anyhow::Result<()> {
let pricing = crate::web::pricing::fallback_pricing();
for p in pricing {
self.upsert_model_pricing(&p)?;
}
Ok(())
}
pub fn cache_github_repo_issues(
&self,
repo: &str,
issues: &[GitHubIssue],
) -> anyhow::Result<()> {
let conn = self.get_connection()?;
let fetched_at = Utc::now().to_rfc3339();
for issue in issues {
let payload_json = serde_json::to_string(issue)?;
conn.execute(
r#"INSERT INTO github_issue_cache (repo, number, state, payload_json, fetched_at)
VALUES (?1, ?2, ?3, ?4, ?5)
ON CONFLICT(repo, number) DO UPDATE SET
state = excluded.state,
payload_json = excluded.payload_json,
fetched_at = excluded.fetched_at"#,
params![repo, issue.number, issue.state, payload_json, fetched_at],
)?;
}
conn.execute(
r#"INSERT INTO github_repo_cache_state (repo, fetched_at, last_error)
VALUES (?1, ?2, NULL)
ON CONFLICT(repo) DO UPDATE SET
fetched_at = excluded.fetched_at,
last_error = NULL"#,
params![repo, fetched_at],
)?;
Ok(())
}
pub fn cache_github_repo_failure(&self, repo: &str, error: &str) -> anyhow::Result<()> {
let conn = self.get_connection()?;
conn.execute(
r#"INSERT INTO github_repo_cache_state (repo, fetched_at, last_error)
VALUES (?1, ?2, ?3)
ON CONFLICT(repo) DO UPDATE SET
fetched_at = excluded.fetched_at,
last_error = excluded.last_error"#,
params![repo, Utc::now().to_rfc3339(), error],
)?;
Ok(())
}
pub fn get_usage_summary(&self, filter: &UsageFilter) -> anyhow::Result<UsageSummary> {
let conn = self.get_connection()?;
let mut sql = String::from(
"SELECT COALESCE(SUM(m.context_tokens), 0) as input_tokens, \
COALESCE(SUM(m.output_tokens), 0) as output_tokens \
FROM messages m \
JOIN sessions s ON s.id = m.session_id \
WHERE s.deleted_at IS NULL",
);
let mut binds: Vec<Box<dyn rusqlite::ToSql>> = Vec::new();
if let Some(ref df) = filter.date_from {
sql.push_str(" AND date(s.ended_at) >= ?");
binds.push(Box::new(df.clone()));
}
if let Some(ref dt) = filter.date_to {
sql.push_str(" AND date(s.ended_at) <= ?");
binds.push(Box::new(dt.clone()));
}
if let Some(ref p) = filter.project {
sql.push_str(" AND s.project = ?");
binds.push(Box::new(p.clone()));
}
if let Some(ref a) = filter.agent {
sql.push_str(" AND s.agent = ?");
binds.push(Box::new(a.clone()));
}
let params_refs: Vec<&dyn rusqlite::ToSql> = binds.iter().map(|b| b.as_ref()).collect();
let row = conn.query_row(&sql, params_refs.as_slice(), |row| {
Ok((
row.get::<_, i64>("input_tokens")?,
row.get::<_, i64>("output_tokens")?,
))
})?;
Ok(UsageSummary {
total_input_tokens: row.0,
total_output_tokens: row.1,
total_cache_creation_tokens: 0,
total_cache_read_tokens: 0,
estimated_cost_usd: 0.0,
})
}
pub fn get_daily_usage(&self, filter: &UsageFilter) -> anyhow::Result<Vec<DailyUsage>> {
let conn = self.get_connection()?;
let mut sql = String::from(
"SELECT date(s.ended_at) as date, \
COUNT(DISTINCT s.id) as session_count, \
COUNT(m.id) as message_count, \
COALESCE(SUM(m.context_tokens), 0) as input_tokens, \
COALESCE(SUM(m.output_tokens), 0) as output_tokens \
FROM sessions s \
LEFT JOIN messages m ON m.session_id = s.id \
WHERE s.deleted_at IS NULL AND s.ended_at IS NOT NULL",
);
let mut binds: Vec<Box<dyn rusqlite::ToSql>> = Vec::new();
if let Some(ref df) = filter.date_from {
sql.push_str(" AND date(s.ended_at) >= ?");
binds.push(Box::new(df.clone()));
}
if let Some(ref dt) = filter.date_to {
sql.push_str(" AND date(s.ended_at) <= ?");
binds.push(Box::new(dt.clone()));
}
if let Some(ref p) = filter.project {
sql.push_str(" AND s.project = ?");
binds.push(Box::new(p.clone()));
}
sql.push_str(" GROUP BY date(s.ended_at) ORDER BY date DESC LIMIT 90");
let params_refs: Vec<&dyn rusqlite::ToSql> = binds.iter().map(|b| b.as_ref()).collect();
let mut stmt = conn.prepare(&sql)?;
let mut rows = stmt.query(params_refs.as_slice())?;
let mut usages = Vec::new();
while let Some(row) = rows.next()? {
usages.push(DailyUsage {
date: row.get("date")?,
session_count: row.get("session_count")?,
message_count: row.get("message_count")?,
input_tokens: row.get("input_tokens")?,
output_tokens: row.get("output_tokens")?,
cache_creation_tokens: 0,
cache_read_tokens: 0,
estimated_cost_usd: 0.0,
});
}
Ok(usages)
}
pub fn get_usage_by_model(&self, filter: &UsageFilter) -> anyhow::Result<Vec<ModelUsage>> {
let conn = self.get_connection()?;
let mut sql = String::from(
"SELECT COALESCE(m.model, 'unknown') as model, \
COUNT(DISTINCT s.id) as session_count, \
COALESCE(SUM(m.context_tokens), 0) as input_tokens, \
COALESCE(SUM(m.output_tokens), 0) as output_tokens \
FROM sessions s \
LEFT JOIN messages m ON m.session_id = s.id \
WHERE s.deleted_at IS NULL",
);
let mut binds: Vec<Box<dyn rusqlite::ToSql>> = Vec::new();
if let Some(ref df) = filter.date_from {
sql.push_str(" AND date(s.ended_at) >= ?");
binds.push(Box::new(df.clone()));
}
if let Some(ref dt) = filter.date_to {
sql.push_str(" AND date(s.ended_at) <= ?");
binds.push(Box::new(dt.clone()));
}
if let Some(ref p) = filter.project {
sql.push_str(" AND s.project = ?");
binds.push(Box::new(p.clone()));
}
sql.push_str(" GROUP BY m.model ORDER BY input_tokens DESC LIMIT 50");
let params_refs: Vec<&dyn rusqlite::ToSql> = binds.iter().map(|b| b.as_ref()).collect();
let mut stmt = conn.prepare(&sql)?;
let mut rows = stmt.query(params_refs.as_slice())?;
let mut usages = Vec::new();
while let Some(row) = rows.next()? {
usages.push(ModelUsage {
model: row.get("model")?,
session_count: row.get("session_count")?,
input_tokens: row.get("input_tokens")?,
output_tokens: row.get("output_tokens")?,
cache_creation_tokens: 0,
cache_read_tokens: 0,
estimated_cost_usd: 0.0,
});
}
Ok(usages)
}
pub fn get_usage_by_project(&self, filter: &UsageFilter) -> anyhow::Result<Vec<ProjectUsage>> {
let conn = self.get_connection()?;
let mut sql = String::from(
"SELECT s.project, \
COUNT(DISTINCT s.id) as session_count, \
COUNT(m.id) as message_count, \
COALESCE(SUM(m.context_tokens), 0) as input_tokens, \
COALESCE(SUM(m.output_tokens), 0) as output_tokens \
FROM sessions s \
LEFT JOIN messages m ON m.session_id = s.id \
WHERE s.deleted_at IS NULL",
);
let mut binds: Vec<Box<dyn rusqlite::ToSql>> = Vec::new();
if let Some(ref df) = filter.date_from {
sql.push_str(" AND date(s.ended_at) >= ?");
binds.push(Box::new(df.clone()));
}
if let Some(ref dt) = filter.date_to {
sql.push_str(" AND date(s.ended_at) <= ?");
binds.push(Box::new(dt.clone()));
}
sql.push_str(" GROUP BY s.project ORDER BY input_tokens DESC");
let params_refs: Vec<&dyn rusqlite::ToSql> = binds.iter().map(|b| b.as_ref()).collect();
let mut stmt = conn.prepare(&sql)?;
let mut rows = stmt.query(params_refs.as_slice())?;
let mut usages = Vec::new();
while let Some(row) = rows.next()? {
usages.push(ProjectUsage {
project: row.get("project")?,
session_count: row.get("session_count")?,
message_count: row.get("message_count")?,
input_tokens: row.get("input_tokens")?,
output_tokens: row.get("output_tokens")?,
cache_creation_tokens: 0,
cache_read_tokens: 0,
estimated_cost_usd: 0.0,
});
}
Ok(usages)
}
}