WhispAssist/src-tauri/src/storage/mod.rs

//! Storage service (Phase 2, FR-STORE-*, FR-REL-*). SQLite index + on-disk
//! audio/transcript/notes files (ADR-0006). Async via sqlx.
//!
//! Invariant: audio is the source of truth. Derived artifacts (transcript/notes/
//! summary) are regenerable; retention never touches an in-progress meeting.

use crate::models::{
    ActionItem, CalendarEvent, ContextExcerpt, FeatureBriefInfo, ImportedEvent, McpAccessEntry,
    MeetingId, MeetingListItem, MeetingStatus, Participant, SearchHit, SpeakerInfo,
    TranscriptSegment,
};
use crate::paths;
use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use sqlx::sqlite::{SqliteConnectOptions, SqlitePool, SqlitePoolOptions};
use sqlx::Row;
use std::collections::HashMap;
use std::path::PathBuf;
use std::time::{SystemTime, UNIX_EPOCH};

#[derive(Debug, thiserror::Error)]
pub enum StoreError {
    #[error("db error: {0}")]
    Db(String),
    #[error("io error: {0}")]
    Io(#[from] std::io::Error),
    #[error("not found: {0}")]
    NotFound(String),
}

impl From<sqlx::Error> for StoreError {
    fn from(e: sqlx::Error) -> Self {
        StoreError::Db(e.to_string())
    }
}

pub struct NewMeeting {
    pub title: String,
    pub calendar_event_id: Option<String>,
    /// Note-template id (Phase 8, T8.1, FR-NOTE-5) — resolved against
    /// `notes::templates::catalog()`, kept alongside the meeting so
    /// re-rendering notes.md on reprocess/resume reapplies the same
    /// section structure instead of losing it.
    pub template_id: Option<String>,
    /// Transcription language requested at recording start (T8.7, FR-TRX-4):
    /// `None` means auto-detect. Recorded immediately (not just at
    /// `finalize_meeting`) so a crash-recovered `recovering` meeting still
    /// knows what was asked for; `finalize_meeting`'s `language` overwrites
    /// this with whatever whisper.cpp actually resolved/detected.
    pub language: Option<String>,
}

/// `list_meetings` filters (Phase 8, FR-SEARCH-2). All fields are ANDed
/// together; each is skipped when `None`.
#[derive(Debug, Default)]
pub struct MeetingFilter {
    /// Title substring match — separate from full-text `search()`.
    pub query: Option<String>,
    pub tag: Option<String>,
    /// Matches either a speaker mapped to this participant or an attendee of
    /// the meeting's linked calendar event.
    pub participant_id: Option<String>,
    pub from: Option<i64>,
    pub to: Option<i64>,
}

/// What actually happened during the session — richer than the doc's bare
/// `&MeetingId` because `finalize_meeting` has to persist what was captured.
pub struct FinalizeMeeting {
    pub segments: Vec<TranscriptSegment>,
    pub speakers: Vec<SpeakerInfo>,
    pub duration_secs: i64,
    pub recorded: bool,
    pub language: Option<String>,
    pub backend_used: Option<String>,
    pub model_used: Option<String>,
    /// How `audio.wav`'s channels are laid out (FR-SPK/FR-CAP): `"split"` =
    /// stereo mic-left/loopback-right, `"summed"`/`None` = mic mixed into every
    /// channel. `None` leaves the stored value unchanged (e.g. on reprocess).
    pub audio_layout: Option<String>,
}

/// Full meeting detail: DB row + transcript + speakers + notes (`get_meeting`'s
/// documented contract — `docs/04-api-contracts.md` — which the old
/// `MeetingListItem`-only stub couldn't represent).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Meeting {
    pub id: MeetingId,
    pub title: String,
    pub started_at: i64,
    pub ended_at: Option<i64>,
    pub duration_secs: Option<i64>,
    pub status: MeetingStatus,
    pub recorded: bool,
    pub language: Option<String>,
    pub backend_used: Option<String>,
    pub model_used: Option<String>,
    /// `audio.wav` channel layout (FR-SPK/FR-CAP): `"split"` (mic-left /
    /// loopback-right) or `"summed"`/`None` (mic mixed into every channel).
    /// Diarization, playback downmix, and bundle export branch on it.
    pub audio_layout: Option<String>,
    pub segments: Vec<TranscriptSegment>,
    pub speakers: Vec<SpeakerInfo>,
    pub notes_markdown: String,
    /// `None` until `generate_summary` has run at least once (Phase 5, FR-LLM-4).
    pub summary: Option<SummaryFile>,
    /// The linked calendar event, if any (T6.3/T6.6, FR-CAL-2/4) — set either
    /// at `start_recording` time or later via `attach_meeting_to_event`.
    pub calendar_event_id: Option<String>,
    /// Tags (Phase 8, FR-SEARCH-2), sorted.
    pub tags: Vec<String>,
    /// Note-template id (Phase 8, T8.1, FR-NOTE-5), if one was picked at
    /// creation — resolved against `notes::templates::catalog()`.
    pub template_id: Option<String>,
    /// Confirmed/edited action items from the `action_items` table — the
    /// source of truth the user manages directly (add/edit/delete via
    /// `confirm_action_items`, FR-LLM-3). Falls back to the drafts parsed
    /// into `summary.json` when the table has none yet, so a freshly
    /// generated summary still shows its suggestions.
    pub action_items: Vec<ActionItem>,
}

/// A calendar event with its attendees (T6.3/T6.4, FR-CAL-1/3) — the
/// "detail" fetch behind the pre-meeting context panel and the
/// participant-aware speaker-naming dropdown (T6.5).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CalendarEventDetail {
    pub event: CalendarEvent,
    pub participants: Vec<Participant>,
}

/// Result of `cleanup_calendar_events` — `protected` is always reported
/// alongside `deleted` so the UI can show the user their meeting-linked
/// events weren't touched, not just a silent lower-than-expected count.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CalendarCleanupResult {
    pub deleted: u32,
    pub protected: u32,
}

/// On-disk shape of `summary.json` (`docs/03-data-model.md`). Drafted action
/// items here are NOT yet rows in the `action_items` table — the user
/// reviews/edits them first; `confirm_action_items` is what persists them
/// (FR-LLM-3).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SummaryFile {
    pub schema: u32,
    pub generated_at: i64,
    pub provider: String,
    pub model: String,
    pub summary_md: String,
    pub decisions: Vec<String>,
    pub action_items: Vec<ActionItem>,
}

pub struct Retention {
    pub max_age_days: Option<u32>,
    pub max_size_gb: Option<u32>,
}

/// A feature brief distilled from a meeting (ADR-0011, M1). Holds only a
/// `credential_ref`-style pointer (`path`) into the meeting's `briefs/`
/// folder — the JSON body is the source of truth; this row is the index
/// `list_feature_briefs`/the MCP scope check reads. Maps 1:1 to the
/// `feature_briefs` table.
#[derive(Debug, Clone, sqlx::FromRow)]
pub struct FeatureBriefRow {
    pub id: String,
    pub meeting_id: MeetingId,
    pub title: String,
    pub target_repo: Option<String>,
    pub path: String, // briefs/<id>.json, relative to the meeting's folder
    pub exposed: bool,
    pub created_at: i64,
}

impl From<FeatureBriefRow> for FeatureBriefInfo {
    fn from(row: FeatureBriefRow) -> Self {
        FeatureBriefInfo {
            id: row.id,
            meeting_id: row.meeting_id,
            title: row.title,
            target_repo: row.target_repo,
            exposed: row.exposed,
        }
    }
}

/// On-disk shape of `briefs/<id>.json` (`docs/03-data-model.md`, ADR-0011) —
/// the schema/provenance envelope around the same fields as the IPC
/// `FeatureBrief` (`models.rs`). Written by `create_feature_brief` (M1.4),
/// sealed at rest with the vault when unlocked, exactly like `summary.json`.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BriefFile {
    pub schema: u32,
    pub id: String,
    pub meeting_id: MeetingId,
    pub generated_at: i64,
    pub provider: String,
    pub model: String,
    pub title: String,
    pub problem: String,
    pub desired_outcome: String,
    pub acceptance_criteria: Vec<String>,
    pub target_repo: Option<String>,
    pub context_excerpts: Vec<ContextExcerpt>,
    pub source: BriefSource,
}

/// `schema`-envelope companion recording which meeting this brief came from,
/// by name and time — distinct from the FK `meeting_id`, which can outlive a
/// renamed/retitled meeting.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BriefSource {
    pub meeting_title: String,
    pub at: i64,
}

/// A configured upload destination as stored in the DB (ADR-0010). Holds only a
/// `credential_ref` into the OS credential store — never the secret itself
/// (FR-SYNC-6). Maps 1:1 to the `sync_targets` table.
#[derive(Debug, Clone, sqlx::FromRow)]
pub struct SyncTargetRow {
    pub id: String,
    pub name: String,
    pub kind: String, // webdav|onedrive|dropbox|box
    pub provider_hint: Option<String>,
    pub base_url: Option<String>,
    pub remote_base_path: String,
    pub username: Option<String>,
    pub credential_ref: String,
    pub enabled: bool,
    pub upload_transcript: bool,
    pub upload_notes: bool,
    pub upload_summary: bool,
    pub upload_recording: bool,
    pub trigger_on_finalize: bool,
    pub allow_plaintext_lan: bool,
    pub encrypt_before_upload: bool,
    pub created_at: i64,
}

/// One durable upload job (per artifact × per target). Maps to `sync_jobs`.
#[derive(Debug, Clone, sqlx::FromRow)]
pub struct SyncJobRow {
    pub id: String,
    pub target_id: String,
    pub meeting_id: MeetingId,
    pub artifact: String, // transcript|notes|summary|recording
    pub local_path: String,
    pub remote_path: String,
    pub sha256: Option<String>,
    pub status: String, // pending|uploading|done|failed|skipped
    pub attempts: i64,
    pub last_error: Option<String>,
    pub next_attempt_at: Option<i64>,
    pub bytes_total: Option<i64>,
    pub bytes_sent: i64,
    pub updated_at: i64,
}

#[async_trait]
pub trait Store: Send + Sync {
    async fn create_meeting(&self, m: NewMeeting) -> Result<MeetingId, StoreError>;
    /// Returns the meeting's `template_id` (set at creation, Phase 8 T8.1) so
    /// callers can re-render notes.md with the same section structure
    /// without a separate fetch.
    async fn finalize_meeting(
        &self,
        id: &MeetingId,
        s: FinalizeMeeting,
    ) -> Result<Option<String>, StoreError>;
    async fn list_meetings(
        &self,
        filter: MeetingFilter,
    ) -> Result<Vec<MeetingListItem>, StoreError>;
    async fn get_meeting(&self, id: &MeetingId) -> Result<Meeting, StoreError>;
    async fn delete_meeting(&self, id: &MeetingId) -> Result<(), StoreError>;
    /// Overwrite a meeting's lifecycle `status` (e.g. mark a background import
    /// `transcribing` while it runs, or `error` if it fails). `finalize_meeting`
    /// is still the only path to `ready`.
    async fn set_meeting_status(&self, id: &MeetingId, status: &str) -> Result<(), StoreError>;
    async fn update_notes(&self, id: &MeetingId, markdown: &str) -> Result<(), StoreError>;
    /// (Re)builds this meeting's FTS index row from the current title and
    /// whatever's on disk/in the DB for transcript/notes/summary/tags (Phase
    /// 8, FR-SEARCH-1). `finalize_meeting`/`update_notes`/`set_tags` already
    /// call this themselves; callers that write searchable content some
    /// other way (e.g. `generate_summary` sealing `summary.json` straight to
    /// disk) must call it afterward so search doesn't silently miss it.
    async fn reindex_fts(&self, id: &MeetingId) -> Result<(), StoreError>;
    /// Set (or create) a speaker's display name; works whether or not the
    /// meeting has finalized yet (T4.4, FR-SPK-2/5).
    async fn rename_speaker(
        &self,
        id: &MeetingId,
        label: &str,
        name: &str,
    ) -> Result<(), StoreError>;
    /// Delete every `speakers` row for a meeting (labels, names, participant
    /// links, merges). Used before a full re-diarization rebuild
    /// (`reprocess_transcript`, FR-SPK): the old labels key to the previous
    /// clustering and are meaningless once the audio is re-clustered, so the
    /// caller re-inserts the fresh set via `finalize_meeting`. Without this,
    /// stale labels from an over-split run linger in the DB (and the
    /// Participants pane) even though the transcript no longer references them.
    async fn clear_speakers(&self, id: &MeetingId) -> Result<(), StoreError>;
    /// Fold over-split speaker labels into one canonical label (T4.5,
    /// FR-SPK-3). Segment speaker IDs in storage are never rewritten
    /// (FR-SPK-5) — `get_meeting` resolves `from` labels to `into` when it
    /// reads segments/speakers back.
    async fn merge_speakers(
        &self,
        id: &MeetingId,
        from: &[String],
        into: &str,
    ) -> Result<(), StoreError>;
    /// Persist reviewed/edited action items as confirmed tasks (T5.6,
    /// FR-LLM-3): items with an `id` update that row, items without one
    /// insert a new row. Drafted (unconfirmed) items from `generate_summary`
    /// only ever live in `summary.json` until this is called. Returns the
    /// saved items with `id` populated (Phase 8, T8.6) — new rows don't have
    /// one yet at call time, but the caller needs it to key reminder scheduling.
    async fn save_action_items(
        &self,
        id: &MeetingId,
        items: &[ActionItem],
    ) -> Result<Vec<ActionItem>, StoreError>;
    /// Persist imported calendar events + their attendees (T6.1/T6.2,
    /// FR-CAL-1). Re-importing the same event (matched by `source` +
    /// `raw_uid`) updates it in place rather than duplicating it. Returns
    /// the number of events imported/updated.
    async fn import_calendar_events(&self, events: Vec<ImportedEvent>) -> Result<u32, StoreError>;
    /// Browse imported calendar events (T6.3, FR-CAL-2), optionally bounded
    /// by unix-epoch start time.
    async fn list_calendar_events(
        &self,
        from: Option<i64>,
        to: Option<i64>,
    ) -> Result<Vec<CalendarEvent>, StoreError>;
    /// A single event with its attendees (T6.3/T6.4, FR-CAL-1/3) — backs the
    /// pre-meeting context panel and the speaker-naming attendee dropdown.
    async fn get_calendar_event(&self, id: &str) -> Result<CalendarEventDetail, StoreError>;
    /// Deletes imported calendar events not linked to any meeting —
    /// `older_than_unix: None` deletes every unlinked event ("Delete all"),
    /// `Some(cutoff)` only those with `starts_at < cutoff`. An event
    /// referenced by `meetings.calendar_event_id` is always kept regardless
    /// of age (FR-CAL-2 pre-meeting context / FR-CAL-4 continuity would
    /// break otherwise). Returns `(deleted, protected)`.
    async fn cleanup_calendar_events(
        &self,
        older_than_unix: Option<i64>,
    ) -> Result<(u32, u32), StoreError>;
    /// Link a meeting (current or historical) to a calendar event (T6.3/T6.6,
    /// FR-CAL-2/4). Errs if either id doesn't exist. Also mirrors the event's
    /// subject onto the meeting's title when it has one — linking is meant to
    /// say "this recording is that meeting," so a recording still sitting at
    /// its default "Untitled meeting" name should follow it.
    async fn attach_meeting_to_event(
        &self,
        meeting_id: &MeetingId,
        event_id: &str,
    ) -> Result<(), StoreError>;
    /// Manually rename a meeting — recordings otherwise default to "Untitled
    /// meeting" with no other way to change that.
    async fn rename_meeting(&self, meeting_id: &MeetingId, title: &str) -> Result<(), StoreError>;
    /// Overwrite a meeting's start/end timestamps — used by bundle import
    /// (FR-STORE-4) so a moved recording keeps its original date rather than
    /// showing the import time (`create_meeting`/`finalize_meeting` both stamp
    /// "now").
    async fn set_meeting_times(
        &self,
        meeting_id: &MeetingId,
        started_at: i64,
        ended_at: Option<i64>,
    ) -> Result<(), StoreError>;
    /// Names a speaker AND links them to a known `Participant` (T6.5/T6.6,
    /// FR-SPK-4): the display name comes from the participant record, and
    /// the shared `participant_id` is what gives naming "continuity" across
    /// meetings — the same person, once identified anywhere, shows up
    /// pre-resolvable wherever they're an attendee again.
    async fn map_speaker_to_participant(
        &self,
        meeting_id: &MeetingId,
        label: &str,
        participant_id: &str,
    ) -> Result<(), StoreError>;
    /// Full-text search across transcripts + notes (Phase 8, FR-SEARCH-1).
    async fn search(&self, query: &str) -> Result<Vec<SearchHit>, StoreError>;
    /// Replaces this meeting's complete tag set (Phase 8, FR-SEARCH-2) —
    /// not incremental add/remove. Unknown tag names are created.
    async fn set_tags(&self, meeting_id: &MeetingId, tags: &[String]) -> Result<(), StoreError>;
    /// All known tag names, sorted, for filter/autocomplete UI.
    async fn list_tags(&self) -> Result<Vec<String>, StoreError>;
    /// Action items with an unfired reminder due in the future (Phase 8,
    /// T8.6, FR-CAL-5) — a startup reconcile against Windows' scheduled-toast
    /// list, since that's the actual source of truth for "already scheduled"
    /// (see `reminders` module) and it can be cleared out from under the app
    /// (uninstall/reinstall, a different AppUserModelID, etc).
    async fn list_pending_reminders(&self) -> Result<Vec<ActionItem>, StoreError>;
    /// Startup reconcile: meetings with audio but no finalized transcript (FR-REL-1).
    async fn recover_scan(&self) -> Result<Vec<MeetingId>, StoreError>;
    /// Enforce retention; returns count removed. Skips in-progress meetings
    /// (anything whose `status` isn't a terminal `ready`/`error`).
    async fn enforce_retention(&self, policy: Retention) -> Result<u32, StoreError>;

    // ---- Sync targets (Phase 9, ADR-0010) ----
    async fn add_sync_target(&self, row: SyncTargetRow) -> Result<(), StoreError>;
    /// Full-row replace (caller loads, edits, and preserves `credential_ref`).
    async fn update_sync_target(&self, row: SyncTargetRow) -> Result<(), StoreError>;
    async fn remove_sync_target(&self, id: &str) -> Result<(), StoreError>;
    async fn list_sync_targets(&self) -> Result<Vec<SyncTargetRow>, StoreError>;
    async fn get_sync_target(&self, id: &str) -> Result<SyncTargetRow, StoreError>;

    // ---- Sync jobs / durable queue (Phase 9, T9.5) ----
    /// Enqueue (or re-enqueue) a job. Idempotent per (target, meeting, artifact):
    /// a job already `done` with the same `sha256` is left untouched (skip-if-
    /// unchanged); anything else is reset to `pending`. Returns true if enqueued.
    async fn upsert_sync_job(&self, job: SyncJobRow) -> Result<bool, StoreError>;
    /// Jobs ready to run now: `pending`/`failed` with no future `next_attempt_at`.
    async fn claim_due_sync_jobs(&self, now: i64) -> Result<Vec<SyncJobRow>, StoreError>;
    async fn update_sync_job(&self, job: SyncJobRow) -> Result<(), StoreError>;
    async fn get_sync_job(&self, id: &str) -> Result<SyncJobRow, StoreError>;
    async fn list_sync_jobs(
        &self,
        meeting_id: Option<&MeetingId>,
    ) -> Result<Vec<SyncJobRow>, StoreError>;

    // ---- Feature briefs (Phase 10 M1, ADR-0011) ----
    /// Indexes a brief already sealed to disk by `create_feature_brief`
    /// (M1.4). Deletion cascades via the meeting FK — `delete_meeting`
    /// already drops the row (and its folder) with the rest of the meeting.
    async fn insert_feature_brief(&self, row: FeatureBriefRow) -> Result<(), StoreError>;
    /// Newest first; `None` lists across all meetings (the MCP "recent
    /// briefs" surface and the frontend's per-meeting list share this call).
    async fn list_feature_briefs(
        &self,
        meeting_id: Option<&MeetingId>,
    ) -> Result<Vec<FeatureBriefInfo>, StoreError>;
    /// The full row (incl. `path`) so a caller can resolve and read the
    /// sealed JSON file itself (`get_feature_brief`, MCP `get_feature_brief`
    /// tool).
    async fn get_feature_brief_row(&self, id: &str) -> Result<FeatureBriefRow, StoreError>;
    /// Scope control: include/exclude a brief from the MCP server (FR-MCP-3).
    async fn set_brief_exposed(&self, id: &str, exposed: bool) -> Result<(), StoreError>;

    // ---- MCP server (Phase 10b, ADR-0011) ----
    /// Confirmed action items for a meeting (FR-MCP-2 `get_action_items`) —
    /// distinct from `list_pending_reminders` (which is filtered to
    /// unfired-reminder rows across *all* meetings for the startup reconcile).
    async fn list_action_items(
        &self,
        meeting_id: &MeetingId,
    ) -> Result<Vec<ActionItem>, StoreError>;
    /// Appends one row to the audit log (FR-MCP-5). Every MCP tool read calls
    /// this, regardless of whether the read was actually allowed to see
    /// anything — the audit trail is "what an agent asked for", not just
    /// "what it received".
    async fn record_mcp_access(
        &self,
        tool: &str,
        meeting_id: Option<&MeetingId>,
        client: Option<&str>,
    ) -> Result<(), StoreError>;
    /// Most recent audit rows first, optionally capped (`mcp_access_log` command).
    async fn list_mcp_access_log(
        &self,
        limit: Option<u32>,
    ) -> Result<Vec<McpAccessEntry>, StoreError>;
}

/// SQLite-backed store. Migrations live in `migrations/` (`sqlx::migrate!`).
pub struct SqliteStore {
    pool: SqlitePool,
}

impl SqliteStore {
    pub async fn connect() -> Result<Self, StoreError> {
        let root = paths::wa_root();
        std::fs::create_dir_all(&root)?;
        let opts = SqliteConnectOptions::new()
            .filename(paths::db_path())
            .create_if_missing(true)
            .foreign_keys(true);
        let pool = SqlitePoolOptions::new()
            .max_connections(5)
            .connect_with(opts)
            .await?;
        sqlx::migrate!("./migrations")
            .run(&pool)
            .await
            .map_err(|e| StoreError::Db(e.to_string()))?;
        let store = Self { pool };
        store.backfill_fts().await?;
        Ok(store)
    }

    /// In-memory store for tests: one connection, migrations applied, foreign
    /// keys off so sync-queue tests can insert jobs without full meeting rows.
    #[cfg(test)]
    pub async fn connect_in_memory() -> Result<Self, StoreError> {
        let opts = SqliteConnectOptions::new().filename(":memory:");
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect_with(opts)
            .await?;
        sqlx::migrate!("./migrations")
            .run(&pool)
            .await
            .map_err(|e| StoreError::Db(e.to_string()))?;
        sqlx::query("PRAGMA foreign_keys = OFF")
            .execute(&pool)
            .await
            .ok();
        Ok(Self { pool })
    }

    /// One-time-per-meeting catch-up for meetings finalized before full-text
    /// search existed (T8.2, FR-SEARCH-1) — `finalize_meeting`/`update_notes`
    /// keep `meeting_fts` current for everything from here on, but that adds
    /// nothing retroactively for meetings that predate this feature. Cheap
    /// and idempotent: only touches meetings missing a row, so after the
    /// first run this is a no-op scan on every later startup.
    async fn backfill_fts(&self) -> Result<(), StoreError> {
        let missing: Vec<String> = sqlx::query_scalar(
            "SELECT id FROM meetings WHERE id NOT IN (SELECT meeting_id FROM meeting_fts)",
        )
        .fetch_all(&self.pool)
        .await?;
        for id in missing {
            if let Err(e) = self.reindex_fts(&id).await {
                tracing::warn!("FTS backfill skipped meeting {id}: {e}");
            }
        }
        Ok(())
    }
}

impl SqliteStore {
    /// `participant_id: None` leaves an existing link untouched (e.g. a
    /// plain free-text `rename_speaker` after `map_speaker_to_participant`
    /// must not silently unlink the participant) — `COALESCE(excluded.…, …)`
    /// falls back to the row's current value when the new one isn't given.
    async fn upsert_speaker(
        &self,
        meeting_id: &MeetingId,
        label: &str,
        display_name: Option<&str>,
        participant_id: Option<&str>,
    ) -> Result<(), StoreError> {
        sqlx::query(
            "INSERT INTO speakers (id, meeting_id, label, display_name, participant_id)
             VALUES (?, ?, ?, ?, ?)
             ON CONFLICT(meeting_id, label) DO UPDATE SET
                 display_name = excluded.display_name,
                 participant_id = COALESCE(excluded.participant_id, participant_id)",
        )
        .bind(uuid::Uuid::new_v4().to_string())
        .bind(meeting_id)
        .bind(label)
        .bind(display_name)
        .bind(participant_id)
        .execute(&self.pool)
        .await?;
        Ok(())
    }

    /// Finds-or-creates a `participants` row. Not a plain `ON CONFLICT`
    /// upsert: `UNIQUE(name, email)` treats two NULL emails as distinct under
    /// standard SQL NULL semantics, so a conflict target wouldn't reliably
    /// catch a repeat import of the same no-email attendee — this looks the
    /// row up explicitly instead.
    async fn upsert_participant(
        &self,
        name: &str,
        email: Option<&str>,
    ) -> Result<String, StoreError> {
        let existing: Option<String> = match email {
            Some(email) => sqlx::query("SELECT id FROM participants WHERE name = ? AND email = ?")
                .bind(name)
                .bind(email)
                .fetch_optional(&self.pool)
                .await?
                .map(|r| r.get("id")),
            None => sqlx::query("SELECT id FROM participants WHERE name = ? AND email IS NULL")
                .bind(name)
                .fetch_optional(&self.pool)
                .await?
                .map(|r| r.get("id")),
        };
        if let Some(id) = existing {
            return Ok(id);
        }
        let id = uuid::Uuid::new_v4().to_string();
        sqlx::query("INSERT INTO participants (id, name, email) VALUES (?, ?, ?)")
            .bind(&id)
            .bind(name)
            .bind(email)
            .execute(&self.pool)
            .await?;
        Ok(id)
    }

    /// Finds-or-creates a `tags` row by name (Phase 8, FR-SEARCH-2). `name` is
    /// `UNIQUE`, so — unlike `upsert_participant`'s nullable `email` — a plain
    /// `INSERT OR IGNORE` conflict target is unambiguous here.
    async fn upsert_tag(&self, name: &str) -> Result<String, StoreError> {
        if let Some(id) = sqlx::query_scalar("SELECT id FROM tags WHERE name = ?")
            .bind(name)
            .fetch_optional(&self.pool)
            .await?
        {
            return Ok(id);
        }
        let id = uuid::Uuid::new_v4().to_string();
        sqlx::query("INSERT OR IGNORE INTO tags (id, name) VALUES (?, ?)")
            .bind(&id)
            .bind(name)
            .execute(&self.pool)
            .await?;
        // Someone else (a concurrent set_tags call) may have won the
        // INSERT OR IGNORE race; re-look-up rather than assume our id landed.
        sqlx::query_scalar("SELECT id FROM tags WHERE name = ?")
            .bind(name)
            .fetch_one(&self.pool)
            .await
            .map_err(Into::into)
    }

    /// Tag names for one meeting, sorted (Phase 8, FR-SEARCH-2).
    async fn tags_for_meeting(&self, meeting_id: &MeetingId) -> Result<Vec<String>, StoreError> {
        Ok(sqlx::query_scalar(
            "SELECT t.name FROM tags t
             JOIN meeting_tags mt ON mt.tag_id = t.id
             WHERE mt.meeting_id = ?
             ORDER BY t.name",
        )
        .bind(meeting_id)
        .fetch_all(&self.pool)
        .await?)
    }
}

/// Follows a `label -> merged_into` chain to its canonical label. Capped at 8
/// hops so a stale/cyclic mapping (shouldn't happen, but merges are
/// user-driven data) can't loop forever; merges normally resolve in one hop.
fn resolve_canonical<'a>(label: &'a str, merge_map: &'a HashMap<String, String>) -> &'a str {
    let mut current = label;
    for _ in 0..8 {
        match merge_map.get(current) {
            Some(next) if next != current => current = next.as_str(),
            _ => break,
        }
    }
    current
}

fn now_unix() -> i64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_secs() as i64)
        .unwrap_or(0)
}

/// Turns free-text user input into a safe FTS5 MATCH query (Phase 8,
/// FR-SEARCH-1). FTS5's query syntax gives special meaning to `" - * AND OR
/// NOT`, so passing a search box's raw text straight to MATCH can either
/// throw a syntax error on stray punctuation or search less literally than
/// the user typed. Quoting each token as its own phrase (doubling embedded
/// quotes) makes every token a literal match, ANDed together by default.
fn fts_query_from_input(input: &str) -> String {
    input
        .split_whitespace()
        .map(|tok| format!("\"{}\"", tok.replace('"', "\"\"")))
        .collect::<Vec<_>>()
        .join(" ")
}

/// On-disk shape of `transcript.json` (`docs/03-data-model.md`).
#[derive(Debug, Serialize, Deserialize, Default)]
struct TranscriptFile {
    schema: u32,
    meeting_id: MeetingId,
    language: Option<String>,
    model: Option<String>,
    backend: Option<String>,
    segments: Vec<TranscriptSegment>,
    speakers: Vec<TranscriptSpeaker>,
}

#[derive(Debug, Serialize, Deserialize)]
struct TranscriptSpeaker {
    label: String,
    display_name: Option<String>,
}

/// Builds everything but `tags` — the caller fills that in with an async
/// `tags_for_meeting` lookup, since sqlx rows can't be read across an await.
fn row_to_list_item(row: &sqlx::sqlite::SqliteRow) -> MeetingListItem {
    MeetingListItem {
        id: row.get("id"),
        title: row.get("title"),
        started_at: row.get("started_at"),
        duration_secs: row.get("duration_secs"),
        status: MeetingStatus::parse(row.get::<String, _>("status").as_str()),
        tags: Vec::new(),
    }
}

fn row_to_calendar_event(row: &sqlx::sqlite::SqliteRow) -> CalendarEvent {
    CalendarEvent {
        id: row.get("id"),
        source: row.get("source"),
        subject: row.get("subject"),
        organizer: row.get("organizer"),
        starts_at: row.get("starts_at"),
        ends_at: row.get("ends_at"),
        description: row.get("description"),
        raw_uid: row.get("raw_uid"),
    }
}

#[async_trait]
impl Store for SqliteStore {
    async fn create_meeting(&self, m: NewMeeting) -> Result<MeetingId, StoreError> {
        let id = uuid::Uuid::new_v4().to_string();
        let folder = paths::meeting_dir(&id);
        std::fs::create_dir_all(&folder)?;
        let audio_path = folder.join("audio.wav");
        let now = now_unix();
        sqlx::query(
            "INSERT INTO meetings (id, title, started_at, folder_path, audio_path, status, calendar_event_id, template_id, language, created_at, updated_at)
             VALUES (?, ?, ?, ?, ?, 'recording', ?, ?, ?, ?, ?)",
        )
        .bind(&id)
        .bind(&m.title)
        .bind(now)
        .bind(folder.display().to_string())
        .bind(audio_path.display().to_string())
        .bind(&m.calendar_event_id)
        .bind(&m.template_id)
        .bind(&m.language)
        .bind(now)
        .bind(now)
        .execute(&self.pool)
        .await?;
        Ok(id)
    }

    async fn finalize_meeting(
        &self,
        id: &MeetingId,
        s: FinalizeMeeting,
    ) -> Result<Option<String>, StoreError> {
        let now = now_unix();
        sqlx::query(
            "UPDATE meetings SET status = 'ready', ended_at = ?, duration_secs = ?, recorded = ?,
                language = ?, backend_used = ?, model_used = ?,
                audio_layout = COALESCE(?, audio_layout), updated_at = ? WHERE id = ?",
        )
        .bind(now)
        .bind(s.duration_secs)
        .bind(s.recorded as i64)
        .bind(&s.language)
        .bind(&s.backend_used)
        .bind(&s.model_used)
        .bind(&s.audio_layout)
        .bind(now)
        .bind(id)
        .execute(&self.pool)
        .await?;

        for speaker in &s.speakers {
            self.upsert_speaker(
                id,
                &speaker.label,
                speaker.display_name.as_deref(),
                speaker.participant_id.as_deref(),
            )
            .await?;
        }

        let transcript = TranscriptFile {
            schema: 1,
            meeting_id: id.clone(),
            language: s.language.clone(),
            model: s.model_used.clone(),
            backend: s.backend_used.clone(),
            segments: s.segments,
            speakers: s
                .speakers
                .iter()
                .map(|sp| TranscriptSpeaker {
                    label: sp.label.clone(),
                    display_name: sp.display_name.clone(),
                })
                .collect(),
        };
        let json =
            serde_json::to_string_pretty(&transcript).map_err(|e| StoreError::Db(e.to_string()))?;
        write_artifact(
            &paths::meeting_dir(id).join("transcript.json"),
            json.as_bytes(),
        )?;
        self.reindex_fts(id).await?;
        let template_id: Option<String> =
            sqlx::query_scalar("SELECT template_id FROM meetings WHERE id = ?")
                .bind(id)
                .fetch_one(&self.pool)
                .await?;
        Ok(template_id)
    }

    async fn list_meetings(
        &self,
        filter: MeetingFilter,
    ) -> Result<Vec<MeetingListItem>, StoreError> {
        // Conditions are pushed in a fixed order (tag, query, participant,
        // from, to) and bound in that exact same order below — there's no
        // named-parameter binding in this sqlx query style, so the two lists
        // must stay in lockstep.
        let mut sql = String::from("SELECT DISTINCT m.id, m.title, m.started_at, m.duration_secs, m.status FROM meetings m");
        if filter.tag.is_some() {
            sql.push_str(
                " JOIN meeting_tags mt ON mt.meeting_id = m.id JOIN tags t ON t.id = mt.tag_id",
            );
        }
        let mut conditions = Vec::new();
        if filter.tag.is_some() {
            conditions.push("t.name = ?");
        }
        let query = filter.query.filter(|q| !q.trim().is_empty());
        if query.is_some() {
            conditions.push("m.title LIKE ?");
        }
        if filter.participant_id.is_some() {
            conditions.push(
                "(EXISTS (SELECT 1 FROM speakers s WHERE s.meeting_id = m.id AND s.participant_id = ?)
                  OR EXISTS (SELECT 1 FROM calendar_event_participants cep WHERE cep.calendar_event_id = m.calendar_event_id AND cep.participant_id = ?))",
            );
        }
        if filter.from.is_some() {
            conditions.push("m.started_at >= ?");
        }
        if filter.to.is_some() {
            conditions.push("m.started_at <= ?");
        }
        if !conditions.is_empty() {
            sql.push_str(" WHERE ");
            sql.push_str(&conditions.join(" AND "));
        }
        sql.push_str(" ORDER BY m.started_at DESC");

        let mut q = sqlx::query(&sql);
        if let Some(tag) = &filter.tag {
            q = q.bind(tag);
        }
        if let Some(query) = &query {
            q = q.bind(format!("%{}%", query.trim()));
        }
        if let Some(pid) = &filter.participant_id {
            q = q.bind(pid).bind(pid);
        }
        if let Some(from) = filter.from {
            q = q.bind(from);
        }
        if let Some(to) = filter.to {
            q = q.bind(to);
        }
        let rows = q.fetch_all(&self.pool).await?;

        let mut items = Vec::with_capacity(rows.len());
        for row in &rows {
            let mut item = row_to_list_item(row);
            item.tags = self.tags_for_meeting(&item.id).await?;
            items.push(item);
        }
        Ok(items)
    }

    async fn get_meeting(&self, id: &MeetingId) -> Result<Meeting, StoreError> {
        let row = sqlx::query(
            "SELECT id, title, started_at, ended_at, duration_secs, status, recorded, language, backend_used, model_used, calendar_event_id, template_id, audio_layout
             FROM meetings WHERE id = ?",
        )
        .bind(id)
        .fetch_optional(&self.pool)
        .await?
        .ok_or_else(|| StoreError::NotFound(id.clone()))?;

        let speaker_rows = sqlx::query(
            "SELECT label, display_name, participant_id, merged_into FROM speakers WHERE meeting_id = ? ORDER BY label",
        )
        .bind(id)
        .fetch_all(&self.pool)
        .await?;
        // T4.5: rows with merged_into set are folded away — only canonical
        // speakers are returned, but their raw label still resolves through
        // `merge_map` below so folded segments render under the right one.
        let merge_map: HashMap<String, String> = speaker_rows
            .iter()
            .filter_map(|r| {
                let merged_into: Option<String> = r.get("merged_into");
                merged_into.map(|into| (r.get::<String, _>("label"), into))
            })
            .collect();
        let speakers: Vec<SpeakerInfo> = speaker_rows
            .iter()
            .filter(|r| r.get::<Option<String>, _>("merged_into").is_none())
            .map(|r| SpeakerInfo {
                label: r.get("label"),
                display_name: r.get("display_name"),
                participant_id: r.get("participant_id"),
            })
            .collect();

        let folder = paths::meeting_dir(id);
        let mut segments = read_artifact(&folder.join("transcript.json"))
            .and_then(|s| serde_json::from_str::<TranscriptFile>(&s).ok())
            .map(|t| t.segments)
            .unwrap_or_default();
        if !merge_map.is_empty() {
            // Resolution only touches this in-memory copy — transcript.json
            // on disk keeps its raw labels, regenerable and non-destructive
            // (FR-SPK-5), same as display names.
            for seg in &mut segments {
                seg.speaker = resolve_canonical(&seg.speaker, &merge_map).to_string();
            }
        }
        let notes_markdown = read_artifact(&folder.join("notes.md")).unwrap_or_default();
        let summary = read_artifact(&folder.join("summary.json"))
            .and_then(|s| serde_json::from_str::<SummaryFile>(&s).ok());
        let tags = self.tags_for_meeting(id).await?;
        // Table is the source of truth for confirmed/edited items; fall back
        // to the summary's parsed drafts only when nothing's been saved yet.
        let mut action_items = self.list_action_items(id).await?;
        if action_items.is_empty() {
            if let Some(s) = &summary {
                action_items = s.action_items.clone();
            }
        }

        Ok(Meeting {
            id: row.get("id"),
            title: row.get("title"),
            started_at: row.get("started_at"),
            ended_at: row.get("ended_at"),
            duration_secs: row.get("duration_secs"),
            status: MeetingStatus::parse(row.get::<String, _>("status").as_str()),
            recorded: row.get::<i64, _>("recorded") != 0,
            language: row.get("language"),
            backend_used: row.get("backend_used"),
            model_used: row.get("model_used"),
            audio_layout: row.get("audio_layout"),
            segments,
            speakers,
            notes_markdown,
            summary,
            calendar_event_id: row.get("calendar_event_id"),
            tags,
            template_id: row.get("template_id"),
            action_items,
        })
    }

    async fn delete_meeting(&self, id: &MeetingId) -> Result<(), StoreError> {
        // meeting_fts is a virtual table — no FK/CASCADE support, so its row
        // would otherwise outlive the meeting and show up as a ghost result.
        sqlx::query("DELETE FROM meeting_fts WHERE meeting_id = ?")
            .bind(id)
            .execute(&self.pool)
            .await?;
        sqlx::query("DELETE FROM meetings WHERE id = ?")
            .bind(id)
            .execute(&self.pool)
            .await?;
        let folder = paths::meeting_dir(id);
        if folder.exists() {
            std::fs::remove_dir_all(folder)?;
        }
        Ok(())
    }

    async fn set_meeting_status(&self, id: &MeetingId, status: &str) -> Result<(), StoreError> {
        sqlx::query("UPDATE meetings SET status = ?, updated_at = ? WHERE id = ?")
            .bind(status)
            .bind(now_unix())
            .bind(id)
            .execute(&self.pool)
            .await?;
        Ok(())
    }

    async fn update_notes(&self, id: &MeetingId, markdown: &str) -> Result<(), StoreError> {
        write_artifact(
            &paths::meeting_dir(id).join("notes.md"),
            markdown.as_bytes(),
        )?;
        sqlx::query("UPDATE meetings SET updated_at = ? WHERE id = ?")
            .bind(now_unix())
            .bind(id)
            .execute(&self.pool)
            .await?;
        self.reindex_fts(id).await?;
        Ok(())
    }

    /// `meeting_fts` is a plain (non-`content=`) FTS5 table, so nothing keeps
    /// it in sync automatically — every write path that changes title,
    /// transcript, notes, summary, or tags calls this afterward so search
    /// (FR-SEARCH-1) can't drift from what's actually stored. Deletes-then-
    /// inserts rather than `INSERT OR REPLACE`: FTS5 has no unique
    /// constraint to conflict on.
    async fn reindex_fts(&self, id: &MeetingId) -> Result<(), StoreError> {
        let title: String = sqlx::query_scalar("SELECT title FROM meetings WHERE id = ?")
            .bind(id)
            .fetch_one(&self.pool)
            .await?;

        let transcript_text = read_artifact(&paths::meeting_dir(id).join("transcript.json"))
            .and_then(|s| serde_json::from_str::<TranscriptFile>(&s).ok())
            .map(|t| {
                t.segments
                    .iter()
                    .map(|s| s.text.as_str())
                    .collect::<Vec<_>>()
                    .join(" ")
            })
            .unwrap_or_default();

        let notes_text =
            read_artifact(&paths::meeting_dir(id).join("notes.md")).unwrap_or_default();

        // `read_artifact` already unseals (T8.8's vault, passthrough when
        // locked/plaintext), same as the transcript/notes reads above.
        let summary_text = read_artifact(&paths::meeting_dir(id).join("summary.json"))
            .and_then(|s| serde_json::from_str::<SummaryFile>(&s).ok())
            .map(|s| {
                let mut text = s.summary_md;
                for decision in &s.decisions {
                    text.push(' ');
                    text.push_str(decision);
                }
                for item in &s.action_items {
                    text.push(' ');
                    text.push_str(&item.text);
                }
                text
            })
            .unwrap_or_default();

        let tags_text = self.tags_for_meeting(id).await?.join(" ");

        sqlx::query("DELETE FROM meeting_fts WHERE meeting_id = ?")
            .bind(id)
            .execute(&self.pool)
            .await?;
        sqlx::query(
            "INSERT INTO meeting_fts (meeting_id, title, transcript_text, notes_text, summary_text, tags_text)
             VALUES (?, ?, ?, ?, ?, ?)",
        )
        .bind(id)
        .bind(&title)
        .bind(&transcript_text)
        .bind(&notes_text)
        .bind(&summary_text)
        .bind(&tags_text)
        .execute(&self.pool)
        .await?;
        Ok(())
    }

    async fn rename_speaker(
        &self,
        id: &MeetingId,
        label: &str,
        name: &str,
    ) -> Result<(), StoreError> {
        self.upsert_speaker(id, label, Some(name), None).await
    }

    async fn list_calendar_events(
        &self,
        from: Option<i64>,
        to: Option<i64>,
    ) -> Result<Vec<CalendarEvent>, StoreError> {
        let rows = match (from, to) {
            (Some(f), Some(t)) => {
                sqlx::query(
                    "SELECT * FROM calendar_events WHERE starts_at >= ? AND starts_at <= ? ORDER BY starts_at",
                )
                .bind(f)
                .bind(t)
                .fetch_all(&self.pool)
                .await?
            }
            (Some(f), None) => {
                sqlx::query("SELECT * FROM calendar_events WHERE starts_at >= ? ORDER BY starts_at")
                    .bind(f)
                    .fetch_all(&self.pool)
                    .await?
            }
            (None, Some(t)) => {
                sqlx::query("SELECT * FROM calendar_events WHERE starts_at <= ? ORDER BY starts_at")
                    .bind(t)
                    .fetch_all(&self.pool)
                    .await?
            }
            (None, None) => {
                sqlx::query("SELECT * FROM calendar_events ORDER BY starts_at")
                    .fetch_all(&self.pool)
                    .await?
            }
        };
        Ok(rows.iter().map(row_to_calendar_event).collect())
    }

    async fn get_calendar_event(&self, id: &str) -> Result<CalendarEventDetail, StoreError> {
        let row = sqlx::query("SELECT * FROM calendar_events WHERE id = ?")
            .bind(id)
            .fetch_optional(&self.pool)
            .await?
            .ok_or_else(|| StoreError::NotFound(id.to_string()))?;
        let event = row_to_calendar_event(&row);

        let participant_rows = sqlx::query(
            "SELECT p.id, p.name, p.email, cep.role FROM calendar_event_participants cep
             JOIN participants p ON p.id = cep.participant_id
             WHERE cep.calendar_event_id = ? ORDER BY p.name",
        )
        .bind(id)
        .fetch_all(&self.pool)
        .await?;
        let participants = participant_rows
            .iter()
            .map(|r| Participant {
                id: r.get("id"),
                name: r.get("name"),
                email: r.get("email"),
                role: r.get("role"),
            })
            .collect();

        Ok(CalendarEventDetail {
            event,
            participants,
        })
    }

    async fn cleanup_calendar_events(
        &self,
        older_than_unix: Option<i64>,
    ) -> Result<(u32, u32), StoreError> {
        // Protected = referenced by any meeting's calendar_event_id, full
        // stop — never deleted regardless of age or the "delete all" choice.
        let protected: i64 = match older_than_unix {
            Some(cutoff) => {
                sqlx::query_scalar(
                    "SELECT COUNT(*) FROM calendar_events
                     WHERE starts_at < ?
                       AND id IN (SELECT calendar_event_id FROM meetings WHERE calendar_event_id IS NOT NULL)",
                )
                .bind(cutoff)
                .fetch_one(&self.pool)
                .await?
            }
            None => {
                sqlx::query_scalar(
                    "SELECT COUNT(*) FROM calendar_events
                     WHERE id IN (SELECT calendar_event_id FROM meetings WHERE calendar_event_id IS NOT NULL)",
                )
                .fetch_one(&self.pool)
                .await?
            }
        };

        // calendar_event_participants cascades via its ON DELETE CASCADE FK.
        let deleted = match older_than_unix {
            Some(cutoff) => {
                sqlx::query(
                    "DELETE FROM calendar_events
                     WHERE starts_at < ?
                       AND id NOT IN (SELECT calendar_event_id FROM meetings WHERE calendar_event_id IS NOT NULL)",
                )
                .bind(cutoff)
                .execute(&self.pool)
                .await?
                .rows_affected()
            }
            None => {
                sqlx::query(
                    "DELETE FROM calendar_events
                     WHERE id NOT IN (SELECT calendar_event_id FROM meetings WHERE calendar_event_id IS NOT NULL)",
                )
                .execute(&self.pool)
                .await?
                .rows_affected()
            }
        };

        Ok((deleted as u32, protected as u32))
    }

    async fn attach_meeting_to_event(
        &self,
        meeting_id: &MeetingId,
        event_id: &str,
    ) -> Result<(), StoreError> {
        let row: Option<(String, Option<String>)> =
            sqlx::query_as("SELECT id, subject FROM calendar_events WHERE id = ?")
                .bind(event_id)
                .fetch_optional(&self.pool)
                .await?;
        let Some((_, subject)) = row else {
            return Err(StoreError::NotFound(format!("calendar event {event_id}")));
        };

        let result = match subject.filter(|s| !s.is_empty()) {
            Some(title) => sqlx::query(
                "UPDATE meetings SET calendar_event_id = ?, title = ?, updated_at = ? WHERE id = ?",
            )
            .bind(event_id)
            .bind(title)
            .bind(now_unix())
            .bind(meeting_id)
            .execute(&self.pool)
            .await?,
            None => {
                sqlx::query(
                    "UPDATE meetings SET calendar_event_id = ?, updated_at = ? WHERE id = ?",
                )
                .bind(event_id)
                .bind(now_unix())
                .bind(meeting_id)
                .execute(&self.pool)
                .await?
            }
        };
        if result.rows_affected() == 0 {
            return Err(StoreError::NotFound(meeting_id.clone()));
        }
        Ok(())
    }

    async fn rename_meeting(&self, meeting_id: &MeetingId, title: &str) -> Result<(), StoreError> {
        let result = sqlx::query("UPDATE meetings SET title = ?, updated_at = ? WHERE id = ?")
            .bind(title)
            .bind(now_unix())
            .bind(meeting_id)
            .execute(&self.pool)
            .await?;
        if result.rows_affected() == 0 {
            return Err(StoreError::NotFound(meeting_id.clone()));
        }
        Ok(())
    }

    async fn set_meeting_times(
        &self,
        meeting_id: &MeetingId,
        started_at: i64,
        ended_at: Option<i64>,
    ) -> Result<(), StoreError> {
        sqlx::query(
            "UPDATE meetings SET started_at = ?, ended_at = ?, updated_at = ? WHERE id = ?",
        )
        .bind(started_at)
        .bind(ended_at)
        .bind(now_unix())
        .bind(meeting_id)
        .execute(&self.pool)
        .await?;
        Ok(())
    }

    async fn map_speaker_to_participant(
        &self,
        meeting_id: &MeetingId,
        label: &str,
        participant_id: &str,
    ) -> Result<(), StoreError> {
        let name: String = sqlx::query_scalar("SELECT name FROM participants WHERE id = ?")
            .bind(participant_id)
            .fetch_optional(&self.pool)
            .await?
            .ok_or_else(|| StoreError::NotFound(format!("participant {participant_id}")))?;
        self.upsert_speaker(meeting_id, label, Some(&name), Some(participant_id))
            .await
    }

    async fn clear_speakers(&self, id: &MeetingId) -> Result<(), StoreError> {
        sqlx::query("DELETE FROM speakers WHERE meeting_id = ?")
            .bind(id)
            .execute(&self.pool)
            .await?;
        Ok(())
    }

    async fn merge_speakers(
        &self,
        id: &MeetingId,
        from: &[String],
        into: &str,
    ) -> Result<(), StoreError> {
        // Ensure the canonical label has a row, without clobbering a name it
        // may already have (plain upsert_speaker would overwrite display_name
        // with `None` if `into` hasn't been named yet).
        sqlx::query(
            "INSERT INTO speakers (id, meeting_id, label) VALUES (?, ?, ?)
             ON CONFLICT(meeting_id, label) DO NOTHING",
        )
        .bind(uuid::Uuid::new_v4().to_string())
        .bind(id)
        .bind(into)
        .execute(&self.pool)
        .await?;

        for label in from {
            if label == into {
                continue; // merging a label into itself is a no-op
            }
            sqlx::query(
                "INSERT INTO speakers (id, meeting_id, label, merged_into) VALUES (?, ?, ?, ?)
                 ON CONFLICT(meeting_id, label) DO UPDATE SET merged_into = excluded.merged_into",
            )
            .bind(uuid::Uuid::new_v4().to_string())
            .bind(id)
            .bind(label)
            .bind(into)
            .execute(&self.pool)
            .await?;
        }
        Ok(())
    }

    async fn save_action_items(
        &self,
        id: &MeetingId,
        items: &[ActionItem],
    ) -> Result<Vec<ActionItem>, StoreError> {
        let now = now_unix();
        // Reconcile deletes: any row previously saved for this meeting that the
        // caller no longer includes was removed in the UI (FR-LLM-3). Action
        // items per meeting number in the low tens, so a per-row delete loop is
        // fine. ponytail: O(n) delete scan, batch it only if n ever gets large.
        let keep: std::collections::HashSet<&str> =
            items.iter().filter_map(|i| i.id.as_deref()).collect();
        let existing_ids: Vec<String> =
            sqlx::query_scalar("SELECT id FROM action_items WHERE meeting_id = ?")
                .bind(id)
                .fetch_all(&self.pool)
                .await?;
        for eid in &existing_ids {
            if !keep.contains(eid.as_str()) {
                sqlx::query("DELETE FROM action_items WHERE id = ? AND meeting_id = ?")
                    .bind(eid)
                    .bind(id)
                    .execute(&self.pool)
                    .await?;
            }
        }
        let mut saved = Vec::with_capacity(items.len());
        for item in items {
            let item_id = match &item.id {
                Some(existing_id) => {
                    sqlx::query(
                        "UPDATE action_items SET text = ?, owner = ?, due_at = ?, confirmed = ?, reminder_set = ?
                         WHERE id = ? AND meeting_id = ?",
                    )
                    .bind(&item.text)
                    .bind(&item.owner)
                    .bind(item.due_at)
                    .bind(item.confirmed as i64)
                    .bind(item.reminder_set as i64)
                    .bind(existing_id)
                    .bind(id)
                    .execute(&self.pool)
                    .await?;
                    existing_id.clone()
                }
                None => {
                    let new_id = uuid::Uuid::new_v4().to_string();
                    sqlx::query(
                        "INSERT INTO action_items (id, meeting_id, text, owner, due_at, confirmed, reminder_set, created_at)
                         VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
                    )
                    .bind(&new_id)
                    .bind(id)
                    .bind(&item.text)
                    .bind(&item.owner)
                    .bind(item.due_at)
                    .bind(item.confirmed as i64)
                    .bind(item.reminder_set as i64)
                    .bind(now)
                    .execute(&self.pool)
                    .await?;
                    new_id
                }
            };
            saved.push(ActionItem {
                id: Some(item_id),
                ..item.clone()
            });
        }
        Ok(saved)
    }

    async fn import_calendar_events(&self, events: Vec<ImportedEvent>) -> Result<u32, StoreError> {
        let mut imported = 0u32;
        for ImportedEvent { event, attendees } in events {
            // Bug fix: `raw_uid = NULL` is explicitly exempt from the
            // (source, raw_uid) unique index below, so any caller that
            // passes one through duplicated that event on every re-import
            // forever. `parse_vevents` no longer produces one, but default
            // here too — defense in depth for any other/future import path.
            let raw_uid = event.raw_uid.clone().or_else(|| {
                Some(crate::calendar::content_uid(
                    &event.subject,
                    &event.organizer,
                    event.starts_at,
                    event.ends_at,
                ))
            });
            sqlx::query(
                "INSERT INTO calendar_events (id, source, subject, organizer, starts_at, ends_at, description, raw_uid)
                 VALUES (?, ?, ?, ?, ?, ?, ?, ?)
                 ON CONFLICT(source, raw_uid) WHERE raw_uid IS NOT NULL DO UPDATE SET
                     subject = excluded.subject,
                     organizer = excluded.organizer,
                     starts_at = excluded.starts_at,
                     ends_at = excluded.ends_at,
                     description = excluded.description",
            )
            .bind(&event.id)
            .bind(&event.source)
            .bind(&event.subject)
            .bind(&event.organizer)
            .bind(event.starts_at)
            .bind(event.ends_at)
            .bind(&event.description)
            .bind(&raw_uid)
            .execute(&self.pool)
            .await?;

            // A re-import may have updated an *existing* row rather than
            // inserting `event.id` — resolve the row that's actually there
            // before linking attendees to it.
            let row_id: String = match &raw_uid {
                Some(raw_uid) => {
                    sqlx::query("SELECT id FROM calendar_events WHERE source = ? AND raw_uid = ?")
                        .bind(&event.source)
                        .bind(raw_uid)
                        .fetch_one(&self.pool)
                        .await?
                        .get("id")
                }
                None => event.id.clone(),
            };

            for attendee in &attendees {
                let participant_id = self
                    .upsert_participant(&attendee.name, attendee.email.as_deref())
                    .await?;
                sqlx::query(
                    "INSERT INTO calendar_event_participants (calendar_event_id, participant_id, role)
                     VALUES (?, ?, ?)
                     ON CONFLICT(calendar_event_id, participant_id) DO UPDATE SET role = excluded.role",
                )
                .bind(&row_id)
                .bind(&participant_id)
                .bind(&attendee.role)
                .execute(&self.pool)
                .await?;
            }
            imported += 1;
        }
        Ok(imported)
    }

    async fn search(&self, query: &str) -> Result<Vec<SearchHit>, StoreError> {
        if query.trim().is_empty() {
            return Ok(Vec::new());
        }
        let fts_query = fts_query_from_input(query);
        let rows = sqlx::query(
            "SELECT m.id, m.title, m.started_at, m.duration_secs, m.status,
                    snippet(meeting_fts, -1, '', '', '…', 12) AS snippet
             FROM meeting_fts
             JOIN meetings m ON m.id = meeting_fts.meeting_id
             WHERE meeting_fts MATCH ?
             ORDER BY meeting_fts.rank
             LIMIT 50",
        )
        .bind(&fts_query)
        .fetch_all(&self.pool)
        .await?;
        let mut hits = Vec::with_capacity(rows.len());
        for row in &rows {
            let item = row_to_list_item(row);
            let tags = self.tags_for_meeting(&item.id).await?;
            hits.push(SearchHit {
                id: item.id,
                title: item.title,
                started_at: item.started_at,
                duration_secs: item.duration_secs,
                status: item.status,
                tags,
                snippet: row.get("snippet"),
            });
        }
        Ok(hits)
    }

    async fn set_tags(&self, meeting_id: &MeetingId, tags: &[String]) -> Result<(), StoreError> {
        sqlx::query("DELETE FROM meeting_tags WHERE meeting_id = ?")
            .bind(meeting_id)
            .execute(&self.pool)
            .await?;
        for name in tags {
            let name = name.trim();
            if name.is_empty() {
                continue;
            }
            let tag_id = self.upsert_tag(name).await?;
            sqlx::query("INSERT OR IGNORE INTO meeting_tags (meeting_id, tag_id) VALUES (?, ?)")
                .bind(meeting_id)
                .bind(&tag_id)
                .execute(&self.pool)
                .await?;
        }
        self.reindex_fts(meeting_id).await?;
        Ok(())
    }

    async fn list_tags(&self) -> Result<Vec<String>, StoreError> {
        Ok(sqlx::query_scalar("SELECT name FROM tags ORDER BY name")
            .fetch_all(&self.pool)
            .await?)
    }

    async fn list_pending_reminders(&self) -> Result<Vec<ActionItem>, StoreError> {
        let rows = sqlx::query(
            "SELECT id, text, owner, due_at, confirmed, reminder_set FROM action_items
             WHERE reminder_set = 1 AND due_at IS NOT NULL AND due_at > ?",
        )
        .bind(now_unix())
        .fetch_all(&self.pool)
        .await?;
        Ok(rows
            .iter()
            .map(|r| ActionItem {
                id: Some(r.get("id")),
                text: r.get("text"),
                owner: r.get("owner"),
                due_at: r.get("due_at"),
                confirmed: r.get::<i64, _>("confirmed") != 0,
                reminder_set: r.get::<i64, _>("reminder_set") != 0,
            })
            .collect())
    }

    async fn recover_scan(&self) -> Result<Vec<MeetingId>, StoreError> {
        let rows = sqlx::query(
            "SELECT id, audio_path FROM meetings WHERE status IN ('recording', 'transcribing')",
        )
        .fetch_all(&self.pool)
        .await?;
        let mut recovered = Vec::new();
        for row in rows {
            let id: MeetingId = row.get("id");
            let audio_path: String = row.get("audio_path");
            if PathBuf::from(&audio_path).exists() {
                sqlx::query(
                    "UPDATE meetings SET status = 'recovering', updated_at = ? WHERE id = ?",
                )
                .bind(now_unix())
                .bind(&id)
                .execute(&self.pool)
                .await?;
                recovered.push(id);
            } else {
                // No working audio and never finalized: nothing to recover from.
                sqlx::query("UPDATE meetings SET status = 'error', updated_at = ? WHERE id = ?")
                    .bind(now_unix())
                    .bind(&id)
                    .execute(&self.pool)
                    .await?;
            }
        }
        Ok(recovered)
    }

    async fn enforce_retention(&self, policy: Retention) -> Result<u32, StoreError> {
        if policy.max_age_days.is_none() && policy.max_size_gb.is_none() {
            return Ok(0);
        }
        let candidates = sqlx::query(
            "SELECT id, folder_path, started_at FROM meetings WHERE status IN ('ready', 'error') ORDER BY started_at ASC",
        )
        .fetch_all(&self.pool)
        .await?;

        let mut removed = 0u32;
        let now = now_unix();
        let max_age_secs = policy.max_age_days.map(|d| d as i64 * 86_400);

        // Age-based pruning first (cheap, no disk walk needed).
        let mut kept = Vec::new();
        for row in &candidates {
            let id: MeetingId = row.get("id");
            let started_at: i64 = row.get("started_at");
            if let Some(max_age) = max_age_secs {
                if now - started_at > max_age {
                    self.delete_meeting(&id).await?;
                    removed += 1;
                    continue;
                }
            }
            kept.push((id, row.get::<String, _>("folder_path")));
        }

        // Size-based pruning, oldest first, until under the cap.
        if let Some(max_gb) = policy.max_size_gb {
            let cap_bytes = max_gb as u64 * 1_073_741_824;
            let sizes: Vec<(MeetingId, u64)> = kept
                .iter()
                .map(|(id, folder)| (id.clone(), dir_size(folder)))
                .collect();
            let mut total: u64 = sizes.iter().map(|(_, s)| s).sum();
            let mut idx = 0;
            while total > cap_bytes && idx < sizes.len() {
                let (id, size) = &sizes[idx];
                self.delete_meeting(id).await?;
                total = total.saturating_sub(*size);
                removed += 1;
                idx += 1;
            }
        }

        Ok(removed)
    }

    async fn add_sync_target(&self, row: SyncTargetRow) -> Result<(), StoreError> {
        sqlx::query(
            "INSERT INTO sync_targets (id, name, kind, provider_hint, base_url, remote_base_path, \
             username, credential_ref, enabled, upload_transcript, upload_notes, upload_summary, \
             upload_recording, trigger_on_finalize, allow_plaintext_lan, encrypt_before_upload, created_at) \
             VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
        )
        .bind(&row.id)
        .bind(&row.name)
        .bind(&row.kind)
        .bind(&row.provider_hint)
        .bind(&row.base_url)
        .bind(&row.remote_base_path)
        .bind(&row.username)
        .bind(&row.credential_ref)
        .bind(row.enabled)
        .bind(row.upload_transcript)
        .bind(row.upload_notes)
        .bind(row.upload_summary)
        .bind(row.upload_recording)
        .bind(row.trigger_on_finalize)
        .bind(row.allow_plaintext_lan)
        .bind(row.encrypt_before_upload)
        .bind(row.created_at)
        .execute(&self.pool)
        .await?;
        Ok(())
    }

    async fn update_sync_target(&self, row: SyncTargetRow) -> Result<(), StoreError> {
        sqlx::query(
            "UPDATE sync_targets SET name=?, kind=?, provider_hint=?, base_url=?, remote_base_path=?, \
             username=?, credential_ref=?, enabled=?, upload_transcript=?, upload_notes=?, \
             upload_summary=?, upload_recording=?, trigger_on_finalize=?, allow_plaintext_lan=?, \
             encrypt_before_upload=? WHERE id=?",
        )
        .bind(&row.name)
        .bind(&row.kind)
        .bind(&row.provider_hint)
        .bind(&row.base_url)
        .bind(&row.remote_base_path)
        .bind(&row.username)
        .bind(&row.credential_ref)
        .bind(row.enabled)
        .bind(row.upload_transcript)
        .bind(row.upload_notes)
        .bind(row.upload_summary)
        .bind(row.upload_recording)
        .bind(row.trigger_on_finalize)
        .bind(row.allow_plaintext_lan)
        .bind(row.encrypt_before_upload)
        .bind(&row.id)
        .execute(&self.pool)
        .await?;
        Ok(())
    }

    async fn remove_sync_target(&self, id: &str) -> Result<(), StoreError> {
        // ON DELETE CASCADE clears this target's sync_jobs too.
        sqlx::query("DELETE FROM sync_targets WHERE id = ?")
            .bind(id)
            .execute(&self.pool)
            .await?;
        Ok(())
    }

    async fn list_sync_targets(&self) -> Result<Vec<SyncTargetRow>, StoreError> {
        let rows = sqlx::query_as::<_, SyncTargetRow>(
            "SELECT * FROM sync_targets ORDER BY created_at ASC",
        )
        .fetch_all(&self.pool)
        .await?;
        Ok(rows)
    }

    async fn get_sync_target(&self, id: &str) -> Result<SyncTargetRow, StoreError> {
        let row = sqlx::query_as::<_, SyncTargetRow>("SELECT * FROM sync_targets WHERE id = ?")
            .bind(id)
            .fetch_optional(&self.pool)
            .await?
            .ok_or_else(|| StoreError::NotFound(format!("sync target {id}")))?;
        Ok(row)
    }

    async fn upsert_sync_job(&self, job: SyncJobRow) -> Result<bool, StoreError> {
        // ON CONFLICT: re-enqueue as pending, EXCEPT when the row is already
        // `done` with the same content hash — that's the skip-if-unchanged case
        // (`IS NOT` is SQLite's null-safe comparison).
        let res = sqlx::query(
            "INSERT INTO sync_jobs (id, target_id, meeting_id, artifact, local_path, remote_path, \
             sha256, status, attempts, last_error, next_attempt_at, bytes_total, bytes_sent, updated_at) \
             VALUES (?,?,?,?,?,?,?, 'pending', 0, NULL, NULL, ?, 0, ?) \
             ON CONFLICT(target_id, meeting_id, artifact) DO UPDATE SET \
               local_path=excluded.local_path, remote_path=excluded.remote_path, sha256=excluded.sha256, \
               status='pending', attempts=0, last_error=NULL, next_attempt_at=NULL, \
               bytes_total=excluded.bytes_total, bytes_sent=0, updated_at=excluded.updated_at \
             WHERE sync_jobs.status != 'done' OR sync_jobs.sha256 IS NOT excluded.sha256",
        )
        .bind(&job.id)
        .bind(&job.target_id)
        .bind(&job.meeting_id)
        .bind(&job.artifact)
        .bind(&job.local_path)
        .bind(&job.remote_path)
        .bind(&job.sha256)
        .bind(job.bytes_total)
        .bind(job.updated_at)
        .execute(&self.pool)
        .await?;
        Ok(res.rows_affected() > 0)
    }

    async fn claim_due_sync_jobs(&self, now: i64) -> Result<Vec<SyncJobRow>, StoreError> {
        let rows = sqlx::query_as::<_, SyncJobRow>(
            "SELECT * FROM sync_jobs WHERE status IN ('pending','failed') \
             AND (next_attempt_at IS NULL OR next_attempt_at <= ?) ORDER BY updated_at ASC",
        )
        .bind(now)
        .fetch_all(&self.pool)
        .await?;
        Ok(rows)
    }

    async fn update_sync_job(&self, job: SyncJobRow) -> Result<(), StoreError> {
        sqlx::query(
            "UPDATE sync_jobs SET status=?, attempts=?, last_error=?, next_attempt_at=?, \
             bytes_total=?, bytes_sent=?, sha256=?, local_path=?, remote_path=?, updated_at=? WHERE id=?",
        )
        .bind(&job.status)
        .bind(job.attempts)
        .bind(&job.last_error)
        .bind(job.next_attempt_at)
        .bind(job.bytes_total)
        .bind(job.bytes_sent)
        .bind(&job.sha256)
        .bind(&job.local_path)
        .bind(&job.remote_path)
        .bind(job.updated_at)
        .bind(&job.id)
        .execute(&self.pool)
        .await?;
        Ok(())
    }

    async fn get_sync_job(&self, id: &str) -> Result<SyncJobRow, StoreError> {
        sqlx::query_as::<_, SyncJobRow>("SELECT * FROM sync_jobs WHERE id = ?")
            .bind(id)
            .fetch_optional(&self.pool)
            .await?
            .ok_or_else(|| StoreError::NotFound(format!("sync job {id}")))
    }

    async fn list_sync_jobs(
        &self,
        meeting_id: Option<&MeetingId>,
    ) -> Result<Vec<SyncJobRow>, StoreError> {
        let rows = match meeting_id {
            Some(m) => {
                sqlx::query_as::<_, SyncJobRow>(
                    "SELECT * FROM sync_jobs WHERE meeting_id = ? ORDER BY updated_at DESC",
                )
                .bind(m)
                .fetch_all(&self.pool)
                .await?
            }
            None => {
                sqlx::query_as::<_, SyncJobRow>("SELECT * FROM sync_jobs ORDER BY updated_at DESC")
                    .fetch_all(&self.pool)
                    .await?
            }
        };
        Ok(rows)
    }

    async fn insert_feature_brief(&self, row: FeatureBriefRow) -> Result<(), StoreError> {
        sqlx::query(
            "INSERT INTO feature_briefs (id, meeting_id, title, target_repo, path, exposed, created_at) \
             VALUES (?,?,?,?,?,?,?)",
        )
        .bind(&row.id)
        .bind(&row.meeting_id)
        .bind(&row.title)
        .bind(&row.target_repo)
        .bind(&row.path)
        .bind(row.exposed)
        .bind(row.created_at)
        .execute(&self.pool)
        .await?;
        Ok(())
    }

    async fn list_feature_briefs(
        &self,
        meeting_id: Option<&MeetingId>,
    ) -> Result<Vec<FeatureBriefInfo>, StoreError> {
        let rows =
            match meeting_id {
                Some(m) => sqlx::query_as::<_, FeatureBriefRow>(
                    "SELECT * FROM feature_briefs WHERE meeting_id = ? ORDER BY created_at DESC",
                )
                .bind(m)
                .fetch_all(&self.pool)
                .await?,
                None => {
                    sqlx::query_as::<_, FeatureBriefRow>(
                        "SELECT * FROM feature_briefs ORDER BY created_at DESC",
                    )
                    .fetch_all(&self.pool)
                    .await?
                }
            };
        Ok(rows.into_iter().map(FeatureBriefInfo::from).collect())
    }

    async fn get_feature_brief_row(&self, id: &str) -> Result<FeatureBriefRow, StoreError> {
        sqlx::query_as::<_, FeatureBriefRow>("SELECT * FROM feature_briefs WHERE id = ?")
            .bind(id)
            .fetch_optional(&self.pool)
            .await?
            .ok_or_else(|| StoreError::NotFound(format!("feature brief {id}")))
    }

    async fn set_brief_exposed(&self, id: &str, exposed: bool) -> Result<(), StoreError> {
        let res = sqlx::query("UPDATE feature_briefs SET exposed = ? WHERE id = ?")
            .bind(exposed)
            .bind(id)
            .execute(&self.pool)
            .await?;
        if res.rows_affected() == 0 {
            return Err(StoreError::NotFound(format!("feature brief {id}")));
        }
        Ok(())
    }

    async fn list_action_items(
        &self,
        meeting_id: &MeetingId,
    ) -> Result<Vec<ActionItem>, StoreError> {
        let rows = sqlx::query(
            "SELECT id, text, owner, due_at, confirmed, reminder_set FROM action_items \
             WHERE meeting_id = ? ORDER BY created_at ASC",
        )
        .bind(meeting_id)
        .fetch_all(&self.pool)
        .await?;
        Ok(rows
            .iter()
            .map(|r| ActionItem {
                id: Some(r.get("id")),
                text: r.get("text"),
                owner: r.get("owner"),
                due_at: r.get("due_at"),
                confirmed: r.get::<i64, _>("confirmed") != 0,
                reminder_set: r.get::<i64, _>("reminder_set") != 0,
            })
            .collect())
    }

    async fn record_mcp_access(
        &self,
        tool: &str,
        meeting_id: Option<&MeetingId>,
        client: Option<&str>,
    ) -> Result<(), StoreError> {
        sqlx::query(
            "INSERT INTO mcp_access_log (id, at, tool, meeting_id, client) VALUES (?,?,?,?,?)",
        )
        .bind(uuid::Uuid::new_v4().to_string())
        .bind(now_unix())
        .bind(tool)
        .bind(meeting_id)
        .bind(client)
        .execute(&self.pool)
        .await?;
        Ok(())
    }

    async fn list_mcp_access_log(
        &self,
        limit: Option<u32>,
    ) -> Result<Vec<McpAccessEntry>, StoreError> {
        let cap: i64 = limit.map(i64::from).unwrap_or(200).max(1);
        let rows = sqlx::query(
            "SELECT at, tool, meeting_id, client FROM mcp_access_log ORDER BY at DESC LIMIT ?",
        )
        .bind(cap)
        .fetch_all(&self.pool)
        .await?;
        Ok(rows
            .iter()
            .map(|r| McpAccessEntry {
                at: r.get("at"),
                tool: r.get("tool"),
                meeting_id: r.get("meeting_id"),
                client: r.get("client"),
            })
            .collect())
    }
}

/// Read a derived artifact, transparently decrypting it if the vault sealed it
/// (T8.8, FR-SEC-3). Returns `None` if the file is missing or can't be
/// decrypted (vault locked / wrong key) — callers already treat a missing
/// artifact as empty, so a locked vault degrades to "no content" rather than a
/// crash.
fn read_artifact(path: &std::path::Path) -> Option<String> {
    let bytes = std::fs::read(path).ok()?;
    let plain = crate::vault::open(&bytes).ok()?;
    Some(String::from_utf8_lossy(&plain).into_owned())
}

/// Write a derived artifact, sealing it when the vault is unlocked (a no-op
/// passthrough otherwise). Audio is not sealed here — it's handled separately.
fn write_artifact(path: &std::path::Path, contents: &[u8]) -> std::io::Result<()> {
    let sealed = crate::vault::seal(contents).map_err(|e| std::io::Error::other(e.to_string()))?;
    std::fs::write(path, sealed)
}

fn dir_size(path: &str) -> u64 {
    std::fs::read_dir(path)
        .into_iter()
        .flatten()
        .flatten()
        .filter_map(|entry| entry.metadata().ok())
        .map(|m| m.len())
        .sum()
}

#[cfg(test)]
mod tests {
    use super::*;

    fn sample_target(id: &str) -> SyncTargetRow {
        SyncTargetRow {
            id: id.to_string(),
            name: "t".to_string(),
            kind: "webdav".to_string(),
            provider_hint: None,
            base_url: Some("https://host/dav".to_string()),
            remote_base_path: "/WhispAssist".to_string(),
            username: Some("u".to_string()),
            credential_ref: format!("wa-sync-{id}"),
            enabled: true,
            upload_transcript: true,
            upload_notes: true,
            upload_summary: true,
            upload_recording: false,
            trigger_on_finalize: true,
            allow_plaintext_lan: false,
            encrypt_before_upload: false,
            created_at: 0,
        }
    }

    fn sample_job(id: &str, sha: &str) -> SyncJobRow {
        SyncJobRow {
            id: id.to_string(),
            target_id: "t1".to_string(),
            meeting_id: "m1".to_string(),
            artifact: "transcript".to_string(),
            local_path: "/x/transcript.json".to_string(),
            remote_path: "/WhispAssist/m1/transcript.json".to_string(),
            sha256: Some(sha.to_string()),
            status: "pending".to_string(),
            attempts: 0,
            last_error: None,
            next_attempt_at: None,
            bytes_total: Some(10),
            bytes_sent: 0,
            updated_at: 0,
        }
    }

    /// T8.7/FR-TRX-4: the language requested at `create_meeting` is visible
    /// immediately (not just after `finalize_meeting`) — a crash-recovered
    /// `recovering` meeting still knows what was asked for.
    #[tokio::test]
    async fn create_meeting_persists_the_requested_language_immediately() {
        let store = SqliteStore::connect_in_memory().await.unwrap();
        let id = store
            .create_meeting(NewMeeting {
                title: "Reunión semanal".to_string(),
                calendar_event_id: None,
                template_id: None,
                language: Some("es".to_string()),
            })
            .await
            .unwrap();
        let meeting = store.get_meeting(&id).await.unwrap();
        assert_eq!(meeting.language.as_deref(), Some("es"));
    }

    /// `finalize_meeting` overwrites whatever `create_meeting` stored with
    /// the language whisper.cpp actually resolved/detected (T8.7, FR-TRX-4)
    /// — e.g. "auto" mode's detected result, or an English-only model's
    /// forced "en".
    #[tokio::test]
    async fn finalize_meeting_overwrites_the_requested_language_with_the_resolved_one() {
        let store = SqliteStore::connect_in_memory().await.unwrap();
        let id = store
            .create_meeting(NewMeeting {
                title: "Auto-detect meeting".to_string(),
                calendar_event_id: None,
                template_id: None,
                language: None, // requested "auto"
            })
            .await
            .unwrap();
        store
            .finalize_meeting(
                &id,
                FinalizeMeeting {
                    segments: Vec::new(),
                    speakers: Vec::new(),
                    duration_secs: 42,
                    recorded: false,
                    language: Some("fr".to_string()), // what auto-detect resolved to
                    backend_used: Some("cpu".to_string()),
                    model_used: Some("small-q5_1".to_string()),
                    audio_layout: None,
                },
            )
            .await
            .unwrap();
        let meeting = store.get_meeting(&id).await.unwrap();
        assert_eq!(meeting.language.as_deref(), Some("fr"));
    }

    /// End-to-end regression for the search bug: `meeting_fts` originally had
    /// no summary/tags columns at all and nothing reindexed on those writes,
    /// so search silently missed anything that wasn't in the title,
    /// transcript, or notes (FR-SEARCH-1).
    #[tokio::test]
    async fn search_finds_hits_via_transcript_notes_summary_and_tags() {
        let store = SqliteStore::connect_in_memory().await.unwrap();
        let id = store
            .create_meeting(NewMeeting {
                title: "Weekly sync".to_string(),
                calendar_event_id: None,
                template_id: None,
                language: None,
            })
            .await
            .unwrap();
        store
            .finalize_meeting(
                &id,
                FinalizeMeeting {
                    segments: vec![TranscriptSegment {
                        id: 0,
                        start_ms: 0,
                        end_ms: 1000,
                        speaker: "S1".to_string(),
                        text: "let's discuss the transcriptword rollout".to_string(),
                        confidence: None,
                        interim: false,
                    }],
                    speakers: Vec::new(),
                    duration_secs: 60,
                    recorded: false,
                    language: Some("en".to_string()),
                    backend_used: Some("cpu".to_string()),
                    model_used: Some("small-q5_1".to_string()),
                    audio_layout: None,
                },
            )
            .await
            .unwrap();
        store
            .update_notes(&id, "action: follow up on notesword")
            .await
            .unwrap();
        store.set_tags(&id, &["tagword".to_string()]).await.unwrap();

        // summary.json isn't written through the store (generate_summary
        // seals it straight to disk in commands.rs), so reindex_fts must
        // pick it up when explicitly told to, same as that command does.
        let summary = SummaryFile {
            schema: 1,
            generated_at: 0,
            provider: "test".to_string(),
            model: "test".to_string(),
            summary_md: "summaryword recap".to_string(),
            decisions: Vec::new(),
            action_items: Vec::new(),
        };
        write_artifact(
            &paths::meeting_dir(&id).join("summary.json"),
            serde_json::to_string(&summary).unwrap().as_bytes(),
        )
        .unwrap();
        store.reindex_fts(&id).await.unwrap();

        for (query, source) in [
            ("transcriptword", "transcript"),
            ("notesword", "notes"),
            ("summaryword", "summary"),
            ("tagword", "tags"),
        ] {
            let hits = store.search(query).await.unwrap();
            assert!(
                hits.iter().any(|h| h.id == id),
                "expected a hit from {source} for query {query:?}, got {hits:?}"
            );
        }
    }

    #[tokio::test]
    async fn rename_meeting_updates_the_title() {
        let store = SqliteStore::connect_in_memory().await.unwrap();
        let id = store
            .create_meeting(NewMeeting {
                title: "Untitled meeting".to_string(),
                calendar_event_id: None,
                template_id: None,
                language: None,
            })
            .await
            .unwrap();
        store.rename_meeting(&id, "Sprint planning").await.unwrap();
        assert_eq!(
            store.get_meeting(&id).await.unwrap().title,
            "Sprint planning"
        );
    }

    #[tokio::test]
    async fn rename_meeting_errs_for_an_unknown_id() {
        let store = SqliteStore::connect_in_memory().await.unwrap();
        let result = store.rename_meeting(&"no-such-id".to_string(), "x").await;
        assert!(matches!(result, Err(StoreError::NotFound(_))));
    }

    #[tokio::test]
    async fn attach_meeting_to_event_mirrors_the_events_subject_onto_the_title() {
        let store = SqliteStore::connect_in_memory().await.unwrap();
        let meeting_id = store
            .create_meeting(NewMeeting {
                title: "Untitled meeting".to_string(),
                calendar_event_id: None,
                template_id: None,
                language: None,
            })
            .await
            .unwrap();
        store
            .import_calendar_events(vec![ImportedEvent {
                event: CalendarEvent {
                    id: "ev1".to_string(),
                    source: "pst".to_string(),
                    subject: Some("Jerry / Daniel - Weekly 1:1".to_string()),
                    organizer: None,
                    starts_at: None,
                    ends_at: None,
                    description: None,
                    raw_uid: Some("uid-1".to_string()),
                },
                attendees: vec![],
            }])
            .await
            .unwrap();
        let event_id = store.list_calendar_events(None, None).await.unwrap()[0]
            .id
            .clone();

        store
            .attach_meeting_to_event(&meeting_id, &event_id)
            .await
            .unwrap();

        let meeting = store.get_meeting(&meeting_id).await.unwrap();
        assert_eq!(meeting.title, "Jerry / Daniel - Weekly 1:1");
        assert_eq!(
            meeting.calendar_event_id.as_deref(),
            Some(event_id.as_str())
        );
    }

    #[tokio::test]
    async fn attach_meeting_to_event_leaves_the_title_alone_when_the_event_has_no_subject() {
        let store = SqliteStore::connect_in_memory().await.unwrap();
        let meeting_id = store
            .create_meeting(NewMeeting {
                title: "Untitled meeting".to_string(),
                calendar_event_id: None,
                template_id: None,
                language: None,
            })
            .await
            .unwrap();
        store
            .import_calendar_events(vec![ImportedEvent {
                event: CalendarEvent {
                    id: "ev1".to_string(),
                    source: "pst".to_string(),
                    subject: None,
                    organizer: None,
                    starts_at: None,
                    ends_at: None,
                    description: None,
                    raw_uid: Some("uid-2".to_string()),
                },
                attendees: vec![],
            }])
            .await
            .unwrap();
        let event_id = store.list_calendar_events(None, None).await.unwrap()[0]
            .id
            .clone();

        store
            .attach_meeting_to_event(&meeting_id, &event_id)
            .await
            .unwrap();

        assert_eq!(
            store.get_meeting(&meeting_id).await.unwrap().title,
            "Untitled meeting"
        );
    }

    /// Regression for the calendar duplicate-import bug: re-importing the
    /// exact same events (same `(source, raw_uid)`) must update the existing
    /// rows in place, not add new ones -- whether or not the source
    /// supplied a `raw_uid` at all.
    #[tokio::test]
    async fn reimporting_the_same_events_does_not_duplicate_rows() {
        let store = SqliteStore::connect_in_memory().await.unwrap();
        let events = || {
            vec![
                ImportedEvent {
                    event: CalendarEvent {
                        id: "ev1".to_string(),
                        source: "pst".to_string(),
                        subject: Some("Weekly 1:1".to_string()),
                        organizer: None,
                        starts_at: Some(1_000),
                        ends_at: Some(2_000),
                        description: None,
                        raw_uid: Some("uid-1".to_string()),
                    },
                    attendees: vec![],
                },
                // No raw_uid at all -- the case that used to duplicate on
                // every re-import (exempt from the unique index).
                ImportedEvent {
                    event: CalendarEvent {
                        id: "ev2".to_string(),
                        source: "pst".to_string(),
                        subject: Some("No UID event".to_string()),
                        organizer: None,
                        starts_at: Some(3_000),
                        ends_at: Some(4_000),
                        description: None,
                        raw_uid: None,
                    },
                    attendees: vec![],
                },
            ]
        };
        store.import_calendar_events(events()).await.unwrap();
        assert_eq!(
            store.list_calendar_events(None, None).await.unwrap().len(),
            2
        );

        // Re-import the identical batch a few times, as a user retrying an
        // import would.
        for _ in 0..3 {
            store.import_calendar_events(events()).await.unwrap();
        }
        assert_eq!(
            store.list_calendar_events(None, None).await.unwrap().len(),
            2
        );
    }

    /// Regression for the "14,686 calendar events" bloat report: cleanup
    /// must delete unlinked old events while never touching one attached to
    /// a recorded meeting, whether age-bounded or "delete all".
    #[tokio::test]
    async fn cleanup_calendar_events_protects_meeting_linked_rows() {
        let store = SqliteStore::connect_in_memory().await.unwrap();
        let event = |id: &str, starts_at: i64| ImportedEvent {
            event: CalendarEvent {
                id: id.to_string(),
                source: "pst".to_string(),
                subject: Some(id.to_string()),
                organizer: None,
                starts_at: Some(starts_at),
                ends_at: Some(starts_at + 1_800),
                description: None,
                raw_uid: Some(id.to_string()),
            },
            attendees: vec![],
        };
        store
            .import_calendar_events(vec![
                event("old-unlinked", 1_000),
                event("old-linked", 1_000),
                event("recent-unlinked", 1_000_000_000),
            ])
            .await
            .unwrap();

        let meeting_id = store
            .create_meeting(NewMeeting {
                title: "Untitled meeting".to_string(),
                calendar_event_id: None,
                template_id: None,
                language: None,
            })
            .await
            .unwrap();
        store
            .attach_meeting_to_event(&meeting_id, "old-linked")
            .await
            .unwrap();

        // Cutoff between the two "old" timestamps and the "recent" one.
        let (deleted, protected) = store.cleanup_calendar_events(Some(500_000)).await.unwrap();
        assert_eq!(deleted, 1, "only old-unlinked should go");
        assert_eq!(protected, 1, "old-linked is attached to a meeting");

        let remaining: Vec<String> = store
            .list_calendar_events(None, None)
            .await
            .unwrap()
            .into_iter()
            .map(|e| e.id)
            .collect();
        assert!(!remaining.contains(&"old-unlinked".to_string()));
        assert!(remaining.contains(&"old-linked".to_string()));
        assert!(remaining.contains(&"recent-unlinked".to_string()));
    }

    #[tokio::test]
    async fn cleanup_calendar_events_delete_all_still_protects_linked_rows() {
        let store = SqliteStore::connect_in_memory().await.unwrap();
        let event = |id: &str| ImportedEvent {
            event: CalendarEvent {
                id: id.to_string(),
                source: "pst".to_string(),
                subject: Some(id.to_string()),
                organizer: None,
                starts_at: Some(1_000),
                ends_at: Some(2_800),
                description: None,
                raw_uid: Some(id.to_string()),
            },
            attendees: vec![],
        };
        store
            .import_calendar_events(vec![event("unlinked"), event("linked")])
            .await
            .unwrap();
        let meeting_id = store
            .create_meeting(NewMeeting {
                title: "Untitled meeting".to_string(),
                calendar_event_id: None,
                template_id: None,
                language: None,
            })
            .await
            .unwrap();
        store
            .attach_meeting_to_event(&meeting_id, "linked")
            .await
            .unwrap();

        let (deleted, protected) = store.cleanup_calendar_events(None).await.unwrap();
        assert_eq!(deleted, 1);
        assert_eq!(protected, 1);
        assert_eq!(
            store.list_calendar_events(None, None).await.unwrap().len(),
            1
        );
    }

    #[tokio::test]
    async fn sync_target_crud_round_trips() {
        let store = SqliteStore::connect_in_memory().await.unwrap();
        store.add_sync_target(sample_target("t1")).await.unwrap();
        let listed = store.list_sync_targets().await.unwrap();
        assert_eq!(listed.len(), 1);
        assert_eq!(listed[0].id, "t1");
        store.remove_sync_target("t1").await.unwrap();
        assert!(store.list_sync_targets().await.unwrap().is_empty());
    }

    #[tokio::test]
    async fn upsert_skips_unchanged_done_job_and_reenqueues_changed() {
        let store = SqliteStore::connect_in_memory().await.unwrap();
        store.add_sync_target(sample_target("t1")).await.unwrap();

        // New job → enqueued.
        assert!(store.upsert_sync_job(sample_job("j1", "A")).await.unwrap());

        // Mark it done, then re-upsert the same content (same sha) → skipped.
        let mut done = store.list_sync_jobs(None).await.unwrap().remove(0);
        done.status = "done".to_string();
        store.update_sync_job(done).await.unwrap();
        assert!(
            !store.upsert_sync_job(sample_job("j2", "A")).await.unwrap(),
            "an unchanged, already-done artifact must not re-enqueue"
        );

        // Changed content (new sha) → re-enqueued as pending, still one row.
        assert!(store.upsert_sync_job(sample_job("j3", "B")).await.unwrap());
        let jobs = store.list_sync_jobs(None).await.unwrap();
        assert_eq!(
            jobs.len(),
            1,
            "UNIQUE(target,meeting,artifact) keeps one row"
        );
        assert_eq!(jobs[0].status, "pending");
        assert_eq!(jobs[0].sha256.as_deref(), Some("B"));
    }

    #[tokio::test]
    async fn claim_due_skips_jobs_with_a_future_backoff() {
        let store = SqliteStore::connect_in_memory().await.unwrap();
        store.add_sync_target(sample_target("t1")).await.unwrap();
        store.upsert_sync_job(sample_job("j1", "A")).await.unwrap();
        // Push it into the future with a failed+backoff state.
        let mut job = store.list_sync_jobs(None).await.unwrap().remove(0);
        job.status = "failed".to_string();
        job.next_attempt_at = Some(10_000);
        store.update_sync_job(job).await.unwrap();
        assert!(store.claim_due_sync_jobs(5_000).await.unwrap().is_empty());
        assert_eq!(store.claim_due_sync_jobs(20_000).await.unwrap().len(), 1);
    }

    #[test]
    fn fts_query_from_input_quotes_each_token_as_a_literal_phrase() {
        assert_eq!(
            fts_query_from_input("standup notes"),
            "\"standup\" \"notes\""
        );
    }

    #[test]
    fn fts_query_from_input_escapes_embedded_quotes() {
        assert_eq!(fts_query_from_input("say \"hi\""), "\"say\" \"\"\"hi\"\"\"");
    }

    #[test]
    fn fts_query_from_input_neutralizes_fts5_operators() {
        // Without quoting, "-" and "OR"/"AND"/"*" carry special FTS5 meaning;
        // quoted, they're just literal tokens that can't blow up MATCH.
        assert_eq!(
            fts_query_from_input("a - b OR c*"),
            "\"a\" \"-\" \"b\" \"OR\" \"c*\""
        );
    }

    #[test]
    fn fts_query_from_input_of_blank_input_is_blank() {
        assert_eq!(fts_query_from_input("   "), "");
    }
}