1. 1//! Bulk import and incremental file-level indexing.
2. 2
3. 3use std::path::Path;
4. 4
5. 5use walkdir::WalkDir;
6. 6
7. 7use crate::{ImportStats, RebuildStats, Store, StoreError};
8. 8use vak_session::types::Entry;
9. 9
10. 10impl Store {
11. 11 /// Scan every `<home>/sessions/<hash>/*.jsonl` file and index entries.
12. 12 pub(crate) fn import_all(
13. 13 &self,
14. 14 conn: &rusqlite::Connection,
15. 15 sessions_home: &Path,
16. 16 ) -> Result<RebuildStats, StoreError> {
17. 17 let mut roots = Vec::new();
18. 18 let direct = sessions_home.join("sessions");
19. 19 if direct.exists() {
20. 20 roots.push((sessions_home.to_path_buf(), direct));
21. 21 }
22. 22 if let Ok(agents) = std::fs::read_dir(sessions_home.join("agents")) {
23. 23 for agent in agents.flatten() {
24. 24 let s = agent.path().join("sessions");
25. 25 if s.exists() {
26. 26 roots.push((agent.path(), s));
27. 27 }
28. 28 }
29. 29 }
30. 30 if let Some(siblings) = sessions_home
31. 31 .parent()
32. 32 .filter(|p| p.file_name().and_then(|s| s.to_str()) == Some("agents"))
33. 33 .and_then(|p| std::fs::read_dir(p).ok())
34. 34 {
35. 35 for sibling in siblings.flatten() {
36. 36 let s = sibling.path().join("sessions");
37. 37 if s.exists() && !roots.iter().any(|(_, r)| r == &s) {
38. 38 roots.push((sibling.path(), s));
39. 39 }
40. 40 }
41. 41 }
42. 42 if roots.is_empty() {
43. 43 return Ok(RebuildStats {
44. 44 files_scanned: 0,
45. 45 entries_indexed: 0,
46. 46 fts_rows: 0,
47. 47 });
48. 48 }
49. 49 let mut files_scanned = 0usize;
50. 50 let mut entries_indexed = 0usize;
51. 51 let mut fts_rows = 0usize;
52. 52
53. 53 for (home, root) in roots {
54. 54 for entry in WalkDir::new(&root)
55. 55 .min_depth(2)
56. 56 .max_depth(2)
57. 57 .follow_links(false)
58. 58 .into_iter()
59. 59 .filter_entry(|e| e.file_type().is_file())
60. 60 {
61. 61 let entry = entry.map_err(|e| std::io::Error::other(e.to_string()))?;
62. 62 let path = entry.path();
63. 63 if path.extension().and_then(|e| e.to_str()) != Some("jsonl") {
64. 64 continue;
65. 65 }
66. 66 let stats = self.import_file(conn, &home, path)?;
67. 67 entries_indexed += stats.entries_indexed;
68. 68 fts_rows += stats.fts_rows;
69. 69 files_scanned += 1;
70. 70 }
71. 71 }
72. 72
73. 73 Ok(RebuildStats {
74. 74 files_scanned,
75. 75 entries_indexed,
76. 76 fts_rows,
77. 77 })
78. 78 }
79. 79
80. 80 /// Parse a single JSONL file and insert all entries into the index.
81. 81 /// Already-indexed entry IDs are skipped.
82. 82 pub(crate) fn import_file(
83. 83 &self,
84. 84 conn: &rusqlite::Connection,
85. 85 _sessions_home: &Path,
86. 86 jsonl_path: &Path,
87. 87 ) -> Result<ImportStats, StoreError> {
88. 88 let session_id = jsonl_path
89. 89 .file_stem()
90. 90 .and_then(|s| s.to_str())
91. 91 .unwrap_or_default()
92. 92 .to_string();
93. 93
94. 94 // Derive project hash from the parent directory name.
95. 95 // Directory structure: <home>/sessions/<hash>/<session>.jsonl
96. 96 let project_hash = jsonl_path
97. 97 .parent()
98. 98 .and_then(|p| p.file_name())
99. 99 .and_then(|n| n.to_str())
100. 100 .unwrap_or_default()
101. 101 .to_string();
102. 102
103. 103 let content = std::fs::read_to_string(jsonl_path)?;
104. 104 let mut entries_indexed = 0usize;
105. 105 let mut fts_rows = 0usize;
106. 106 let mut skipped = 0usize;
107. 107
108. 108 // Prepare check and insert statements once.
109. 109 let mut check_stmt = conn.prepare("SELECT 1 FROM entries WHERE entry_id = ?1 LIMIT 1")?;
110. 110 let mut insert_entry = conn.prepare(
111. 111 "INSERT OR IGNORE INTO entries
112. 112 (entry_id, session_id, project_hash, parent_id, ts, kind, role,
113. 113 provider, model, tool_name, content_text, is_error)
114. 114 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)",
115. 115 )?;
116. 116 let mut insert_fts = conn.prepare(
117. 117 "INSERT INTO entries_fts(entry_id, session_id, project_hash, ts, kind, role, content)
118. 118 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
119. 119 )?;
120. 120
121. 121 for (i, line) in content.lines().enumerate() {
122. 122 let line = line.trim();
123. 123 if line.is_empty() {
124. 124 continue;
125. 125 }
126. 126 let entry: Entry = match serde_json::from_str(line) {
127. 127 Ok(e) => e,
128. 128 Err(_) => continue,
129. 129 };
130. 130
131. 131 // Skip already indexed.
132. 132 let already: bool = check_stmt
133. 133 .query_row(rusqlite::params![entry.id], |row| row.get::<_, i32>(0))
134. 134 .is_ok();
135. 135 if already {
136. 136 skipped += 1;
137. 137 continue;
138. 138 }
139. 139
140. 140 if let Some(meta) = self.extract_meta_with_hash(&session_id, &project_hash, &entry) {
141. 141 let has_text = !meta.content_text.trim().is_empty();
142. 142
143. 143 insert_entry.execute(rusqlite::params![
144. 144 meta.entry_id,
145. 145 meta.session_id,
146. 146 meta.project_hash,
147. 147 meta.parent_id,
148. 148 meta.ts,
149. 149 meta.kind.as_str(),
150. 150 meta.role,
151. 151 meta.provider,
152. 152 meta.model,
153. 153 meta.tool_name,
154. 154 meta.content_text,
155. 155 meta.is_error as i32,
156. 156 ])?;
157. 157
158. 158 if has_text {
159. 159 insert_fts.execute(rusqlite::params![
160. 160 meta.entry_id,
161. 161 meta.session_id,
162. 162 meta.project_hash,
163. 163 meta.ts,
164. 164 meta.kind.as_str(),
165. 165 meta.role.as_deref().unwrap_or(""),
166. 166 meta.content_text,
167. 167 ])?;
168. 168 fts_rows += 1;
169. 169 }
170. 170
171. 171 entries_indexed += 1;
172. 172 }
173. 173
174. 174 if i % 1000 == 999 {
175. 175 conn.execute("PRAGMA optimize", [])?;
176. 176 }
177. 177 }
178. 178
179. 179 Ok(ImportStats {
180. 180 entries_indexed,
181. 181 fts_rows,
182. 182 skipped,
183. 183 })
184. 184 }
185. 185
186. 186 /// Like `extract_meta` but attaches project_hash.
187. 187 fn extract_meta_with_hash(
188. 188 &self,
189. 189 session_id: &str,
190. 190 project_hash: &str,
191. 191 entry: &Entry,
192. 192 ) -> Option<crate::IndexedEntry> {
193. 193 let mut meta = self.extract_meta(session_id, entry)?;
194. 194 meta.project_hash = project_hash.to_string();
195. 195 Some(meta)
196. 196 }
197. 197}
198. 198