kcode-audio-ingress 0.1.1

//! Durable, automatic audio transcription.
//!
//! [`AudioIngress`] accepts owned WAV bytes, persists them before returning,
//! and automatically transcribes them in the background.

#![deny(missing_docs)]
#![forbid(unsafe_code)]

use std::{
    collections::HashMap,
    fs,
    path::{Path, PathBuf},
    sync::{Arc, Mutex, Weak},
    time::Duration,
};

use anyhow::{Context, ensure};
use chrono::{DateTime, Duration as ChronoDuration, Utc};
use rusqlite::{Connection, OptionalExtension, params};
use serde::Serialize;
use serde_json::Value;
use sha2::{Digest, Sha256};
use tokio::io::AsyncWriteExt;
pub use transcribe::{
    AudioTranscriber, JobState, RECONCILIATION_MODEL, RECONCILIATION_REASONING, Step, StepError,
    StepState, StepStatus, TRANSCRIPTION_MODEL, TranscriptionJob, TranscriptionStatus,
};
use transcribe::{ChunkPlan, ChunkTranscript, PIECE_CACHE_REVISION, PieceCache, PieceSink};
use uuid::Uuid;

mod transcribe;

const INITIAL_MIGRATION: &str = include_str!("../migrations/001_initial.sql");
const RELEASE_DEFERRED_INGRESS_MIGRATION: &str =
    include_str!("../migrations/002_release_deferred_ingress.sql");
const TRANSCRIPTION_STATUS_MIGRATION: &str =
    include_str!("../migrations/003_transcription_status.sql");
const RETRY_ROUNDED_WAV_INTERVALS_MIGRATION: &str =
    include_str!("../migrations/004_retry_rounded_wav_intervals.sql");
const UNIFIED_INGRESS_QUEUE_MIGRATION: &str =
    include_str!("../migrations/005_unified_ingress_queue.sql");
const STANDALONE_LIBRARY_MIGRATION: &str = include_str!("../migrations/006_standalone_library.sql");
const DURABLE_TRANSCRIPT_PIECES_MIGRATION: &str =
    include_str!("../migrations/007_durable_transcript_pieces.sql");
const UNIQUE_TRANSCRIPTION_ATTEMPTS_MIGRATION: &str =
    include_str!("../migrations/008_unique_transcription_attempts.sql");

const DATABASE_FILENAME: &str = "state.sqlite3";
const ORIGINALS_DIRECTORY: &str = "originals";
const FAILURE_LIMIT: i64 = 5;
const RETRY_DELAY_SECONDS: i64 = 15;
const LATEST_SCHEMA_VERSION: i64 = 8;

const FRESH_SCHEMA: &str = r#"
CREATE TABLE audio_recordings (
    id TEXT PRIMARY KEY NOT NULL,
    sha256 TEXT NOT NULL UNIQUE CHECK(length(sha256) = 64),
    original_filename TEXT NOT NULL,
    content_type TEXT NOT NULL,
    size_bytes INTEGER NOT NULL CHECK(size_bytes >= 0),
    source_created_at TEXT NOT NULL,
    received_at TEXT NOT NULL,
    updated_at TEXT NOT NULL,
    original_relative_path TEXT NOT NULL,
    status TEXT NOT NULL CHECK(status IN (
        'uploaded', 'chunking', 'transcribing', 'reconciling',
        'ready_for_ingress', 'ingressing', 'ingress_failed', 'complete', 'failed'
    )),
    gemini_model TEXT NOT NULL,
    reconciliation_model TEXT NOT NULL,
    reconciliation_reasoning TEXT NOT NULL,
    final_transcript TEXT,
    attempt_count INTEGER NOT NULL DEFAULT 0 CHECK(attempt_count >= 0),
    next_attempt_at TEXT,
    last_error TEXT,
    transcription_status_json TEXT,
    failure_retryable INTEGER NOT NULL DEFAULT 1
        CHECK(failure_retryable IN (0, 1))
);

CREATE INDEX audio_recordings_work_queue
ON audio_recordings(status, next_attempt_at, received_at);

CREATE TABLE audio_transcript_pieces (
    recording_id TEXT NOT NULL REFERENCES audio_recordings(id) ON DELETE CASCADE,
    attempt_id TEXT NOT NULL CHECK(length(attempt_id) > 0),
    cache_revision TEXT NOT NULL CHECK(length(cache_revision) > 0),
    piece_index INTEGER NOT NULL CHECK(piece_index >= 0),
    piece_count INTEGER NOT NULL CHECK(piece_count > 0 AND piece_index < piece_count),
    audio_start_ms INTEGER NOT NULL CHECK(audio_start_ms >= 0),
    audio_end_ms INTEGER NOT NULL CHECK(audio_end_ms > audio_start_ms),
    transcript_json TEXT NOT NULL,
    created_at TEXT NOT NULL,
    PRIMARY KEY(recording_id, attempt_id, piece_index)
);

CREATE INDEX audio_transcript_pieces_cache_lookup
ON audio_transcript_pieces(
    recording_id,
    cache_revision,
    piece_index,
    piece_count,
    audio_start_ms,
    audio_end_ms,
    created_at
);

PRAGMA user_version = 8;
"#;

/// Owned audio and its source metadata.
#[derive(Clone, Debug)]
pub struct AudioInput {
    /// Complete WAV bytes.
    pub bytes: Vec<u8>,
    /// Instant at which recording began.
    pub recorded_at: DateTime<Utc>,
    /// Original leaf filename, when known.
    pub original_filename: Option<String>,
}

/// Result of durably submitting audio.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
pub struct Submission {
    /// Stable recording identifier.
    pub recording_id: Uuid,
    /// Whether the same audio bytes were already known.
    pub deduplicated: bool,
}

/// Complete current library state.
#[derive(Clone, Debug, Serialize)]
pub struct Status {
    /// Recordings ordered by recording time, newest first.
    pub recordings: Vec<RecordingStatus>,
}

/// Current state and immutable identity of one recording.
#[derive(Clone, Debug, Serialize)]
pub struct RecordingStatus {
    /// Stable recording identifier.
    pub id: Uuid,
    /// Lowercase SHA-256 digest of the original bytes.
    pub sha256: String,
    /// Sanitized original filename.
    pub original_filename: String,
    /// Original byte length.
    pub size_bytes: u64,
    /// Instant at which recording began.
    pub recorded_at: DateTime<Utc>,
    /// Instant at which AudioIngress accepted the recording.
    pub received_at: DateTime<Utc>,
    /// Gemini transcription model attribution.
    pub transcription_model: String,
    /// Codex reconciliation model attribution.
    pub reconciliation_model: String,
    /// Codex reconciliation reasoning attribution.
    pub reconciliation_reasoning: String,
    /// Current processing state.
    pub state: RecordingState,
}

/// Automatic processing state of one recording.
#[derive(Clone, Debug, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum RecordingState {
    /// Persisted and waiting for automatic processing.
    Queued,
    /// An in-memory transcription attempt is active.
    Processing {
        /// One-based full-job attempt number.
        attempt: u8,
        /// Current dependency status. Its transcript is always omitted.
        progress: TranscriptionStatus,
    },
    /// The canonical transcript is durable.
    Complete {
        /// Canonical reconciled Markdown.
        transcript: String,
    },
    /// Automatic processing stopped.
    Failed {
        /// Attempts used in the current manual-attempt budget.
        attempts: u8,
        /// Concise diagnostic.
        error: String,
        /// Whether another attempt can reasonably succeed.
        retryable: bool,
    },
}

/// Stable library error category.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ErrorKind {
    /// Supplied data is invalid.
    InvalidInput,
    /// The requested recording does not exist.
    NotFound,
    /// The requested transition is not valid in the current state.
    Conflict,
    /// Persistence or internal processing failed unexpectedly.
    Internal,
}

/// Error returned by the minimal AudioIngress API.
#[derive(Debug)]
pub struct Error {
    kind: ErrorKind,
    message: String,
}

impl Error {
    fn invalid(message: impl Into<String>) -> Self {
        Self {
            kind: ErrorKind::InvalidInput,
            message: message.into(),
        }
    }

    fn not_found() -> Self {
        Self {
            kind: ErrorKind::NotFound,
            message: "Audio recording not found.".into(),
        }
    }

    fn conflict(message: impl Into<String>) -> Self {
        Self {
            kind: ErrorKind::Conflict,
            message: message.into(),
        }
    }

    fn internal(error: impl std::fmt::Display) -> Self {
        tracing::error!(%error, "AudioIngress operation failed");
        Self {
            kind: ErrorKind::Internal,
            message: "An unexpected AudioIngress error occurred.".into(),
        }
    }

    /// Returns the stable error category.
    pub fn kind(&self) -> ErrorKind {
        self.kind
    }
}

impl std::fmt::Display for Error {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter.write_str(&self.message)
    }
}

impl std::error::Error for Error {}

struct TemporaryUpload(PathBuf);

impl Drop for TemporaryUpload {
    fn drop(&mut self) {
        let _ = fs::remove_file(&self.0);
    }
}

struct Inner {
    root: PathBuf,
    db: Arc<Mutex<Connection>>,
    transcriber: AudioTranscriber,
    jobs: Mutex<HashMap<Uuid, TranscriptionJob>>,
}

/// Cloneable handle to durable, automatically processed audio.
#[derive(Clone)]
pub struct AudioIngress {
    inner: Arc<Inner>,
}

impl AudioIngress {
    /// Opens the owned persistence root and starts automatic processing.
    pub async fn open(
        persistence_root: impl AsRef<Path>,
        transcriber: AudioTranscriber,
    ) -> Result<Self, Error> {
        let root = persistence_root.as_ref().to_path_buf();
        ensure_private_directory(&root).map_err(Error::internal)?;
        ensure_private_directory(&root.join(ORIGINALS_DIRECTORY)).map_err(Error::internal)?;
        let connection = Connection::open(root.join(DATABASE_FILENAME)).map_err(Error::internal)?;
        connection
            .execute_batch(
                "PRAGMA foreign_keys=ON; PRAGMA journal_mode=WAL; PRAGMA busy_timeout=5000;",
            )
            .map_err(Error::internal)?;
        apply_migrations(&connection).map_err(Error::internal)?;
        recover_interrupted_attempts(&connection).map_err(Error::internal)?;

        let inner = Arc::new(Inner {
            root,
            db: Arc::new(Mutex::new(connection)),
            transcriber,
            jobs: Mutex::new(HashMap::new()),
        });
        tokio::spawn(worker_loop(Arc::downgrade(&inner)));
        Ok(Self { inner })
    }

    /// Durably accepts complete WAV bytes and schedules automatic processing.
    pub async fn submit(&self, input: AudioInput) -> Result<Submission, Error> {
        if input.bytes.is_empty() {
            return Err(Error::invalid("Audio bytes must not be empty."));
        }
        let size_bytes = i64::try_from(input.bytes.len())
            .map_err(|_| Error::invalid("Audio is too large for this platform."))?;
        let sha256 = format!("{:x}", Sha256::digest(&input.bytes));
        if let Some(id) = self.recording_id_by_sha(&sha256)? {
            return Ok(Submission {
                recording_id: id,
                deduplicated: true,
            });
        }

        let upload_id = Uuid::new_v4();
        let temporary = TemporaryUpload(self.inner.root.join(format!(".upload-{upload_id}.tmp")));
        let mut file = tokio::fs::OpenOptions::new()
            .write(true)
            .create_new(true)
            .open(&temporary.0)
            .await
            .map_err(Error::internal)?;
        set_private_file(&temporary.0).map_err(Error::internal)?;
        file.write_all(&input.bytes)
            .await
            .map_err(Error::internal)?;
        file.sync_all().await.map_err(Error::internal)?;
        drop(file);

        let relative_path = format!("{ORIGINALS_DIRECTORY}/{sha256}.wav");
        let final_path = self.inner.root.join(&relative_path);
        if final_path.exists() {
            tokio::fs::remove_file(&temporary.0)
                .await
                .map_err(Error::internal)?;
        } else {
            tokio::fs::rename(&temporary.0, &final_path)
                .await
                .map_err(Error::internal)?;
        }
        set_private_file(&final_path).map_err(Error::internal)?;
        sync_file(&final_path).map_err(Error::internal)?;
        sync_directory(final_path.parent().unwrap_or(&self.inner.root)).map_err(Error::internal)?;
        sync_directory(&self.inner.root).map_err(Error::internal)?;

        let id = Uuid::new_v4();
        let now = Utc::now().to_rfc3339();
        let filename = safe_filename(input.original_filename.as_deref());
        let insert = {
            let db = self.inner.db.lock().map_err(Error::internal)?;
            db.execute(
                "INSERT INTO audio_recordings(
                    id,sha256,original_filename,content_type,size_bytes,
                    source_created_at,received_at,updated_at,original_relative_path,
                    status,gemini_model,reconciliation_model,reconciliation_reasoning
                 ) VALUES(?1,?2,?3,'audio/wav',?4,?5,?6,?6,?7,'uploaded',?8,?9,?10)",
                params![
                    id.to_string(),
                    sha256,
                    filename,
                    size_bytes,
                    input.recorded_at.to_rfc3339(),
                    now,
                    relative_path,
                    TRANSCRIPTION_MODEL,
                    RECONCILIATION_MODEL,
                    RECONCILIATION_REASONING,
                ],
            )
        };
        if let Err(error) = insert {
            if let Some(existing) = self.recording_id_by_sha(&sha256)? {
                return Ok(Submission {
                    recording_id: existing,
                    deduplicated: true,
                });
            }
            return Err(Error::internal(error));
        }
        tracing::info!(recording_id=%id, %sha256, bytes=size_bytes, "Durably accepted audio");
        Ok(Submission {
            recording_id: id,
            deduplicated: false,
        })
    }

    /// Returns all current recording states in one snapshot.
    pub fn status(&self) -> Result<Status, Error> {
        let db = self.inner.db.lock().map_err(Error::internal)?;
        let mut statement = db
            .prepare(
                "SELECT id,sha256,original_filename,size_bytes,source_created_at,received_at,
                        status,gemini_model,reconciliation_model,reconciliation_reasoning,
                        attempt_count,last_error,failure_retryable,transcription_status_json,
                        final_transcript
                 FROM audio_recordings
                 ORDER BY datetime(source_created_at) DESC,datetime(received_at) DESC,id DESC",
            )
            .map_err(Error::internal)?;
        let recordings = statement
            .query_map([], row_recording_status)
            .map_err(Error::internal)?
            .collect::<Result<Vec<_>, _>>()
            .map_err(Error::internal)?;
        Ok(Status { recordings })
    }

    /// Gives a failed recording a fresh five-attempt processing budget.
    pub fn retry(&self, recording_id: Uuid) -> Result<(), Error> {
        let db = self.inner.db.lock().map_err(Error::internal)?;
        let changed = db
            .execute(
                "UPDATE audio_recordings
                 SET status='uploaded',attempt_count=0,next_attempt_at=NULL,last_error=NULL,
                     transcription_status_json=NULL,failure_retryable=1,updated_at=?1
                 WHERE id=?2 AND status='failed'",
                params![Utc::now().to_rfc3339(), recording_id.to_string()],
            )
            .map_err(Error::internal)?;
        if changed == 1 {
            return Ok(());
        }
        let exists = db
            .query_row(
                "SELECT 1 FROM audio_recordings WHERE id=?1",
                [recording_id.to_string()],
                |row| row.get::<_, i64>(0),
            )
            .optional()
            .map_err(Error::internal)?
            .is_some();
        if exists {
            Err(Error::conflict("Only a failed recording can be retried."))
        } else {
            Err(Error::not_found())
        }
    }

    fn recording_id_by_sha(&self, sha256: &str) -> Result<Option<Uuid>, Error> {
        let db = self.inner.db.lock().map_err(Error::internal)?;
        db.query_row(
            "SELECT id FROM audio_recordings WHERE sha256=?1",
            [sha256],
            |row| row.get::<_, String>(0),
        )
        .optional()
        .map_err(Error::internal)?
        .map(|id| Uuid::parse_str(&id).map_err(Error::internal))
        .transpose()
    }
}

fn row_recording_status(row: &rusqlite::Row<'_>) -> rusqlite::Result<RecordingStatus> {
    let id: String = row.get(0)?;
    let recorded_at: String = row.get(4)?;
    let received_at: String = row.get(5)?;
    let durable_status: String = row.get(6)?;
    let attempts: i64 = row.get(10)?;
    let last_error: Option<String> = row.get(11)?;
    let retryable: i64 = row.get(12)?;
    let progress_json: Option<String> = row.get(13)?;
    let transcript: Option<String> = row.get(14)?;
    let parse_time = |index, value: &str| {
        DateTime::parse_from_rfc3339(value)
            .map(|value| value.with_timezone(&Utc))
            .map_err(|error| {
                rusqlite::Error::FromSqlConversionFailure(
                    index,
                    rusqlite::types::Type::Text,
                    Box::new(error),
                )
            })
    };
    let state = match durable_status.as_str() {
        "uploaded" => RecordingState::Queued,
        "chunking" | "transcribing" | "reconciling" => {
            let progress = progress_json
                .as_deref()
                .map(serde_json::from_str)
                .transpose()
                .map_err(|error| {
                    rusqlite::Error::FromSqlConversionFailure(
                        13,
                        rusqlite::types::Type::Text,
                        Box::new(error),
                    )
                })?
                .unwrap_or_else(initial_progress);
            RecordingState::Processing {
                attempt: attempts.clamp(0, i64::from(u8::MAX)) as u8,
                progress: without_transcript(progress),
            }
        }
        "ready_for_ingress" | "ingressing" | "ingress_failed" | "complete" => {
            RecordingState::Complete {
                transcript: transcript.unwrap_or_default(),
            }
        }
        "failed" => RecordingState::Failed {
            attempts: attempts.clamp(0, i64::from(u8::MAX)) as u8,
            error: last_error.unwrap_or_else(|| "Audio processing failed.".into()),
            retryable: retryable != 0,
        },
        other => {
            return Err(rusqlite::Error::FromSqlConversionFailure(
                6,
                rusqlite::types::Type::Text,
                format!("unknown audio status {other:?}").into(),
            ));
        }
    };
    Ok(RecordingStatus {
        id: Uuid::parse_str(&id).map_err(|error| {
            rusqlite::Error::FromSqlConversionFailure(
                0,
                rusqlite::types::Type::Text,
                Box::new(error),
            )
        })?,
        sha256: row.get(1)?,
        original_filename: row.get(2)?,
        size_bytes: u64::try_from(row.get::<_, i64>(3)?).map_err(|error| {
            rusqlite::Error::FromSqlConversionFailure(
                3,
                rusqlite::types::Type::Integer,
                Box::new(error),
            )
        })?,
        recorded_at: parse_time(4, &recorded_at)?,
        received_at: parse_time(5, &received_at)?,
        transcription_model: row.get(7)?,
        reconciliation_model: row.get(8)?,
        reconciliation_reasoning: row.get(9)?,
        state,
    })
}

async fn worker_loop(inner: Weak<Inner>) {
    loop {
        let Some(inner) = inner.upgrade() else {
            return;
        };
        let worked = match process_next_recording(&inner).await {
            Ok(worked) => worked,
            Err(error) => {
                tracing::error!(error=%error, "AudioIngress worker iteration failed");
                false
            }
        };
        drop(inner);
        tokio::time::sleep(if worked {
            Duration::from_millis(100)
        } else {
            Duration::from_secs(5)
        })
        .await;
    }
}

#[derive(Debug)]
struct WorkRecording {
    id: Uuid,
    original_relative_path: String,
    attempt_count: i64,
}

async fn process_next_recording(inner: &Inner) -> anyhow::Result<bool> {
    let recording = {
        let db = inner
            .db
            .lock()
            .map_err(|_| anyhow::anyhow!("AudioIngress database lock was poisoned"))?;
        fetch_work_recording(&db)?
    };
    let Some(recording) = recording else {
        return Ok(false);
    };
    poll_transcription(inner, recording).await?;
    Ok(true)
}

fn fetch_work_recording(db: &Connection) -> anyhow::Result<Option<WorkRecording>> {
    db.query_row(
        "SELECT id,original_relative_path,attempt_count
         FROM audio_recordings
         WHERE status IN ('uploaded','chunking','transcribing','reconciling')
           AND (next_attempt_at IS NULL OR datetime(next_attempt_at)<=datetime('now'))
         ORDER BY datetime(received_at),id
         LIMIT 1",
        [],
        |row| {
            let id: String = row.get(0)?;
            Ok((id, row.get(1)?, row.get(2)?))
        },
    )
    .optional()?
    .map(|(id, original_relative_path, attempt_count)| {
        Ok(WorkRecording {
            id: Uuid::parse_str(&id)?,
            original_relative_path,
            attempt_count,
        })
    })
    .transpose()
}

async fn poll_transcription(inner: &Inner, mut recording: WorkRecording) -> anyhow::Result<()> {
    let existing = inner
        .jobs
        .lock()
        .map_err(|_| anyhow::anyhow!("audio transcription job lock was poisoned"))?
        .get(&recording.id)
        .cloned();
    let job = if let Some(job) = existing {
        job
    } else {
        if recording.attempt_count >= FAILURE_LIMIT {
            mark_failed(
                inner,
                recording.id,
                recording.attempt_count,
                true,
                "Audio transcription exhausted its five automatic attempts.",
                None,
            )?;
            return Ok(());
        }
        let source = inner.root.join(&recording.original_relative_path);
        let audio = tokio::fs::read(&source)
            .await
            .with_context(|| format!("reading retained audio {}", source.display()))?;
        recording.attempt_count += 1;
        {
            let db = inner
                .db
                .lock()
                .map_err(|_| anyhow::anyhow!("AudioIngress database lock was poisoned"))?;
            db.execute(
                "UPDATE audio_recordings
                 SET status='chunking',attempt_count=?1,next_attempt_at=NULL,last_error=NULL,
                     failure_retryable=1,updated_at=?2
                 WHERE id=?3",
                params![
                    recording.attempt_count,
                    Utc::now().to_rfc3339(),
                    recording.id.to_string()
                ],
            )?;
        }

        let cache_db = inner.db.clone();
        let cache_recording_id = recording.id;
        let piece_cache: PieceCache =
            Arc::new(move |plan| load_cached_transcript_piece(&cache_db, cache_recording_id, plan));

        let sink_db = inner.db.clone();
        let sink_recording_id = recording.id;
        let attempt_id = Uuid::new_v4().to_string();
        let piece_sink: PieceSink = Arc::new(move |piece| {
            persist_transcript_piece(&sink_db, sink_recording_id, &attempt_id, piece)
        });

        let job = inner
            .transcriber
            .transcribe_durably(audio, piece_cache, piece_sink);
        inner
            .jobs
            .lock()
            .map_err(|_| anyhow::anyhow!("audio transcription job lock was poisoned"))?
            .insert(recording.id, job.clone());
        job
    };

    let snapshot = job.status();
    persist_progress(inner, recording.id, &snapshot)?;
    match snapshot.state {
        JobState::Queued | JobState::Running => Ok(()),
        JobState::Completed => {
            remove_job(inner, recording.id)?;
            let transcript = snapshot
                .transcript
                .clone()
                .context("completed transcription omitted its transcript")?;
            ensure!(
                !transcript.trim().is_empty(),
                "completed transcription is empty"
            );
            let progress = serde_json::to_string(&without_transcript(snapshot))?;
            let db = inner
                .db
                .lock()
                .map_err(|_| anyhow::anyhow!("AudioIngress database lock was poisoned"))?;
            db.execute(
                "UPDATE audio_recordings
                 SET status='ready_for_ingress',final_transcript=?1,
                     transcription_status_json=?2,next_attempt_at=NULL,last_error=NULL,
                     failure_retryable=1,updated_at=?3
                 WHERE id=?4",
                params![
                    transcript.trim(),
                    progress,
                    Utc::now().to_rfc3339(),
                    recording.id.to_string()
                ],
            )?;
            tracing::info!(recording_id=%recording.id, "Audio transcript completed");
            Ok(())
        }
        JobState::Failed => {
            remove_job(inner, recording.id)?;
            let error = snapshot
                .steps
                .iter()
                .find(|step| step.state == StepState::Failed)
                .and_then(|step| step.error.as_ref());
            let message = error
                .map(|error| error.message.clone())
                .unwrap_or_else(|| "Audio transcription failed without detail.".into());
            let retryable = error.is_none_or(|error| error.retryable);
            record_attempt_failure(
                inner,
                recording.id,
                recording.attempt_count,
                retryable,
                &message,
                Some(snapshot),
            )
        }
    }
}

fn record_attempt_failure(
    inner: &Inner,
    id: Uuid,
    attempts: i64,
    retryable: bool,
    message: &str,
    progress: Option<TranscriptionStatus>,
) -> anyhow::Result<()> {
    if !retryable || attempts >= FAILURE_LIMIT {
        return mark_failed(inner, id, attempts, retryable, message, progress);
    }
    let db = inner
        .db
        .lock()
        .map_err(|_| anyhow::anyhow!("AudioIngress database lock was poisoned"))?;
    db.execute(
        "UPDATE audio_recordings
         SET status='uploaded',next_attempt_at=?1,last_error=?2,failure_retryable=1,
             transcription_status_json=?3,updated_at=?4
         WHERE id=?5",
        params![
            (Utc::now() + ChronoDuration::seconds(RETRY_DELAY_SECONDS)).to_rfc3339(),
            concise(message, 2_000),
            progress
                .map(without_transcript)
                .map(|progress| serde_json::to_string(&progress))
                .transpose()?,
            Utc::now().to_rfc3339(),
            id.to_string()
        ],
    )?;
    tracing::warn!(recording_id=%id, attempt=attempts, "Audio transcription will retry");
    Ok(())
}

fn mark_failed(
    inner: &Inner,
    id: Uuid,
    attempts: i64,
    retryable: bool,
    message: &str,
    progress: Option<TranscriptionStatus>,
) -> anyhow::Result<()> {
    let db = inner
        .db
        .lock()
        .map_err(|_| anyhow::anyhow!("AudioIngress database lock was poisoned"))?;
    db.execute(
        "UPDATE audio_recordings
         SET status='failed',attempt_count=?1,next_attempt_at=NULL,last_error=?2,
             failure_retryable=?3,transcription_status_json=?4,updated_at=?5
         WHERE id=?6",
        params![
            attempts,
            concise(message, 2_000),
            i64::from(retryable),
            progress
                .map(without_transcript)
                .map(|progress| serde_json::to_string(&progress))
                .transpose()?,
            Utc::now().to_rfc3339(),
            id.to_string()
        ],
    )?;
    tracing::error!(recording_id=%id, attempts, retryable, "Audio transcription stopped");
    Ok(())
}

fn persist_progress(inner: &Inner, id: Uuid, snapshot: &TranscriptionStatus) -> anyhow::Result<()> {
    let durable_status = transcription_stage(snapshot);
    let serialized = serde_json::to_string(&without_transcript(snapshot.clone()))?;
    let db = inner
        .db
        .lock()
        .map_err(|_| anyhow::anyhow!("AudioIngress database lock was poisoned"))?;
    db.execute(
        "UPDATE audio_recordings
         SET status=?1,transcription_status_json=?2,updated_at=?3
         WHERE id=?4 AND (status<>?1 OR COALESCE(transcription_status_json,'')<>?2)",
        params![
            durable_status,
            serialized,
            Utc::now().to_rfc3339(),
            id.to_string()
        ],
    )?;
    Ok(())
}

fn load_cached_transcript_piece(
    db: &Mutex<Connection>,
    recording_id: Uuid,
    plan: ChunkPlan,
) -> anyhow::Result<Option<String>> {
    let piece_index = i64::try_from(plan.index).context("piece index exceeds SQLite limits")?;
    let piece_count = i64::try_from(plan.total).context("piece count exceeds SQLite limits")?;
    let audio_start_ms =
        i64::try_from(plan.start_ms).context("piece start exceeds SQLite limits")?;
    let audio_end_ms = i64::try_from(plan.end_ms).context("piece end exceeds SQLite limits")?;
    let transcript = {
        let db = db
            .lock()
            .map_err(|_| anyhow::anyhow!("AudioIngress database lock was poisoned"))?;
        db.query_row(
            "SELECT transcript_json
             FROM audio_transcript_pieces
             WHERE recording_id=?1
               AND cache_revision=?2
               AND piece_index=?3
               AND piece_count=?4
               AND audio_start_ms=?5
               AND audio_end_ms=?6
             ORDER BY datetime(created_at) DESC,attempt_id DESC
             LIMIT 1",
            params![
                recording_id.to_string(),
                PIECE_CACHE_REVISION,
                piece_index,
                piece_count,
                audio_start_ms,
                audio_end_ms,
            ],
            |row| row.get::<_, String>(0),
        )
        .optional()?
    };
    if let Some(value) = transcript.as_deref() {
        let parsed: Value =
            serde_json::from_str(value).context("cached piece transcript is invalid JSON")?;
        ensure!(
            parsed.get("utterances").and_then(Value::as_array).is_some(),
            "cached piece transcript omitted the utterances array"
        );
    }
    Ok(transcript)
}

fn persist_transcript_piece(
    db: &Mutex<Connection>,
    recording_id: Uuid,
    attempt_id: &str,
    piece: &ChunkTranscript,
) -> anyhow::Result<()> {
    ensure!(!attempt_id.is_empty(), "piece attempt identity is empty");
    let piece_index =
        i64::try_from(piece.plan.index).context("piece index exceeds SQLite limits")?;
    let piece_count =
        i64::try_from(piece.plan.total).context("piece count exceeds SQLite limits")?;
    let audio_start_ms =
        i64::try_from(piece.plan.start_ms).context("piece start exceeds SQLite limits")?;
    let audio_end_ms =
        i64::try_from(piece.plan.end_ms).context("piece end exceeds SQLite limits")?;
    let db = db
        .lock()
        .map_err(|_| anyhow::anyhow!("AudioIngress database lock was poisoned"))?;
    db.execute(
        "INSERT INTO audio_transcript_pieces(
            recording_id,attempt_id,cache_revision,piece_index,piece_count,
            audio_start_ms,audio_end_ms,transcript_json,created_at
         ) VALUES(?1,?2,?3,?4,?5,?6,?7,?8,?9)
         ON CONFLICT(recording_id,attempt_id,piece_index) DO UPDATE SET
            cache_revision=excluded.cache_revision,
            piece_count=excluded.piece_count,
            audio_start_ms=excluded.audio_start_ms,
            audio_end_ms=excluded.audio_end_ms,
            transcript_json=excluded.transcript_json",
        params![
            recording_id.to_string(),
            attempt_id,
            PIECE_CACHE_REVISION,
            piece_index,
            piece_count,
            audio_start_ms,
            audio_end_ms,
            piece.transcript,
            Utc::now().to_rfc3339(),
        ],
    )?;
    Ok(())
}

fn remove_job(inner: &Inner, id: Uuid) -> anyhow::Result<()> {
    inner
        .jobs
        .lock()
        .map_err(|_| anyhow::anyhow!("audio transcription job lock was poisoned"))?
        .remove(&id);
    Ok(())
}

fn transcription_stage(snapshot: &TranscriptionStatus) -> &'static str {
    let plan_complete = snapshot
        .steps
        .iter()
        .any(|entry| entry.step == Step::PlanChunks && entry.state == StepState::Completed);
    if !plan_complete {
        return "chunking";
    }
    let chunks_complete = snapshot
        .steps
        .iter()
        .filter(|entry| matches!(entry.step, Step::TranscribeChunk { .. }))
        .all(|entry| entry.state == StepState::Completed);
    if chunks_complete {
        "reconciling"
    } else {
        "transcribing"
    }
}

fn without_transcript(mut status: TranscriptionStatus) -> TranscriptionStatus {
    status.transcript = None;
    status
}

fn initial_progress() -> TranscriptionStatus {
    TranscriptionStatus {
        state: JobState::Queued,
        steps: Vec::new(),
        transcript: None,
    }
}

fn recover_interrupted_attempts(connection: &Connection) -> rusqlite::Result<()> {
    connection.execute(
        "UPDATE audio_recordings
         SET status=CASE WHEN attempt_count>=?1 THEN 'failed' ELSE 'uploaded' END,
             next_attempt_at=NULL,
             last_error=CASE WHEN attempt_count>=?1
                 THEN 'Audio transcription stopped after its fifth attempt was interrupted.'
                 ELSE 'Audio transcription was interrupted and will restart automatically.'
             END,
             failure_retryable=1,
             updated_at=?2
         WHERE status IN ('chunking','transcribing','reconciling')",
        params![FAILURE_LIMIT, Utc::now().to_rfc3339()],
    )?;
    Ok(())
}

fn apply_migrations(connection: &Connection) -> anyhow::Result<()> {
    let version: i64 = connection.query_row("PRAGMA user_version", [], |row| row.get(0))?;
    ensure!(
        version <= LATEST_SCHEMA_VERSION,
        "audio-ingress database schema version {version} is newer than supported version {LATEST_SCHEMA_VERSION}"
    );
    if version == 0 {
        let has_recordings = connection.query_row(
            "SELECT EXISTS(
                SELECT 1 FROM sqlite_schema
                WHERE type='table' AND name='audio_recordings'
             )",
            [],
            |row| row.get::<_, i64>(0),
        )? == 1;
        if !has_recordings {
            connection.execute_batch(FRESH_SCHEMA)?;
            return Ok(());
        }
    }
    if version < 1 {
        connection.execute_batch(INITIAL_MIGRATION)?;
    }
    if version < 2 {
        connection.execute_batch(RELEASE_DEFERRED_INGRESS_MIGRATION)?;
    }
    if version < 3 {
        connection.execute_batch(TRANSCRIPTION_STATUS_MIGRATION)?;
    }
    if version < 4 {
        connection.execute_batch(RETRY_ROUNDED_WAV_INTERVALS_MIGRATION)?;
    }
    if version < 5 {
        connection.execute_batch(UNIFIED_INGRESS_QUEUE_MIGRATION)?;
    }
    if version < 6 {
        connection.execute_batch(STANDALONE_LIBRARY_MIGRATION)?;
    }
    if version < 7 {
        connection.execute_batch(DURABLE_TRANSCRIPT_PIECES_MIGRATION)?;
    }
    if version < 8 {
        connection.execute_batch(UNIQUE_TRANSCRIPTION_ATTEMPTS_MIGRATION)?;
    }
    Ok(())
}

fn ensure_private_directory(path: &Path) -> anyhow::Result<()> {
    fs::create_dir_all(path).with_context(|| format!("creating {}", path.display()))?;
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        fs::set_permissions(path, fs::Permissions::from_mode(0o700))
            .with_context(|| format!("setting private permissions on {}", path.display()))?;
    }
    Ok(())
}

fn set_private_file(path: &Path) -> anyhow::Result<()> {
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        fs::set_permissions(path, fs::Permissions::from_mode(0o600))
            .with_context(|| format!("setting private permissions on {}", path.display()))?;
    }
    Ok(())
}

fn sync_file(path: &Path) -> anyhow::Result<()> {
    fs::OpenOptions::new()
        .read(true)
        .write(true)
        .open(path)
        .with_context(|| format!("opening {} for sync", path.display()))?
        .sync_all()
        .with_context(|| format!("syncing {}", path.display()))
}

fn sync_directory(path: &Path) -> anyhow::Result<()> {
    #[cfg(unix)]
    fs::File::open(path)
        .with_context(|| format!("opening directory {} for sync", path.display()))?
        .sync_all()
        .with_context(|| format!("syncing directory {}", path.display()))?;
    Ok(())
}

fn safe_filename(value: Option<&str>) -> String {
    let name = value
        .and_then(|value| Path::new(value).file_name())
        .and_then(|value| value.to_str())
        .unwrap_or("audio.wav");
    let clean = name
        .chars()
        .map(|character| {
            if character.is_ascii_alphanumeric() || matches!(character, '.' | '-' | '_') {
                character
            } else {
                '_'
            }
        })
        .take(200)
        .collect::<String>();
    if clean.is_empty() {
        "audio.wav".into()
    } else {
        clean
    }
}

fn concise(value: &str, limit: usize) -> String {
    let normalized = value.split_whitespace().collect::<Vec<_>>().join(" ");
    let bounded = normalized.chars().take(limit).collect::<String>();
    if bounded.is_empty() {
        "Audio transcription failed without an error message.".into()
    } else {
        bounded
    }
}

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

    fn database() -> Connection {
        let connection = Connection::open_in_memory().unwrap();
        apply_migrations(&connection).unwrap();
        connection
    }

    fn insert_recording(connection: &Connection, id: Uuid, status: &str) {
        connection
            .execute(
                "INSERT INTO audio_recordings(
                id,sha256,original_filename,content_type,size_bytes,source_created_at,
                received_at,updated_at,original_relative_path,status,gemini_model,
                reconciliation_model,reconciliation_reasoning
             ) VALUES(?1,?2,'note.wav','audio/wav',4,?3,?3,?3,?4,?5,?6,?7,?8)",
                params![
                    id.to_string(),
                    format!("{:064x}", 1),
                    "2026-01-01T00:00:00Z",
                    format!("originals/{id}.wav"),
                    status,
                    TRANSCRIPTION_MODEL,
                    RECONCILIATION_MODEL,
                    RECONCILIATION_REASONING,
                ],
            )
            .unwrap();
    }

    fn sample_piece() -> ChunkTranscript {
        ChunkTranscript {
            plan: ChunkPlan {
                index: 0,
                total: 2,
                start_ms: 0,
                end_ms: 120_000,
            },
            transcript: r#"{"utterances":[{"original_text":"hello"}]}"#.into(),
        }
    }

    #[test]
    fn fresh_schema_contains_recordings_and_durable_pieces() {
        let connection = database();
        let tables = connection
            .prepare(
                "SELECT name FROM sqlite_schema
                 WHERE type='table' AND name NOT LIKE 'sqlite_%'
                 ORDER BY name",
            )
            .unwrap()
            .query_map([], |row| row.get::<_, String>(0))
            .unwrap()
            .collect::<Result<Vec<_>, _>>()
            .unwrap();
        let version: i64 = connection
            .query_row("PRAGMA user_version", [], |row| row.get(0))
            .unwrap();
        assert_eq!(tables, vec!["audio_recordings", "audio_transcript_pieces"]);
        assert_eq!(version, LATEST_SCHEMA_VERSION);
    }

    #[test]
    fn version_five_databases_upgrade_without_losing_legacy_queue_data() {
        let connection = Connection::open_in_memory().unwrap();
        connection.execute_batch(INITIAL_MIGRATION).unwrap();
        connection
            .execute_batch(RELEASE_DEFERRED_INGRESS_MIGRATION)
            .unwrap();
        connection
            .execute_batch(TRANSCRIPTION_STATUS_MIGRATION)
            .unwrap();
        connection
            .execute_batch(RETRY_ROUNDED_WAV_INTERVALS_MIGRATION)
            .unwrap();
        connection
            .execute_batch(UNIFIED_INGRESS_QUEUE_MIGRATION)
            .unwrap();
        let id = Uuid::new_v4();
        insert_recording(&connection, id, "failed");

        apply_migrations(&connection).unwrap();

        let version: i64 = connection
            .query_row("PRAGMA user_version", [], |row| row.get(0))
            .unwrap();
        let retryable: i64 = connection
            .query_row(
                "SELECT failure_retryable FROM audio_recordings WHERE id=?1",
                [id.to_string()],
                |row| row.get(0),
            )
            .unwrap();
        let legacy_queue_exists: i64 = connection
            .query_row(
                "SELECT EXISTS(
                    SELECT 1 FROM sqlite_schema
                    WHERE type='table' AND name='audio_ingress_pieces'
                 )",
                [],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(version, LATEST_SCHEMA_VERSION);
        assert_eq!(retryable, 1);
        assert_eq!(legacy_queue_exists, 1);
    }

    #[test]
    fn version_seven_piece_rows_migrate_without_loss() {
        let connection = Connection::open_in_memory().unwrap();
        connection.execute_batch(INITIAL_MIGRATION).unwrap();
        connection
            .execute_batch(RELEASE_DEFERRED_INGRESS_MIGRATION)
            .unwrap();
        connection
            .execute_batch(TRANSCRIPTION_STATUS_MIGRATION)
            .unwrap();
        connection
            .execute_batch(RETRY_ROUNDED_WAV_INTERVALS_MIGRATION)
            .unwrap();
        connection
            .execute_batch(UNIFIED_INGRESS_QUEUE_MIGRATION)
            .unwrap();
        connection
            .execute_batch(STANDALONE_LIBRARY_MIGRATION)
            .unwrap();
        connection
            .execute_batch(DURABLE_TRANSCRIPT_PIECES_MIGRATION)
            .unwrap();
        let id = Uuid::new_v4();
        insert_recording(&connection, id, "transcribing");
        connection
            .execute(
                "INSERT INTO audio_transcript_pieces(
                    recording_id,attempt,piece_index,piece_count,audio_start_ms,
                    audio_end_ms,transcript_json,created_at
                 ) VALUES(?1,1,0,2,0,120000,?2,?3)",
                params![
                    id.to_string(),
                    r#"{"utterances":[{"original_text":"hello"}]}"#,
                    "2026-01-01T00:00:00Z",
                ],
            )
            .unwrap();

        apply_migrations(&connection).unwrap();

        let migrated: (String, String, String) = connection
            .query_row(
                "SELECT attempt_id,cache_revision,transcript_json
                 FROM audio_transcript_pieces
                 WHERE recording_id=?1 AND piece_index=0",
                [id.to_string()],
                |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
            )
            .unwrap();
        assert_eq!(migrated.0, "version-7-attempt-1");
        assert_eq!(migrated.1, PIECE_CACHE_REVISION);
        assert!(migrated.2.contains("hello"));
    }

    #[test]
    fn status_exposes_one_complete_transcript() {
        let connection = database();
        let id = Uuid::new_v4();
        insert_recording(&connection, id, "ready_for_ingress");
        connection
            .execute(
                "UPDATE audio_recordings SET final_transcript='hello' WHERE id=?1",
                [id.to_string()],
            )
            .unwrap();
        let status = connection
            .query_row(
                "SELECT id,sha256,original_filename,size_bytes,source_created_at,received_at,
                        status,gemini_model,reconciliation_model,reconciliation_reasoning,
                        attempt_count,last_error,failure_retryable,transcription_status_json,
                        final_transcript
                 FROM audio_recordings WHERE id=?1",
                [id.to_string()],
                row_recording_status,
            )
            .unwrap();
        assert!(matches!(
            status.state,
            RecordingState::Complete { ref transcript } if transcript == "hello"
        ));
    }

    #[test]
    fn transcript_pieces_are_durable_and_attempt_scoped() {
        let connection = database();
        let id = Uuid::new_v4();
        insert_recording(&connection, id, "transcribing");
        let db = Mutex::new(connection);
        let piece = sample_piece();
        persist_transcript_piece(&db, id, "attempt-a", &piece).unwrap();
        persist_transcript_piece(&db, id, "attempt-b", &piece).unwrap();

        let db = db.lock().unwrap();
        let rows: i64 = db
            .query_row(
                "SELECT COUNT(*) FROM audio_transcript_pieces WHERE recording_id=?1",
                [id.to_string()],
                |row| row.get(0),
            )
            .unwrap();
        let stored: String = db
            .query_row(
                "SELECT transcript_json FROM audio_transcript_pieces
                 WHERE recording_id=?1 AND attempt_id='attempt-a' AND piece_index=0",
                [id.to_string()],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(rows, 2);
        assert_eq!(stored, piece.transcript);
    }

    #[test]
    fn cache_reuses_only_exact_matching_piece_plans() {
        let connection = database();
        let id = Uuid::new_v4();
        insert_recording(&connection, id, "transcribing");
        let db = Mutex::new(connection);
        let piece = sample_piece();
        persist_transcript_piece(&db, id, "attempt-a", &piece).unwrap();

        let exact = load_cached_transcript_piece(&db, id, piece.plan).unwrap();
        let changed = load_cached_transcript_piece(
            &db,
            id,
            ChunkPlan {
                end_ms: piece.plan.end_ms + 1,
                ..piece.plan
            },
        )
        .unwrap();

        assert_eq!(exact.as_deref(), Some(piece.transcript.as_str()));
        assert_eq!(changed, None);
    }

    #[test]
    fn future_schema_versions_are_rejected() {
        let connection = Connection::open_in_memory().unwrap();
        connection
            .execute_batch("PRAGMA user_version = 9;")
            .unwrap();
        let error = apply_migrations(&connection).unwrap_err().to_string();
        assert!(error.contains("newer than supported"));
    }

    #[test]
    fn interrupted_attempts_consume_the_fixed_budget() {
        let connection = database();
        let id = Uuid::new_v4();
        insert_recording(&connection, id, "transcribing");
        connection
            .execute(
                "UPDATE audio_recordings SET attempt_count=5 WHERE id=?1",
                [id.to_string()],
            )
            .unwrap();
        recover_interrupted_attempts(&connection).unwrap();
        let state: String = connection
            .query_row(
                "SELECT status FROM audio_recordings WHERE id=?1",
                [id.to_string()],
                |row| row.get(0),
            )
            .unwrap();
        assert_eq!(state, "failed");
    }

    #[test]
    fn filenames_cannot_escape_the_persistence_root() {
        assert_eq!(safe_filename(Some("../../secret.wav")), "secret.wav");
        assert_eq!(safe_filename(Some("meeting note.wav")), "meeting_note.wav");
        assert_eq!(safe_filename(None), "audio.wav");
    }
}