storage: serialize validation in write transactions

crates/consolebook-server/src/assignments.rs

@@ -16,6 +16,7 @@ use crate::audit::{self, EventKind, Subject};
 use crate::capabilities::{self, Capability};
 use crate::lifecycle::{self, EnrollmentStatus};
 use crate::notices::{self, NoticeKind};
+use crate::storage;
 
 /// One assignment on an enrollment, with the trainer resolved.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize)]
@@ -101,20 +102,22 @@ pub async fn create(
     if !capabilities::user_has(pool, actor_user_id, Capability::AssignTraining).await? {
         return Ok(Err(AssignRefusal::CapabilityRequired));
     }
-    let mut tx = pool.begin().await.context("starting assignment")?;
+    let mut tx = storage::write_tx(pool)
+        .await
+        .context("starting assignment")?;
     let Some(status) = lifecycle::status(&mut tx, enrollment_id).await? else {
-        return Ok(Err(AssignRefusal::NoSuchEnrollment));
+        return storage::refuse(tx, AssignRefusal::NoSuchEnrollment).await;
     };
     if status != EnrollmentStatus::Active {
-        return Ok(Err(AssignRefusal::EnrollmentInactive));
+        return storage::refuse(tx, AssignRefusal::EnrollmentInactive).await;
     }
     let trainer_exists: Option<i64> = sqlx::query_scalar("SELECT 1 FROM user WHERE id = ?1")
         .bind(trainer_user_id)
         .fetch_optional(&mut *tx)
         .await
         .context("checking trainer")?;
     if trainer_exists.is_none() {
-        return Ok(Err(AssignRefusal::NoSuchUser));
+        return storage::refuse(tx, AssignRefusal::NoSuchUser).await;
     }
     let can_view: Option<i64> =
         sqlx::query_scalar("SELECT 1 FROM capability_grant WHERE user_id = ?1 AND capability = ?2")
@@ -124,7 +127,7 @@ pub async fn create(
             .await
             .context("checking trainer capability")?;
     if can_view.is_none() {
-        return Ok(Err(AssignRefusal::TrainerLacksCapability));
+        return storage::refuse(tx, AssignRefusal::TrainerLacksCapability).await;
     }
     let duplicate: Option<i64> = sqlx::query_scalar(
         "SELECT 1 FROM training_assignment
@@ -136,7 +139,7 @@ pub async fn create(
     .await
     .context("checking duplicate assignment")?;
     if duplicate.is_some() {
-        return Ok(Err(AssignRefusal::AlreadyAssigned));
+        return storage::refuse(tx, AssignRefusal::AlreadyAssigned).await;
     }
 
     let result = sqlx::query(
@@ -195,16 +198,18 @@ pub async fn end(
     if !capabilities::user_has(pool, actor_user_id, Capability::AssignTraining).await? {
         return Ok(Err(AssignRefusal::CapabilityRequired));
     }
-    let mut tx = pool.begin().await.context("starting assignment end")?;
+    let mut tx = storage::write_tx(pool)
+        .await
+        .context("starting assignment end")?;
     let ended_at: Option<Option<i64>> =
         sqlx::query_scalar("SELECT ended_at FROM training_assignment WHERE id = ?1")
             .bind(assignment_id)
             .fetch_optional(&mut *tx)
             .await
             .context("checking assignment")?;
     match ended_at {
-        None => return Ok(Err(AssignRefusal::NoSuchAssignment)),
-        Some(Some(_)) => return Ok(Err(AssignRefusal::AlreadyEnded)),
+        None => return storage::refuse(tx, AssignRefusal::NoSuchAssignment).await,
+        Some(Some(_)) => return storage::refuse(tx, AssignRefusal::AlreadyEnded).await,
         Some(None) => {}
     }
     sqlx::query("UPDATE training_assignment SET ended_at = ?1, ended_by = ?2 WHERE id = ?3")

crates/consolebook-server/src/enrollments.rs

@@ -13,6 +13,7 @@ use time::OffsetDateTime;
 
 use crate::audit::{self, EventKind, Subject};
 use crate::capabilities::{self, Capability, TRAINEE_BUNDLE};
+use crate::storage;
 
 /// One enrollee of a program version, with presentation fields resolved.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize)]
@@ -50,18 +51,20 @@ pub async fn enroll(
     if !holds_assign_training(pool, actor_user_id).await? {
         return Ok(Err(EnrollRefusal::CapabilityRequired));
     }
-    let mut tx = pool.begin().await.context("starting enrollment")?;
+    let mut tx = storage::write_tx(pool)
+        .await
+        .context("starting enrollment")?;
     let version = sqlx::query("SELECT published_at FROM program_version WHERE id = ?1")
         .bind(version_id)
         .fetch_optional(&mut *tx)
         .await
         .context("checking version")?;
     match version {
-        None => return Ok(Err(EnrollRefusal::NoSuchVersion)),
+        None => return storage::refuse(tx, EnrollRefusal::NoSuchVersion).await,
         Some(row) => {
             let published_at: Option<i64> = row.get("published_at");
             if published_at.is_none() {
-                return Ok(Err(EnrollRefusal::NotPublished));
+                return storage::refuse(tx, EnrollRefusal::NotPublished).await;
             }
         }
     }
@@ -71,7 +74,7 @@ pub async fn enroll(
         .await
         .context("checking user")?;
     if user_exists.is_none() {
-        return Ok(Err(EnrollRefusal::NoSuchUser));
+        return storage::refuse(tx, EnrollRefusal::NoSuchUser).await;
     }
     let duplicate: Option<i64> = sqlx::query_scalar(
         "SELECT 1 FROM enrollment WHERE user_id = ?1 AND program_version_id = ?2",
@@ -82,7 +85,7 @@ pub async fn enroll(
     .await
     .context("checking enrollment")?;
     if duplicate.is_some() {
-        return Ok(Err(EnrollRefusal::AlreadyEnrolled));
+        return storage::refuse(tx, EnrollRefusal::AlreadyEnrolled).await;
     }
     let now = OffsetDateTime::now_utc().unix_timestamp();
     let result = sqlx::query(

crates/consolebook-server/src/lifecycle.rs

@@ -24,6 +24,7 @@ use time::OffsetDateTime;
 use crate::assignments;
 use crate::audit::{self, EventKind, Subject};
 use crate::capabilities::{self, Capability};
+use crate::storage;
 
 /// Enrollment status, derived from the event stream and never stored
 /// beside it.
@@ -347,19 +348,21 @@ pub async fn record_enrollment_event(
         return Ok(Err(LifecycleRefusal::ReasonRequired));
     }
 
-    let mut tx = pool.begin().await.context("starting lifecycle event")?;
+    let mut tx = storage::write_tx(pool)
+        .await
+        .context("starting lifecycle event")?;
     let Some(current_status) = status(&mut tx, enrollment_id).await? else {
-        return Ok(Err(LifecycleRefusal::NoSuchEnrollment));
+        return storage::refuse(tx, LifecycleRefusal::NoSuchEnrollment).await;
     };
     match kind {
         EnrollmentEventKind::Reinstate => {
             if current_status == EnrollmentStatus::Active {
-                return Ok(Err(LifecycleRefusal::AlreadyActive));
+                return storage::refuse(tx, LifecycleRefusal::AlreadyActive).await;
             }
         }
         _ => {
             if current_status != EnrollmentStatus::Active {
-                return Ok(Err(LifecycleRefusal::NotActive));
+                return storage::refuse(tx, LifecycleRefusal::NotActive).await;
             }
         }
     }
@@ -380,12 +383,12 @@ pub async fn record_enrollment_event(
                     .await
                     .context("reading enrollment pin")?;
             let Some(to) = to_version_id else {
-                return Ok(Err(LifecycleRefusal::NoSuchVersion));
+                return storage::refuse(tx, LifecycleRefusal::NoSuchVersion).await;
             };
             if let Some(refusal) =
                 version_change_refusal(&mut tx, enrollment_id, trainee, from, to).await?
             {
-                return Ok(Err(refusal));
+                return storage::refuse(tx, refusal).await;
             }
             (Some(from), Some(to), EventKind::EnrollmentVersionChanged)
         }
@@ -464,12 +467,14 @@ pub async fn record_phase_event(
         return Ok(Err(LifecycleRefusal::EffectiveInFuture));
     }
 
-    let mut tx = pool.begin().await.context("starting phase event")?;
+    let mut tx = storage::write_tx(pool)
+        .await
+        .context("starting phase event")?;
     let Some(current_status) = status(&mut tx, enrollment_id).await? else {
-        return Ok(Err(LifecycleRefusal::NoSuchEnrollment));
+        return storage::refuse(tx, LifecycleRefusal::NoSuchEnrollment).await;
     };
     if current_status != EnrollmentStatus::Active {
-        return Ok(Err(LifecycleRefusal::NotActive));
+        return storage::refuse(tx, LifecycleRefusal::NotActive).await;
     }
     let latest_effective: Option<i64> =
         sqlx::query_scalar("SELECT MAX(effective_at) FROM phase_event WHERE enrollment_id = ?1")
@@ -478,7 +483,7 @@ pub async fn record_phase_event(
             .await
             .context("reading latest effective instant")?;
     if latest_effective.is_some_and(|latest| effective < latest) {
-        return Ok(Err(LifecycleRefusal::OutOfOrder));
+        return storage::refuse(tx, LifecycleRefusal::OutOfOrder).await;
     }
     // The version-change event that opened the current epoch is recorded
     // history too: a phase event cannot take effect before its epoch
@@ -494,7 +499,7 @@ pub async fn record_phase_event(
     .await
     .context("reading epoch boundary")?;
     if epoch_opened.is_some_and(|opened| effective < opened) {
-        return Ok(Err(LifecycleRefusal::OutOfOrder));
+        return storage::refuse(tx, LifecycleRefusal::OutOfOrder).await;
     }
 
     let pinned: i64 = sqlx::query_scalar("SELECT program_version_id FROM enrollment WHERE id = ?1")
@@ -508,10 +513,10 @@ pub async fn record_phase_event(
     let (from_phase, to_phase) = match kind {
         PhaseEventKind::Advance | PhaseEventKind::Return | PhaseEventKind::Restart => {
             if paused {
-                return Ok(Err(LifecycleRefusal::Paused));
+                return storage::refuse(tx, LifecycleRefusal::Paused).await;
             }
             let Some(to) = to_phase_id else {
-                return Ok(Err(LifecycleRefusal::NoSuchPhase));
+                return storage::refuse(tx, LifecycleRefusal::NoSuchPhase).await;
             };
             let target_in_version: Option<i64> =
                 sqlx::query_scalar("SELECT 1 FROM phase WHERE id = ?1 AND program_version_id = ?2")
@@ -521,14 +526,14 @@ pub async fn record_phase_event(
                     .await
                     .context("checking target phase")?;
             if target_in_version.is_none() {
-                return Ok(Err(LifecycleRefusal::NoSuchPhase));
+                return storage::refuse(tx, LifecycleRefusal::NoSuchPhase).await;
             }
             match &current {
                 // Entry: no current phase, any phase of the pinned
                 // version; return and restart need somewhere to come from.
                 None if matches!(kind, PhaseEventKind::Advance) => (None, Some(to)),
                 None => {
-                    return Ok(Err(LifecycleRefusal::NoCurrentPhase));
+                    return storage::refuse(tx, LifecycleRefusal::NoCurrentPhase).await;
                 }
                 Some((from, _)) => {
                     let edge_kind: Option<String> = sqlx::query_scalar(
@@ -549,25 +554,25 @@ pub async fn record_phase_event(
                         _ => edge_kind.as_deref() == Some("restart"),
                     };
                     if !allowed {
-                        return Ok(Err(LifecycleRefusal::TransitionNotAllowed));
+                        return storage::refuse(tx, LifecycleRefusal::TransitionNotAllowed).await;
                     }
                     (Some(*from), Some(to))
                 }
             }
         }
         PhaseEventKind::Pause | PhaseEventKind::Resume | PhaseEventKind::Complete => {
             let Some((from, _)) = current else {
-                return Ok(Err(LifecycleRefusal::NoCurrentPhase));
+                return storage::refuse(tx, LifecycleRefusal::NoCurrentPhase).await;
             };
             match kind {
                 PhaseEventKind::Pause if paused => {
-                    return Ok(Err(LifecycleRefusal::AlreadyPaused));
+                    return storage::refuse(tx, LifecycleRefusal::AlreadyPaused).await;
                 }
                 PhaseEventKind::Resume if !paused => {
-                    return Ok(Err(LifecycleRefusal::NotPaused));
+                    return storage::refuse(tx, LifecycleRefusal::NotPaused).await;
                 }
                 PhaseEventKind::Complete if paused => {
-                    return Ok(Err(LifecycleRefusal::Paused));
+                    return storage::refuse(tx, LifecycleRefusal::Paused).await;
                 }
                 _ => {}
             }

crates/consolebook-server/src/program_export.rs

@@ -17,6 +17,7 @@ use sqlx::SqlitePool;
 use crate::audit::EventKind;
 use crate::capabilities::{self, Capability};
 use crate::programs::{self, VersionContent};
+use crate::storage;
 
 /// Envelope discriminator for this document family.
 pub const FORMAT: &str = "consolebook-program-version";
@@ -105,7 +106,7 @@ pub async fn import_version(
         return Ok(Err(ImportRefusal::Invalid(problems)));
     }
 
-    let mut tx = pool.begin().await.context("starting import")?;
+    let mut tx = storage::write_tx(pool).await.context("starting import")?;
     let program_id = match target {
         ImportTarget::NewProgram => {
             let name = envelope.content.name.trim();
@@ -116,7 +117,7 @@ pub async fn import_version(
                     .await
                     .context("checking program name")?;
             if taken.is_some() {
-                return Ok(Err(ImportRefusal::ProgramNameTaken));
+                return storage::refuse(tx, ImportRefusal::ProgramNameTaken).await;
             }
             programs::insert_program(&mut tx, name, actor_user_id).await?
         }
@@ -127,7 +128,7 @@ pub async fn import_version(
                 .await
                 .context("checking program")?;
             if exists.is_none() {
-                return Ok(Err(ImportRefusal::NoSuchProgram));
+                return storage::refuse(tx, ImportRefusal::NoSuchProgram).await;
             }
             program_id
         }