mi6_core/input/transcript/

scanner.rs

//! Transcript file scanner with storage integration.
//!
//! Provides high-level scanning functionality that orchestrates parsing,
//! deduplication, and storage operations.

use std::path::Path;

use crate::model::error::ScanError;
use crate::model::{EventBuilder, EventType, Storage};

use super::{TranscriptParser, extract_first_prompt};

/// Result of scanning a transcript file.
#[derive(Debug, Clone, Default)]
pub struct ScanResult {
    /// Number of events inserted
    pub inserted: usize,
    /// Number of lines that failed to parse
    pub parse_errors: u64,
}

/// Orchestrates transcript file scanning with storage integration.
///
/// `TranscriptScanner` provides a high-level API for scanning Claude Code
/// transcript files and inserting events into storage. It handles:
/// - Incremental parsing (only reads new content since last scan)
/// - Deduplication by UUID
/// - Position tracking for future scans
///
/// # Example
///
/// ```ignore
/// use mi6_core::{Storage, TranscriptScanner};
/// use mi6_storage_sqlite::SqliteStorage;
/// use std::path::Path;
///
/// let storage = SqliteStorage::open(Path::new("mi6.db"))?;
/// let scanner = TranscriptScanner::new(&storage, "my-machine-id");
///
/// let result = scanner.scan_file(Path::new("/path/to/transcript.jsonl"))?;
/// println!("Inserted {} events", result.inserted);
/// ```
pub struct TranscriptScanner<'a, S: Storage> {
    storage: &'a S,
    machine_id: String,
}

impl<'a, S: Storage> TranscriptScanner<'a, S> {
    /// Create a new scanner with the given storage and machine ID.
    ///
    /// The machine ID is used to tag events with their source machine,
    /// enabling multi-machine aggregation.
    pub fn new(storage: &'a S, machine_id: impl Into<String>) -> Self {
        Self {
            storage,
            machine_id: machine_id.into(),
        }
    }

    /// Scan a transcript file and insert events into storage.
    ///
    /// This method:
    /// 1. Loads the last scanned position for the file
    /// 2. Parses new entries since that position
    /// 3. Inserts new events (checking for duplicates by UUID)
    /// 4. Updates the position for future incremental scans
    ///
    /// # Returns
    ///
    /// Returns a [`ScanResult`] with the count of inserted events and
    /// parse errors encountered.
    ///
    /// # Errors
    ///
    /// Returns a [`ScanError`] if:
    /// - The file cannot be read
    /// - Storage operations fail
    pub fn scan_file(&self, path: &Path) -> Result<ScanResult, ScanError> {
        // Get last scanned position (or start from beginning)
        let start_position = self
            .storage
            .get_transcript_position(path)?
            .unwrap_or_default();

        // Parse the transcript file
        let parser = TranscriptParser::new(&self.machine_id);
        let result = parser.parse_incremental(path, &start_position)?;

        let parse_errors = result.parse_errors;

        // Insert new events, skipping duplicates
        let mut inserted = 0;
        for event in result.events {
            // Check for duplicate by UUID in metadata
            if let Some(metadata) = &event.metadata
                && let Ok(meta_json) = serde_json::from_str::<serde_json::Value>(metadata)
                && let Some(uuid) = meta_json.get("uuid").and_then(|v| v.as_str())
            {
                // Skip if already exists
                if self.storage.event_exists_by_uuid(&event.session_id, uuid)? {
                    continue;
                }
            }

            self.storage.insert(&event)?;
            inserted += 1;
        }

        // Update position for next scan
        self.storage
            .set_transcript_position(path, &result.position)?;

        Ok(ScanResult {
            inserted,
            parse_errors,
        })
    }

    /// Scan transcripts for specific sessions by ID.
    ///
    /// Looks up each session, checks if it has a `transcript_path` that exists,
    /// and scans it. Sessions without transcript paths or with missing files
    /// are silently skipped.
    ///
    /// # Arguments
    ///
    /// * `session_ids` - Slice of session IDs to scan
    ///
    /// # Returns
    ///
    /// Returns an aggregated [`ScanResult`] with the total count of inserted
    /// events and parse errors across all scanned files.
    ///
    /// # Errors
    ///
    /// Returns a [`ScanError`] if a storage operation fails or a transcript
    /// file cannot be parsed.
    pub fn scan_sessions(&self, session_ids: &[&str]) -> Result<ScanResult, ScanError> {
        let mut total = ScanResult::default();
        for session_id in session_ids {
            if let Some(session) = self.storage.get_session(session_id)?
                && let Some(ref transcript_path) = session.transcript_path
            {
                let path = Path::new(transcript_path);
                if path.exists() {
                    let result = self.scan_file(path)?;
                    total.inserted += result.inserted;
                    total.parse_errors += result.parse_errors;
                }
            }
        }
        Ok(total)
    }

    /// Backfill the initial prompt for a session from its transcript file.
    ///
    /// This is a fallback for when the UserPromptSubmit hook doesn't fire,
    /// such as when Claude is started with an inline prompt like
    /// `claude 'initial prompt'`.
    ///
    /// The method:
    /// 1. Checks if the session exists and has no first_user_message
    /// 2. Reads the transcript file to find the first user prompt
    /// 3. Creates a UserPromptSubmit event with the prompt
    ///
    /// This is idempotent - if the session already has a first_user_message,
    /// this method does nothing.
    ///
    /// # Arguments
    ///
    /// * `session_id` - The session ID to backfill
    /// * `transcript_path` - Path to the transcript file
    ///
    /// # Returns
    ///
    /// Returns `true` if a prompt was backfilled, `false` otherwise.
    pub fn backfill_initial_prompt(
        &self,
        session_id: &str,
        transcript_path: &Path,
    ) -> Result<bool, ScanError> {
        // Check if session exists and needs backfill
        let session = match self.storage.get_session(session_id)? {
            Some(s) => s,
            None => return Ok(false),
        };

        // If session already has a first_user_message, nothing to do
        if session.first_user_message.is_some() {
            return Ok(false);
        }

        // Try to extract the first prompt from the transcript
        let prompt = match extract_first_prompt(transcript_path) {
            Some(p) => p,
            None => return Ok(false),
        };

        // Create a UserPromptSubmit event with the prompt
        let event = EventBuilder::new(&self.machine_id, EventType::UserPromptSubmit, session_id)
            .source("transcript")
            .framework("claude")
            .payload(serde_json::json!({"prompt": prompt}).to_string())
            .build();

        // Insert the event (this will update first_user_message via session update logic)
        self.storage.insert(&event)?;

        Ok(true)
    }
}