zeph_memory/store/
mem_scenes.rs

1// SPDX-FileCopyrightText: 2026 Andrei G <bug-ops>
2// SPDX-License-Identifier: MIT OR Apache-2.0
3
4//! `SQLite` operations for `MemScene` consolidation (#2332).
5
6use crate::scenes::MemScene;
7use crate::types::{MemSceneId, MessageId};
8#[allow(unused_imports)]
9use zeph_db::sql;
10
11use zeph_db::ActiveDialect;
12
13use crate::error::MemoryError;
14
15use super::SqliteStore;
16
17impl SqliteStore {
18    /// Fetch semantic-tier messages not yet assigned to any scene.
19    ///
20    /// Returns `(message_id, content)` pairs ordered by `id` ASC.
21    ///
22    /// # Errors
23    ///
24    /// Returns an error if the `SQLite` query fails.
25    pub async fn find_unscened_semantic_messages(
26        &self,
27        limit: usize,
28    ) -> Result<Vec<(MessageId, String)>, MemoryError> {
29        let limit_i64 = i64::try_from(limit).unwrap_or(i64::MAX);
30        let rows: Vec<(i64, String)> = zeph_db::query_as(sql!(
31            "SELECT m.id, m.content
32             FROM messages m
33             WHERE m.tier = 'semantic'
34               AND m.deleted_at IS NULL
35               AND m.id NOT IN (SELECT message_id FROM mem_scene_members)
36             ORDER BY m.id ASC
37             LIMIT ?"
38        ))
39        .bind(limit_i64)
40        .fetch_all(&self.pool)
41        .await?;
42
43        Ok(rows
44            .into_iter()
45            .map(|(id, content)| (MessageId(id), content))
46            .collect())
47    }
48
49    /// Insert a new `MemScene` and link member messages to it.
50    ///
51    /// Returns the new scene's ID.
52    ///
53    /// # Errors
54    ///
55    /// Returns an error if the `SQLite` insert fails.
56    pub async fn insert_mem_scene(
57        &self,
58        label: &str,
59        profile: &str,
60        member_ids: &[MessageId],
61    ) -> Result<MemSceneId, MemoryError> {
62        let member_count = i64::try_from(member_ids.len()).unwrap_or(0);
63        let mut tx = self.pool.begin().await?;
64
65        let row: (i64,) = zeph_db::query_as(sql!(
66            "INSERT INTO mem_scenes (label, profile, member_count) VALUES (?, ?, ?) RETURNING id"
67        ))
68        .bind(label)
69        .bind(profile)
70        .bind(member_count)
71        .fetch_one(&mut *tx)
72        .await?;
73        let scene_id = row.0;
74
75        let member_sql = zeph_db::rewrite_placeholders(&format!(
76            "{} INTO mem_scene_members (scene_id, message_id) VALUES (?, ?){}",
77            <ActiveDialect as zeph_db::dialect::Dialect>::INSERT_IGNORE,
78            <ActiveDialect as zeph_db::dialect::Dialect>::CONFLICT_NOTHING,
79        ));
80        for &msg_id in member_ids {
81            zeph_db::query(sqlx::AssertSqlSafe(member_sql.as_str()))
82                .bind(scene_id)
83                .bind(msg_id.0)
84                .execute(&mut *tx)
85                .await?;
86        }
87
88        tx.commit().await?;
89        Ok(MemSceneId(scene_id))
90    }
91
92    /// List all `MemScenes` ordered by creation time descending.
93    ///
94    /// # Errors
95    ///
96    /// Returns an error if the `SQLite` query fails.
97    pub async fn list_mem_scenes(&self) -> Result<Vec<MemScene>, MemoryError> {
98        let rows: Vec<(i64, String, String, i64, i64, i64)> = zeph_db::query_as(sql!(
99            "SELECT id, label, profile, member_count, created_at, updated_at \
100             FROM mem_scenes ORDER BY created_at DESC"
101        ))
102        .fetch_all(&self.pool)
103        .await?;
104
105        Ok(rows
106            .into_iter()
107            .map(
108                |(id, label, profile, member_count, created_at, updated_at)| MemScene {
109                    id: MemSceneId(id),
110                    label,
111                    profile,
112                    member_count: u32::try_from(member_count).unwrap_or(0),
113                    created_at,
114                    updated_at,
115                },
116            )
117            .collect())
118    }
119
120    /// Fetch member message IDs for a given scene.
121    ///
122    /// # Errors
123    ///
124    /// Returns an error if the `SQLite` query fails.
125    pub async fn scene_member_ids(
126        &self,
127        scene_id: MemSceneId,
128    ) -> Result<Vec<MessageId>, MemoryError> {
129        let rows: Vec<(i64,)> = zeph_db::query_as(sql!(
130            "SELECT message_id FROM mem_scene_members WHERE scene_id = ?"
131        ))
132        .bind(scene_id.0)
133        .fetch_all(&self.pool)
134        .await?;
135
136        Ok(rows.into_iter().map(|(id,)| MessageId(id)).collect())
137    }
138
139    /// Delete all `MemScenes` and their memberships (reset for re-clustering).
140    ///
141    /// # Errors
142    ///
143    /// Returns an error if the `SQLite` delete fails.
144    pub async fn reset_mem_scenes(&self) -> Result<u64, MemoryError> {
145        let result = zeph_db::query(sql!("DELETE FROM mem_scenes"))
146            .execute(&self.pool)
147            .await?;
148        Ok(result.rows_affected())
149    }
150}
151
152#[cfg(test)]
153mod tests {
154    use super::*;
155
156    async fn make_store() -> SqliteStore {
157        // pool_size=1 is required: SQLite :memory: creates an isolated database per
158        // connection, so a multi-connection pool would split inserts across empty DBs,
159        // causing FK violations when querying across connections.
160        SqliteStore::with_pool_size(":memory:", 1).await.unwrap()
161    }
162
163    /// Create N real messages in the DB and return their IDs.
164    async fn seed_messages(store: &SqliteStore, n: usize) -> Vec<MessageId> {
165        let cid = store.create_conversation().await.unwrap();
166        let mut ids = Vec::with_capacity(n);
167        for i in 0..n {
168            let id = store
169                .save_message(cid, "user", &format!("msg {i}"))
170                .await
171                .unwrap();
172            ids.push(id);
173        }
174        ids
175    }
176
177    // Test: insert_mem_scene creates a scene and links member messages.
178    #[tokio::test]
179    async fn insert_and_list_scene() {
180        let store = make_store().await;
181        let ids = seed_messages(&store, 2).await;
182
183        let scene_id = store
184            .insert_mem_scene("Rust Auth", "JWT tokens used for RS256.", &ids)
185            .await
186            .unwrap();
187        assert!(scene_id.0 > 0, "scene id must be positive");
188
189        let scenes = store.list_mem_scenes().await.unwrap();
190        assert_eq!(scenes.len(), 1);
191        assert_eq!(scenes[0].label, "Rust Auth");
192        assert_eq!(scenes[0].member_count, 2);
193    }
194
195    // Test: scene_member_ids returns linked message IDs (scene member expansion on demand).
196    #[tokio::test]
197    async fn scene_member_ids_expansion() {
198        let store = make_store().await;
199        let ids = seed_messages(&store, 3).await;
200
201        let scene_id = store
202            .insert_mem_scene("Topic A", "Profile text.", &ids)
203            .await
204            .unwrap();
205
206        let members = store.scene_member_ids(scene_id).await.unwrap();
207        assert_eq!(members.len(), 3);
208        for id in &ids {
209            assert!(members.contains(id), "member {id} must be in expansion");
210        }
211    }
212
213    // Test: find_unscened_semantic_messages excludes already-assigned messages.
214    #[tokio::test]
215    async fn find_unscened_excludes_assigned_members() {
216        let store = make_store().await;
217        let ids = seed_messages(&store, 3).await;
218
219        // Promote all to semantic tier.
220        for id in &ids {
221            zeph_db::query(sql!("UPDATE messages SET tier = 'semantic' WHERE id = ?"))
222                .bind(id.0)
223                .execute(store.pool())
224                .await
225                .unwrap();
226        }
227
228        // All three should appear as unscened.
229        let unscened = store.find_unscened_semantic_messages(100).await.unwrap();
230        assert_eq!(unscened.len(), 3);
231
232        // Assign first two to a scene.
233        store
234            .insert_mem_scene("Partial Scene", "Some profile", &ids[..2])
235            .await
236            .unwrap();
237
238        // Now only the third should be unscened.
239        let unscened_after = store.find_unscened_semantic_messages(100).await.unwrap();
240        assert_eq!(unscened_after.len(), 1);
241        assert_eq!(unscened_after[0].0, ids[2]);
242    }
243
244    // Test: reset_mem_scenes clears all scenes and allows re-clustering.
245    #[tokio::test]
246    async fn reset_scenes_clears_all() {
247        let store = make_store().await;
248        let ids1 = seed_messages(&store, 1).await;
249        let ids2 = seed_messages(&store, 1).await;
250
251        store
252            .insert_mem_scene("Scene 1", "Profile 1", &ids1)
253            .await
254            .unwrap();
255        store
256            .insert_mem_scene("Scene 2", "Profile 2", &ids2)
257            .await
258            .unwrap();
259
260        let deleted = store.reset_mem_scenes().await.unwrap();
261        assert_eq!(deleted, 2);
262
263        let scenes = store.list_mem_scenes().await.unwrap();
264        assert!(scenes.is_empty());
265    }
266
267    // Test: list_mem_scenes returns newest first (by id DESC as proxy for created_at DESC).
268    // unixepoch() has 1-second resolution — use explicit created_at override via INSERT to
269    // guarantee ordering even when both inserts happen within the same second.
270    #[tokio::test]
271    async fn list_scenes_ordered_newest_first() {
272        let store = make_store().await;
273        let ids1 = seed_messages(&store, 1).await;
274        let ids2 = seed_messages(&store, 1).await;
275
276        // Insert directly with distinct created_at values to avoid single-second collision.
277        zeph_db::query(
278            sql!("INSERT INTO mem_scenes (label, profile, member_count, created_at, updated_at) VALUES (?, ?, ?, ?, ?)"),
279        )
280        .bind("First")
281        .bind("Profile first")
282        .bind(1i64)
283        .bind(1_000_000i64)
284        .bind(1_000_000i64)
285        .execute(store.pool())
286        .await
287        .unwrap();
288        let scene1_id: (i64,) = zeph_db::query_as(sql!("SELECT last_insert_rowid()"))
289            .fetch_one(store.pool())
290            .await
291            .unwrap();
292
293        zeph_db::query(
294            sql!("INSERT INTO mem_scenes (label, profile, member_count, created_at, updated_at) VALUES (?, ?, ?, ?, ?)"),
295        )
296        .bind("Second")
297        .bind("Profile second")
298        .bind(1i64)
299        .bind(2_000_000i64)
300        .bind(2_000_000i64)
301        .execute(store.pool())
302        .await
303        .unwrap();
304        let scene2_id: (i64,) = zeph_db::query_as(sql!("SELECT last_insert_rowid()"))
305            .fetch_one(store.pool())
306            .await
307            .unwrap();
308
309        // Link messages to satisfy FK.
310        zeph_db::query(sql!(
311            "INSERT INTO mem_scene_members (scene_id, message_id) VALUES (?, ?)"
312        ))
313        .bind(scene1_id.0)
314        .bind(ids1[0].0)
315        .execute(store.pool())
316        .await
317        .unwrap();
318        zeph_db::query(sql!(
319            "INSERT INTO mem_scene_members (scene_id, message_id) VALUES (?, ?)"
320        ))
321        .bind(scene2_id.0)
322        .bind(ids2[0].0)
323        .execute(store.pool())
324        .await
325        .unwrap();
326
327        let scenes = store.list_mem_scenes().await.unwrap();
328        // "Second" has created_at=2_000_000 > "First" created_at=1_000_000 → Second comes first.
329        assert_eq!(scenes.len(), 2);
330        assert_eq!(scenes[0].label, "Second");
331        assert_eq!(scenes[1].label, "First");
332    }
333}
zeph_memory/store/mem_scenes.rs

zeph_memory/store/
mem_scenes.rs
