use anyhow::{Context, ensure};
use chrono::{DateTime, Utc};
use rusqlite::{Connection, OptionalExtension, params};
use serde_json::Value;
use uuid::Uuid;
use crate::LegacyReviewDisposition;
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum Outcome {
Applied,
Unchanged,
Missing,
Ineligible,
}
pub(crate) fn resolve(
connection: &mut Connection,
recording_id: Uuid,
disposition: LegacyReviewDisposition,
now: DateTime<Utc>,
) -> anyhow::Result<Outcome> {
let transaction = connection
.transaction()
.context("starting legacy speaker-review resolution")?;
let stored = transaction
.query_row(
"SELECT status,final_transcript,correction_packet_json
FROM audio_recordings WHERE id=?1",
[recording_id.to_string()],
|row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, Option<String>>(1)?,
row.get::<_, Option<String>>(2)?,
))
},
)
.optional()
.context("reading legacy speaker-review state")?;
let Some((status, transcript, packet)) = stored else {
return Ok(Outcome::Missing);
};
if already_resolved(
&transaction,
recording_id,
&status,
&transcript,
&packet,
disposition,
)? {
return Ok(Outcome::Unchanged);
}
let (Some(transcript), Some(packet)) = (transcript, packet) else {
return Ok(Outcome::Ineligible);
};
let confirmation_state = confirmation_state(&packet);
if status != "ready_for_ingress"
|| transcript.trim().is_empty()
|| !matches!(
confirmation_state.as_deref(),
Some("unconfirmed" | "automatically_trained")
)
{
return Ok(Outcome::Ineligible);
}
match disposition {
LegacyReviewDisposition::Reprocess => {
transaction
.execute(
"INSERT INTO audio_legacy_review_archive(
recording_id,final_transcript,correction_packet_json,archived_at
) VALUES(?1,?2,?3,?4)
ON CONFLICT(recording_id) DO NOTHING",
params![
recording_id.to_string(),
transcript,
packet,
now.to_rfc3339()
],
)
.context("archiving legacy transcript and speaker packet")?;
let archived_matches = transaction
.query_row(
"SELECT final_transcript=?1 AND correction_packet_json=?2
FROM audio_legacy_review_archive WHERE recording_id=?3",
params![transcript, packet, recording_id.to_string()],
|row| row.get::<_, bool>(0),
)
.context("verifying archived legacy speaker-review state")?;
ensure!(
archived_matches,
"legacy speaker-review archive conflicts with the active recording"
);
transaction
.execute(
"UPDATE audio_recordings
SET status='uploaded',final_transcript=NULL,correction_packet_json=NULL,
attempt_count=0,next_attempt_at=NULL,last_error=NULL,
transcription_status_json=NULL,failure_retryable=1,updated_at=?1
WHERE id=?2",
params![now.to_rfc3339(), recording_id.to_string()],
)
.context("queueing legacy recording for current analysis")?;
}
LegacyReviewDisposition::Complete => {
transaction
.execute(
"UPDATE audio_recordings
SET status='complete',next_attempt_at=NULL,last_error=NULL,
failure_retryable=1,updated_at=?1
WHERE id=?2",
params![now.to_rfc3339(), recording_id.to_string()],
)
.context("marking ingressed legacy recording complete")?;
}
}
transaction
.commit()
.context("committing legacy speaker-review resolution")?;
Ok(Outcome::Applied)
}
fn already_resolved(
connection: &Connection,
recording_id: Uuid,
status: &str,
transcript: &Option<String>,
packet: &Option<String>,
disposition: LegacyReviewDisposition,
) -> anyhow::Result<bool> {
match disposition {
LegacyReviewDisposition::Complete => {
Ok(status == "complete" && transcript.is_some() && packet.is_some())
}
LegacyReviewDisposition::Reprocess => {
if status != "uploaded" || transcript.is_some() || packet.is_some() {
return Ok(false);
}
connection
.query_row(
"SELECT 1 FROM audio_legacy_review_archive WHERE recording_id=?1",
[recording_id.to_string()],
|_| Ok(true),
)
.optional()
.map(|found| found.unwrap_or(false))
.context("checking prior legacy speaker-review reset")
}
}
}
fn confirmation_state(packet: &str) -> Option<String> {
serde_json::from_str::<Value>(packet)
.ok()?
.get("confirmation_state")?
.as_str()
.map(str::to_owned)
}
#[cfg(test)]
mod tests {
use chrono::{TimeZone, Utc};
use rusqlite::Connection;
use uuid::Uuid;
use super::{Outcome, resolve};
use crate::{LegacyReviewDisposition, apply_migrations};
fn database() -> Connection {
let connection = Connection::open_in_memory().unwrap();
connection
.execute_batch(
"CREATE TABLE audio_recordings (
id TEXT PRIMARY KEY,
status TEXT NOT NULL,
final_transcript TEXT,
correction_packet_json TEXT,
attempt_count INTEGER NOT NULL,
next_attempt_at TEXT,
last_error TEXT,
transcription_status_json TEXT,
failure_retryable INTEGER NOT NULL,
updated_at TEXT NOT NULL
);
CREATE TABLE audio_legacy_review_archive (
recording_id TEXT PRIMARY KEY,
final_transcript TEXT NOT NULL,
correction_packet_json TEXT NOT NULL,
archived_at TEXT NOT NULL
);",
)
.unwrap();
connection
}
fn legacy_recording(connection: &Connection, id: Uuid) {
connection
.execute(
"INSERT INTO audio_recordings VALUES(
?1,'ready_for_ingress','Old transcript',
'{\"confirmation_state\":\"unconfirmed\"}',3,'later','old error','{}',0,'old'
)",
[id.to_string()],
)
.unwrap();
}
#[test]
fn reprocess_archives_exact_state_before_queueing_current_analysis() {
let mut connection = database();
let id = Uuid::new_v4();
legacy_recording(&connection, id);
let now = Utc.with_ymd_and_hms(2026, 8, 5, 4, 0, 0).unwrap();
assert_eq!(
resolve(&mut connection, id, LegacyReviewDisposition::Reprocess, now).unwrap(),
Outcome::Applied
);
let active = connection
.query_row(
"SELECT status,final_transcript,correction_packet_json,attempt_count,
transcription_status_json,last_error
FROM audio_recordings WHERE id=?1",
[id.to_string()],
|row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, Option<String>>(1)?,
row.get::<_, Option<String>>(2)?,
row.get::<_, i64>(3)?,
row.get::<_, Option<String>>(4)?,
row.get::<_, Option<String>>(5)?,
))
},
)
.unwrap();
assert_eq!(active, ("uploaded".into(), None, None, 0, None, None));
let archived = connection
.query_row(
"SELECT final_transcript,correction_packet_json
FROM audio_legacy_review_archive WHERE recording_id=?1",
[id.to_string()],
|row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)),
)
.unwrap();
assert_eq!(archived.0, "Old transcript");
assert_eq!(archived.1, "{\"confirmation_state\":\"unconfirmed\"}");
assert_eq!(
resolve(&mut connection, id, LegacyReviewDisposition::Reprocess, now).unwrap(),
Outcome::Unchanged
);
}
#[test]
fn ingressed_resolution_preserves_payload_and_marks_complete() {
let mut connection = database();
let id = Uuid::new_v4();
legacy_recording(&connection, id);
assert_eq!(
resolve(
&mut connection,
id,
LegacyReviewDisposition::Complete,
Utc::now()
)
.unwrap(),
Outcome::Applied
);
let stored = connection
.query_row(
"SELECT status,final_transcript,correction_packet_json
FROM audio_recordings WHERE id=?1",
[id.to_string()],
|row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
))
},
)
.unwrap();
assert_eq!(stored.0, "complete");
assert_eq!(stored.1, "Old transcript");
assert_eq!(stored.2, "{\"confirmation_state\":\"unconfirmed\"}");
assert_eq!(
resolve(
&mut connection,
id,
LegacyReviewDisposition::Complete,
Utc::now()
)
.unwrap(),
Outcome::Unchanged
);
}
#[test]
fn durable_complete_status_suppresses_the_obsolete_packet() {
let connection = Connection::open_in_memory().unwrap();
let id = Uuid::new_v4();
let timestamp = "2026-08-05T04:00:00+00:00";
let recording = connection
.query_row(
"SELECT ?1,'user',?2,'legacy.wav',42,?3,?3,'complete',
'gemini','gpt','xhigh',0,NULL,1,NULL,'Legacy transcript',
'{\"confirmation_state\":\"unconfirmed\"}'",
rusqlite::params![id.to_string(), "0".repeat(64), timestamp],
crate::row_recording_status,
)
.unwrap();
assert!(matches!(
recording.state,
crate::RecordingState::Complete { .. }
));
assert!(recording.correction_packet.is_none());
}
#[test]
fn version_ten_database_adds_the_offline_recovery_archive() {
let connection = Connection::open_in_memory().unwrap();
connection
.execute_batch(
"PRAGMA foreign_keys=ON;
CREATE TABLE audio_recordings(id TEXT PRIMARY KEY);
PRAGMA user_version=10;",
)
.unwrap();
apply_migrations(&connection).unwrap();
assert_eq!(
connection
.query_row("PRAGMA user_version", [], |row| row.get::<_, i64>(0))
.unwrap(),
11
);
assert_eq!(
connection
.query_row(
"SELECT count(*) FROM sqlite_schema
WHERE type='table' AND name='audio_legacy_review_archive'",
[],
|row| row.get::<_, i64>(0),
)
.unwrap(),
1
);
}
}