1use std::collections::BTreeSet;
2
3use anyhow::{bail, Result};
4use rusqlite::{params, Connection, OptionalExtension};
5
6use super::ExtractionTaskKind;
7use extraction_task::{
8 coalesce_extraction_task, extraction_task_for_replayed_event, with_capture_savepoint,
9};
10
11mod extraction_task;
12
13const DIRECT_CONTENT_BYTES: usize = 16 * 1024;
14
15pub struct CaptureEventInput<'a> {
16 pub host: &'a str,
17 pub session_id: &'a str,
18 pub project: &'a str,
19 pub cwd: Option<&'a str>,
20 pub event_type: &'a str,
21 pub role: Option<&'a str>,
22 pub tool_name: Option<&'a str>,
23 pub content: &'a str,
24 pub task_kind: Option<ExtractionTaskKind>,
25}
26
27#[derive(Debug, Clone, PartialEq, Eq)]
28pub struct CaptureEventOutcome {
29 pub event_row_id: i64,
30 pub event_id: String,
31 pub extraction_task_id: Option<i64>,
32}
33
34#[derive(Debug, Clone, Copy)]
35struct IdentityIds {
36 host_id: i64,
37 workspace_id: i64,
38 project_id: i64,
39 session_row_id: i64,
40}
41
42#[derive(Clone, Copy)]
43enum CaptureGitBranch<'a> {
44 DetectFromCwd,
45 Precomputed(Option<&'a str>),
46}
47
48pub fn record_captured_event(
49 conn: &Connection,
50 input: &CaptureEventInput<'_>,
51) -> Result<CaptureEventOutcome> {
52 record_captured_event_with_id(conn, input, None)
53}
54
55pub fn record_captured_event_with_id(
56 conn: &Connection,
57 input: &CaptureEventInput<'_>,
58 event_id_override: Option<&str>,
59) -> Result<CaptureEventOutcome> {
60 let now = chrono::Utc::now().timestamp();
61 record_captured_event_inner(
62 conn,
63 input,
64 event_id_override,
65 None,
66 now,
67 now,
68 None,
69 None,
70 CaptureGitBranch::DetectFromCwd,
71 )
72}
73
74pub(crate) fn record_captured_event_with_id_and_turn_id(
75 conn: &Connection,
76 input: &CaptureEventInput<'_>,
77 event_id_override: Option<&str>,
78 turn_id: Option<&str>,
79) -> Result<CaptureEventOutcome> {
80 let now = chrono::Utc::now().timestamp();
81 record_captured_event_inner(
82 conn,
83 input,
84 event_id_override,
85 turn_id,
86 now,
87 now,
88 None,
89 None,
90 CaptureGitBranch::DetectFromCwd,
91 )
92}
93
94pub fn record_captured_event_with_id_and_reference_time(
95 conn: &Connection,
96 input: &CaptureEventInput<'_>,
97 event_id_override: Option<&str>,
98 reference_time_epoch: Option<i64>,
99) -> Result<CaptureEventOutcome> {
100 let now = chrono::Utc::now().timestamp();
101 let created_at_epoch = reference_time_epoch.unwrap_or(now);
102 record_captured_event_inner(
103 conn,
104 input,
105 event_id_override,
106 None,
107 created_at_epoch,
108 now,
109 reference_time_epoch,
110 None,
111 CaptureGitBranch::DetectFromCwd,
112 )
113}
114
115pub fn record_captured_event_with_id_and_reference_time_and_git_evidence(
116 conn: &Connection,
117 input: &CaptureEventInput<'_>,
118 event_id_override: Option<&str>,
119 reference_time_epoch: Option<i64>,
120 git_evidence: &[crate::git_util::GitCommitEvidence],
121) -> Result<CaptureEventOutcome> {
122 let now = chrono::Utc::now().timestamp();
123 let created_at_epoch = reference_time_epoch.unwrap_or(now);
124 record_captured_event_inner(
125 conn,
126 input,
127 event_id_override,
128 None,
129 created_at_epoch,
130 now,
131 reference_time_epoch,
132 Some(git_evidence),
133 CaptureGitBranch::DetectFromCwd,
134 )
135}
136
137pub fn record_captured_event_with_id_and_created_at(
138 conn: &Connection,
139 input: &CaptureEventInput<'_>,
140 event_id_override: Option<&str>,
141 created_at_epoch: i64,
142) -> Result<CaptureEventOutcome> {
143 let now = chrono::Utc::now().timestamp();
144 record_captured_event_inner(
145 conn,
146 input,
147 event_id_override,
148 None,
149 created_at_epoch,
150 now,
151 Some(created_at_epoch),
152 None,
153 CaptureGitBranch::DetectFromCwd,
154 )
155}
156
157pub(crate) fn record_captured_event_with_id_and_created_at_and_precomputed_git_branch(
158 conn: &Connection,
159 input: &CaptureEventInput<'_>,
160 event_id_override: Option<&str>,
161 created_at_epoch: i64,
162 git_branch: Option<&str>,
163) -> Result<CaptureEventOutcome> {
164 record_captured_event_with_precomputed_git_branch(
165 conn,
166 input,
167 event_id_override,
168 Some(created_at_epoch),
169 &[],
170 git_branch,
171 )
172}
173
174pub(crate) fn record_captured_event_with_precomputed_git_branch(
177 conn: &Connection,
178 input: &CaptureEventInput<'_>,
179 event_id_override: Option<&str>,
180 reference_time_epoch: Option<i64>,
181 git_evidence: &[crate::git_util::GitCommitEvidence],
182 git_branch: Option<&str>,
183) -> Result<CaptureEventOutcome> {
184 let now = chrono::Utc::now().timestamp();
185 let created_at_epoch = reference_time_epoch.unwrap_or(now);
186 record_captured_event_inner(
187 conn,
188 input,
189 event_id_override,
190 None,
191 created_at_epoch,
192 now,
193 reference_time_epoch,
194 Some(git_evidence),
195 CaptureGitBranch::Precomputed(git_branch),
196 )
197}
198
199fn record_captured_event_inner(
200 conn: &Connection,
201 input: &CaptureEventInput<'_>,
202 event_id_override: Option<&str>,
203 turn_id: Option<&str>,
204 created_at_epoch: i64,
205 now: i64,
206 reference_time_epoch: Option<i64>,
207 git_evidence: Option<&[crate::git_util::GitCommitEvidence]>,
208 git_branch: CaptureGitBranch<'_>,
209) -> Result<CaptureEventOutcome> {
210 let inserted_at = now;
211 let sanitized_content = redact_capture_content(input.content);
212 let content_hash = exact_hash(&sanitized_content);
213 let event_id = event_id_override
214 .map(ToString::to_string)
215 .unwrap_or_else(|| synthesize_event_id(input.event_type, &content_hash));
216 let sanitized_git_evidence = git_evidence
217 .unwrap_or_default()
218 .iter()
219 .cloned()
220 .map(|mut evidence| {
221 evidence.metadata = crate::git_util::sanitize_commit_metadata(evidence.metadata);
222 evidence
223 })
224 .collect::<Vec<_>>();
225 let git_branch = match git_branch {
226 CaptureGitBranch::DetectFromCwd => input.cwd.and_then(crate::db::detect_git_branch),
227 CaptureGitBranch::Precomputed(git_branch) => git_branch.map(ToString::to_string),
228 };
229 with_capture_savepoint(conn, || {
230 let identity = upsert_identity(conn, input, git_branch.as_deref(), now)?;
231 let existing_event_row_id: Option<i64> = conn
232 .query_row(
233 "SELECT id FROM captured_events
234 WHERE host_id = ?1 AND session_id = ?2 AND event_id = ?3",
235 params![identity.host_id, input.session_id, event_id],
236 |row| row.get(0),
237 )
238 .optional()?;
239 let (content_text, content_blob_id, retention_class) =
240 store_content(conn, &sanitized_content, &content_hash, now)?;
241 let token_estimate = estimate_tokens(&sanitized_content);
242 conn.execute(
243 "INSERT INTO captured_events
244 (host_id, workspace_id, project_id, session_row_id, session_id, turn_id,
245 event_id, event_type, role, tool_name, content_text, content_blob_id,
246 content_hash, token_estimate, retention_class, created_at_epoch, inserted_at_epoch,
247 reference_time_epoch)
248 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18)
249 ON CONFLICT(host_id, session_id, event_id) DO UPDATE SET
250 inserted_at_epoch = excluded.inserted_at_epoch,
251 turn_id = COALESCE(excluded.turn_id, captured_events.turn_id),
252 reference_time_epoch = COALESCE(excluded.reference_time_epoch, captured_events.reference_time_epoch)",
253 params![
254 identity.host_id,
255 identity.workspace_id,
256 identity.project_id,
257 identity.session_row_id,
258 input.session_id,
259 turn_id,
260 event_id,
261 input.event_type,
262 input.role,
263 input.tool_name,
264 content_text,
265 content_blob_id,
266 content_hash,
267 token_estimate,
268 retention_class,
269 created_at_epoch,
270 inserted_at,
271 reference_time_epoch
272 ],
273 )?;
274
275 let event_row_id = conn.query_row(
276 "SELECT id FROM captured_events WHERE host_id = ?1 AND session_id = ?2 AND event_id = ?3",
277 params![identity.host_id, input.session_id, event_id],
278 |row| row.get(0),
279 )?;
280
281 let mut inserted_git_evidence_keys = BTreeSet::new();
282 for evidence in &sanitized_git_evidence {
283 let inserted = conn.execute(
284 "INSERT INTO captured_event_commits
285 (event_row_id, sha, metadata_json, evidence_kind, evidence_locator)
286 VALUES (?1, ?2, ?3, ?4, ?5)
287 ON CONFLICT(event_row_id, sha, evidence_kind) DO NOTHING",
288 params![
289 event_row_id,
290 evidence.metadata.sha,
291 serde_json::to_string(&evidence.metadata)?,
292 evidence.kind.as_str(),
293 evidence.locator
294 ],
295 )?;
296 if inserted > 0 {
297 inserted_git_evidence_keys.insert(format!(
298 "{}:{}",
299 evidence.kind.as_str(),
300 evidence.metadata.sha.trim().to_ascii_lowercase()
301 ));
302 }
303 }
304 let late_git_evidence_key = (!inserted_git_evidence_keys.is_empty()).then(|| {
305 exact_hash(
306 &inserted_git_evidence_keys
307 .into_iter()
308 .collect::<Vec<_>>()
309 .join("\n"),
310 )
311 });
312
313 let extraction_task_id = if let Some(kind) = input.task_kind {
314 if existing_event_row_id.is_some() {
315 Some(extraction_task_for_replayed_event(
316 conn,
317 identity,
318 kind,
319 event_row_id,
320 late_git_evidence_key.as_deref(),
321 now,
322 )?)
323 } else {
324 Some(coalesce_extraction_task(
325 conn,
326 identity,
327 kind,
328 event_row_id,
329 now,
330 )?)
331 }
332 } else {
333 None
334 };
335
336 Ok(CaptureEventOutcome {
337 event_row_id,
338 event_id,
339 extraction_task_id,
340 })
341 })
342}
343
344pub(crate) fn ensure_project_row(conn: &Connection, project_path: &str) -> Result<i64> {
348 let now = chrono::Utc::now().timestamp();
349 let project_path = crate::project_alias::canonical_project_path_for_write(conn, project_path)?;
350 let workspace_id = upsert_workspace(conn, &project_path, None, now)?;
351 upsert_project(conn, workspace_id, &project_path, now)
352}
353
354fn upsert_identity(
355 conn: &Connection,
356 input: &CaptureEventInput<'_>,
357 git_branch: Option<&str>,
358 now: i64,
359) -> Result<IdentityIds> {
360 let host_id = upsert_host(conn, normalize_host(input.host)?, now)?;
361 let root_path = crate::project_alias::canonical_project_path_for_write(conn, input.project)?;
362 let workspace_id = upsert_workspace(conn, &root_path, git_branch, now)?;
363 let project_id = upsert_project(conn, workspace_id, &root_path, now)?;
364 let session_row_id = upsert_session_row(
365 conn,
366 host_id,
367 workspace_id,
368 project_id,
369 input.session_id,
370 now,
371 )?;
372 Ok(IdentityIds {
373 host_id,
374 workspace_id,
375 project_id,
376 session_row_id,
377 })
378}
379
380fn normalize_host(host: &str) -> Result<&str> {
381 match host {
382 "claude-code" | "codex-cli" | "cursor" => Ok(host),
383 other => {
384 bail!("invalid capture host '{other}'; expected claude-code, codex-cli, or cursor")
385 }
386 }
387}
388
389fn upsert_host(conn: &Connection, name: &str, now: i64) -> Result<i64> {
390 conn.execute(
391 "INSERT OR IGNORE INTO hosts(name, enabled, created_at_epoch) VALUES (?1, 1, ?2)",
392 params![name, now],
393 )?;
394 Ok(conn.query_row(
395 "SELECT id FROM hosts WHERE name = ?1",
396 params![name],
397 |row| row.get(0),
398 )?)
399}
400
401fn upsert_workspace(
402 conn: &Connection,
403 root_path: &str,
404 git_branch: Option<&str>,
405 now: i64,
406) -> Result<i64> {
407 conn.execute(
408 "INSERT INTO workspaces(root_path, git_remote, git_branch, created_at_epoch, updated_at_epoch)
409 VALUES (?1, NULL, ?2, ?3, ?3)
410 ON CONFLICT(root_path) DO UPDATE SET
411 git_branch = COALESCE(excluded.git_branch, workspaces.git_branch),
412 updated_at_epoch = excluded.updated_at_epoch",
413 params![root_path, git_branch, now],
414 )?;
415 Ok(conn.query_row(
416 "SELECT id FROM workspaces WHERE root_path = ?1",
417 params![root_path],
418 |row| row.get(0),
419 )?)
420}
421
422fn upsert_project(
423 conn: &Connection,
424 workspace_id: i64,
425 project_path: &str,
426 now: i64,
427) -> Result<i64> {
428 let project_key = project_path
429 .rsplit('/')
430 .find(|part| !part.is_empty())
431 .unwrap_or(project_path);
432 conn.execute(
433 "INSERT INTO projects(workspace_id, project_path, project_key, created_at_epoch, updated_at_epoch)
434 VALUES (?1, ?2, ?3, ?4, ?4)
435 ON CONFLICT(workspace_id, project_path) DO UPDATE SET
436 project_key = excluded.project_key,
437 updated_at_epoch = excluded.updated_at_epoch",
438 params![workspace_id, project_path, project_key, now],
439 )?;
440 Ok(conn.query_row(
441 "SELECT id FROM projects WHERE workspace_id = ?1 AND project_path = ?2",
442 params![workspace_id, project_path],
443 |row| row.get(0),
444 )?)
445}
446
447fn upsert_session_row(
448 conn: &Connection,
449 host_id: i64,
450 workspace_id: i64,
451 project_id: i64,
452 session_id: &str,
453 now: i64,
454) -> Result<i64> {
455 conn.execute(
456 "INSERT INTO sessions(host_id, workspace_id, project_id, session_id, started_at_epoch, last_seen_at_epoch, status)
457 VALUES (?1, ?2, ?3, ?4, ?5, ?5, 'active')
458 ON CONFLICT(host_id, project_id, session_id) DO UPDATE SET
459 last_seen_at_epoch = excluded.last_seen_at_epoch,
460 status = 'active'",
461 params![host_id, workspace_id, project_id, session_id, now],
462 )?;
463 Ok(conn.query_row(
464 "SELECT id FROM sessions WHERE host_id = ?1 AND project_id = ?2 AND session_id = ?3",
465 params![host_id, project_id, session_id],
466 |row| row.get(0),
467 )?)
468}
469
470fn store_content(
471 conn: &Connection,
472 content: &str,
473 content_hash: &str,
474 now: i64,
475) -> Result<(String, Option<i64>, &'static str)> {
476 if content.len() <= DIRECT_CONTENT_BYTES {
477 return Ok((content.to_string(), None, "raw_keep"));
478 }
479
480 let bytes = content.as_bytes();
481 if let Some(blob_id) = matching_legacy_blob_id(conn, content)? {
482 return Ok((
483 compact_preview(content, DIRECT_CONTENT_BYTES),
484 Some(blob_id),
485 "raw_compact",
486 ));
487 }
488
489 conn.execute(
490 "INSERT INTO event_blobs(content_hash, content_encoding, content_bytes, original_bytes, stored_bytes, created_at_epoch)
491 VALUES (?1, 'plain', ?2, ?3, ?3, ?4)
492 ON CONFLICT(content_hash) DO NOTHING",
493 params![content_hash, bytes, bytes.len() as i64, now],
494 )?;
495 let blob_id: i64 = conn
496 .query_row(
497 "SELECT id FROM event_blobs WHERE content_hash = ?1",
498 params![content_hash],
499 |row| row.get(0),
500 )
501 .optional()?
502 .expect("event blob row should exist after insert");
503 Ok((
504 compact_preview(content, DIRECT_CONTENT_BYTES),
505 Some(blob_id),
506 "raw_compact",
507 ))
508}
509
510fn matching_legacy_blob_id(conn: &Connection, content: &str) -> Result<Option<i64>> {
511 let legacy_hash = legacy_exact_hash(content);
512 let Some((id, encoding, bytes)) = conn
513 .query_row(
514 "SELECT id, content_encoding, content_bytes FROM event_blobs WHERE content_hash = ?1",
515 params![legacy_hash],
516 |row| {
517 Ok((
518 row.get::<_, i64>(0)?,
519 row.get::<_, String>(1)?,
520 row.get::<_, Vec<u8>>(2)?,
521 ))
522 },
523 )
524 .optional()?
525 else {
526 return Ok(None);
527 };
528
529 if encoding == "plain" && bytes == content.as_bytes() {
530 Ok(Some(id))
531 } else {
532 Ok(None)
533 }
534}
535
536fn exact_hash(content: &str) -> String {
537 crate::db::content_identity_hash(content.as_bytes())
538}
539
540fn legacy_exact_hash(content: &str) -> String {
541 crate::db::legacy_content_identity_hash(content.as_bytes())
542}
543
544pub fn unique_capture_event_id(event_type: &str, content: &str) -> String {
545 let sanitized_content = redact_capture_content(content);
546 let nanos = chrono::Utc::now()
547 .timestamp_nanos_opt()
548 .unwrap_or_else(|| chrono::Utc::now().timestamp() * 1_000_000_000);
549 format!(
550 "{}-{}-{}",
551 event_type,
552 nanos,
553 exact_hash(&sanitized_content)
554 )
555}
556
557fn synthesize_event_id(event_type: &str, content_hash: &str) -> String {
558 let nanos = chrono::Utc::now()
559 .timestamp_nanos_opt()
560 .unwrap_or_else(|| chrono::Utc::now().timestamp() * 1_000_000_000);
561 format!("{}-{}-{}", event_type, nanos, content_hash)
562}
563
564fn estimate_tokens(content: &str) -> i64 {
565 ((content.len() as i64) + 3) / 4
566}
567
568pub(crate) fn redact_capture_content(content: &str) -> String {
569 if let Ok(value) = serde_json::from_str::<serde_json::Value>(content) {
570 let mut redacted = crate::adapter::common::redact_sensitive_value(&value);
571 preserve_capture_path_field(&mut redacted, &value, "cwd");
572 preserve_capture_path_field(&mut redacted, &value, "transcript_path");
573 return serde_json::to_string(&redacted)
574 .unwrap_or_else(|_| crate::adapter::common::redact_sensitive_text(content));
575 }
576 crate::adapter::common::redact_sensitive_text(content)
577}
578
579fn preserve_capture_path_field(
580 redacted: &mut serde_json::Value,
581 original: &serde_json::Value,
582 key: &str,
583) {
584 let (Some(redacted_obj), Some(original_obj)) = (redacted.as_object_mut(), original.as_object())
585 else {
586 return;
587 };
588 if let Some(original_value) = original_obj.get(key).and_then(serde_json::Value::as_str) {
589 redacted_obj.insert(
590 key.to_string(),
591 serde_json::Value::String(original_value.to_string()),
592 );
593 }
594}
595
596fn compact_preview(content: &str, max_bytes: usize) -> String {
597 let half = (max_bytes / 2).saturating_sub(128);
598 let prefix = crate::db::truncate_str(content, half).to_string();
599 let suffix_start = content.len().saturating_sub(half);
600 let suffix = if content.is_char_boundary(suffix_start) {
601 &content[suffix_start..]
602 } else {
603 let mut start = suffix_start;
604 while start < content.len() && !content.is_char_boundary(start) {
605 start += 1;
606 }
607 &content[start..]
608 };
609 format!(
610 "{}\n\n[remem raw event compacted: original_bytes={}]\n\n{}",
611 prefix,
612 content.len(),
613 suffix
614 )
615}
616
617#[cfg(test)]
618#[path = "capture/tests.rs"]
619mod tests;