feat(scoring): persist dispatch request job and outbox atomically

src/postgres_scoring_request.rs

@@ -6,13 +6,20 @@
 //! `fast-mlsirm` and does not store numeric scores. Replay requires
 //! `READ COMMITTED`.
 
+use crate::integration::IntegrationEvent;
+use crate::postgres_integration::{enqueue_outbox_event, PersistenceDisposition, PersistenceError};
+use crate::postgres_scoring_job::{
+    persist_scoring_job, ScoringJobPersistenceDisposition, ScoringJobPersistenceError,
+};
 use crate::reference::normalized_reference;
 use crate::scoring::ScoringRequest;
+use crate::scoring_job::ScoringJob;
 use postgres::Transaction;
 use std::error::Error;
 use std::fmt::{Display, Formatter};
 
 const SCORING_REQUEST_MIGRATION: &str = include_str!("../migrations/0011_scoring_request.sql");
+const PRODUCT_EVENT_SOURCE: &str = "psychometrics_commons";
 
 /// Outcome of persisting one immutable scoring request.
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
@@ -24,6 +31,81 @@ pub enum ScoringRequestPersistenceDisposition {
     Duplicate,
 }
 
+/// Durable dispositions produced by one atomic scoring-dispatch persistence call.
+///
+/// Mixed dispositions are valid when this transaction safely reconciles pre-existing
+/// exact evidence from an older write path. Every newly inserted row is still committed
+/// or rolled back together by the caller-owned transaction.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub struct ScoringDispatchPersistence {
+    scoring_request: ScoringRequestPersistenceDisposition,
+    scoring_job: ScoringJobPersistenceDisposition,
+    outbox: PersistenceDisposition,
+}
+
+impl ScoringDispatchPersistence {
+    /// Return whether immutable scoring-request evidence was inserted or replayed.
+    #[must_use]
+    pub const fn scoring_request(self) -> ScoringRequestPersistenceDisposition {
+        self.scoring_request
+    }
+
+    /// Return whether immutable scoring-job evidence was inserted or replayed.
+    #[must_use]
+    pub const fn scoring_job(self) -> ScoringJobPersistenceDisposition {
+        self.scoring_job
+    }
+
+    /// Return whether immutable outbox evidence was inserted or replayed.
+    #[must_use]
+    pub const fn outbox(self) -> PersistenceDisposition {
+        self.outbox
+    }
+}
+
+/// Fail-closed error for one atomic scoring-dispatch persistence operation.
+#[derive(Debug)]
+#[non_exhaustive]
+pub enum ScoringDispatchPersistenceError {
+    /// The scoring job names a different immutable scoring request.
+    MismatchedScoringRequest,
+    /// The dispatch outbox envelope is not causally bound to this local request/job pair.
+    InvalidDispatchEnvelope,
+    /// Immutable scoring-request persistence failed.
+    Request(ScoringRequestPersistenceError),
+    /// Durable scoring-job persistence failed.
+    Job(ScoringJobPersistenceError),
+    /// Transactional outbox persistence failed.
+    Outbox(PersistenceError),
+}
+
+impl Display for ScoringDispatchPersistenceError {
+    fn fmt(&self, formatter: &mut Formatter<'_>) -> std::fmt::Result {
+        formatter.write_str(match self {
+            Self::MismatchedScoringRequest => {
+                "scoring job must reference the immutable scoring request in the same dispatch"
+            }
+            Self::InvalidDispatchEnvelope => {
+                "scoring dispatch outbox envelope must identify the local job and response snapshot"
+            }
+            Self::Request(_) => "scoring dispatch request persistence failed",
+            Self::Job(_) => "scoring dispatch job persistence failed",
+            Self::Outbox(_) => "scoring dispatch outbox persistence failed",
+        })
+    }
+}
+
+impl Error for ScoringDispatchPersistenceError {
+    fn source(&self) -> Option<&(dyn Error + 'static)> {
+        match self {
+            Self::MismatchedScoringRequest | Self::InvalidDispatchEnvelope => None,
+            Self::Request(error) => Some(error),
+            Self::Job(error) => Some(error),
+            Self::Outbox(error) => Some(error),
+        }
+    }
+}
+
 /// Fail-closed error for durable scoring-request persistence.
 #[derive(Debug)]
 #[non_exhaustive]
@@ -86,6 +168,58 @@ pub fn apply_scoring_request_migration(
     client.batch_execute(SCORING_REQUEST_MIGRATION)
 }
 
+/// Persist a scoring request, its fresh asynchronous job, and one outbox event atomically.
+///
+/// The caller owns the transaction and therefore the final commit/rollback decision. This
+/// function composes the existing immutable request, job, and transactional-outbox adapters
+/// without introducing a second transaction boundary. The job must name exactly the supplied
+/// scoring request. The outbox envelope must be emitted by this bounded context, use the scoring
+/// job as its subject, and identify the immutable response snapshot as its causation reference.
+/// Event type/schema, tenant, correlation, and payload semantics remain the caller's versioned
+/// integration-contract responsibility after that causal binding is established.
+///
+/// Exact replay is idempotent. Pre-existing exact evidence may produce mixed dispositions,
+/// allowing a caller to reconcile a legacy partial state without rewriting immutable rows.
+/// If any stage returns an error, callers must roll back the transaction rather than commit
+/// earlier successful stages.
+///
+/// # Errors
+///
+/// Returns [`ScoringDispatchPersistenceError::MismatchedScoringRequest`] before writing when
+/// the job is bound to another request, [`ScoringDispatchPersistenceError::InvalidDispatchEnvelope`]
+/// when the outbox envelope is not causally bound to this local dispatch, or the typed request,
+/// job, or outbox persistence error when one of those stages fails.
+pub fn persist_scoring_dispatch(
+    transaction: &mut Transaction<'_>,
+    request: &ScoringRequest,
+    job: &ScoringJob,
+    dispatch_event: &IntegrationEvent,
+    outbox_max_attempts: usize,
+) -> Result<ScoringDispatchPersistence, ScoringDispatchPersistenceError> {
+    if job.scoring_request_ref() != request.scoring_request_ref() {
+        return Err(ScoringDispatchPersistenceError::MismatchedScoringRequest);
+    }
+    if dispatch_event.source() != PRODUCT_EVENT_SOURCE
+        || dispatch_event.subject_ref() != job.scoring_job_ref()
+        || dispatch_event.causation_ref() != Some(request.response_snapshot_ref())
+    {
+        return Err(ScoringDispatchPersistenceError::InvalidDispatchEnvelope);
+    }
+
+    let scoring_request = persist_scoring_request(transaction, request)
+        .map_err(ScoringDispatchPersistenceError::Request)?;
+    let scoring_job =
+        persist_scoring_job(transaction, job).map_err(ScoringDispatchPersistenceError::Job)?;
+    let outbox = enqueue_outbox_event(transaction, dispatch_event, outbox_max_attempts)
+        .map_err(ScoringDispatchPersistenceError::Outbox)?;
+
+    Ok(ScoringDispatchPersistence {
+        scoring_request,
+        scoring_job,
+        outbox,
+    })
+}
+
 /// Persist one immutable scoring-request identity.
 ///
 /// Exact replay of the same request identity and version bundle is idempotent.

tests/postgres_scoring_dispatch_envelope.rs

@@ -0,0 +1,153 @@
+//! Scoring dispatch outbox evidence must be causally bound to the request and job being persisted.
+
+use postgres::{Client, NoTls};
+use psychometrics_commons_runtime::integration::IntegrationEvent;
+use psychometrics_commons_runtime::postgres_integration::apply_integration_migration;
+use psychometrics_commons_runtime::postgres_scoring_job::apply_scoring_job_migration;
+use psychometrics_commons_runtime::postgres_scoring_request::{
+    apply_scoring_request_migration, persist_scoring_dispatch, ScoringDispatchPersistenceError,
+};
+use psychometrics_commons_runtime::response::{ResponseLedger, ResponseWrite};
+use psychometrics_commons_runtime::scoring::{ScoringRequest, ScoringRequestInput};
+use psychometrics_commons_runtime::scoring_job::ScoringJob;
+use psychometrics_commons_runtime::session::SessionState;
+
+const SCHEMA: &str = "scoring_dispatch_envelope_test";
+const DATABASE_TEST_LOCK_KEY: i64 = 0x5343_4453_5045_4E56;
+const DIGEST: &str = "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
+
+fn ready_client() -> Client {
+    let connection = std::env::var("TEST_DATABASE_URL")
+        .expect("TEST_DATABASE_URL must identify the isolated CI PostgreSQL database");
+    let mut client = Client::connect(&connection, NoTls)
+        .expect("isolated CI PostgreSQL database must be reachable");
+    client
+        .query_one("SELECT pg_advisory_lock($1)", &[&DATABASE_TEST_LOCK_KEY])
+        .expect("shared scoring-dispatch envelope test lock should be acquired");
+    client
+        .batch_execute(&format!(
+            "DROP SCHEMA IF EXISTS {SCHEMA} CASCADE;
+             CREATE SCHEMA {SCHEMA};
+             SET search_path TO {SCHEMA};"
+        ))
+        .unwrap();
+    apply_integration_migration(&mut client).unwrap();
+    apply_scoring_job_migration(&mut client).unwrap();
+    apply_scoring_request_migration(&mut client).unwrap();
+    client
+}
+
+fn request() -> ScoringRequest {
+    let mut ledger = ResponseLedger::new("session_dispatch_envelope_alpha").unwrap();
+    ledger
+        .record(
+            SessionState::Active,
+            ResponseWrite {
+                server_event_ref: "server_event_dispatch_envelope_alpha",
+                client_event_ref: "client_event_dispatch_envelope_alpha",
+                item_version_ref: "item_version_dispatch_envelope_alpha",
+                payload_digest: DIGEST,
+            },
+        )
+        .unwrap();
+    let snapshot = ledger
+        .freeze_as(
+            SessionState::Completed,
+            "response_snapshot_dispatch_envelope_alpha",
+        )
+        .unwrap();
+    ScoringRequest::from_snapshot(
+        &snapshot,
+        ScoringRequestInput {
+            scoring_request_ref: "scoring_request_dispatch_envelope_alpha",
+            response_snapshot_ref: "response_snapshot_dispatch_envelope_alpha",
+            assessment_spec_ref: "assessment_spec_big_five_v1",
+            instrument_version_ref: "instrument_version_big_five_ko_v1",
+            scoring_version_ref: "scoring_version_big_five_v1",
+            calibration_reference: "calibration_big_five_ko_v1",
+            norm_version_ref: Some("norm_version_big_five_ko_v1"),
+            requested_output_schema_version: 1,
+        },
+    )
+    .unwrap()
+}
+
+fn event(source: &str, subject: &str, causation: Option<&str>) -> IntegrationEvent {
+    IntegrationEvent::new(
+        "event_scoring_dispatch_envelope_alpha",
+        "scoring.dispatch.requested",
+        "v1",
+        source,
+        "tenant_dispatch_envelope_alpha",
+        subject,
+        10_000,
+        "correlation_dispatch_envelope_alpha",
+        causation,
+        DIGEST,
+    )
+    .unwrap()
+}
+
+#[test]
+fn unrelated_source_subject_or_snapshot_causation_is_rejected_before_writes() {
+    let cases = [
+        (
+            "other_source",
+            "scoring_job_dispatch_envelope_alpha",
+            Some("response_snapshot_dispatch_envelope_alpha"),
+        ),
+        (
+            "psychometrics_commons",
+            "scoring_job_dispatch_envelope_other",
+            Some("response_snapshot_dispatch_envelope_alpha"),
+        ),
+        (
+            "psychometrics_commons",
+            "scoring_job_dispatch_envelope_alpha",
+            Some("response_snapshot_dispatch_envelope_other"),
+        ),
+        (
+            "psychometrics_commons",
+            "scoring_job_dispatch_envelope_alpha",
+            None,
+        ),
+    ];
+
+    for (index, (source, subject, causation)) in cases.into_iter().enumerate() {
+        let mut client = ready_client();
+        let request = request();
+        let job = ScoringJob::new(
+            "scoring_job_dispatch_envelope_alpha",
+            request.scoring_request_ref(),
+            3,
+        )
+        .unwrap();
+        let dispatch_event = event(source, subject, causation);
+
+        let mut transaction = client.transaction().unwrap();
+        assert!(matches!(
+            persist_scoring_dispatch(&mut transaction, &request, &job, &dispatch_event, 3),
+            Err(ScoringDispatchPersistenceError::InvalidDispatchEnvelope)
+        ));
+        transaction.rollback().unwrap();
+
+        for table in ["scoring_request", "scoring_job_state", "integration_outbox"] {
+            let count: i64 = client
+                .query_one(&format!("SELECT count(*) FROM {table}"), &[])
+                .unwrap()
+                .get(0);
+            assert_eq!(
+                count, 0,
+                "invalid envelope case {index} must not write {table}"
+            );
+        }
+    }
+
+    let mut client = ready_client();
+    client
+        .batch_execute(&format!(
+            "SET search_path TO public;
+             DROP SCHEMA IF EXISTS {SCHEMA} CASCADE;"
+        ))
+        .unwrap();
+}