From 06bdc1ce4b506b093692bb01803800abee4427ab Mon Sep 17 00:00:00 2001 From: Peter Permenter <41281403+TusanHomichi@users.noreply.github.com> Date: Sat, 5 Sep 2026 19:54:02 +0200 Subject: [PATCH 1/2] chore(storage): start issue 27 transaction retrofit From b1e8d45484108cf8faa730b4c8e211899d3903b5 Mon Sep 17 00:00:00 2001 From: Peter Permenter <41281403+TusanHomichi@users.noreply.github.com> Date: Sat, 5 Sep 2026 20:04:58 +0200 Subject: [PATCH 2/2] storage: reserve writes before validating concurrent operations --- crates/consolebook-server/src/assignments.rs | 23 +- crates/consolebook-server/src/enrollments.rs | 13 +- crates/consolebook-server/src/lifecycle.rs | 45 +- .../consolebook-server/src/program_export.rs | 7 +- crates/consolebook-server/src/programs.rs | 1218 +---------------- .../src/programs/content.rs | 498 +++++++ .../src/programs/persistence.rs | 679 +++++++++ .../src/session_membership.rs | 21 +- crates/consolebook-server/src/setup.rs | 21 +- crates/consolebook-server/src/storage.rs | 20 +- .../src/training_sessions.rs | 41 +- crates/consolebook-server/src/users.rs | 19 +- .../tests/write_transactions.rs | 183 +++ .../tests/write_transactions/accounts.rs | 210 +++ .../tests/write_transactions/programs.rs | 172 +++ .../tests/write_transactions/training.rs | 395 ++++++ .../0003-sqlite-connection-invariants.md | 2 + .../0019-immediate-write-transactions.md | 98 ++ docs/development.md | 15 +- 19 files changed, 2413 insertions(+), 1267 deletions(-) create mode 100644 crates/consolebook-server/src/programs/content.rs create mode 100644 crates/consolebook-server/src/programs/persistence.rs create mode 100644 crates/consolebook-server/tests/write_transactions.rs create mode 100644 crates/consolebook-server/tests/write_transactions/accounts.rs create mode 100644 crates/consolebook-server/tests/write_transactions/programs.rs create mode 100644 crates/consolebook-server/tests/write_transactions/training.rs create mode 100644 docs/decisions/0019-immediate-write-transactions.md diff --git a/crates/consolebook-server/src/assignments.rs b/crates/consolebook-server/src/assignments.rs index c26bbe4..4d44198 100644 --- a/crates/consolebook-server/src/assignments.rs +++ b/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,12 +102,14 @@ 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 = sqlx::query_scalar("SELECT 1 FROM user WHERE id = ?1") .bind(trainer_user_id) @@ -114,7 +117,7 @@ pub async fn create( .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 = 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 = 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,7 +198,9 @@ 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> = sqlx::query_scalar("SELECT ended_at FROM training_assignment WHERE id = ?1") .bind(assignment_id) @@ -203,8 +208,8 @@ pub async fn end( .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") diff --git a/crates/consolebook-server/src/enrollments.rs b/crates/consolebook-server/src/enrollments.rs index 1fe64db..2c823f7 100644 --- a/crates/consolebook-server/src/enrollments.rs +++ b/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 = 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 = 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( diff --git a/crates/consolebook-server/src/lifecycle.rs b/crates/consolebook-server/src/lifecycle.rs index 5164109..ea6b49e 100644 --- a/crates/consolebook-server/src/lifecycle.rs +++ b/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 = 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 = 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 ¤t { // 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 = sqlx::query_scalar( @@ -549,7 +554,7 @@ 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)) } @@ -557,17 +562,17 @@ pub async fn record_phase_event( } 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; } _ => {} } diff --git a/crates/consolebook-server/src/program_export.rs b/crates/consolebook-server/src/program_export.rs index 3c3a52e..7aaf7f0 100644 --- a/crates/consolebook-server/src/program_export.rs +++ b/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 } diff --git a/crates/consolebook-server/src/programs.rs b/crates/consolebook-server/src/programs.rs index 94ece59..df612b9 100644 --- a/crates/consolebook-server/src/programs.rs +++ b/crates/consolebook-server/src/programs.rs @@ -8,256 +8,22 @@ //! is written, so the composite foreign keys that enforce domain //! invariant 5 never see a dangling reference. -use std::collections::HashMap; -use std::collections::HashSet; +mod content; +mod persistence; -use anyhow::{Context, Result, bail}; -use serde::{Deserialize, Serialize}; -use sqlx::{Row, SqliteConnection, SqlitePool}; +pub use content::*; +pub use persistence::load_content; +use persistence::{delete_content, insert_content}; +pub(crate) use persistence::{insert_program, insert_version}; + +use anyhow::{Context, Result}; +use serde::Serialize; +use sqlx::{Row, SqlitePool}; use time::OffsetDateTime; use crate::audit::{self, EventKind, Subject}; use crate::capabilities::{self, Capability}; - -// ---- content document -// -// The same typed document is the authoring input (`replace_draft`), the -// read model (`load_content`), and the export/import payload -// (`program_export`). Strings are stored verbatim; required fields must -// contain non-whitespace content. References between parts use exact -// names, and name uniqueness is ASCII-case-insensitive to match the -// database's NOCASE indexes. - -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct VersionContent { - /// Snapshot of the program name as presented by this version. A later - /// program rename never rewrites it. - pub name: String, - /// Agency-visible free-text label; presentation, never identity. - pub label: String, - pub description: String, - pub phases: Vec, - pub phase_transitions: Vec, - pub competencies: Vec, - pub rating_scales: Vec, - pub rating_modifiers: Vec, - pub evaluation_forms: Vec, - /// Version-level standards citations; competency- and task-level - /// citations nest under their owners. - pub citations: Vec, - /// Completion rules gating finalization (ADR 0011), versioned like - /// every other piece of configuration. Absent in older exports; - /// the conservative defaults apply. - #[serde(default)] - pub finalization_policy: PolicyDef, -} - -/// The closed v1 completion-rule set (#32 decision 2). -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct PolicyDef { - pub review_approved: bool, - pub required_narratives: bool, - pub ratings_complete: bool, -} - -impl Default for PolicyDef { - fn default() -> Self { - Self { - review_approved: true, - required_narratives: true, - ratings_complete: true, - } - } -} - -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct PhaseDef { - pub name: String, - pub description: String, - /// Presentation data (docs/domain-model.md): ordering, never progress. - pub presentation_number: i64, -} - -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct TransitionDef { - pub from_phase: String, - pub to_phase: String, - pub kind: TransitionKind, -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] -#[serde(rename_all = "snake_case")] -pub enum TransitionKind { - Advance, - Remediation, - Skip, - Restart, -} - -impl TransitionKind { - #[must_use] - pub fn as_str(self) -> &'static str { - match self { - Self::Advance => "advance", - Self::Remediation => "remediation", - Self::Skip => "skip", - Self::Restart => "restart", - } - } - - fn from_db(value: &str) -> Result { - match value { - "advance" => Ok(Self::Advance), - "remediation" => Ok(Self::Remediation), - "skip" => Ok(Self::Skip), - "restart" => Ok(Self::Restart), - other => bail!("unknown transition kind in database: {other}"), - } - } -} - -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct CompetencyDef { - /// Free-text grouping label; empty means uncategorized. - pub category: String, - pub name: String, - pub description: String, - pub tasks: Vec, - pub citations: Vec, -} - -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct TaskDef { - pub prompt: String, - pub citations: Vec, -} - -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct ScaleDef { - pub name: String, - pub kind: ScaleKind, - /// Present exactly when `kind` is `anchored_numeric`. - pub min_value: Option, - /// Present exactly when `kind` is `anchored_numeric`. - pub max_value: Option, - pub anchors: Vec, -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] -#[serde(rename_all = "snake_case")] -pub enum ScaleKind { - AnchoredNumeric, - PassFail, - NarrativeOnly, -} - -impl ScaleKind { - #[must_use] - pub fn as_str(self) -> &'static str { - match self { - Self::AnchoredNumeric => "anchored_numeric", - Self::PassFail => "pass_fail", - Self::NarrativeOnly => "narrative_only", - } - } - - fn from_db(value: &str) -> Result { - match value { - "anchored_numeric" => Ok(Self::AnchoredNumeric), - "pass_fail" => Ok(Self::PassFail), - "narrative_only" => Ok(Self::NarrativeOnly), - other => bail!("unknown rating scale kind in database: {other}"), - } - } -} - -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct AnchorDef { - pub value: i64, - pub label: String, - pub definition: String, -} - -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct ModifierDef { - pub code: String, - pub label: String, - pub description: String, -} - -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct FormDef { - pub record_type: RecordType, - pub name: String, - pub instructions: String, - pub competencies: Vec, - pub narratives: Vec, -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] -#[serde(rename_all = "snake_case")] -pub enum RecordType { - DailyReport, - WeeklySummary, - PhaseEvaluation, -} - -impl RecordType { - #[must_use] - pub fn as_str(self) -> &'static str { - match self { - Self::DailyReport => "daily_report", - Self::WeeklySummary => "weekly_summary", - Self::PhaseEvaluation => "phase_evaluation", - } - } - - fn from_db(value: &str) -> Result { - match value { - "daily_report" => Ok(Self::DailyReport), - "weekly_summary" => Ok(Self::WeeklySummary), - "phase_evaluation" => Ok(Self::PhaseEvaluation), - other => bail!("unknown record type in database: {other}"), - } - } -} - -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct FormCompetencyDef { - /// Exact name of a competency defined in this version. - pub competency: String, - /// Exact name of a rating scale defined in this version. - pub rating_scale: String, -} - -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct NarrativeDef { - pub prompt: String, - pub required: bool, -} - -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct CitationDef { - /// Standards body, e.g. an accreditation program name. - pub body: String, - /// Edition or revision of the cited standard; may be empty. - pub edition: String, - pub clause: String, - pub note: String, -} +use crate::storage; // ---- summaries @@ -308,258 +74,6 @@ pub enum PublishRefusal { Incomplete(Vec), } -// ---- validation - -fn blank(value: &str) -> bool { - value.trim().is_empty() -} - -/// Detects a duplicate under the database's ASCII-case-insensitive -/// uniqueness rules. -fn note_duplicate(seen: &mut HashSet, value: &str) -> bool { - !seen.insert(value.to_ascii_lowercase()) -} - -fn validate_citations(problems: &mut Vec, owner: &str, citations: &[CitationDef]) { - for citation in citations { - if blank(&citation.body) { - problems.push(format!("{owner}: citation has an empty standards body")); - } - if blank(&citation.clause) { - problems.push(format!("{owner}: citation has an empty clause")); - } - } -} - -fn validate_phases(problems: &mut Vec, content: &VersionContent) { - let mut names = HashSet::new(); - for phase in &content.phases { - if blank(&phase.name) { - problems.push("phase has an empty name".to_owned()); - } else if note_duplicate(&mut names, &phase.name) { - problems.push(format!("duplicate phase name '{}'", phase.name)); - } - } - let defined: HashSet<&str> = content.phases.iter().map(|p| p.name.as_str()).collect(); - let mut edges = HashSet::new(); - for transition in &content.phase_transitions { - for endpoint in [&transition.from_phase, &transition.to_phase] { - if !defined.contains(endpoint.as_str()) { - problems.push(format!("transition references unknown phase '{endpoint}'")); - } - } - if !edges.insert((transition.from_phase.clone(), transition.to_phase.clone())) { - problems.push(format!( - "duplicate transition from '{}' to '{}'", - transition.from_phase, transition.to_phase - )); - } - } -} - -fn validate_competencies(problems: &mut Vec, content: &VersionContent) { - let mut names = HashSet::new(); - for competency in &content.competencies { - if blank(&competency.name) { - problems.push("competency has an empty name".to_owned()); - continue; - } - if note_duplicate(&mut names, &competency.name) { - problems.push(format!("duplicate competency name '{}'", competency.name)); - } - let mut prompts = HashSet::new(); - for task in &competency.tasks { - if blank(&task.prompt) { - problems.push(format!( - "competency '{}': task has an empty prompt", - competency.name - )); - } else if note_duplicate(&mut prompts, &task.prompt) { - problems.push(format!( - "competency '{}': duplicate task prompt '{}'", - competency.name, task.prompt - )); - } - validate_citations( - problems, - &format!("task '{}'", task.prompt), - &task.citations, - ); - } - validate_citations( - problems, - &format!("competency '{}'", competency.name), - &competency.citations, - ); - } -} - -fn validate_scale(problems: &mut Vec, scale: &ScaleDef) { - let mut values = HashSet::new(); - for anchor in &scale.anchors { - if blank(&anchor.label) { - problems.push(format!("scale '{}': anchor has an empty label", scale.name)); - } - if !values.insert(anchor.value) { - problems.push(format!( - "scale '{}': duplicate anchor value {}", - scale.name, anchor.value - )); - } - } - match scale.kind { - ScaleKind::AnchoredNumeric => { - let (Some(min), Some(max)) = (scale.min_value, scale.max_value) else { - problems.push(format!( - "scale '{}': anchored_numeric requires min_value and max_value", - scale.name - )); - return; - }; - if min >= max { - problems.push(format!( - "scale '{}': min_value must be less than max_value", - scale.name - )); - } - if scale.anchors.is_empty() { - problems.push(format!( - "scale '{}': anchored_numeric requires at least one anchor", - scale.name - )); - } - for anchor in &scale.anchors { - if anchor.value < min || anchor.value > max { - problems.push(format!( - "scale '{}': anchor value {} is outside {min}..={max}", - scale.name, anchor.value - )); - } - } - } - ScaleKind::PassFail => { - if scale.min_value.is_some() || scale.max_value.is_some() { - problems.push(format!( - "scale '{}': pass_fail does not take numeric bounds", - scale.name - )); - } - let values: Vec = scale.anchors.iter().map(|a| a.value).collect(); - if !(values.len() == 2 && values.contains(&0) && values.contains(&1)) { - problems.push(format!( - "scale '{}': pass_fail requires exactly two anchors with values 0 and 1", - scale.name - )); - } - } - ScaleKind::NarrativeOnly => { - if scale.min_value.is_some() || scale.max_value.is_some() { - problems.push(format!( - "scale '{}': narrative_only does not take numeric bounds", - scale.name - )); - } - if !scale.anchors.is_empty() { - problems.push(format!( - "scale '{}': narrative_only takes no anchors", - scale.name - )); - } - } - } -} - -fn validate_forms(problems: &mut Vec, content: &VersionContent) { - let competencies: HashSet<&str> = content - .competencies - .iter() - .map(|c| c.name.as_str()) - .collect(); - let scales: HashSet<&str> = content - .rating_scales - .iter() - .map(|s| s.name.as_str()) - .collect(); - let mut names = HashSet::new(); - for form in &content.evaluation_forms { - if blank(&form.name) { - problems.push("evaluation form has an empty name".to_owned()); - continue; - } - if note_duplicate(&mut names, &form.name) { - problems.push(format!("duplicate evaluation form name '{}'", form.name)); - } - let mut bound = HashSet::new(); - for binding in &form.competencies { - if !competencies.contains(binding.competency.as_str()) { - problems.push(format!( - "form '{}': references unknown competency '{}'", - form.name, binding.competency - )); - } - if !scales.contains(binding.rating_scale.as_str()) { - problems.push(format!( - "form '{}': references unknown rating scale '{}'", - form.name, binding.rating_scale - )); - } - if !bound.insert(binding.competency.clone()) { - problems.push(format!( - "form '{}': competency '{}' is bound more than once", - form.name, binding.competency - )); - } - } - for narrative in &form.narratives { - if blank(&narrative.prompt) { - problems.push(format!( - "form '{}': narrative has an empty prompt", - form.name - )); - } - } - } -} - -/// Structural validation of a content document: required text present, -/// names unique under the database's case-insensitive rules, and every -/// cross-reference resolving inside the document. Returns problems; -/// empty means valid. -#[must_use] -pub fn validate_content(content: &VersionContent) -> Vec { - let mut problems = Vec::new(); - if blank(&content.name) { - problems.push("version has an empty program name".to_owned()); - } - validate_phases(&mut problems, content); - validate_competencies(&mut problems, content); - let mut scale_names = HashSet::new(); - for scale in &content.rating_scales { - if blank(&scale.name) { - problems.push("rating scale has an empty name".to_owned()); - continue; - } - if note_duplicate(&mut scale_names, &scale.name) { - problems.push(format!("duplicate rating scale name '{}'", scale.name)); - } - validate_scale(&mut problems, scale); - } - let mut codes = HashSet::new(); - for modifier in &content.rating_modifiers { - if blank(&modifier.code) || blank(&modifier.label) { - problems.push("rating modifier has an empty code or label".to_owned()); - } else if note_duplicate(&mut codes, &modifier.code) { - problems.push(format!( - "duplicate rating modifier code '{}'", - modifier.code - )); - } - } - validate_forms(&mut problems, content); - validate_citations(&mut problems, "version", &content.citations); - problems -} - // ---- services async fn holds_manage_programs(pool: &SqlitePool, user_id: i64) -> Result { @@ -580,7 +94,9 @@ pub async fn create_program( if name.is_empty() { return Ok(Err(ProgramRefusal::NameEmpty)); } - let mut tx = pool.begin().await.context("starting program creation")?; + let mut tx = storage::write_tx(pool) + .await + .context("starting program creation")?; let taken: Option = sqlx::query_scalar("SELECT id FROM program WHERE name = ?1 COLLATE NOCASE") .bind(name) @@ -588,38 +104,13 @@ pub async fn create_program( .await .context("checking program name")?; if taken.is_some() { - return Ok(Err(ProgramRefusal::NameTaken)); + return storage::refuse(tx, ProgramRefusal::NameTaken).await; } let program_id = insert_program(&mut tx, name, actor_user_id).await?; tx.commit().await.context("committing program creation")?; Ok(Ok(program_id)) } -pub(crate) async fn insert_program( - tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>, - name: &str, - actor_user_id: i64, -) -> Result { - let result = - sqlx::query("INSERT INTO program (name, created_at, created_by) VALUES (?1, ?2, ?3)") - .bind(name) - .bind(OffsetDateTime::now_utc().unix_timestamp()) - .bind(actor_user_id) - .execute(&mut **tx) - .await - .context("creating program")?; - let program_id = result.last_insert_rowid(); - audit::record_for_subject( - &mut **tx, - EventKind::ProgramCreated, - Some(actor_user_id), - None, - Subject::Program(program_id), - ) - .await?; - Ok(program_id) -} - pub async fn list_programs(pool: &SqlitePool) -> Result> { let rows = sqlx::query("SELECT id, name, created_at FROM program ORDER BY name COLLATE NOCASE") .fetch_all(pool) @@ -708,14 +199,16 @@ pub async fn create_version( if !problems.is_empty() { return Ok(Err(AuthorRefusal::Invalid(problems))); } - let mut tx = pool.begin().await.context("starting version creation")?; + let mut tx = storage::write_tx(pool) + .await + .context("starting version creation")?; let exists: Option = sqlx::query_scalar("SELECT 1 FROM program WHERE id = ?1") .bind(program_id) .fetch_optional(&mut *tx) .await .context("checking program")?; if exists.is_none() { - return Ok(Err(AuthorRefusal::NoSuchProgram)); + return storage::refuse(tx, AuthorRefusal::NoSuchProgram).await; } let version_id = insert_version( &mut tx, @@ -729,50 +222,6 @@ pub async fn create_version( Ok(Ok(version_id)) } -/// Inserts a draft version row plus its content and records the lifecycle -/// audit event. Callers have already validated the content. -pub(crate) async fn insert_version( - tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>, - program_id: i64, - content: &VersionContent, - actor_user_id: i64, - kind: EventKind, -) -> Result { - let next_number: i64 = sqlx::query_scalar( - "SELECT COALESCE(MAX(version_number), 0) + 1 FROM program_version WHERE program_id = ?1", - ) - .bind(program_id) - .fetch_one(&mut **tx) - .await - .context("numbering version")?; - let result = sqlx::query( - "INSERT INTO program_version - (program_id, version_number, label, name, description, created_at, created_by) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)", - ) - .bind(program_id) - .bind(next_number) - .bind(&content.label) - .bind(&content.name) - .bind(&content.description) - .bind(OffsetDateTime::now_utc().unix_timestamp()) - .bind(actor_user_id) - .execute(&mut **tx) - .await - .context("creating program version")?; - let version_id = result.last_insert_rowid(); - insert_content(tx, version_id, content).await?; - audit::record_for_subject( - &mut **tx, - kind, - Some(actor_user_id), - None, - Subject::ProgramVersion(version_id), - ) - .await?; - Ok(version_id) -} - /// Replaces a draft's entire content — single editor, honest last write /// (ADR 0007). Refused once the version is published. pub async fn replace_draft( @@ -788,10 +237,14 @@ pub async fn replace_draft( if !problems.is_empty() { return Ok(Err(AuthorRefusal::Invalid(problems))); } - let mut tx = pool.begin().await.context("starting draft replacement")?; + let mut tx = storage::write_tx(pool) + .await + .context("starting draft replacement")?; match version_state(&mut tx, version_id).await? { - VersionState::Missing => return Ok(Err(AuthorRefusal::NoSuchVersion)), - VersionState::Published => return Ok(Err(AuthorRefusal::AlreadyPublished)), + VersionState::Missing => return storage::refuse(tx, AuthorRefusal::NoSuchVersion).await, + VersionState::Published => { + return storage::refuse(tx, AuthorRefusal::AlreadyPublished).await; + } VersionState::Draft => {} } delete_content(&mut tx, version_id).await?; @@ -818,10 +271,12 @@ pub async fn publish_version( if !holds_manage_programs(pool, actor_user_id).await? { return Ok(Err(PublishRefusal::CapabilityRequired)); } - let mut tx = pool.begin().await.context("starting publish")?; + let mut tx = storage::write_tx(pool).await.context("starting publish")?; match version_state(&mut tx, version_id).await? { - VersionState::Missing => return Ok(Err(PublishRefusal::NoSuchVersion)), - VersionState::Published => return Ok(Err(PublishRefusal::AlreadyPublished)), + VersionState::Missing => return storage::refuse(tx, PublishRefusal::NoSuchVersion).await, + VersionState::Published => { + return storage::refuse(tx, PublishRefusal::AlreadyPublished).await; + } VersionState::Draft => {} } let empty_forms: Vec = sqlx::query_scalar( @@ -840,7 +295,7 @@ pub async fn publish_version( .iter() .map(|name| format!("form '{name}' has no competencies and no narratives")) .collect(); - return Ok(Err(PublishRefusal::Incomplete(problems))); + return storage::refuse(tx, PublishRefusal::Incomplete(problems)).await; } let stamped = sqlx::query( "UPDATE program_version SET published_at = ?2, published_by = ?3 @@ -853,7 +308,7 @@ pub async fn publish_version( .await .context("publishing version")?; if stamped.rows_affected() != 1 { - return Ok(Err(PublishRefusal::AlreadyPublished)); + return storage::refuse(tx, PublishRefusal::AlreadyPublished).await; } audit::record_for_subject( &mut *tx, @@ -877,10 +332,14 @@ pub async fn discard_draft( if !holds_manage_programs(pool, actor_user_id).await? { return Ok(Err(AuthorRefusal::CapabilityRequired)); } - let mut tx = pool.begin().await.context("starting draft discard")?; + let mut tx = storage::write_tx(pool) + .await + .context("starting draft discard")?; match version_state(&mut tx, version_id).await? { - VersionState::Missing => return Ok(Err(AuthorRefusal::NoSuchVersion)), - VersionState::Published => return Ok(Err(AuthorRefusal::AlreadyPublished)), + VersionState::Missing => return storage::refuse(tx, AuthorRefusal::NoSuchVersion).await, + VersionState::Published => { + return storage::refuse(tx, AuthorRefusal::AlreadyPublished).await; + } VersionState::Draft => {} } delete_content(&mut tx, version_id).await?; @@ -928,600 +387,3 @@ async fn version_state( } }) } - -// ---- content writes - -async fn insert_citation( - conn: &mut SqliteConnection, - version_id: i64, - competency_id: Option, - task_id: Option, - citation: &CitationDef, -) -> Result<()> { - sqlx::query( - "INSERT INTO standards_citation - (program_version_id, competency_id, task_id, body, edition, clause, note) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)", - ) - .bind(version_id) - .bind(competency_id) - .bind(task_id) - .bind(&citation.body) - .bind(&citation.edition) - .bind(&citation.clause) - .bind(&citation.note) - .execute(conn) - .await - .context("inserting standards citation")?; - Ok(()) -} - -async fn insert_phases( - conn: &mut SqliteConnection, - version_id: i64, - content: &VersionContent, -) -> Result<()> { - let mut phase_ids: HashMap<&str, i64> = HashMap::new(); - for phase in &content.phases { - let result = sqlx::query( - "INSERT INTO phase (program_version_id, name, description, presentation_number) - VALUES (?1, ?2, ?3, ?4)", - ) - .bind(version_id) - .bind(&phase.name) - .bind(&phase.description) - .bind(phase.presentation_number) - .execute(&mut *conn) - .await - .context("inserting phase")?; - phase_ids.insert(phase.name.as_str(), result.last_insert_rowid()); - } - for transition in &content.phase_transitions { - sqlx::query( - "INSERT INTO phase_transition - (program_version_id, from_phase_id, to_phase_id, kind) - VALUES (?1, ?2, ?3, ?4)", - ) - .bind(version_id) - .bind(phase_ids[transition.from_phase.as_str()]) - .bind(phase_ids[transition.to_phase.as_str()]) - .bind(transition.kind.as_str()) - .execute(&mut *conn) - .await - .context("inserting phase transition")?; - } - Ok(()) -} - -async fn insert_competencies( - conn: &mut SqliteConnection, - version_id: i64, - content: &VersionContent, -) -> Result> { - let mut competency_ids: HashMap = HashMap::new(); - for (order, competency) in (0_i64..).zip(&content.competencies) { - let result = sqlx::query( - "INSERT INTO competency (program_version_id, category, name, description, sort_order) - VALUES (?1, ?2, ?3, ?4, ?5)", - ) - .bind(version_id) - .bind(&competency.category) - .bind(&competency.name) - .bind(&competency.description) - .bind(order) - .execute(&mut *conn) - .await - .context("inserting competency")?; - let competency_id = result.last_insert_rowid(); - competency_ids.insert(competency.name.clone(), competency_id); - for (task_order, task) in (0_i64..).zip(&competency.tasks) { - let inserted = sqlx::query( - "INSERT INTO task (program_version_id, competency_id, prompt, sort_order) - VALUES (?1, ?2, ?3, ?4)", - ) - .bind(version_id) - .bind(competency_id) - .bind(&task.prompt) - .bind(task_order) - .execute(&mut *conn) - .await - .context("inserting task")?; - let task_id = inserted.last_insert_rowid(); - for citation in &task.citations { - insert_citation(&mut *conn, version_id, None, Some(task_id), citation).await?; - } - } - for citation in &competency.citations { - insert_citation(&mut *conn, version_id, Some(competency_id), None, citation).await?; - } - } - Ok(competency_ids) -} - -async fn insert_scales( - conn: &mut SqliteConnection, - version_id: i64, - content: &VersionContent, -) -> Result> { - let mut scale_ids: HashMap = HashMap::new(); - for scale in &content.rating_scales { - let result = sqlx::query( - "INSERT INTO rating_scale (program_version_id, name, kind, min_value, max_value) - VALUES (?1, ?2, ?3, ?4, ?5)", - ) - .bind(version_id) - .bind(&scale.name) - .bind(scale.kind.as_str()) - .bind(scale.min_value) - .bind(scale.max_value) - .execute(&mut *conn) - .await - .context("inserting rating scale")?; - let scale_id = result.last_insert_rowid(); - scale_ids.insert(scale.name.clone(), scale_id); - for anchor in &scale.anchors { - sqlx::query( - "INSERT INTO rating_anchor - (program_version_id, rating_scale_id, value, label, definition) - VALUES (?1, ?2, ?3, ?4, ?5)", - ) - .bind(version_id) - .bind(scale_id) - .bind(anchor.value) - .bind(&anchor.label) - .bind(&anchor.definition) - .execute(&mut *conn) - .await - .context("inserting rating anchor")?; - } - } - for modifier in &content.rating_modifiers { - sqlx::query( - "INSERT INTO rating_modifier (program_version_id, code, label, description) - VALUES (?1, ?2, ?3, ?4)", - ) - .bind(version_id) - .bind(&modifier.code) - .bind(&modifier.label) - .bind(&modifier.description) - .execute(&mut *conn) - .await - .context("inserting rating modifier")?; - } - Ok(scale_ids) -} - -async fn insert_forms( - conn: &mut SqliteConnection, - version_id: i64, - content: &VersionContent, - competency_ids: &HashMap, - scale_ids: &HashMap, -) -> Result<()> { - for form in &content.evaluation_forms { - let result = sqlx::query( - "INSERT INTO evaluation_form (program_version_id, record_type, name, instructions) - VALUES (?1, ?2, ?3, ?4)", - ) - .bind(version_id) - .bind(form.record_type.as_str()) - .bind(&form.name) - .bind(&form.instructions) - .execute(&mut *conn) - .await - .context("inserting evaluation form")?; - let form_id = result.last_insert_rowid(); - for (order, binding) in (0_i64..).zip(&form.competencies) { - sqlx::query( - "INSERT INTO form_competency - (program_version_id, evaluation_form_id, competency_id, rating_scale_id, sort_order) - VALUES (?1, ?2, ?3, ?4, ?5)", - ) - .bind(version_id) - .bind(form_id) - .bind(competency_ids[&binding.competency]) - .bind(scale_ids[&binding.rating_scale]) - .bind(order) - .execute(&mut *conn) - .await - .context("inserting form competency")?; - } - for (order, narrative) in (0_i64..).zip(&form.narratives) { - sqlx::query( - "INSERT INTO form_narrative - (program_version_id, evaluation_form_id, prompt, required, sort_order) - VALUES (?1, ?2, ?3, ?4, ?5)", - ) - .bind(version_id) - .bind(form_id) - .bind(&narrative.prompt) - .bind(i64::from(narrative.required)) - .bind(order) - .execute(&mut *conn) - .await - .context("inserting form narrative")?; - } - } - Ok(()) -} - -/// Writes every owned row of a validated content document. -async fn insert_content( - conn: &mut SqliteConnection, - version_id: i64, - content: &VersionContent, -) -> Result<()> { - insert_phases(&mut *conn, version_id, content).await?; - let competency_ids = insert_competencies(&mut *conn, version_id, content).await?; - let scale_ids = insert_scales(&mut *conn, version_id, content).await?; - insert_forms(&mut *conn, version_id, content, &competency_ids, &scale_ids).await?; - for citation in &content.citations { - insert_citation(&mut *conn, version_id, None, None, citation).await?; - } - sqlx::query( - "INSERT INTO finalization_policy - (program_version_id, review_approved, required_narratives, ratings_complete) - VALUES (?1, ?2, ?3, ?4)", - ) - .bind(version_id) - .bind(i64::from(content.finalization_policy.review_approved)) - .bind(i64::from(content.finalization_policy.required_narratives)) - .bind(i64::from(content.finalization_policy.ratings_complete)) - .execute(&mut *conn) - .await - .context("writing finalization policy")?; - Ok(()) -} - -/// Deletes every owned row of a draft version, children before parents so -/// foreign keys hold throughout. -async fn delete_content(conn: &mut SqliteConnection, version_id: i64) -> Result<()> { - for statement in [ - "DELETE FROM finalization_policy WHERE program_version_id = ?1", - "DELETE FROM standards_citation WHERE program_version_id = ?1", - "DELETE FROM form_narrative WHERE program_version_id = ?1", - "DELETE FROM form_competency WHERE program_version_id = ?1", - "DELETE FROM evaluation_form WHERE program_version_id = ?1", - "DELETE FROM rating_anchor WHERE program_version_id = ?1", - "DELETE FROM rating_scale WHERE program_version_id = ?1", - "DELETE FROM rating_modifier WHERE program_version_id = ?1", - "DELETE FROM task WHERE program_version_id = ?1", - "DELETE FROM competency WHERE program_version_id = ?1", - "DELETE FROM phase_transition WHERE program_version_id = ?1", - "DELETE FROM phase WHERE program_version_id = ?1", - ] { - sqlx::query(statement) - .bind(version_id) - .execute(&mut *conn) - .await - .context("deleting draft content")?; - } - Ok(()) -} - -// ---- content reads - -/// Loads a version's complete content document, or `None` when the -/// version does not exist. Arrays come back in the deterministic export -/// order (authored order where one exists, content order otherwise). -pub async fn load_content(pool: &SqlitePool, version_id: i64) -> Result> { - // One transaction so every query reads the same snapshot. - let mut tx = pool.begin().await.context("starting content load")?; - let Some(header) = - sqlx::query("SELECT name, label, description FROM program_version WHERE id = ?1") - .bind(version_id) - .fetch_optional(&mut *tx) - .await - .context("loading version row")? - else { - return Ok(None); - }; - let mut content = VersionContent { - name: header.get("name"), - label: header.get("label"), - description: header.get("description"), - phases: Vec::new(), - phase_transitions: Vec::new(), - competencies: Vec::new(), - rating_scales: Vec::new(), - rating_modifiers: Vec::new(), - evaluation_forms: Vec::new(), - citations: Vec::new(), - finalization_policy: PolicyDef::default(), - }; - if let Some(policy) = sqlx::query( - "SELECT review_approved, required_narratives, ratings_complete - FROM finalization_policy WHERE program_version_id = ?1", - ) - .bind(version_id) - .fetch_optional(&mut *tx) - .await - .context("loading finalization policy")? - { - content.finalization_policy = PolicyDef { - review_approved: policy.get::("review_approved") != 0, - required_narratives: policy.get::("required_narratives") != 0, - ratings_complete: policy.get::("ratings_complete") != 0, - }; - } - load_phases(&mut tx, version_id, &mut content).await?; - let competency_index = load_competencies(&mut tx, version_id, &mut content).await?; - load_scales(&mut tx, version_id, &mut content).await?; - load_forms(&mut tx, version_id, &mut content).await?; - load_citations(&mut tx, version_id, &mut content, &competency_index).await?; - Ok(Some(content)) -} - -async fn load_phases( - tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>, - version_id: i64, - content: &mut VersionContent, -) -> Result<()> { - let rows = sqlx::query( - "SELECT name, description, presentation_number FROM phase - WHERE program_version_id = ?1 ORDER BY presentation_number, name", - ) - .bind(version_id) - .fetch_all(&mut **tx) - .await - .context("loading phases")?; - content.phases = rows - .iter() - .map(|row| PhaseDef { - name: row.get("name"), - description: row.get("description"), - presentation_number: row.get("presentation_number"), - }) - .collect(); - let rows = sqlx::query( - "SELECT f.name AS from_name, t.name AS to_name, pt.kind - FROM phase_transition pt - JOIN phase f ON f.id = pt.from_phase_id - JOIN phase t ON t.id = pt.to_phase_id - WHERE pt.program_version_id = ?1 - ORDER BY f.name, t.name", - ) - .bind(version_id) - .fetch_all(&mut **tx) - .await - .context("loading phase transitions")?; - for row in &rows { - content.phase_transitions.push(TransitionDef { - from_phase: row.get("from_name"), - to_phase: row.get("to_name"), - kind: TransitionKind::from_db(row.get("kind"))?, - }); - } - Ok(()) -} - -/// Loads competencies and their tasks; returns row-id lookup maps used to -/// route citations. -async fn load_competencies( - tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>, - version_id: i64, - content: &mut VersionContent, -) -> Result { - let rows = sqlx::query( - "SELECT id, category, name, description FROM competency - WHERE program_version_id = ?1 ORDER BY sort_order, name", - ) - .bind(version_id) - .fetch_all(&mut **tx) - .await - .context("loading competencies")?; - let mut index = CompetencyIndex::default(); - for row in &rows { - let id: i64 = row.get("id"); - index - .by_competency_row - .insert(id, content.competencies.len()); - content.competencies.push(CompetencyDef { - category: row.get("category"), - name: row.get("name"), - description: row.get("description"), - tasks: Vec::new(), - citations: Vec::new(), - }); - } - let rows = sqlx::query( - "SELECT id, competency_id, prompt FROM task - WHERE program_version_id = ?1 ORDER BY sort_order, prompt", - ) - .bind(version_id) - .fetch_all(&mut **tx) - .await - .context("loading tasks")?; - for row in &rows { - let competency_row: i64 = row.get("competency_id"); - let competency_slot = index.by_competency_row[&competency_row]; - let tasks = &mut content.competencies[competency_slot].tasks; - index - .by_task_row - .insert(row.get("id"), (competency_slot, tasks.len())); - tasks.push(TaskDef { - prompt: row.get("prompt"), - citations: Vec::new(), - }); - } - Ok(index) -} - -#[derive(Default)] -struct CompetencyIndex { - /// competency row id -> index into `content.competencies` - by_competency_row: HashMap, - /// task row id -> (competency index, task index) - by_task_row: HashMap, -} - -async fn load_scales( - tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>, - version_id: i64, - content: &mut VersionContent, -) -> Result<()> { - let rows = sqlx::query( - "SELECT id, name, kind, min_value, max_value FROM rating_scale - WHERE program_version_id = ?1 ORDER BY name", - ) - .bind(version_id) - .fetch_all(&mut **tx) - .await - .context("loading rating scales")?; - let mut slot_by_row: HashMap = HashMap::new(); - for row in &rows { - slot_by_row.insert(row.get("id"), content.rating_scales.len()); - content.rating_scales.push(ScaleDef { - name: row.get("name"), - kind: ScaleKind::from_db(row.get("kind"))?, - min_value: row.get("min_value"), - max_value: row.get("max_value"), - anchors: Vec::new(), - }); - } - let rows = sqlx::query( - "SELECT rating_scale_id, value, label, definition FROM rating_anchor - WHERE program_version_id = ?1 ORDER BY value", - ) - .bind(version_id) - .fetch_all(&mut **tx) - .await - .context("loading rating anchors")?; - for row in &rows { - let scale_row: i64 = row.get("rating_scale_id"); - content.rating_scales[slot_by_row[&scale_row]] - .anchors - .push(AnchorDef { - value: row.get("value"), - label: row.get("label"), - definition: row.get("definition"), - }); - } - let rows = sqlx::query( - "SELECT code, label, description FROM rating_modifier - WHERE program_version_id = ?1 ORDER BY code", - ) - .bind(version_id) - .fetch_all(&mut **tx) - .await - .context("loading rating modifiers")?; - content.rating_modifiers = rows - .iter() - .map(|row| ModifierDef { - code: row.get("code"), - label: row.get("label"), - description: row.get("description"), - }) - .collect(); - Ok(()) -} - -async fn load_forms( - tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>, - version_id: i64, - content: &mut VersionContent, -) -> Result<()> { - let rows = sqlx::query( - "SELECT id, record_type, name, instructions FROM evaluation_form - WHERE program_version_id = ?1 ORDER BY name", - ) - .bind(version_id) - .fetch_all(&mut **tx) - .await - .context("loading evaluation forms")?; - let mut slot_by_row: HashMap = HashMap::new(); - for row in &rows { - slot_by_row.insert(row.get("id"), content.evaluation_forms.len()); - content.evaluation_forms.push(FormDef { - record_type: RecordType::from_db(row.get("record_type"))?, - name: row.get("name"), - instructions: row.get("instructions"), - competencies: Vec::new(), - narratives: Vec::new(), - }); - } - let rows = sqlx::query( - "SELECT fc.evaluation_form_id, c.name AS competency, s.name AS rating_scale - FROM form_competency fc - JOIN competency c ON c.id = fc.competency_id - JOIN rating_scale s ON s.id = fc.rating_scale_id - WHERE fc.program_version_id = ?1 - ORDER BY fc.sort_order, c.name", - ) - .bind(version_id) - .fetch_all(&mut **tx) - .await - .context("loading form competencies")?; - for row in &rows { - let form_row: i64 = row.get("evaluation_form_id"); - content.evaluation_forms[slot_by_row[&form_row]] - .competencies - .push(FormCompetencyDef { - competency: row.get("competency"), - rating_scale: row.get("rating_scale"), - }); - } - let rows = sqlx::query( - "SELECT evaluation_form_id, prompt, required FROM form_narrative - WHERE program_version_id = ?1 ORDER BY sort_order, prompt", - ) - .bind(version_id) - .fetch_all(&mut **tx) - .await - .context("loading form narratives")?; - for row in &rows { - let form_row: i64 = row.get("evaluation_form_id"); - let required: i64 = row.get("required"); - content.evaluation_forms[slot_by_row[&form_row]] - .narratives - .push(NarrativeDef { - prompt: row.get("prompt"), - required: required != 0, - }); - } - Ok(()) -} - -async fn load_citations( - tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>, - version_id: i64, - content: &mut VersionContent, - index: &CompetencyIndex, -) -> Result<()> { - let rows = sqlx::query( - "SELECT competency_id, task_id, body, edition, clause, note - FROM standards_citation WHERE program_version_id = ?1 - ORDER BY body, edition, clause, note", - ) - .bind(version_id) - .fetch_all(&mut **tx) - .await - .context("loading standards citations")?; - for row in &rows { - let citation = CitationDef { - body: row.get("body"), - edition: row.get("edition"), - clause: row.get("clause"), - note: row.get("note"), - }; - let competency_id: Option = row.get("competency_id"); - let task_id: Option = row.get("task_id"); - match (competency_id, task_id) { - (Some(competency_row), None) => { - let slot = index.by_competency_row[&competency_row]; - content.competencies[slot].citations.push(citation); - } - (None, Some(task_row)) => { - let (competency_slot, task_slot) = index.by_task_row[&task_row]; - content.competencies[competency_slot].tasks[task_slot] - .citations - .push(citation); - } - (None, None) => content.citations.push(citation), - (Some(_), Some(_)) => { - bail!("citation targets both a competency and a task; the schema forbids this") - } - } - } - Ok(()) -} diff --git a/crates/consolebook-server/src/programs/content.rs b/crates/consolebook-server/src/programs/content.rs new file mode 100644 index 0000000..d7411a9 --- /dev/null +++ b/crates/consolebook-server/src/programs/content.rs @@ -0,0 +1,498 @@ +//! Program configuration vocabulary and structural validation. + +use std::collections::HashSet; + +use anyhow::{Result, bail}; +use serde::{Deserialize, Serialize}; + +// ---- content document +// +// The same typed document is the authoring input (`replace_draft`), the +// read model (`load_content`), and the export/import payload +// (`program_export`). Strings are stored verbatim; required fields must +// contain non-whitespace content. References between parts use exact +// names, and name uniqueness is ASCII-case-insensitive to match the +// database's NOCASE indexes. + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct VersionContent { + /// Snapshot of the program name as presented by this version. A later + /// program rename never rewrites it. + pub name: String, + /// Agency-visible free-text label; presentation, never identity. + pub label: String, + pub description: String, + pub phases: Vec, + pub phase_transitions: Vec, + pub competencies: Vec, + pub rating_scales: Vec, + pub rating_modifiers: Vec, + pub evaluation_forms: Vec, + /// Version-level standards citations; competency- and task-level + /// citations nest under their owners. + pub citations: Vec, + /// Completion rules gating finalization (ADR 0011), versioned like + /// every other piece of configuration. Absent in older exports; + /// the conservative defaults apply. + #[serde(default)] + pub finalization_policy: PolicyDef, +} + +/// The closed v1 completion-rule set (#32 decision 2). +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct PolicyDef { + pub review_approved: bool, + pub required_narratives: bool, + pub ratings_complete: bool, +} + +impl Default for PolicyDef { + fn default() -> Self { + Self { + review_approved: true, + required_narratives: true, + ratings_complete: true, + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct PhaseDef { + pub name: String, + pub description: String, + /// Presentation data (docs/domain-model.md): ordering, never progress. + pub presentation_number: i64, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct TransitionDef { + pub from_phase: String, + pub to_phase: String, + pub kind: TransitionKind, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum TransitionKind { + Advance, + Remediation, + Skip, + Restart, +} + +impl TransitionKind { + #[must_use] + pub fn as_str(self) -> &'static str { + match self { + Self::Advance => "advance", + Self::Remediation => "remediation", + Self::Skip => "skip", + Self::Restart => "restart", + } + } + + pub(super) fn from_db(value: &str) -> Result { + match value { + "advance" => Ok(Self::Advance), + "remediation" => Ok(Self::Remediation), + "skip" => Ok(Self::Skip), + "restart" => Ok(Self::Restart), + other => bail!("unknown transition kind in database: {other}"), + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct CompetencyDef { + /// Free-text grouping label; empty means uncategorized. + pub category: String, + pub name: String, + pub description: String, + pub tasks: Vec, + pub citations: Vec, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct TaskDef { + pub prompt: String, + pub citations: Vec, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ScaleDef { + pub name: String, + pub kind: ScaleKind, + /// Present exactly when `kind` is `anchored_numeric`. + pub min_value: Option, + /// Present exactly when `kind` is `anchored_numeric`. + pub max_value: Option, + pub anchors: Vec, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ScaleKind { + AnchoredNumeric, + PassFail, + NarrativeOnly, +} + +impl ScaleKind { + #[must_use] + pub fn as_str(self) -> &'static str { + match self { + Self::AnchoredNumeric => "anchored_numeric", + Self::PassFail => "pass_fail", + Self::NarrativeOnly => "narrative_only", + } + } + + pub(super) fn from_db(value: &str) -> Result { + match value { + "anchored_numeric" => Ok(Self::AnchoredNumeric), + "pass_fail" => Ok(Self::PassFail), + "narrative_only" => Ok(Self::NarrativeOnly), + other => bail!("unknown rating scale kind in database: {other}"), + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct AnchorDef { + pub value: i64, + pub label: String, + pub definition: String, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ModifierDef { + pub code: String, + pub label: String, + pub description: String, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct FormDef { + pub record_type: RecordType, + pub name: String, + pub instructions: String, + pub competencies: Vec, + pub narratives: Vec, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum RecordType { + DailyReport, + WeeklySummary, + PhaseEvaluation, +} + +impl RecordType { + #[must_use] + pub fn as_str(self) -> &'static str { + match self { + Self::DailyReport => "daily_report", + Self::WeeklySummary => "weekly_summary", + Self::PhaseEvaluation => "phase_evaluation", + } + } + + pub(super) fn from_db(value: &str) -> Result { + match value { + "daily_report" => Ok(Self::DailyReport), + "weekly_summary" => Ok(Self::WeeklySummary), + "phase_evaluation" => Ok(Self::PhaseEvaluation), + other => bail!("unknown record type in database: {other}"), + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct FormCompetencyDef { + /// Exact name of a competency defined in this version. + pub competency: String, + /// Exact name of a rating scale defined in this version. + pub rating_scale: String, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct NarrativeDef { + pub prompt: String, + pub required: bool, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct CitationDef { + /// Standards body, e.g. an accreditation program name. + pub body: String, + /// Edition or revision of the cited standard; may be empty. + pub edition: String, + pub clause: String, + pub note: String, +} + +// ---- validation + +fn blank(value: &str) -> bool { + value.trim().is_empty() +} + +/// Detects a duplicate under the database's ASCII-case-insensitive +/// uniqueness rules. +fn note_duplicate(seen: &mut HashSet, value: &str) -> bool { + !seen.insert(value.to_ascii_lowercase()) +} + +fn validate_citations(problems: &mut Vec, owner: &str, citations: &[CitationDef]) { + for citation in citations { + if blank(&citation.body) { + problems.push(format!("{owner}: citation has an empty standards body")); + } + if blank(&citation.clause) { + problems.push(format!("{owner}: citation has an empty clause")); + } + } +} + +fn validate_phases(problems: &mut Vec, content: &VersionContent) { + let mut names = HashSet::new(); + for phase in &content.phases { + if blank(&phase.name) { + problems.push("phase has an empty name".to_owned()); + } else if note_duplicate(&mut names, &phase.name) { + problems.push(format!("duplicate phase name '{}'", phase.name)); + } + } + let defined: HashSet<&str> = content.phases.iter().map(|p| p.name.as_str()).collect(); + let mut edges = HashSet::new(); + for transition in &content.phase_transitions { + for endpoint in [&transition.from_phase, &transition.to_phase] { + if !defined.contains(endpoint.as_str()) { + problems.push(format!("transition references unknown phase '{endpoint}'")); + } + } + if !edges.insert((transition.from_phase.clone(), transition.to_phase.clone())) { + problems.push(format!( + "duplicate transition from '{}' to '{}'", + transition.from_phase, transition.to_phase + )); + } + } +} + +fn validate_competencies(problems: &mut Vec, content: &VersionContent) { + let mut names = HashSet::new(); + for competency in &content.competencies { + if blank(&competency.name) { + problems.push("competency has an empty name".to_owned()); + continue; + } + if note_duplicate(&mut names, &competency.name) { + problems.push(format!("duplicate competency name '{}'", competency.name)); + } + let mut prompts = HashSet::new(); + for task in &competency.tasks { + if blank(&task.prompt) { + problems.push(format!( + "competency '{}': task has an empty prompt", + competency.name + )); + } else if note_duplicate(&mut prompts, &task.prompt) { + problems.push(format!( + "competency '{}': duplicate task prompt '{}'", + competency.name, task.prompt + )); + } + validate_citations( + problems, + &format!("task '{}'", task.prompt), + &task.citations, + ); + } + validate_citations( + problems, + &format!("competency '{}'", competency.name), + &competency.citations, + ); + } +} + +fn validate_scale(problems: &mut Vec, scale: &ScaleDef) { + let mut values = HashSet::new(); + for anchor in &scale.anchors { + if blank(&anchor.label) { + problems.push(format!("scale '{}': anchor has an empty label", scale.name)); + } + if !values.insert(anchor.value) { + problems.push(format!( + "scale '{}': duplicate anchor value {}", + scale.name, anchor.value + )); + } + } + match scale.kind { + ScaleKind::AnchoredNumeric => { + let (Some(min), Some(max)) = (scale.min_value, scale.max_value) else { + problems.push(format!( + "scale '{}': anchored_numeric requires min_value and max_value", + scale.name + )); + return; + }; + if min >= max { + problems.push(format!( + "scale '{}': min_value must be less than max_value", + scale.name + )); + } + if scale.anchors.is_empty() { + problems.push(format!( + "scale '{}': anchored_numeric requires at least one anchor", + scale.name + )); + } + for anchor in &scale.anchors { + if anchor.value < min || anchor.value > max { + problems.push(format!( + "scale '{}': anchor value {} is outside {min}..={max}", + scale.name, anchor.value + )); + } + } + } + ScaleKind::PassFail => { + if scale.min_value.is_some() || scale.max_value.is_some() { + problems.push(format!( + "scale '{}': pass_fail does not take numeric bounds", + scale.name + )); + } + let values: Vec = scale.anchors.iter().map(|a| a.value).collect(); + if !(values.len() == 2 && values.contains(&0) && values.contains(&1)) { + problems.push(format!( + "scale '{}': pass_fail requires exactly two anchors with values 0 and 1", + scale.name + )); + } + } + ScaleKind::NarrativeOnly => { + if scale.min_value.is_some() || scale.max_value.is_some() { + problems.push(format!( + "scale '{}': narrative_only does not take numeric bounds", + scale.name + )); + } + if !scale.anchors.is_empty() { + problems.push(format!( + "scale '{}': narrative_only takes no anchors", + scale.name + )); + } + } + } +} + +fn validate_forms(problems: &mut Vec, content: &VersionContent) { + let competencies: HashSet<&str> = content + .competencies + .iter() + .map(|c| c.name.as_str()) + .collect(); + let scales: HashSet<&str> = content + .rating_scales + .iter() + .map(|s| s.name.as_str()) + .collect(); + let mut names = HashSet::new(); + for form in &content.evaluation_forms { + if blank(&form.name) { + problems.push("evaluation form has an empty name".to_owned()); + continue; + } + if note_duplicate(&mut names, &form.name) { + problems.push(format!("duplicate evaluation form name '{}'", form.name)); + } + let mut bound = HashSet::new(); + for binding in &form.competencies { + if !competencies.contains(binding.competency.as_str()) { + problems.push(format!( + "form '{}': references unknown competency '{}'", + form.name, binding.competency + )); + } + if !scales.contains(binding.rating_scale.as_str()) { + problems.push(format!( + "form '{}': references unknown rating scale '{}'", + form.name, binding.rating_scale + )); + } + if !bound.insert(binding.competency.clone()) { + problems.push(format!( + "form '{}': competency '{}' is bound more than once", + form.name, binding.competency + )); + } + } + for narrative in &form.narratives { + if blank(&narrative.prompt) { + problems.push(format!( + "form '{}': narrative has an empty prompt", + form.name + )); + } + } + } +} + +/// Structural validation of a content document: required text present, +/// names unique under the database's case-insensitive rules, and every +/// cross-reference resolving inside the document. Returns problems; +/// empty means valid. +#[must_use] +pub fn validate_content(content: &VersionContent) -> Vec { + let mut problems = Vec::new(); + if blank(&content.name) { + problems.push("version has an empty program name".to_owned()); + } + validate_phases(&mut problems, content); + validate_competencies(&mut problems, content); + let mut scale_names = HashSet::new(); + for scale in &content.rating_scales { + if blank(&scale.name) { + problems.push("rating scale has an empty name".to_owned()); + continue; + } + if note_duplicate(&mut scale_names, &scale.name) { + problems.push(format!("duplicate rating scale name '{}'", scale.name)); + } + validate_scale(&mut problems, scale); + } + let mut codes = HashSet::new(); + for modifier in &content.rating_modifiers { + if blank(&modifier.code) || blank(&modifier.label) { + problems.push("rating modifier has an empty code or label".to_owned()); + } else if note_duplicate(&mut codes, &modifier.code) { + problems.push(format!( + "duplicate rating modifier code '{}'", + modifier.code + )); + } + } + validate_forms(&mut problems, content); + validate_citations(&mut problems, "version", &content.citations); + problems +} diff --git a/crates/consolebook-server/src/programs/persistence.rs b/crates/consolebook-server/src/programs/persistence.rs new file mode 100644 index 0000000..8b1d564 --- /dev/null +++ b/crates/consolebook-server/src/programs/persistence.rs @@ -0,0 +1,679 @@ +//! Program content persistence and transaction-owned inserts. + +use super::content::{ + AnchorDef, CitationDef, CompetencyDef, FormCompetencyDef, FormDef, ModifierDef, NarrativeDef, + PhaseDef, PolicyDef, RecordType, ScaleDef, ScaleKind, TaskDef, TransitionDef, TransitionKind, + VersionContent, +}; +use crate::audit::{self, EventKind, Subject}; +use std::collections::HashMap; + +use anyhow::{Context, Result, bail}; +use sqlx::{Row, SqliteConnection, SqlitePool}; +use time::OffsetDateTime; + +pub(crate) async fn insert_program( + tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>, + name: &str, + actor_user_id: i64, +) -> Result { + let result = + sqlx::query("INSERT INTO program (name, created_at, created_by) VALUES (?1, ?2, ?3)") + .bind(name) + .bind(OffsetDateTime::now_utc().unix_timestamp()) + .bind(actor_user_id) + .execute(&mut **tx) + .await + .context("creating program")?; + let program_id = result.last_insert_rowid(); + audit::record_for_subject( + &mut **tx, + EventKind::ProgramCreated, + Some(actor_user_id), + None, + Subject::Program(program_id), + ) + .await?; + Ok(program_id) +} + +/// Inserts a draft version row plus its content and records the lifecycle +/// audit event. Callers have already validated the content. +pub(crate) async fn insert_version( + tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>, + program_id: i64, + content: &VersionContent, + actor_user_id: i64, + kind: EventKind, +) -> Result { + let next_number: i64 = sqlx::query_scalar( + "SELECT COALESCE(MAX(version_number), 0) + 1 FROM program_version WHERE program_id = ?1", + ) + .bind(program_id) + .fetch_one(&mut **tx) + .await + .context("numbering version")?; + let result = sqlx::query( + "INSERT INTO program_version + (program_id, version_number, label, name, description, created_at, created_by) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)", + ) + .bind(program_id) + .bind(next_number) + .bind(&content.label) + .bind(&content.name) + .bind(&content.description) + .bind(OffsetDateTime::now_utc().unix_timestamp()) + .bind(actor_user_id) + .execute(&mut **tx) + .await + .context("creating program version")?; + let version_id = result.last_insert_rowid(); + insert_content(tx, version_id, content).await?; + audit::record_for_subject( + &mut **tx, + kind, + Some(actor_user_id), + None, + Subject::ProgramVersion(version_id), + ) + .await?; + Ok(version_id) +} + +// ---- content writes + +async fn insert_citation( + conn: &mut SqliteConnection, + version_id: i64, + competency_id: Option, + task_id: Option, + citation: &CitationDef, +) -> Result<()> { + sqlx::query( + "INSERT INTO standards_citation + (program_version_id, competency_id, task_id, body, edition, clause, note) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)", + ) + .bind(version_id) + .bind(competency_id) + .bind(task_id) + .bind(&citation.body) + .bind(&citation.edition) + .bind(&citation.clause) + .bind(&citation.note) + .execute(conn) + .await + .context("inserting standards citation")?; + Ok(()) +} + +async fn insert_phases( + conn: &mut SqliteConnection, + version_id: i64, + content: &VersionContent, +) -> Result<()> { + let mut phase_ids: HashMap<&str, i64> = HashMap::new(); + for phase in &content.phases { + let result = sqlx::query( + "INSERT INTO phase (program_version_id, name, description, presentation_number) + VALUES (?1, ?2, ?3, ?4)", + ) + .bind(version_id) + .bind(&phase.name) + .bind(&phase.description) + .bind(phase.presentation_number) + .execute(&mut *conn) + .await + .context("inserting phase")?; + phase_ids.insert(phase.name.as_str(), result.last_insert_rowid()); + } + for transition in &content.phase_transitions { + sqlx::query( + "INSERT INTO phase_transition + (program_version_id, from_phase_id, to_phase_id, kind) + VALUES (?1, ?2, ?3, ?4)", + ) + .bind(version_id) + .bind(phase_ids[transition.from_phase.as_str()]) + .bind(phase_ids[transition.to_phase.as_str()]) + .bind(transition.kind.as_str()) + .execute(&mut *conn) + .await + .context("inserting phase transition")?; + } + Ok(()) +} + +async fn insert_competencies( + conn: &mut SqliteConnection, + version_id: i64, + content: &VersionContent, +) -> Result> { + let mut competency_ids: HashMap = HashMap::new(); + for (order, competency) in (0_i64..).zip(&content.competencies) { + let result = sqlx::query( + "INSERT INTO competency (program_version_id, category, name, description, sort_order) + VALUES (?1, ?2, ?3, ?4, ?5)", + ) + .bind(version_id) + .bind(&competency.category) + .bind(&competency.name) + .bind(&competency.description) + .bind(order) + .execute(&mut *conn) + .await + .context("inserting competency")?; + let competency_id = result.last_insert_rowid(); + competency_ids.insert(competency.name.clone(), competency_id); + for (task_order, task) in (0_i64..).zip(&competency.tasks) { + let inserted = sqlx::query( + "INSERT INTO task (program_version_id, competency_id, prompt, sort_order) + VALUES (?1, ?2, ?3, ?4)", + ) + .bind(version_id) + .bind(competency_id) + .bind(&task.prompt) + .bind(task_order) + .execute(&mut *conn) + .await + .context("inserting task")?; + let task_id = inserted.last_insert_rowid(); + for citation in &task.citations { + insert_citation(&mut *conn, version_id, None, Some(task_id), citation).await?; + } + } + for citation in &competency.citations { + insert_citation(&mut *conn, version_id, Some(competency_id), None, citation).await?; + } + } + Ok(competency_ids) +} + +async fn insert_scales( + conn: &mut SqliteConnection, + version_id: i64, + content: &VersionContent, +) -> Result> { + let mut scale_ids: HashMap = HashMap::new(); + for scale in &content.rating_scales { + let result = sqlx::query( + "INSERT INTO rating_scale (program_version_id, name, kind, min_value, max_value) + VALUES (?1, ?2, ?3, ?4, ?5)", + ) + .bind(version_id) + .bind(&scale.name) + .bind(scale.kind.as_str()) + .bind(scale.min_value) + .bind(scale.max_value) + .execute(&mut *conn) + .await + .context("inserting rating scale")?; + let scale_id = result.last_insert_rowid(); + scale_ids.insert(scale.name.clone(), scale_id); + for anchor in &scale.anchors { + sqlx::query( + "INSERT INTO rating_anchor + (program_version_id, rating_scale_id, value, label, definition) + VALUES (?1, ?2, ?3, ?4, ?5)", + ) + .bind(version_id) + .bind(scale_id) + .bind(anchor.value) + .bind(&anchor.label) + .bind(&anchor.definition) + .execute(&mut *conn) + .await + .context("inserting rating anchor")?; + } + } + for modifier in &content.rating_modifiers { + sqlx::query( + "INSERT INTO rating_modifier (program_version_id, code, label, description) + VALUES (?1, ?2, ?3, ?4)", + ) + .bind(version_id) + .bind(&modifier.code) + .bind(&modifier.label) + .bind(&modifier.description) + .execute(&mut *conn) + .await + .context("inserting rating modifier")?; + } + Ok(scale_ids) +} + +async fn insert_forms( + conn: &mut SqliteConnection, + version_id: i64, + content: &VersionContent, + competency_ids: &HashMap, + scale_ids: &HashMap, +) -> Result<()> { + for form in &content.evaluation_forms { + let result = sqlx::query( + "INSERT INTO evaluation_form (program_version_id, record_type, name, instructions) + VALUES (?1, ?2, ?3, ?4)", + ) + .bind(version_id) + .bind(form.record_type.as_str()) + .bind(&form.name) + .bind(&form.instructions) + .execute(&mut *conn) + .await + .context("inserting evaluation form")?; + let form_id = result.last_insert_rowid(); + for (order, binding) in (0_i64..).zip(&form.competencies) { + sqlx::query( + "INSERT INTO form_competency + (program_version_id, evaluation_form_id, competency_id, rating_scale_id, sort_order) + VALUES (?1, ?2, ?3, ?4, ?5)", + ) + .bind(version_id) + .bind(form_id) + .bind(competency_ids[&binding.competency]) + .bind(scale_ids[&binding.rating_scale]) + .bind(order) + .execute(&mut *conn) + .await + .context("inserting form competency")?; + } + for (order, narrative) in (0_i64..).zip(&form.narratives) { + sqlx::query( + "INSERT INTO form_narrative + (program_version_id, evaluation_form_id, prompt, required, sort_order) + VALUES (?1, ?2, ?3, ?4, ?5)", + ) + .bind(version_id) + .bind(form_id) + .bind(&narrative.prompt) + .bind(i64::from(narrative.required)) + .bind(order) + .execute(&mut *conn) + .await + .context("inserting form narrative")?; + } + } + Ok(()) +} + +/// Writes every owned row of a validated content document. +pub(super) async fn insert_content( + conn: &mut SqliteConnection, + version_id: i64, + content: &VersionContent, +) -> Result<()> { + insert_phases(&mut *conn, version_id, content).await?; + let competency_ids = insert_competencies(&mut *conn, version_id, content).await?; + let scale_ids = insert_scales(&mut *conn, version_id, content).await?; + insert_forms(&mut *conn, version_id, content, &competency_ids, &scale_ids).await?; + for citation in &content.citations { + insert_citation(&mut *conn, version_id, None, None, citation).await?; + } + sqlx::query( + "INSERT INTO finalization_policy + (program_version_id, review_approved, required_narratives, ratings_complete) + VALUES (?1, ?2, ?3, ?4)", + ) + .bind(version_id) + .bind(i64::from(content.finalization_policy.review_approved)) + .bind(i64::from(content.finalization_policy.required_narratives)) + .bind(i64::from(content.finalization_policy.ratings_complete)) + .execute(&mut *conn) + .await + .context("writing finalization policy")?; + Ok(()) +} + +/// Deletes every owned row of a draft version, children before parents so +/// foreign keys hold throughout. +pub(super) async fn delete_content(conn: &mut SqliteConnection, version_id: i64) -> Result<()> { + for statement in [ + "DELETE FROM finalization_policy WHERE program_version_id = ?1", + "DELETE FROM standards_citation WHERE program_version_id = ?1", + "DELETE FROM form_narrative WHERE program_version_id = ?1", + "DELETE FROM form_competency WHERE program_version_id = ?1", + "DELETE FROM evaluation_form WHERE program_version_id = ?1", + "DELETE FROM rating_anchor WHERE program_version_id = ?1", + "DELETE FROM rating_scale WHERE program_version_id = ?1", + "DELETE FROM rating_modifier WHERE program_version_id = ?1", + "DELETE FROM task WHERE program_version_id = ?1", + "DELETE FROM competency WHERE program_version_id = ?1", + "DELETE FROM phase_transition WHERE program_version_id = ?1", + "DELETE FROM phase WHERE program_version_id = ?1", + ] { + sqlx::query(statement) + .bind(version_id) + .execute(&mut *conn) + .await + .context("deleting draft content")?; + } + Ok(()) +} + +// ---- content reads + +/// Loads a version's complete content document, or `None` when the +/// version does not exist. Arrays come back in the deterministic export +/// order (authored order where one exists, content order otherwise). +pub async fn load_content(pool: &SqlitePool, version_id: i64) -> Result> { + // One transaction so every query reads the same snapshot. + let mut tx = pool.begin().await.context("starting content load")?; + let Some(header) = + sqlx::query("SELECT name, label, description FROM program_version WHERE id = ?1") + .bind(version_id) + .fetch_optional(&mut *tx) + .await + .context("loading version row")? + else { + return Ok(None); + }; + let mut content = VersionContent { + name: header.get("name"), + label: header.get("label"), + description: header.get("description"), + phases: Vec::new(), + phase_transitions: Vec::new(), + competencies: Vec::new(), + rating_scales: Vec::new(), + rating_modifiers: Vec::new(), + evaluation_forms: Vec::new(), + citations: Vec::new(), + finalization_policy: PolicyDef::default(), + }; + if let Some(policy) = sqlx::query( + "SELECT review_approved, required_narratives, ratings_complete + FROM finalization_policy WHERE program_version_id = ?1", + ) + .bind(version_id) + .fetch_optional(&mut *tx) + .await + .context("loading finalization policy")? + { + content.finalization_policy = PolicyDef { + review_approved: policy.get::("review_approved") != 0, + required_narratives: policy.get::("required_narratives") != 0, + ratings_complete: policy.get::("ratings_complete") != 0, + }; + } + load_phases(&mut tx, version_id, &mut content).await?; + let competency_index = load_competencies(&mut tx, version_id, &mut content).await?; + load_scales(&mut tx, version_id, &mut content).await?; + load_forms(&mut tx, version_id, &mut content).await?; + load_citations(&mut tx, version_id, &mut content, &competency_index).await?; + Ok(Some(content)) +} + +async fn load_phases( + tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>, + version_id: i64, + content: &mut VersionContent, +) -> Result<()> { + let rows = sqlx::query( + "SELECT name, description, presentation_number FROM phase + WHERE program_version_id = ?1 ORDER BY presentation_number, name", + ) + .bind(version_id) + .fetch_all(&mut **tx) + .await + .context("loading phases")?; + content.phases = rows + .iter() + .map(|row| PhaseDef { + name: row.get("name"), + description: row.get("description"), + presentation_number: row.get("presentation_number"), + }) + .collect(); + let rows = sqlx::query( + "SELECT f.name AS from_name, t.name AS to_name, pt.kind + FROM phase_transition pt + JOIN phase f ON f.id = pt.from_phase_id + JOIN phase t ON t.id = pt.to_phase_id + WHERE pt.program_version_id = ?1 + ORDER BY f.name, t.name", + ) + .bind(version_id) + .fetch_all(&mut **tx) + .await + .context("loading phase transitions")?; + for row in &rows { + content.phase_transitions.push(TransitionDef { + from_phase: row.get("from_name"), + to_phase: row.get("to_name"), + kind: TransitionKind::from_db(row.get("kind"))?, + }); + } + Ok(()) +} + +/// Loads competencies and their tasks; returns row-id lookup maps used to +/// route citations. +async fn load_competencies( + tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>, + version_id: i64, + content: &mut VersionContent, +) -> Result { + let rows = sqlx::query( + "SELECT id, category, name, description FROM competency + WHERE program_version_id = ?1 ORDER BY sort_order, name", + ) + .bind(version_id) + .fetch_all(&mut **tx) + .await + .context("loading competencies")?; + let mut index = CompetencyIndex::default(); + for row in &rows { + let id: i64 = row.get("id"); + index + .by_competency_row + .insert(id, content.competencies.len()); + content.competencies.push(CompetencyDef { + category: row.get("category"), + name: row.get("name"), + description: row.get("description"), + tasks: Vec::new(), + citations: Vec::new(), + }); + } + let rows = sqlx::query( + "SELECT id, competency_id, prompt FROM task + WHERE program_version_id = ?1 ORDER BY sort_order, prompt", + ) + .bind(version_id) + .fetch_all(&mut **tx) + .await + .context("loading tasks")?; + for row in &rows { + let competency_row: i64 = row.get("competency_id"); + let competency_slot = index.by_competency_row[&competency_row]; + let tasks = &mut content.competencies[competency_slot].tasks; + index + .by_task_row + .insert(row.get("id"), (competency_slot, tasks.len())); + tasks.push(TaskDef { + prompt: row.get("prompt"), + citations: Vec::new(), + }); + } + Ok(index) +} + +#[derive(Default)] +struct CompetencyIndex { + /// competency row id -> index into `content.competencies` + by_competency_row: HashMap, + /// task row id -> (competency index, task index) + by_task_row: HashMap, +} + +async fn load_scales( + tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>, + version_id: i64, + content: &mut VersionContent, +) -> Result<()> { + let rows = sqlx::query( + "SELECT id, name, kind, min_value, max_value FROM rating_scale + WHERE program_version_id = ?1 ORDER BY name", + ) + .bind(version_id) + .fetch_all(&mut **tx) + .await + .context("loading rating scales")?; + let mut slot_by_row: HashMap = HashMap::new(); + for row in &rows { + slot_by_row.insert(row.get("id"), content.rating_scales.len()); + content.rating_scales.push(ScaleDef { + name: row.get("name"), + kind: ScaleKind::from_db(row.get("kind"))?, + min_value: row.get("min_value"), + max_value: row.get("max_value"), + anchors: Vec::new(), + }); + } + let rows = sqlx::query( + "SELECT rating_scale_id, value, label, definition FROM rating_anchor + WHERE program_version_id = ?1 ORDER BY value", + ) + .bind(version_id) + .fetch_all(&mut **tx) + .await + .context("loading rating anchors")?; + for row in &rows { + let scale_row: i64 = row.get("rating_scale_id"); + content.rating_scales[slot_by_row[&scale_row]] + .anchors + .push(AnchorDef { + value: row.get("value"), + label: row.get("label"), + definition: row.get("definition"), + }); + } + let rows = sqlx::query( + "SELECT code, label, description FROM rating_modifier + WHERE program_version_id = ?1 ORDER BY code", + ) + .bind(version_id) + .fetch_all(&mut **tx) + .await + .context("loading rating modifiers")?; + content.rating_modifiers = rows + .iter() + .map(|row| ModifierDef { + code: row.get("code"), + label: row.get("label"), + description: row.get("description"), + }) + .collect(); + Ok(()) +} + +async fn load_forms( + tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>, + version_id: i64, + content: &mut VersionContent, +) -> Result<()> { + let rows = sqlx::query( + "SELECT id, record_type, name, instructions FROM evaluation_form + WHERE program_version_id = ?1 ORDER BY name", + ) + .bind(version_id) + .fetch_all(&mut **tx) + .await + .context("loading evaluation forms")?; + let mut slot_by_row: HashMap = HashMap::new(); + for row in &rows { + slot_by_row.insert(row.get("id"), content.evaluation_forms.len()); + content.evaluation_forms.push(FormDef { + record_type: RecordType::from_db(row.get("record_type"))?, + name: row.get("name"), + instructions: row.get("instructions"), + competencies: Vec::new(), + narratives: Vec::new(), + }); + } + let rows = sqlx::query( + "SELECT fc.evaluation_form_id, c.name AS competency, s.name AS rating_scale + FROM form_competency fc + JOIN competency c ON c.id = fc.competency_id + JOIN rating_scale s ON s.id = fc.rating_scale_id + WHERE fc.program_version_id = ?1 + ORDER BY fc.sort_order, c.name", + ) + .bind(version_id) + .fetch_all(&mut **tx) + .await + .context("loading form competencies")?; + for row in &rows { + let form_row: i64 = row.get("evaluation_form_id"); + content.evaluation_forms[slot_by_row[&form_row]] + .competencies + .push(FormCompetencyDef { + competency: row.get("competency"), + rating_scale: row.get("rating_scale"), + }); + } + let rows = sqlx::query( + "SELECT evaluation_form_id, prompt, required FROM form_narrative + WHERE program_version_id = ?1 ORDER BY sort_order, prompt", + ) + .bind(version_id) + .fetch_all(&mut **tx) + .await + .context("loading form narratives")?; + for row in &rows { + let form_row: i64 = row.get("evaluation_form_id"); + let required: i64 = row.get("required"); + content.evaluation_forms[slot_by_row[&form_row]] + .narratives + .push(NarrativeDef { + prompt: row.get("prompt"), + required: required != 0, + }); + } + Ok(()) +} + +async fn load_citations( + tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>, + version_id: i64, + content: &mut VersionContent, + index: &CompetencyIndex, +) -> Result<()> { + let rows = sqlx::query( + "SELECT competency_id, task_id, body, edition, clause, note + FROM standards_citation WHERE program_version_id = ?1 + ORDER BY body, edition, clause, note", + ) + .bind(version_id) + .fetch_all(&mut **tx) + .await + .context("loading standards citations")?; + for row in &rows { + let citation = CitationDef { + body: row.get("body"), + edition: row.get("edition"), + clause: row.get("clause"), + note: row.get("note"), + }; + let competency_id: Option = row.get("competency_id"); + let task_id: Option = row.get("task_id"); + match (competency_id, task_id) { + (Some(competency_row), None) => { + let slot = index.by_competency_row[&competency_row]; + content.competencies[slot].citations.push(citation); + } + (None, Some(task_row)) => { + let (competency_slot, task_slot) = index.by_task_row[&task_row]; + content.competencies[competency_slot].tasks[task_slot] + .citations + .push(citation); + } + (None, None) => content.citations.push(citation), + (Some(_), Some(_)) => { + bail!("citation targets both a competency and a task; the schema forbids this") + } + } + } + Ok(()) +} diff --git a/crates/consolebook-server/src/session_membership.rs b/crates/consolebook-server/src/session_membership.rs index 202dc43..5deeb0c 100644 --- a/crates/consolebook-server/src/session_membership.rs +++ b/crates/consolebook-server/src/session_membership.rs @@ -13,6 +13,7 @@ use time::OffsetDateTime; use crate::audit::{self, EventKind, Subject}; use crate::capabilities::{self, Capability}; +use crate::storage; use crate::training_sessions::SessionRefusal; /// One trainer on a session. @@ -138,14 +139,16 @@ pub async fn add_trainer( if !may_work(pool, actor_user_id, session_id).await? { return Ok(Err(SessionRefusal::CapabilityRequired)); } - let mut tx = pool.begin().await.context("starting trainer add")?; + let mut tx = storage::write_tx(pool) + .await + .context("starting trainer add")?; let exists: Option = sqlx::query_scalar("SELECT 1 FROM training_session WHERE id = ?1") .bind(session_id) .fetch_optional(&mut *tx) .await .context("checking session")?; if exists.is_none() { - return Ok(Err(SessionRefusal::NoSuchSession)); + return storage::refuse(tx, SessionRefusal::NoSuchSession).await; } let user_exists: Option = sqlx::query_scalar("SELECT 1 FROM user WHERE id = ?1") .bind(trainer_user_id) @@ -153,7 +156,7 @@ pub async fn add_trainer( .await .context("checking trainer")?; if user_exists.is_none() { - return Ok(Err(SessionRefusal::NoSuchUser)); + return storage::refuse(tx, SessionRefusal::NoSuchUser).await; } let can_author: Option = sqlx::query_scalar("SELECT 1 FROM capability_grant WHERE user_id = ?1 AND capability = ?2") @@ -163,7 +166,7 @@ pub async fn add_trainer( .await .context("checking trainer capability")?; if can_author.is_none() { - return Ok(Err(SessionRefusal::TrainerLacksCapability)); + return storage::refuse(tx, SessionRefusal::TrainerLacksCapability).await; } let member: Option = sqlx::query_scalar( "SELECT 1 FROM session_trainer WHERE session_id = ?1 AND trainer_user_id = ?2", @@ -174,7 +177,7 @@ pub async fn add_trainer( .await .context("checking membership")?; if member.is_some() { - return Ok(Err(SessionRefusal::AlreadyMember)); + return storage::refuse(tx, SessionRefusal::AlreadyMember).await; } let now = OffsetDateTime::now_utc().unix_timestamp(); insert_member(&mut tx, session_id, trainer_user_id, actor_user_id, now).await?; @@ -193,7 +196,9 @@ pub async fn remove_trainer( if !capabilities::user_has(pool, actor_user_id, Capability::AssignTraining).await? { return Ok(Err(SessionRefusal::CapabilityRequired)); } - let mut tx = pool.begin().await.context("starting trainer removal")?; + let mut tx = storage::write_tx(pool) + .await + .context("starting trainer removal")?; let member: Option = sqlx::query_scalar( "SELECT 1 FROM session_trainer WHERE session_id = ?1 AND trainer_user_id = ?2", ) @@ -203,7 +208,7 @@ pub async fn remove_trainer( .await .context("checking membership")?; if member.is_none() { - return Ok(Err(SessionRefusal::NotMember)); + return storage::refuse(tx, SessionRefusal::NotMember).await; } let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM session_trainer WHERE session_id = ?1") @@ -212,7 +217,7 @@ pub async fn remove_trainer( .await .context("counting trainers")?; if count <= 1 { - return Ok(Err(SessionRefusal::LastTrainer)); + return storage::refuse(tx, SessionRefusal::LastTrainer).await; } sqlx::query("DELETE FROM session_trainer WHERE session_id = ?1 AND trainer_user_id = ?2") .bind(session_id) diff --git a/crates/consolebook-server/src/setup.rs b/crates/consolebook-server/src/setup.rs index 796d300..923c2d0 100644 --- a/crates/consolebook-server/src/setup.rs +++ b/crates/consolebook-server/src/setup.rs @@ -12,6 +12,7 @@ use time::OffsetDateTime; use crate::audit::{self, EventKind}; use crate::capabilities::{self, ADMINISTRATOR_BUNDLE}; use crate::secrets::{self, OpaqueSecret}; +use crate::storage; use crate::users; /// Whether the installation has completed first-run setup. @@ -32,10 +33,17 @@ pub async fn agency_name(pool: &SqlitePool) -> Result> { /// initialized. The raw code is shown once (server log or command output) /// and only its digest is stored. pub async fn issue_setup_code(pool: &SqlitePool) -> Result> { - if is_initialized(pool).await? { + let code = secrets::generate_one_time_code()?; + let mut tx = storage::write_tx(pool).await?; + let initialized: Option = sqlx::query_scalar("SELECT 1 FROM agency WHERE id = 1") + .fetch_optional(&mut *tx) + .await?; + if initialized.is_some() { + tx.rollback() + .await + .context("rolling back setup-code refusal")?; return Ok(None); } - let code = secrets::generate_one_time_code()?; let expires_at = OffsetDateTime::now_utc().unix_timestamp() + users::CODE_TTL_SECONDS; sqlx::query( "INSERT INTO setup_code (id, code_hash, expires_at) VALUES (1, ?1, ?2) @@ -43,9 +51,10 @@ pub async fn issue_setup_code(pool: &SqlitePool) -> Result = sqlx::query_scalar("SELECT 1 FROM agency WHERE id = 1") .fetch_optional(&mut *tx) .await?; if initialized.is_some() { - return Ok(Err(SetupRefusal::AlreadyInitialized)); + return storage::refuse(tx, SetupRefusal::AlreadyInitialized).await; } let valid: Option = sqlx::query_scalar( "SELECT 1 FROM setup_code WHERE id = 1 AND code_hash = ?1 AND expires_at > ?2", @@ -104,7 +113,7 @@ pub async fn initialize( .fetch_optional(&mut *tx) .await?; if valid.is_none() { - return Ok(Err(SetupRefusal::InvalidCode)); + return storage::refuse(tx, SetupRefusal::InvalidCode).await; } sqlx::query("INSERT INTO agency (id, name, created_at) VALUES (1, ?1, ?2)") diff --git a/crates/consolebook-server/src/storage.rs b/crates/consolebook-server/src/storage.rs index bffe087..501daac 100644 --- a/crates/consolebook-server/src/storage.rs +++ b/crates/consolebook-server/src/storage.rs @@ -20,23 +20,17 @@ use uuid::Uuid; /// Busy timeout applied to every connection. pub const BUSY_TIMEOUT: Duration = Duration::from_secs(5); -/// Begins an immediate (write) transaction: the write lock is taken up -/// front, so a check-then-write path validates against the committed -/// state and a concurrent writer waits out the busy timeout instead of -/// failing its read snapshot mid-transaction — typed refusals stay typed -/// under concurrency. New write paths use this; #27 tracks retrofitting -/// the earlier deferred ones. +/// Begins an immediate write transaction before transactional validation. +/// All application-owned write transactions use this reservation (ADR 0019). +/// A concurrent writer waits up to the connection's busy timeout, then checks +/// committed state; exceeding that timeout remains an operational error. pub async fn write_tx(pool: &SqlitePool) -> sqlx::Result> { pool.begin_with("BEGIN IMMEDIATE").await } -/// Ends a write transaction on a refusal path with the rollback awaited, -/// so the write lock never outlives the decision. A dropped transaction -/// only queues its rollback on the connection's worker thread; until that -/// runs, the lock lingers — and a deferred writer elsewhere meets it as -/// an immediate `SQLITE_BUSY`, because `SQLite` does not consult the busy -/// timeout when promoting an open read transaction to a write (#27 tracks -/// converting those deferred paths themselves). +/// Ends a write transaction on a typed refusal with rollback awaited, so +/// the write lock is released before the refusal returns. Dropping a `SQLx` +/// transaction only queues rollback on its connection's worker (ADR 0019). pub async fn refuse( tx: sqlx::Transaction<'static, sqlx::Sqlite>, refusal: E, diff --git a/crates/consolebook-server/src/training_sessions.rs b/crates/consolebook-server/src/training_sessions.rs index cc08850..81abb16 100644 --- a/crates/consolebook-server/src/training_sessions.rs +++ b/crates/consolebook-server/src/training_sessions.rs @@ -25,6 +25,7 @@ use crate::capabilities::{self, Capability}; use crate::lifecycle::{self, EnrollmentStatus}; use crate::session_membership::{self, SessionTrainerRow}; use crate::session_time::{self, TimeRefusal}; +use crate::storage; /// Session dispositions: a closed set, like scale kinds (ADR 0007's /// pattern). Completed and interrupted training occupied their interval; @@ -261,12 +262,12 @@ pub async fn create( Err(refusal) => return Ok(Err(refusal.into())), }; - let mut tx = pool.begin().await.context("starting session")?; + let mut tx = storage::write_tx(pool).await.context("starting session")?; let Some(status) = lifecycle::status(&mut tx, enrollment_id).await? else { - return Ok(Err(SessionRefusal::NoSuchEnrollment)); + return storage::refuse(tx, SessionRefusal::NoSuchEnrollment).await; }; if status != EnrollmentStatus::Active { - return Ok(Err(SessionRefusal::EnrollmentInactive)); + return storage::refuse(tx, SessionRefusal::EnrollmentInactive).await; } // The session stamps the pin at creation (migration 0007), so its // program and phase context stay historic across version changes. @@ -284,7 +285,7 @@ pub async fn create( .await .context("checking session phase")?; if in_version.is_none() { - return Ok(Err(SessionRefusal::NoSuchPhase)); + return storage::refuse(tx, SessionRefusal::NoSuchPhase).await; } } let trainers = match session_membership::validate_trainers( @@ -295,10 +296,10 @@ pub async fn create( .await? { Ok(trainers) => trainers, - Err(refusal) => return Ok(Err(refusal)), + Err(refusal) => return storage::refuse(tx, refusal).await, }; if overlaps(&mut tx, enrollment_id, times.utc_start, times.utc_end, 0).await? { - return Ok(Err(SessionRefusal::Overlap)); + return storage::refuse(tx, SessionRefusal::Overlap).await; } let now = OffsetDateTime::now_utc().unix_timestamp(); @@ -386,7 +387,9 @@ pub async fn update_open( Ok(times) => times, Err(refusal) => return Ok(Err(refusal.into())), }; - let mut tx = pool.begin().await.context("starting session update")?; + let mut tx = storage::write_tx(pool) + .await + .context("starting session update")?; let Some(row) = sqlx::query( "SELECT enrollment_id, disposition, phase_id FROM training_session WHERE id = ?1", ) @@ -395,11 +398,11 @@ pub async fn update_open( .await .context("reading session")? else { - return Ok(Err(SessionRefusal::NoSuchSession)); + return storage::refuse(tx, SessionRefusal::NoSuchSession).await; }; let disposition: Option = row.get("disposition"); if disposition.is_some() { - return Ok(Err(SessionRefusal::SessionClosed)); + return storage::refuse(tx, SessionRefusal::SessionClosed).await; } let enrollment_id: i64 = row.get("enrollment_id"); // A session's phase context comes from the version it was recorded @@ -420,11 +423,11 @@ pub async fn update_open( .await .context("checking session phase")?; if in_version.is_none() { - return Ok(Err(SessionRefusal::NoSuchPhase)); + return storage::refuse(tx, SessionRefusal::NoSuchPhase).await; } } if overlaps(&mut tx, enrollment_id, times.utc_start, None, session_id).await? { - return Ok(Err(SessionRefusal::Overlap)); + return storage::refuse(tx, SessionRefusal::Overlap).await; } sqlx::query( "UPDATE training_session @@ -466,7 +469,9 @@ pub async fn close( if !session_membership::may_work(pool, actor_user_id, session_id).await? { return Ok(Err(SessionRefusal::CapabilityRequired)); } - let mut tx = pool.begin().await.context("starting session close")?; + let mut tx = storage::write_tx(pool) + .await + .context("starting session close")?; let Some(row) = sqlx::query("SELECT timezone, utc_start, disposition FROM training_session WHERE id = ?1") .bind(session_id) @@ -474,11 +479,11 @@ pub async fn close( .await .context("reading session")? else { - return Ok(Err(SessionRefusal::NoSuchSession)); + return storage::refuse(tx, SessionRefusal::NoSuchSession).await; }; let already: Option = row.get("disposition"); if already.is_some() { - return Ok(Err(SessionRefusal::SessionClosed)); + return storage::refuse(tx, SessionRefusal::SessionClosed).await; } let local_end = local_end.map(str::trim).filter(|value| !value.is_empty()); @@ -494,22 +499,22 @@ pub async fn close( .await .context("checking coverage")?; if covered.is_some() { - return Ok(Err(SessionRefusal::SessionDocumented)); + return storage::refuse(tx, SessionRefusal::SessionDocumented).await; } if local_end.is_some() { - return Ok(Err(SessionRefusal::EndNotAllowed)); + return storage::refuse(tx, SessionRefusal::EndNotAllowed).await; } (None, None) } Disposition::Completed | Disposition::Interrupted => { let Some(value) = local_end else { - return Ok(Err(SessionRefusal::EndRequired)); + return storage::refuse(tx, SessionRefusal::EndRequired).await; }; let timezone: String = row.get("timezone"); let utc_start: i64 = row.get("utc_start"); match session_time::resolve_end(&timezone, value, utc_start) { Ok(instant) => (Some(value.to_owned()), Some(instant)), - Err(refusal) => return Ok(Err(refusal.into())), + Err(refusal) => return storage::refuse(tx, refusal.into()).await, } } }; diff --git a/crates/consolebook-server/src/users.rs b/crates/consolebook-server/src/users.rs index 25f5ab3..40dcdbd 100644 --- a/crates/consolebook-server/src/users.rs +++ b/crates/consolebook-server/src/users.rs @@ -12,6 +12,7 @@ use crate::audit::{self, EventKind}; use crate::capabilities::{self, Capability}; use crate::secrets::{self, OpaqueSecret}; use crate::sessions; +use crate::storage; /// Lifetime of setup and password-reset codes. pub const CODE_TTL_SECONDS: i64 = 15 * 60; @@ -197,7 +198,18 @@ pub async fn create_with_reset_code( let unusable = secrets::generate_one_time_code()?; let password_hash = secrets::hash_password(&unusable.raw)?; - let mut tx = pool.begin().await?; + let mut tx = storage::write_tx(pool).await?; + // The early lookup saves hashing for an existing name; only this + // reserved-transaction check can decide uniqueness under concurrency. + let taken: Option = + sqlx::query_scalar("SELECT 1 FROM user WHERE username = ?1 COLLATE NOCASE") + .bind(username) + .fetch_optional(&mut *tx) + .await + .context("checking username in write transaction")?; + if taken.is_some() { + return storage::refuse(tx, CreateUserRefusal::UsernameTaken).await; + } let user_id = create( &mut tx, username, @@ -291,7 +303,7 @@ pub async fn issue_reset_code( ResetOrigin::Recovery => ("recovery", None, EventKind::RecoveryCodeIssued), }; - let mut tx = pool.begin().await?; + let mut tx = storage::write_tx(pool).await?; sqlx::query( "INSERT INTO password_reset_code (user_id, code_hash, issued_via, issued_by, issued_at, expires_at) @@ -346,7 +358,7 @@ pub async fn use_reset_code( // transaction open. let password_hash = secrets::hash_password(new_password)?; - let mut tx = pool.begin().await?; + let mut tx = storage::write_tx(pool).await?; let code_id: Option = sqlx::query_scalar( "SELECT id FROM password_reset_code WHERE user_id = ?1 AND code_hash = ?2 AND used_at IS NULL AND expires_at > ?3", @@ -358,6 +370,7 @@ pub async fn use_reset_code( .await .context("looking up reset code")?; let Some(code_id) = code_id else { + tx.rollback().await.context("rolling back invalid reset")?; return Ok(ResetOutcome::Invalid); }; sqlx::query("UPDATE password_reset_code SET used_at = ?1 WHERE id = ?2") diff --git a/crates/consolebook-server/tests/write_transactions.rs b/crates/consolebook-server/tests/write_transactions.rs new file mode 100644 index 0000000..6d4fdd0 --- /dev/null +++ b/crates/consolebook-server/tests/write_transactions.rs @@ -0,0 +1,183 @@ +//! Contended writes must validate after reserving SQLite's writer. All data +//! is invented. Child modules own program, training, and account scenarios. + +use std::{fmt::Debug, future::Future, time::Duration}; + +use consolebook_server::{capabilities, programs, storage, users}; +use sqlx::{ConnectOptions, Connection, SqlitePool, sqlite::SqliteConnectOptions}; + +#[path = "write_transactions/accounts.rs"] +mod account_writes; +#[path = "write_transactions/programs.rs"] +mod program_writes; +#[path = "write_transactions/training.rs"] +mod training_writes; + +const ACTOR: i64 = 1; +const TRAINEE: i64 = 2; +const TRAINER: i64 = 3; +const OTHER_TRAINER: i64 = 4; +const PASSWORD: &str = "invented-passphrase-1"; + +struct Fixture { + tmp: tempfile::TempDir, + pool: SqlitePool, +} + +impl Fixture { + async fn empty() -> Self { + let tmp = tempfile::tempdir().expect("scratch"); + let pool = storage::open(&tmp.path().join("consolebook.db")) + .await + .expect("open"); + Self { tmp, pool } + } + + async fn new() -> Self { + let fx = Self::empty().await; + let mut tx = storage::write_tx(&fx.pool).await.expect("seed transaction"); + for (name, bundle) in [ + ("avery.admin", capabilities::ADMINISTRATOR_BUNDLE.as_slice()), + ("taylor.trainee", capabilities::TRAINEE_BUNDLE.as_slice()), + ("jordan.trainer", capabilities::TRAINER_BUNDLE.as_slice()), + ("rowan.trainer", capabilities::TRAINER_BUNDLE.as_slice()), + ] { + let id = users::create(&mut tx, name, name, "", "", "unused-invented-hash") + .await + .expect("seed user"); + capabilities::grant_bundle(&mut tx, id, bundle, None) + .await + .expect("grants"); + } + tx.commit().await.expect("seed commit"); + fx + } + + async fn draft(&self) -> (i64, i64) { + let program = programs::create_program(&self.pool, ACTOR, "Invented County Program") + .await + .expect("call") + .expect("program"); + let version = programs::create_version(&self.pool, ACTOR, program, &content()) + .await + .expect("call") + .expect("version"); + (program, version) + } + + async fn published(&self) -> (i64, i64) { + let (program, version) = self.draft().await; + programs::publish_version(&self.pool, ACTOR, version) + .await + .expect("call") + .expect("publish"); + (program, version) + } + + async fn probe(&self) { + let mut conn = SqliteConnectOptions::new() + .filename(self.tmp.path().join("consolebook.db")) + .busy_timeout(Duration::ZERO) + .connect() + .await + .expect("probe connection"); + sqlx::query("BEGIN IMMEDIATE") + .execute(&mut conn) + .await + .expect("refusal released write lock"); + sqlx::query("ROLLBACK") + .execute(&mut conn) + .await + .expect("probe rollback"); + conn.close().await.expect("probe close"); + } +} + +fn content() -> programs::VersionContent { + serde_json::from_value(serde_json::json!({ + "name": "Invented County Program", "label": "rev A", "description": "Invented fixture", + "phases": [{"name": "Phase One", "description": "", "presentation_number": 1}], + "phase_transitions": [], "competencies": [], "rating_scales": [], + "rating_modifiers": [], "evaluation_forms": [{ + "record_type": "daily_report", "name": "Invented Daily", "instructions": "", + "competencies": [], "narratives": [{"prompt": "Invented observations", "required": false}] + }], "citations": [] + })) + .expect("content") +} + +/// Force contention before allowing either service to commit. Both futures +/// are polled while a separate connection owns the write reservation. Deferred +/// check-then-write paths fail on promotion during this interval; immediate +/// writers wait, then validate serially. No production test hooks are needed. +fn contend( + pool: &SqlitePool, + left: A, + right: B, +) -> impl Future +where + A: Future, + B: Future, +{ + let (left, right) = (Box::pin(left), Box::pin(right)); + async move { + let blocker = storage::write_tx(pool).await.expect("hold writer"); + let pair = async { tokio::join!(left, right) }; + tokio::pin!(pair); + assert!( + tokio::time::timeout(Duration::from_millis(200), &mut pair) + .await + .is_err(), + "competing writers must wait for the reservation, not fail on read promotion" + ); + blocker.rollback().await.expect("release writer"); + tokio::time::timeout(Duration::from_secs(10), pair) + .await + .expect("bounded completion") + } +} + +type OutcomePair = (anyhow::Result>, anyhow::Result>); + +fn one_winner(pair: OutcomePair, refusal: &E) -> T { + let left = pair + .0 + .expect("left returns a domain outcome, never an internal error"); + let right = pair + .1 + .expect("right returns a domain outcome, never an internal error"); + match (left, right) { + (Ok(value), Err(error)) | (Err(error), Ok(value)) => { + assert_eq!(&error, refusal); + value + } + outcomes => panic!("expected one success and one refusal: {outcomes:?}"), + } +} + +#[tokio::test] +async fn wal_deferred_snapshot_reproduction() { + let fx = Fixture::new().await; + let mut stale = fx.pool.begin().await.expect("deferred reader"); + let _: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM program") + .fetch_one(&mut *stale) + .await + .expect("snapshot"); + programs::create_program(&fx.pool, ACTOR, "Concurrent invented program") + .await + .expect("call") + .expect("winner"); + let err = + sqlx::query("INSERT INTO program (name, created_at) VALUES ('Stale invented program', 1)") + .execute(&mut *stale) + .await + .expect_err("stale snapshot cannot promote"); + assert_eq!( + err.as_database_error() + .expect("SQLite error") + .code() + .as_deref(), + Some("517") + ); + stale.rollback().await.expect("rollback stale reader"); +} diff --git a/crates/consolebook-server/tests/write_transactions/accounts.rs b/crates/consolebook-server/tests/write_transactions/accounts.rs new file mode 100644 index 0000000..db94c64 --- /dev/null +++ b/crates/consolebook-server/tests/write_transactions/accounts.rs @@ -0,0 +1,210 @@ +use super::*; +use consolebook_server::{sessions, setup}; +use users::{CreateUserRefusal, ResetOrigin, ResetOutcome}; + +#[tokio::test] +async fn duplicate_usernames_are_rechecked_after_hashing() { + let fx = Fixture::new().await; + let created = one_winner( + contend( + &fx.pool, + users::create_with_reset_code( + &fx.pool, + ACTOR, + "casey.example", + "Casey Example", + "", + "", + capabilities::RoleBundle::Trainee, + ), + users::create_with_reset_code( + &fx.pool, + ACTOR, + "CASEY.EXAMPLE", + "Casey Example", + "", + "", + capabilities::RoleBundle::Trainee, + ), + ) + .await, + &CreateUserRefusal::UsernameTaken, + ); + let count: i64 = + sqlx::query_scalar("SELECT COUNT(*) FROM password_reset_code WHERE user_id = ?1") + .bind(created.id) + .fetch_one(&fx.pool) + .await + .expect("initial code"); + assert_eq!(count, 1); + fx.probe().await; +} + +#[tokio::test] +async fn reset_issuance_serializes_without_discarding_valid_codes() { + let fx = Fixture::new().await; + let (left, right) = contend( + &fx.pool, + users::issue_reset_code( + &fx.pool, + "taylor.trainee", + ResetOrigin::Administrator { issued_by: ACTOR }, + ), + users::issue_reset_code( + &fx.pool, + "taylor.trainee", + ResetOrigin::Administrator { issued_by: ACTOR }, + ), + ) + .await; + let left = left.expect("left").expect("issued"); + let right = right.expect("right").expect("issued"); + assert_ne!(left.code.digest_hex, right.code.digest_hex); + let count: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM password_reset_code WHERE user_id = ?1 AND used_at IS NULL", + ) + .bind(TRAINEE) + .fetch_one(&fx.pool) + .await + .expect("codes"); + assert_eq!( + count, 2, + "issuance permits multiple independent codes as before" + ); +} + +#[tokio::test] +async fn reset_code_is_consumed_once_and_revokes_sessions() { + let fx = Fixture::new().await; + let issued = users::issue_reset_code( + &fx.pool, + "taylor.trainee", + ResetOrigin::Administrator { issued_by: ACTOR }, + ) + .await + .expect("call") + .expect("issued"); + let token = sessions::create(&fx.pool, TRAINEE) + .await + .expect("session") + .0; + let (left, right) = contend( + &fx.pool, + users::use_reset_code(&fx.pool, "taylor.trainee", &issued.code.raw, PASSWORD), + users::use_reset_code(&fx.pool, "taylor.trainee", &issued.code.raw, PASSWORD), + ) + .await; + let outcomes = (left.expect("left"), right.expect("right")); + assert!( + matches!( + outcomes, + (ResetOutcome::Done, ResetOutcome::Invalid) + | (ResetOutcome::Invalid, ResetOutcome::Done) + ), + "{outcomes:?}" + ); + assert!( + sessions::validate(&fx.pool, &token.raw) + .await + .expect("validate") + .is_none() + ); + let count: i64 = + sqlx::query_scalar("SELECT COUNT(*) FROM audit_event WHERE kind = 'reset_code_used'") + .fetch_one(&fx.pool) + .await + .expect("audit"); + assert_eq!(count, 1); + fx.probe().await; +} + +#[tokio::test] +async fn setup_has_one_administrator_and_consumes_its_code_once() { + let fx = Fixture::empty().await; + let code = setup::issue_setup_code(&fx.pool) + .await + .expect("issue") + .expect("uninitialized") + .0; + one_winner( + contend( + &fx.pool, + setup::initialize( + &fx.pool, + &code.raw, + "Invented County", + "avery.admin", + "Avery Admin", + PASSWORD, + ), + setup::initialize( + &fx.pool, + &code.raw, + "Invented County", + "rowan.admin", + "Rowan Admin", + PASSWORD, + ), + ) + .await, + &setup::SetupRefusal::AlreadyInitialized, + ); + let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM user") + .fetch_one(&fx.pool) + .await + .expect("users"); + assert_eq!(count, 1); + assert!( + setup::issue_setup_code(&fx.pool) + .await + .expect("issue after setup") + .is_none() + ); + fx.probe().await; +} + +#[tokio::test] +async fn setup_code_rotation_serializes_with_initialization() { + let fx = Fixture::empty().await; + let code = setup::issue_setup_code(&fx.pool) + .await + .expect("issue") + .expect("uninitialized") + .0; + let (initialized, rotated) = contend( + &fx.pool, + setup::initialize( + &fx.pool, + &code.raw, + "Invented County", + "avery.admin", + "Avery Admin", + PASSWORD, + ), + setup::issue_setup_code(&fx.pool), + ) + .await; + match (initialized.expect("initialize"), rotated.expect("rotate")) { + (Ok(_), None) => assert!(setup::is_initialized(&fx.pool).await.expect("initialized")), + (Err(setup::SetupRefusal::InvalidCode), Some((replacement, _))) => { + setup::initialize( + &fx.pool, + &replacement.raw, + "Invented County", + "avery.admin", + "Avery Admin", + PASSWORD, + ) + .await + .expect("call") + .expect("replacement accepted"); + } + _ => panic!("rotation either precedes setup or is refused after setup"), + } + let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM setup_code") + .fetch_one(&fx.pool) + .await + .expect("codes"); + assert_eq!(count, 0, "no setup code survives initialization"); + fx.probe().await; +} diff --git a/crates/consolebook-server/tests/write_transactions/programs.rs b/crates/consolebook-server/tests/write_transactions/programs.rs new file mode 100644 index 0000000..879534b --- /dev/null +++ b/crates/consolebook-server/tests/write_transactions/programs.rs @@ -0,0 +1,172 @@ +use super::*; +use consolebook_server::program_export::{self, ImportRefusal, ImportTarget}; +use programs::{AuthorRefusal, ProgramRefusal, PublishRefusal}; + +#[tokio::test] +async fn duplicate_program_name_has_one_winner() { + let fx = Fixture::new().await; + one_winner( + contend( + &fx.pool, + programs::create_program(&fx.pool, ACTOR, "Invented Program"), + programs::create_program(&fx.pool, ACTOR, "INVENTED PROGRAM"), + ) + .await, + &ProgramRefusal::NameTaken, + ); + assert_eq!( + programs::list_programs(&fx.pool) + .await + .expect("programs") + .len(), + 1 + ); + fx.probe().await; +} + +#[tokio::test] +async fn version_creation_assigns_distinct_monotonic_numbers() { + let fx = Fixture::new().await; + let (program, _) = fx.draft().await; + let content = content(); + let (left, right) = contend( + &fx.pool, + programs::create_version(&fx.pool, ACTOR, program, &content), + programs::create_version(&fx.pool, ACTOR, program, &content), + ) + .await; + assert_ne!( + left.expect("left").expect("version"), + right.expect("right").expect("version") + ); + let numbers: Vec<_> = programs::list_versions(&fx.pool, program) + .await + .expect("versions") + .into_iter() + .map(|v| v.version_number) + .collect(); + assert_eq!(numbers, vec![1, 2, 3]); +} + +#[tokio::test] +async fn publication_has_one_winner_and_one_audit_event() { + let fx = Fixture::new().await; + let (_, version) = fx.draft().await; + one_winner( + contend( + &fx.pool, + programs::publish_version(&fx.pool, ACTOR, version), + programs::publish_version(&fx.pool, ACTOR, version), + ) + .await, + &PublishRefusal::AlreadyPublished, + ); + let count: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM audit_event WHERE kind = 'program_version_published'", + ) + .fetch_one(&fx.pool) + .await + .expect("audit"); + assert_eq!(count, 1); + fx.probe().await; +} + +#[tokio::test] +async fn draft_replacements_remain_whole_last_write_content() { + let fx = Fixture::new().await; + let (_, version) = fx.draft().await; + let mut first = content(); + first.label = "first".into(); + first.phases[0].name = "First phase".into(); + let mut second = content(); + second.label = "second".into(); + second.phases[0].name = "Second phase".into(); + let (left, right) = contend( + &fx.pool, + programs::replace_draft(&fx.pool, ACTOR, version, &first), + programs::replace_draft(&fx.pool, ACTOR, version, &second), + ) + .await; + left.expect("left").expect("replace"); + right.expect("right").expect("replace"); + let stored = programs::load_content(&fx.pool, version) + .await + .expect("load") + .expect("content"); + assert!(stored == first || stored == second, "no mixed content"); +} + +#[tokio::test] +async fn draft_discard_has_one_winner() { + let fx = Fixture::new().await; + let (_, version) = fx.draft().await; + one_winner( + contend( + &fx.pool, + programs::discard_draft(&fx.pool, ACTOR, version), + programs::discard_draft(&fx.pool, ACTOR, version), + ) + .await, + &AuthorRefusal::NoSuchVersion, + ); + assert!( + programs::load_content(&fx.pool, version) + .await + .expect("load") + .is_none() + ); + fx.probe().await; +} + +#[tokio::test] +async fn import_serializes_names_and_version_numbers() { + let fx = Fixture::new().await; + let (_, version) = fx.draft().await; + let export = program_export::export_version(&fx.pool, version) + .await + .expect("export") + .expect("exists"); + let doc = export.replace("Invented County Program", "Imported Invented Program"); + let imported = one_winner( + contend( + &fx.pool, + program_export::import_version(&fx.pool, ACTOR, &doc, ImportTarget::NewProgram), + program_export::import_version(&fx.pool, ACTOR, &doc, ImportTarget::NewProgram), + ) + .await, + &ImportRefusal::ProgramNameTaken, + ); + fx.probe().await; + let program = programs::version_summary(&fx.pool, imported) + .await + .expect("summary") + .expect("version") + .program_id; + let (left, right) = contend( + &fx.pool, + program_export::import_version(&fx.pool, ACTOR, &doc, ImportTarget::VersionOf(program)), + program_export::import_version(&fx.pool, ACTOR, &doc, ImportTarget::VersionOf(program)), + ) + .await; + let ids = [ + left.expect("left").expect("import"), + right.expect("right").expect("import"), + ]; + assert_ne!(ids[0], ids[1]); + for id in ids { + assert_eq!( + program_export::export_version(&fx.pool, id) + .await + .expect("export") + .expect("exists"), + doc + ); + } + let numbers: Vec<_> = programs::list_versions(&fx.pool, program) + .await + .expect("versions") + .into_iter() + .map(|v| v.version_number) + .collect(); + assert_eq!(numbers, vec![1, 2, 3]); +} diff --git a/crates/consolebook-server/tests/write_transactions/training.rs b/crates/consolebook-server/tests/write_transactions/training.rs new file mode 100644 index 0000000..ea41b04 --- /dev/null +++ b/crates/consolebook-server/tests/write_transactions/training.rs @@ -0,0 +1,395 @@ +use super::*; +use assignments::AssignRefusal; +use consolebook_server::{ + assignments, enrollments, lifecycle, session_membership, training_sessions, +}; +use enrollments::EnrollRefusal; +use lifecycle::{EnrollmentEventKind, LifecycleRefusal, PhaseEventKind}; +use training_sessions::{Disposition, SessionInput, SessionRefusal, SessionUpdate}; + +async fn enrolled(fx: &Fixture) -> i64 { + let (_, version) = fx.published().await; + enrollments::enroll(&fx.pool, ACTOR, version, TRAINEE) + .await + .expect("call") + .expect("enroll") +} + +fn input() -> SessionInput { + SessionInput { + business_date: "2026-06-02".into(), + timezone: "UTC".into(), + local_start: "2026-06-02T08:00".into(), + local_end: None, + disposition: None, + phase_id: None, + trainer_user_ids: vec![TRAINER], + } +} + +async fn session(fx: &Fixture) -> i64 { + let enrollment = enrolled(fx).await; + training_sessions::create(&fx.pool, ACTOR, enrollment, &input()) + .await + .expect("call") + .expect("session") +} + +#[tokio::test] +async fn duplicate_enrollment_has_one_winner() { + let fx = Fixture::new().await; + let (_, version) = fx.published().await; + one_winner( + contend( + &fx.pool, + enrollments::enroll(&fx.pool, ACTOR, version, TRAINEE), + enrollments::enroll(&fx.pool, ACTOR, version, TRAINEE), + ) + .await, + &EnrollRefusal::AlreadyEnrolled, + ); + fx.probe().await; +} + +#[tokio::test] +async fn assignments_serialize_creation_and_ending() { + let fx = Fixture::new().await; + let enrollment = enrolled(&fx).await; + let assignment = one_winner( + contend( + &fx.pool, + assignments::create(&fx.pool, ACTOR, enrollment, TRAINER), + assignments::create(&fx.pool, ACTOR, enrollment, TRAINER), + ) + .await, + &AssignRefusal::AlreadyAssigned, + ); + fx.probe().await; + one_winner( + contend( + &fx.pool, + assignments::end(&fx.pool, ACTOR, assignment), + assignments::end(&fx.pool, ACTOR, assignment), + ) + .await, + &AssignRefusal::AlreadyEnded, + ); + fx.probe().await; + let notices: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM notice WHERE user_id = ?1") + .bind(TRAINER) + .fetch_one(&fx.pool) + .await + .expect("notices"); + assert_eq!(notices, 1, "losing create leaves no duplicate notice"); +} + +#[tokio::test] +async fn enrollment_events_validate_the_committed_status_and_pin() { + let fx = Fixture::new().await; + let enrollment = enrolled(&fx).await; + one_winner( + contend( + &fx.pool, + lifecycle::record_enrollment_event( + &fx.pool, + ACTOR, + enrollment, + EnrollmentEventKind::Withdraw, + "Invented withdrawal", + None, + ), + lifecycle::record_enrollment_event( + &fx.pool, + ACTOR, + enrollment, + EnrollmentEventKind::Withdraw, + "Invented withdrawal", + None, + ), + ) + .await, + &LifecycleRefusal::NotActive, + ); + fx.probe().await; + one_winner( + contend( + &fx.pool, + lifecycle::record_enrollment_event( + &fx.pool, + ACTOR, + enrollment, + EnrollmentEventKind::Reinstate, + "Invented return", + None, + ), + lifecycle::record_enrollment_event( + &fx.pool, + ACTOR, + enrollment, + EnrollmentEventKind::Reinstate, + "Invented return", + None, + ), + ) + .await, + &LifecycleRefusal::AlreadyActive, + ); + let next = programs::create_version(&fx.pool, ACTOR, 1, &content()) + .await + .expect("call") + .expect("version"); + programs::publish_version(&fx.pool, ACTOR, next) + .await + .expect("call") + .expect("publish"); + one_winner( + contend( + &fx.pool, + lifecycle::record_enrollment_event( + &fx.pool, + ACTOR, + enrollment, + EnrollmentEventKind::VersionChange, + "Invented revision", + Some(next), + ), + lifecycle::record_enrollment_event( + &fx.pool, + ACTOR, + enrollment, + EnrollmentEventKind::VersionChange, + "Invented revision", + Some(next), + ), + ) + .await, + &LifecycleRefusal::SameVersion, + ); + fx.probe().await; + let count: i64 = + sqlx::query_scalar("SELECT COUNT(*) FROM enrollment_event WHERE enrollment_id = ?1") + .bind(enrollment) + .fetch_one(&fx.pool) + .await + .expect("history"); + assert_eq!(count, 3); +} + +#[tokio::test] +async fn phase_events_validate_the_committed_pause_state() { + let fx = Fixture::new().await; + let enrollment = enrolled(&fx).await; + let phase: i64 = sqlx::query_scalar("SELECT id FROM phase LIMIT 1") + .fetch_one(&fx.pool) + .await + .expect("phase"); + lifecycle::record_phase_event( + &fx.pool, + ACTOR, + enrollment, + PhaseEventKind::Advance, + Some(phase), + None, + "", + ) + .await + .expect("call") + .expect("entry"); + one_winner( + contend( + &fx.pool, + lifecycle::record_phase_event( + &fx.pool, + ACTOR, + enrollment, + PhaseEventKind::Pause, + None, + None, + "Invented pause", + ), + lifecycle::record_phase_event( + &fx.pool, + ACTOR, + enrollment, + PhaseEventKind::Pause, + None, + None, + "Invented pause", + ), + ) + .await, + &LifecycleRefusal::AlreadyPaused, + ); + fx.probe().await; +} + +#[tokio::test] +async fn overlapping_session_creation_has_one_winner() { + let fx = Fixture::new().await; + let enrollment = enrolled(&fx).await; + let input = input(); + one_winner( + contend( + &fx.pool, + training_sessions::create(&fx.pool, ACTOR, enrollment, &input), + training_sessions::create(&fx.pool, ACTOR, enrollment, &input), + ) + .await, + &SessionRefusal::Overlap, + ); + fx.probe().await; +} + +#[tokio::test] +async fn session_updates_and_closes_keep_existing_outcomes() { + let fx = Fixture::new().await; + let session = session(&fx).await; + let first = SessionUpdate { + business_date: "2026-06-02".into(), + timezone: "UTC".into(), + local_start: "2026-06-02T07:00".into(), + phase_id: None, + }; + let second = SessionUpdate { + local_start: "2026-06-02T06:00".into(), + ..first.clone() + }; + let (left, right) = contend( + &fx.pool, + training_sessions::update_open(&fx.pool, ACTOR, session, &first), + training_sessions::update_open(&fx.pool, ACTOR, session, &second), + ) + .await; + left.expect("left").expect("update"); + right.expect("right").expect("update"); + one_winner( + contend( + &fx.pool, + training_sessions::close( + &fx.pool, + ACTOR, + session, + Disposition::Completed, + Some("2026-06-02T16:00"), + ), + training_sessions::close( + &fx.pool, + ACTOR, + session, + Disposition::Completed, + Some("2026-06-02T16:00"), + ), + ) + .await, + &SessionRefusal::SessionClosed, + ); + fx.probe().await; +} + +#[tokio::test] +async fn membership_additions_and_removals_have_typed_losers() { + let fx = Fixture::new().await; + let session = session(&fx).await; + one_winner( + contend( + &fx.pool, + session_membership::add_trainer(&fx.pool, ACTOR, session, OTHER_TRAINER), + session_membership::add_trainer(&fx.pool, ACTOR, session, OTHER_TRAINER), + ) + .await, + &SessionRefusal::AlreadyMember, + ); + fx.probe().await; + one_winner( + contend( + &fx.pool, + session_membership::remove_trainer(&fx.pool, ACTOR, session, TRAINER), + session_membership::remove_trainer(&fx.pool, ACTOR, session, OTHER_TRAINER), + ) + .await, + &SessionRefusal::LastTrainer, + ); + fx.probe().await; + let count: i64 = + sqlx::query_scalar("SELECT COUNT(*) FROM session_trainer WHERE session_id = ?1") + .bind(session) + .fetch_one(&fx.pool) + .await + .expect("trainers"); + assert_eq!(count, 1); +} + +#[tokio::test] +async fn competing_session_requests_return_created_and_typed_conflict() { + use axum::{ + body::Body, + http::{ + Request, StatusCode, + header::{CONTENT_TYPE, COOKIE}, + }, + }; + use http_body_util::BodyExt; + use tower::ServiceExt; + let fx = Fixture::new().await; + let enrollment = enrolled(&fx).await; + let token = consolebook_server::sessions::create(&fx.pool, ACTOR) + .await + .expect("login token") + .0; + let app = consolebook_server::http::router(consolebook_server::http::AppState { + pool: fx.pool.clone(), + }); + let request = || { + Request::builder().method("POST").uri(format!("/api/enrollments/{enrollment}/sessions")) + .header(CONTENT_TYPE, "application/json") + .header(COOKIE, format!("{}={}", consolebook_server::http::SESSION_COOKIE, token.raw)) + .body(Body::from(serde_json::json!({"business_date":"2026-06-02", "timezone":"UTC", "local_start":"2026-06-02T08:00", "trainer_user_ids":[TRAINER]}).to_string())).expect("request") + }; + let (left, right) = contend( + &fx.pool, + app.clone().oneshot(request()), + app.oneshot(request()), + ) + .await; + let (left, right) = (left.expect("response"), right.expect("response")); + let refused = match (left.status(), right.status()) { + (StatusCode::CREATED, StatusCode::CONFLICT) => right, + (StatusCode::CONFLICT, StatusCode::CREATED) => left, + statuses => panic!("expected 201 and 409, got {statuses:?}"), + }; + let bytes = refused + .into_body() + .collect() + .await + .expect("body") + .to_bytes(); + let body: serde_json::Value = serde_json::from_slice(&bytes).expect("typed JSON"); + assert_eq!(body["error"], "interval_overlap"); + fx.probe().await; +} + +#[tokio::test] +async fn cancellation_and_draft_creation_cannot_both_commit() { + use consolebook_server::evaluation_drafts::{self, DraftRefusal}; + let fx = Fixture::new().await; + let session = session(&fx).await; + let (cancelled, documented) = contend( + &fx.pool, + training_sessions::close(&fx.pool, ACTOR, session, Disposition::Cancelled, None), + evaluation_drafts::create(&fx.pool, TRAINER, session, None), + ) + .await; + match ( + cancelled.expect("cancel outcome"), + documented.expect("draft outcome"), + ) { + (Ok(()), Err(DraftRefusal::SessionCancelled)) + | (Err(SessionRefusal::SessionDocumented), Ok(_)) => {} + outcomes => panic!("cancellation and coverage are mutually exclusive: {outcomes:?}"), + } + let contradictions: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM training_session ts JOIN evaluation_session es ON es.training_session_id = ts.id WHERE ts.disposition = 'cancelled'" + ).fetch_one(&fx.pool).await.expect("coverage"); + assert_eq!(contradictions, 0); + fx.probe().await; +} diff --git a/docs/decisions/0003-sqlite-connection-invariants.md b/docs/decisions/0003-sqlite-connection-invariants.md index 3008f6f..72de206 100644 --- a/docs/decisions/0003-sqlite-connection-invariants.md +++ b/docs/decisions/0003-sqlite-connection-invariants.md @@ -4,6 +4,8 @@ - **Date:** 2026-08-28 - **Amended by:** [ADR 0016](0016-read-only-diagnostics.md), which separates diagnostic connection options and defines WAL-sidecar and PRAGMA scope. +- **Amended by:** [ADR 0019](0019-immediate-write-transactions.md), which + completes immediate write reservations and awaited refusal rollback. ## Context diff --git a/docs/decisions/0019-immediate-write-transactions.md b/docs/decisions/0019-immediate-write-transactions.md new file mode 100644 index 0000000..bd12595 --- /dev/null +++ b/docs/decisions/0019-immediate-write-transactions.md @@ -0,0 +1,98 @@ +# ADR 0019: Reserve the writer before transactional validation + +- **Status:** Accepted +- **Date:** 2026-09-05 +- **Issue:** [#27](https://github.com/FieldmouseWorks/consolebook/issues/27) +- **Amends:** [ADR 0003](0003-sqlite-connection-invariants.md) + +## Context + +The draft services introduced `storage::write_tx` in #29 and awaited refusal +rollback in #31. Earlier services still opened deferred transactions. Under +WAL, their validation reads could establish a snapshot that another writer +made stale before the write. SQLite then refuses promotion with +`SQLITE_BUSY_SNAPSHOT`; the service reports an internal error instead of its +existing domain conflict. A deferred reader can also fail promotion while +another writer still owns the reservation. + +SQLite documents this behavior in [Isolation in SQLite](https://www.sqlite.org/isolation.html) +and its [transaction rules](https://www.sqlite.org/lang_transaction.html). +The issue's approved direction is to complete the immediate-transaction +retrofit, retaining validation semantics and refusal vocabulary. + +## Decision + +Every application-owned multi-statement write transaction starts through +`storage::write_tx`, whose sole implementation is `BEGIN IMMEDIATE`. +Reservation precedes the transactional validation reads; writes, notices, +and audit events commit together as before. A competing writer waits for +reservation, then evaluates those checks against committed state. + +This covers program creation, version creation/replacement/publication/discard, +program imports, enrollments, enrollment and phase events, assignments, +training-session creation/update/close, session membership, user creation, +password-reset issuance/consumption, and setup issuance/initialization, as +well as the draft and record services that already use the helper. +Single-statement writes remain atomic SQLite statements; read-only snapshots +in program content, enrollment detail, draft workspace, and trainee packets +remain deferred and do not reserve the writer. Migration and backup mechanics +remain governed by their existing owners. + +A typed refusal reached after reservation ends through `storage::refuse`, +which awaits rollback before returning it. Operations whose existing return +type is an outcome enum or `Option` await rollback directly. Unexpected +errors and cancellation retain SQLx's transaction-drop cleanup; they are not +translated into a fabricated domain refusal. + +User creation retains its early username lookup to avoid unnecessary password +hashing, but repeats the case-insensitive uniqueness check inside the write +transaction. Setup-code issuance checks initialization and stores its code +under one reservation, so it cannot leave a setup code behind a successful +initialization. Password hashing remains outside write transactions. + +## Boundaries and costs + +The five-second connection busy timeout remains the bound on waiting for a +SQLite writer. A lock held beyond that limit, I/O failure, or pool exhaustion +can still produce an operational error. This change prevents stale-snapshot +promotion failures; it does not promise successful service under unlimited +contention, add retries, or change HTTP error vocabulary. + +Existing capability and scope gates retain their placement. An immediate +transaction does not move a preceding authorization decision into its +snapshot; the qualification in the domain model still applies. This retrofit +does not establish a new authorization contract. + +The reservation lasts through validation, including refusal paths, so those +reads briefly serialize with writers. Pure input validation and expensive +hashing stay outside it. Read-only exports and workspace loads keep their +existing snapshots. No schema, migration checksum, public service signature, +HTTP payload, configuration serialization, or canonical record format changes. + +## Program ownership + +To retrofit the large programs owner under CONTRIBUTING.md and #58: + +- `programs/content.rs` owns configuration vocabulary and structural validation; +- `programs/persistence.rs` owns content loading/replacement helpers and + transaction-owned program/version insertion; and +- `programs.rs` retains policy, typed refusals, summaries, and transaction + orchestration, with compatible type/function re-exports. + +This is #58's programs slice. Draft-workspace decomposition remains separate. + +## Proof + +`tests/write_transactions.rs` owns the contention harness and a direct WAL +stale-snapshot reproduction; its `programs`, `training`, and `accounts` child +modules own the service scenarios. A separate connection holds the writer +while competing operations are polled, then releases it. The scenarios assert +existing typed losers, or both legitimate successes (version numbering, +whole-draft replacement, session editing, and independent reset-code issuance). +An HTTP overlap race requires one 201 and one 409 with `interval_overlap`. +Cancellation racing with draft creation cannot commit both outcomes. Refusal +probes immediately reserve a separate connection with zero busy timeout. + +Existing program/API, lifecycle/session, authentication, draft/review, +finalization, export, and browser suites provide regression coverage for the +retained service and serialization contracts. diff --git a/docs/development.md b/docs/development.md index adea896..e7758e7 100644 --- a/docs/development.md +++ b/docs/development.md @@ -16,7 +16,7 @@ tests show what is implemented. [Roadmap](roadmap.md) owns milestone status. | Backups and restore | `backup.rs`, `scheduler.rs`, `restore.rs`, `serve_lock.rs` | [ADR 0006](decisions/0006-backup-scheduling-and-restore.md) | | Setup, login, recovery | `setup.rs`, `users.rs`, `sessions.rs`, `secrets.rs` | [ADR 0004](decisions/0004-local-authentication.md) | | Capabilities and assignments | `capabilities.rs`, `assignments.rs`, `draft_access.rs` | [ADR 0010](decisions/0010-service-owned-authorization-boundary.md), [Domain model](domain-model.md) | -| Program configuration | `programs.rs`, `program_export.rs` | [ADR 0007](decisions/0007-program-version-configuration-model.md), [Program format](formats/program-version-export.md) | +| Program configuration | `programs.rs`, `programs/content.rs`, `programs/persistence.rs`, `program_export.rs` | [ADR 0007](decisions/0007-program-version-configuration-model.md), [Program format](formats/program-version-export.md) | | Enrollment and training sessions | `enrollments.rs`, `lifecycle.rs`, `training_sessions.rs`, `session_membership.rs`, `session_time.rs` | [ADR 0008](decisions/0008-session-draft-and-attribution-model.md), [ADR 0009](decisions/0009-session-local-time-resolution.md), [ADR 0018](decisions/0018-enrollment-event-reference-shape.md) | | Drafts and review | `evaluation_drafts.rs`, `draft_content.rs`, `draft_review.rs` | [ADR 0008](decisions/0008-session-draft-and-attribution-model.md), [ADR 0010](decisions/0010-service-owned-authorization-boundary.md) | | Finalization and canonical bytes | `finalization.rs`, `canonical.rs`, `record_envelope.rs` | [Integrity](records-integrity.md), [ADR 0011](decisions/0011-canonical-record-format-and-finalization.md) | @@ -59,9 +59,16 @@ Policy belongs in services; persisted constraints also have database backstops. `sessions.rs` owns login sessions; `training_sessions.rs` owns periods of training. Do not infer policy from a role name or a UI guard. -New write paths use `storage::write_tx` and await rollback on refusal through -`storage::refuse`. Earlier deferred write transactions still need the -[#27 retrofit](https://github.com/FieldmouseWorks/consolebook/issues/27). +Application-owned write transactions use `storage::write_tx` and await rollback +on refusal through `storage::refuse` (or directly for outcome/optional returns). +Read-only snapshots remain deferred. [ADR 0019](decisions/0019-immediate-write-transactions.md) +owns the transaction discipline and contention limits; `tests/write_transactions.rs` +and its domain child modules own concurrency proof. + +`programs.rs` owns policy and transaction orchestration; `programs/content.rs` +owns configuration vocabulary and validation; `programs/persistence.rs` owns +content persistence and caller-transaction inserts. Public imports remain under +`programs`. A transaction's presence alone does not prove authorization shares its snapshot; check where the decision is evaluated. See the [domain-model qualification](domain-model.md#application-service-invariants).