sac/store/
threads.rs

1use super::*;
2
3pub fn append_episode(
4    path: &Path,
5    session_id: &str,
6    thread_name: &str,
7    action: &str,
8    content: &str,
9) -> Result<()> {
10    tracing::debug!(
11        db_path = %path.display(),
12        session_id = %session_id,
13        thread_name = %thread_name,
14        action_len = action.len(),
15        content_len = content.len(),
16        "appending retained episode"
17    );
18    let mut conn = open_connection(path)?;
19    let tx = conn.transaction()?;
20    ensure_thread_in_tx(&tx, session_id, thread_name)?;
21
22    tx.execute(
23        "INSERT INTO episodes (thread_name, session_id, action, content, created_at)
24         VALUES (?1, ?2, ?3, ?4, ?5)",
25        params![thread_name, session_id, action, content, now_utc()],
26    )?;
27
28    tx.execute(
29        "UPDATE threads
30         SET updated_at = ?1
31         WHERE name = ?2 AND session_id = ?3",
32        params![now_utc(), thread_name, session_id],
33    )?;
34
35    tx.commit()?;
36    tracing::info!(db_path = %path.display(), session_id = %session_id, thread_name = %thread_name, "retained episode appended");
37    Ok(())
38}
39
40pub fn load_worker_context(
41    path: &Path,
42    session_id: &str,
43    thread_name: &str,
44    source_threads: &[String],
45) -> Result<WorkerContext> {
46    tracing::debug!(
47        db_path = %path.display(),
48        session_id = %session_id,
49        thread_name = %thread_name,
50        source_threads = ?source_threads,
51        "loading worker context"
52    );
53    let conn = open_connection(path)?;
54    let self_episodes = load_thread_episodes(&conn, session_id, thread_name)?;
55    let mut source_episodes = Vec::with_capacity(source_threads.len());
56
57    for source_thread in source_threads {
58        let episode = latest_episode(&conn, session_id, source_thread)?
59            .ok_or_else(|| anyhow!("Source thread '{}' has no retained episode", source_thread))?;
60        source_episodes.push(episode);
61    }
62
63    let context = WorkerContext {
64        self_episodes,
65        source_episodes,
66    };
67    tracing::info!(
68        db_path = %path.display(),
69        session_id = %session_id,
70        thread_name = %thread_name,
71        self_episode_count = context.self_episodes.len(),
72        source_episode_count = context.source_episodes.len(),
73        "worker context loaded"
74    );
75    Ok(context)
76}
77
78/// Load all episodes for all threads in one query, grouped by thread_name.
79/// Episodes are ordered by id ASC (chronological order).
80pub fn load_all_episodes(
81    store_path: &Path,
82    session_id: &str,
83) -> Result<HashMap<String, Vec<EpisodeRecord>>> {
84    tracing::debug!(db_path = %store_path.display(), session_id = %session_id, "loading all retained episodes");
85    let conn = open_connection(store_path)?;
86    let mut stmt = conn.prepare(
87        "SELECT e.id, e.thread_name, e.session_id, e.action, e.content, e.created_at
88         FROM episodes e
89         INNER JOIN threads t ON e.thread_name = t.name AND e.session_id = t.session_id
90         WHERE e.session_id = ?
91         ORDER BY e.thread_name, e.id",
92    )?;
93    let rows = stmt.query_map(params![session_id], row_to_episode)?;
94
95    let mut grouped: HashMap<String, Vec<EpisodeRecord>> = HashMap::new();
96    for row in rows {
97        let episode = row?;
98        grouped
99            .entry(episode.thread_name.clone())
100            .or_default()
101            .push(episode);
102    }
103    tracing::info!(db_path = %store_path.display(), session_id = %session_id, thread_count = grouped.len(), "loaded all retained episodes");
104    Ok(grouped)
105}
106
107pub fn list_threads(path: &Path, session_id: &str) -> Result<Vec<ThreadRecord>> {
108    tracing::debug!(db_path = %path.display(), session_id = %session_id, "listing retained threads");
109    let conn = open_connection(path)?;
110    let mut stmt = conn.prepare(
111        "SELECT t.name, t.session_id, t.created_at, t.updated_at,
112                (SELECT COUNT(*) FROM episodes e
113                 WHERE e.thread_name = t.name AND e.session_id = t.session_id) AS episode_count,
114                (SELECT e.action FROM episodes e
115                 WHERE e.thread_name = t.name AND e.session_id = t.session_id
116                 ORDER BY e.id DESC
117                 LIMIT 1) AS latest_action
118         FROM threads t
119         WHERE t.session_id = ?1
120         ORDER BY t.updated_at DESC, t.name ASC",
121    )?;
122
123    let mut rows = stmt.query([session_id])?;
124    let mut threads = Vec::new();
125    while let Some(row) = rows.next()? {
126        threads.push(ThreadRecord {
127            name: row.get(0)?,
128            session_id: row.get(1)?,
129            created_at: row.get(2)?,
130            updated_at: row.get(3)?,
131            episode_count: row.get(4)?,
132            latest_action: row.get(5)?,
133        });
134    }
135    tracing::info!(db_path = %path.display(), session_id = %session_id, thread_count = threads.len(), "listed retained threads");
136    Ok(threads)
137}
138
139pub fn thread_read(path: &Path, session_id: &str, thread_name: &str) -> Result<Vec<EpisodeRecord>> {
140    tracing::debug!(db_path = %path.display(), session_id = %session_id, thread_name = %thread_name, "reading retained thread episodes");
141    let conn = open_connection(path)?;
142    let episodes = load_thread_episodes(&conn, session_id, thread_name)?;
143    tracing::info!(db_path = %path.display(), session_id = %session_id, thread_name = %thread_name, episode_count = episodes.len(), "read retained thread episodes");
144    Ok(episodes)
145}
146
147pub fn delete_thread(path: &Path, session_id: &str, thread_name: &str) -> Result<bool> {
148    tracing::debug!(db_path = %path.display(), session_id = %session_id, thread_name = %thread_name, "deleting retained thread");
149    let mut conn = open_connection(path)?;
150    let tx = conn.transaction()?;
151    tx.execute(
152        "DELETE FROM episodes WHERE thread_name = ?1 AND session_id = ?2",
153        params![thread_name, session_id],
154    )?;
155    let deleted = tx.execute(
156        "DELETE FROM threads WHERE name = ?1 AND session_id = ?2",
157        params![thread_name, session_id],
158    )?;
159    tx.commit()?;
160    tracing::info!(db_path = %path.display(), session_id = %session_id, thread_name = %thread_name, deleted = deleted > 0, "retained thread deletion finished");
161    Ok(deleted > 0)
162}
163
164fn ensure_thread_in_tx(tx: &Transaction<'_>, session_id: &str, thread_name: &str) -> Result<()> {
165    let now = now_utc();
166    tx.execute(
167        "INSERT OR IGNORE INTO threads (name, session_id, created_at, updated_at)
168         VALUES (?1, ?2, ?3, ?3)",
169        params![thread_name, session_id, now],
170    )?;
171    Ok(())
172}
173
174fn load_thread_episodes(
175    conn: &Connection,
176    session_id: &str,
177    thread_name: &str,
178) -> Result<Vec<EpisodeRecord>> {
179    let mut stmt = conn.prepare(
180        "SELECT id, thread_name, session_id, action, content, created_at
181         FROM episodes
182         WHERE thread_name = ?1 AND session_id = ?2
183         ORDER BY id ASC",
184    )?;
185    let mut rows = stmt.query(params![thread_name, session_id])?;
186    let mut episodes = Vec::new();
187    while let Some(row) = rows.next()? {
188        episodes.push(row_to_episode(row)?);
189    }
190    Ok(episodes)
191}
192
193fn latest_episode(
194    conn: &Connection,
195    session_id: &str,
196    thread_name: &str,
197) -> Result<Option<EpisodeRecord>> {
198    conn.query_row(
199        "SELECT id, thread_name, session_id, action, content, created_at
200         FROM episodes
201         WHERE thread_name = ?1 AND session_id = ?2
202         ORDER BY id DESC
203         LIMIT 1",
204        params![thread_name, session_id],
205        row_to_episode,
206    )
207    .optional()
208    .map_err(Into::into)
209}
210
211fn row_to_episode(row: &rusqlite::Row<'_>) -> rusqlite::Result<EpisodeRecord> {
212    Ok(EpisodeRecord {
213        id: row.get(0)?,
214        thread_name: row.get(1)?,
215        session_id: row.get(2)?,
216        action: row.get(3)?,
217        content: row.get(4)?,
218        created_at: row.get(5)?,
219    })
220}
sac/store/threads.rs

sac/store/
threads.rs
