diff --git a/Cargo.lock b/Cargo.lock index 87cf4ca..af4eed9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8099,7 +8099,7 @@ dependencies = [ [[package]] name = "schema-forge-acton" -version = "0.37.3" +version = "0.38.1" dependencies = [ "acton-service", "arc-swap", @@ -8156,7 +8156,7 @@ dependencies = [ [[package]] name = "schema-forge-backend" -version = "0.14.0" +version = "0.15.0" dependencies = [ "acton-service", "argon2", @@ -8187,7 +8187,7 @@ dependencies = [ [[package]] name = "schema-forge-cli" -version = "0.38.3" +version = "0.39.1" dependencies = [ "acton-service", "assert_cmd", @@ -8270,7 +8270,7 @@ dependencies = [ [[package]] name = "schema-forge-postgres" -version = "0.9.1" +version = "0.10.0" dependencies = [ "arc-swap", "argon2", diff --git a/crates/schema-forge-acton/Cargo.toml b/crates/schema-forge-acton/Cargo.toml index 6eab07f..4e4d69e 100644 --- a/crates/schema-forge-acton/Cargo.toml +++ b/crates/schema-forge-acton/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "schema-forge-acton" -version = "0.37.3" +version = "0.38.1" edition = "2021" [dependencies] diff --git a/crates/schema-forge-acton/src/actor.rs b/crates/schema-forge-acton/src/actor.rs index 6a7e749..d2fb682 100644 --- a/crates/schema-forge-acton/src/actor.rs +++ b/crates/schema-forge-acton/src/actor.rs @@ -405,6 +405,18 @@ fn no_backend_error() -> BackendError { } fn configure_backend_operations(actor: &mut ManagedActor) { + actor.act_on::(|actor, ctx| { + let backend = actor.model.backend.clone(); + let request = ctx.message().clone(); + Reply::pending(async move { + let result = match backend { + Some(backend) => backend.create_intent(&request.request).await, + None => Err(schema_forge_backend::create_intent::CreateIntentError::Unsupported), + }; + request.reply.send(result).await; + }) + }); + actor.act_on::(|actor, ctx| { let backend = actor.model.backend.clone(); let entity = ctx.message().entity.clone(); diff --git a/crates/schema-forge-acton/src/messages.rs b/crates/schema-forge-acton/src/messages.rs index 7b73224..6ba90fe 100644 --- a/crates/schema-forge-acton/src/messages.rs +++ b/crates/schema-forge-acton/src/messages.rs @@ -323,3 +323,15 @@ pub struct DeleteEntityIf { pub reply: ReplyChannel>, } + +/// Execute one optional create-reconciliation operation under supervision. +#[derive(Clone, Debug)] +pub struct ProcessCreateIntent { + pub request: schema_forge_backend::create_intent::CreateIntentRequest, + pub reply: ReplyChannel< + Result< + schema_forge_backend::create_intent::CreateIntentReceipt, + schema_forge_backend::create_intent::CreateIntentError, + >, + >, +} diff --git a/crates/schema-forge-acton/src/routes/create_intents.rs b/crates/schema-forge-acton/src/routes/create_intents.rs new file mode 100644 index 0000000..242392c --- /dev/null +++ b/crates/schema-forge-acton/src/routes/create_intents.rs @@ -0,0 +1,471 @@ +//! Bounded, currently authorized create reconciliation for hook-free schemas. +use super::entities::{self, EntityRequest}; +use crate::{ + access::{check_schema_access, AccessAction, OptionalClaims}, + actor::ForgeActor, + config::SchemaForgeConfig, + error::ForgeError, + messages::{GetSchema, ProcessCreateIntent, ReplyChannel}, +}; +use acton_service::{middleware::Claims, prelude::ActorHandleInterface, state::AppState}; +use axum::{ + extract::{Path, Query, State}, + http::{HeaderMap, StatusCode}, + response::{IntoResponse, Response}, + Json, +}; +use schema_forge_backend::{ + create_intent::{ + CreateFingerprint, CreateIntentError, CreateIntentId, CreateIntentReceipt, + CreateIntentRequest, CreateIntentScope, + }, + Entity, TenantRef, +}; +use schema_forge_core::types::{SchemaDefinition, SchemaName}; +use sha2::{Digest, Sha256}; +use std::{ + collections::{BTreeMap, HashMap}, + time::Duration, +}; +use tokio::sync::oneshot; + +pub(super) const HEADER: &str = "create-intent"; +pub(super) fn error(error: CreateIntentError) -> ForgeError { + let reason = match error { + CreateIntentError::Unsupported => "create_intent_unsupported", + CreateIntentError::Invalid => { + return ForgeError::InvalidQuery { + message: "Invalid create intent.".into(), + } + } + CreateIntentError::Unavailable => "create_intent_unavailable", + CreateIntentError::ContentConflict => "create_intent_content_conflict", + CreateIntentError::SchemaChanged => "create_intent_schema_changed", + CreateIntentError::Backend(schema_forge_backend::BackendError::UniqueViolation { + .. + }) => { + return ForgeError::Conflict { + reason: "unique_violation", + message: "A record with these unique values already exists.".into(), + }; + } + CreateIntentError::Backend(_) => return ForgeError::BackendUnavailable { + message: + "Create outcome could not be determined. Reconcile the same intent before retrying." + .into(), + }, + }; + ForgeError::Conflict { + reason, + message: + "Create intent cannot be used for this request. No replacement create was attempted." + .into(), + } +} + +/// Sort objects recursively. Arrays, missing/null and JSON number kinds remain distinct. +fn canonical(value: &serde_json::Value) -> serde_json::Value { + match value { + serde_json::Value::Object(fields) => { + let sorted: BTreeMap<_, _> = fields + .iter() + .map(|(key, value)| (key.clone(), canonical(value))) + .collect(); + serde_json::Value::Object(sorted.into_iter().collect()) + } + serde_json::Value::Array(values) => { + serde_json::Value::Array(values.iter().map(canonical).collect()) + } + _ => value.clone(), + } +} +fn fingerprint(body: &EntityRequest) -> Result { + let bytes = serde_json::to_vec(&canonical(&serde_json::Value::Object(body.fields.clone()))) + .map_err(|_| error(CreateIntentError::Invalid))?; + CreateFingerprint::parse(hex::encode(Sha256::digest(bytes))).map_err(error) +} + +pub(super) async fn process( + state: &AppState, + request: CreateIntentRequest, +) -> Result { + let forge = state + .actor::() + .ok_or_else(|| error(CreateIntentError::Unsupported))?; + let (tx, rx) = oneshot::channel(); + forge + .send(ProcessCreateIntent { + request, + reply: ReplyChannel::new(tx), + }) + .await; + tokio::time::timeout(Duration::from_secs(5), rx) + .await + .map_err(|_| ForgeError::BackendUnavailable { + message: "Create outcome is uncertain. Reconcile this intent before retrying.".into(), + })? + .map_err(|_| ForgeError::BackendUnavailable { + message: "Create outcome is uncertain. Reconcile this intent before retrying.".into(), + })? + .map_err(error) +} + +async fn scope( + state: &AppState, + schema: SchemaDefinition, + claims: Option<&Claims>, +) -> Result { + let store = entities::fetch_export_policy_store(state).await?; + check_schema_access(&store, &schema, claims, AccessAction::Create)?; + let caller = claims.ok_or_else(|| ForgeError::Unauthorized { + message: "Authentication required.".into(), + })?; + let tenants: Vec = caller.custom_claim_as("tenant_chain").unwrap_or_default(); + let tenant = tenants + .last() + .map(|tenant| format!("{}:{}", tenant.schema, tenant.entity_id)) + .unwrap_or_default(); + if schema.is_tenanted() && tenant.is_empty() { + return Err(ForgeError::Forbidden { + message: "An active tenant is required.".into(), + }); + } + Ok(CreateIntentScope { + principal: crate::authz::adapters::user_id_from_sub(&caller.sub).to_string(), + tenant, + schema, + }) +} +fn supported( + state: &AppState, + schema: &SchemaDefinition, +) -> Result<(), ForgeError> { + if schema.has_hooks() || state.config().custom.schema_forge.webhooks.enabled { + return Err(error(CreateIntentError::Unsupported)); + } + Ok(()) +} +async fn load_scope( + state: &AppState, + schema: &str, + claims: Option<&Claims>, +) -> Result { + let name = SchemaName::new(schema).map_err(|_| ForgeError::InvalidSchemaName { + name: schema.into(), + })?; + let forge = state + .actor::() + .ok_or_else(|| error(CreateIntentError::Unsupported))?; + let (tx, rx) = oneshot::channel(); + forge + .send(GetSchema { + name: name.to_string(), + reply: ReplyChannel::new(tx), + }) + .await; + let schema = tokio::time::timeout(Duration::from_secs(5), rx) + .await + .map_err(|_| error(CreateIntentError::Unsupported))? + .map_err(|_| error(CreateIntentError::Unsupported))? + .ok_or(ForgeError::SchemaNotFound { + name: name.to_string(), + })?; + scope(state, schema, claims).await +} +async fn authorize_input( + state: &AppState, + scope: &CreateIntentScope, + claims: Option<&Claims>, + body: &EntityRequest, +) -> Result<(), ForgeError> { + entities::reject_hidden_fields_in_body(&scope.schema, &body.fields)?; + if body + .fields + .keys() + .any(|name| scope.schema.field(name).is_none()) + { + return Err(ForgeError::InvalidQuery { + message: "Unknown field in create intent.".into(), + }); + } + let mut fields = entities::json_to_entity_fields(&scope.schema, &body.fields) + .map_err(|details| ForgeError::ValidationFailed { details })?; + crate::access::inject_owner_on_create(&mut fields, &scope.schema, claims); + if let Some((_, tenant)) = scope.tenant.split_once(':') { + fields.insert( + "_tenant".into(), + schema_forge_core::types::DynamicValue::Text(tenant.into()), + ); + } + let baseline = Entity::new(scope.schema.name.clone(), fields); + let store = entities::fetch_export_policy_store(state).await?; + entities::authorize_conditional_fields(&store, &scope.schema, &baseline, claims, &body.fields) +} + +/// Reserve a server-generated intent without creating a record. +pub async fn reserve( + State(state): State>, + Path(schema): Path, + OptionalClaims(claims): OptionalClaims, + Json(body): Json, +) -> Result { + let scope = load_scope(&state, &schema, claims.as_ref()).await?; + supported(&state, &scope.schema)?; + authorize_input(&state, &scope, claims.as_ref(), &body).await?; + let receipt = process( + &state, + CreateIntentRequest::Reserve { + scope, + fingerprint: fingerprint(&body)?, + }, + ) + .await?; + Ok(( + StatusCode::CREATED, + Json(receipt_json(&receipt, "pending", None)), + ) + .into_response()) +} +fn receipt_json( + receipt: &CreateIntentReceipt, + state: &str, + entity_id: Option<&str>, +) -> serde_json::Value { + serde_json::json!({"id":receipt.id.as_str(),"state":state,"expires_at":receipt.expires_at,"recover_until":receipt.recover_until,"entity_id":entity_id}) +} +/// Read a receipt under current scope and current record authorization. +pub async fn read( + State(state): State>, + Path((schema, id)): Path<(String, String)>, + OptionalClaims(claims): OptionalClaims, +) -> Result { + let scope = load_scope(&state, &schema, claims.as_ref()).await?; + let receipt = process( + &state, + CreateIntentRequest::Read { + scope, + id: id.parse().map_err(error)?, + }, + ) + .await?; + let Some(entity_id) = receipt.entity_id.as_ref() else { + return Ok(Json(receipt_json(&receipt, "pending", None)).into_response()); + }; + match entities::get_entity( + State(state), + Path((schema, entity_id.to_string())), + OptionalClaims(claims), + Query(HashMap::new()), + ) + .await + { + Ok(_) => Ok(Json(receipt_json( + &receipt, + "committed", + Some(entity_id.as_str()), + )) + .into_response()), + Err(ForgeError::EntityNotFound { .. }) => { + Ok(Json(receipt_json(&receipt, "committed_unavailable", None)).into_response()) + } + Err(error) => Err(error), + } +} + +pub(super) struct IntentCommit { + scope: CreateIntentScope, + id: CreateIntentId, + fingerprint: CreateFingerprint, + pub receipt: CreateIntentReceipt, +} +pub(super) async fn preflight( + state: &AppState, + schema: &SchemaDefinition, + claims: Option<&Claims>, + headers: &HeaderMap, + body: &EntityRequest, +) -> Result, ForgeError> { + if !headers.contains_key(HEADER) { + return Ok(None); + } + let scope = scope(state, schema.clone(), claims).await?; + authorize_input(state, &scope, claims, body).await?; + if headers.get_all(HEADER).iter().count() != 1 { + return Err(error(CreateIntentError::Invalid)); + } + let id: CreateIntentId = headers[HEADER] + .to_str() + .map_err(|_| error(CreateIntentError::Invalid))? + .parse() + .map_err(error)?; + let fingerprint = fingerprint(body)?; + let receipt = process( + state, + CreateIntentRequest::Read { + scope: scope.clone(), + id: id.clone(), + }, + ) + .await?; + if receipt.fingerprint != fingerprint { + return Err(error(CreateIntentError::ContentConflict)); + } + if receipt.entity_id.is_none() { + supported(state, &scope.schema)?; + } + if receipt.entity_id.is_none() && !receipt.definition_matches { + return Err(error(CreateIntentError::SchemaChanged)); + } + Ok(Some(IntentCommit { + scope, + id, + fingerprint, + receipt, + })) +} +pub(super) async fn commit( + state: &AppState, + intent: IntentCommit, + entity: Entity, +) -> Result { + process( + state, + CreateIntentRequest::Commit { + scope: intent.scope, + id: intent.id, + fingerprint: intent.fingerprint, + entity, + }, + ) + .await +} +pub(super) async fn result( + state: AppState, + schema: String, + claims: Option, + receipt: &CreateIntentReceipt, +) -> Result { + let id = receipt + .entity_id + .as_ref() + .ok_or_else(|| error(CreateIntentError::Unavailable))?; + let response = entities::get_entity( + State(state), + Path((schema, id.to_string())), + OptionalClaims(claims), + Query(HashMap::new()), + ) + .await; + let mut response = match response { + Ok(response) => response.into_response(), + Err(ForgeError::EntityNotFound { .. }) => return Err(ForgeError::Conflict { + reason: "create_result_unavailable", + message: + "The create committed, but its result is no longer available. It was not recreated." + .into(), + }), + Err(error) => return Err(error), + }; + *response.status_mut() = if receipt.created { + StatusCode::CREATED + } else { + StatusCode::OK + }; + response.headers_mut().insert( + HEADER, + receipt + .id + .as_str() + .parse() + .map_err(|_| error(CreateIntentError::Invalid))?, + ); + Ok(response) +} + +#[cfg(test)] +mod tests { + use super::*; + fn digest(value: serde_json::Value) -> CreateFingerprint { + fingerprint(&EntityRequest { + fields: value.as_object().unwrap().clone(), + }) + .unwrap() + } + #[test] + fn input_fingerprints_preserve_content_distinctions() { + assert_eq!( + digest(serde_json::from_str(r#"{"a":{"z":1,"b":2},"b":3}"#).unwrap()), + digest(serde_json::from_str(r#"{"b":3,"a":{"b":2,"z":1}}"#).unwrap()) + ); + for (a, b) in [ + (serde_json::json!({}), serde_json::json!({"a":null})), + ( + serde_json::json!({"a":[1,2]}), + serde_json::json!({"a":[2,1]}), + ), + (serde_json::json!({"a":1}), serde_json::json!({"a":1.0})), + ] { + assert_ne!(digest(a), digest(b)); + } + } + #[tokio::test] + async fn configured_side_effects_refuse_pending_protocol() { + use acton_service::{config::Config, service_builder::ServiceBuilder}; + use schema_forge_core::types::{ + Annotation, FieldDefinition, FieldName, FieldType, HookEvent, SchemaId, TextConstraints, + }; + let mut schema = SchemaDefinition::new( + SchemaId::new(), + SchemaName::new("Note").unwrap(), + vec![FieldDefinition::new( + FieldName::new("title").unwrap(), + FieldType::Text(TextConstraints::unconstrained()), + )], + vec![], + ) + .unwrap(); + let service = ServiceBuilder::new() + .with_config(Config::::default()) + .build(); + assert!(supported(service.state(), &schema).is_ok()); + schema.annotations.push(Annotation::Hook { + event: HookEvent::AfterChange, + intent: "notify".into(), + }); + assert!(matches!( + supported(service.state(), &schema), + Err(ForgeError::Conflict { + reason: "create_intent_unsupported", + .. + }) + )); + schema.annotations.clear(); + let mut config = Config::::default(); + config.custom.schema_forge.webhooks.enabled = true; + let service = ServiceBuilder::new().with_config(config).build(); + assert!(matches!( + supported(service.state(), &schema), + Err(ForgeError::Conflict { + reason: "create_intent_unsupported", + .. + }) + )); + } + #[test] + fn uniqueness_conflicts_do_not_disclose_hidden_constraints() { + let mapped = error(CreateIntentError::Backend( + schema_forge_backend::BackendError::UniqueViolation { + schema: "Note".into(), + field: "hidden_name_key".into(), + }, + )); + assert!(matches!( + mapped, + ForgeError::Conflict { + reason: "unique_violation", + .. + } + )); + assert!(!mapped.to_string().contains("hidden_name_key")); + } +} diff --git a/crates/schema-forge-acton/src/routes/entities.rs b/crates/schema-forge-acton/src/routes/entities.rs index 6b18a81..09d90ad 100644 --- a/crates/schema-forge-acton/src/routes/entities.rs +++ b/crates/schema-forge-acton/src/routes/entities.rs @@ -236,7 +236,7 @@ async fn load_mutation_baseline( Ok((ask_forge(rx).await?.map_err(ForgeError::from)?, None)) } -fn authorize_conditional_fields( +pub(super) fn authorize_conditional_fields( store: &Arc, schema: &SchemaDefinition, baseline: &Entity, @@ -1154,7 +1154,7 @@ fn entity_to_response(entity: &Entity, schema: &SchemaDefinition) -> EntityRespo /// and must never accept input from a user-facing endpoint. This guard /// lives at the request-deserialization boundary so `create`, `update`, /// and `patch` all share the same enforcement. -fn reject_hidden_fields_in_body( +pub(super) fn reject_hidden_fields_in_body( schema: &SchemaDefinition, body_fields: &serde_json::Map, ) -> Result<(), ForgeError> { @@ -2166,8 +2166,9 @@ pub async fn create_entity( State(state): State>, Path(schema): Path, OptionalClaims(claims): OptionalClaims, + headers: HeaderMap, Json(body): Json, -) -> Result { +) -> Result { let schema_name = validate_schema_name(&schema)?; let forge = state .actor::() @@ -2210,6 +2211,15 @@ pub async fn create_entity( return Err(e); } + let intent = + super::create_intents::preflight(&state, &schema_def, claims.as_ref(), &headers, &body) + .await?; + if let Some(intent) = &intent { + if intent.receipt.entity_id.is_some() { + return super::create_intents::result(state, schema, claims, &intent.receipt).await; + } + } + // Reject any client-supplied @hidden fields up front. reject_hidden_fields_in_body(&schema_def, &body.fields)?; @@ -2311,6 +2321,16 @@ pub async fn create_entity( ); check_field_constraints(&schema_def, &entity.fields)?; + if let Some(intent) = intent { + let receipt = super::create_intents::commit(&state, intent, entity).await?; + if receipt.created { + if let Some(logger) = state.audit_logger() { + logger.log_custom("forge.entity.created", acton_service::audit::AuditSeverity::Informational, Some(serde_json::json!({"schema":schema,"intent_id":receipt.id.as_str(),"entity_id":receipt.entity_id.as_ref().map(|id|id.as_str())}))).await; + } + } + return super::create_intents::result(state, schema, claims, &receipt).await; + } + // Create entity via actor (supervised backend call) let (tx, rx) = oneshot::channel(); forge @@ -2374,7 +2394,8 @@ pub async fn create_entity( Ok(( StatusCode::CREATED, Json(entity_to_response(&created, &schema_def)), - )) + ) + .into_response()) } /// GET /schemas/{schema}/entities -- List/query entities. diff --git a/crates/schema-forge-acton/src/routes/meta.rs b/crates/schema-forge-acton/src/routes/meta.rs index bc3705d..6de9e70 100644 --- a/crates/schema-forge-acton/src/routes/meta.rs +++ b/crates/schema-forge-acton/src/routes/meta.rs @@ -83,6 +83,14 @@ pub async fn get_meta(meta: Option>>) -> Response { Some(Extension(info)) => { let mut body = serde_json::json!(info.as_ref()); body["capabilities"] = serde_json::json!({ + "create_reconciliation": { + "protocol": "create-intent-v1", + "backend_supported": info.backend == "postgres", + "request_header": "Create-Intent", + "schema_support": "hook-free", + "admission_seconds": 900, + "recovery_seconds": 86400 + }, "conditional_entity_mutations": { "protocol": "record-revision-v1", "backend_supported": info.backend == "postgres", diff --git a/crates/schema-forge-acton/src/routes/mod.rs b/crates/schema-forge-acton/src/routes/mod.rs index 4efbf42..83dc71c 100644 --- a/crates/schema-forge-acton/src/routes/mod.rs +++ b/crates/schema-forge-acton/src/routes/mod.rs @@ -1,4 +1,5 @@ pub mod auth; +pub mod create_intents; pub mod entities; pub mod export; pub mod files; @@ -45,6 +46,14 @@ pub fn forge_routes() -> Router> { .put(schemas::update_schema) .delete(schemas::delete_schema), ) + .route( + "/schemas/{schema}/create-intents", + post(create_intents::reserve), + ) + .route( + "/schemas/{schema}/create-intents/{id}", + get(create_intents::read), + ) // Entity CRUD (nested under schema) .route( "/schemas/{schema}/entities", diff --git a/crates/schema-forge-acton/src/state.rs b/crates/schema-forge-acton/src/state.rs index 435b978..57ce057 100644 --- a/crates/schema-forge-acton/src/state.rs +++ b/crates/schema-forge-acton/src/state.rs @@ -133,6 +133,25 @@ impl DynSchemaBackend for T { /// /// Same pattern as `DynSchemaBackend`: boxed futures for dynamic dispatch. pub trait DynEntityStore: Send + Sync { + /// Durable optional create reconciliation. + fn create_intent<'a>( + &'a self, + _request: &'a schema_forge_backend::create_intent::CreateIntentRequest, + ) -> Pin< + Box< + dyn Future< + Output = Result< + schema_forge_backend::create_intent::CreateIntentReceipt, + schema_forge_backend::create_intent::CreateIntentError, + >, + > + Send + + Sync + + 'a, + >, + > { + Box::pin(async { Err(schema_forge_backend::create_intent::CreateIntentError::Unsupported) }) + } + /// Read a record and its storage revision, if supported. fn get_versioned<'a>( &'a self, @@ -234,6 +253,24 @@ pub trait DynEntityStore: Send + Sync { /// `Send + Sync` `FutureBox` bound, so backend calls can be awaited /// directly inside `act_on` handlers without an inner `tokio::spawn`. impl DynEntityStore for T { + fn create_intent<'a>( + &'a self, + request: &'a schema_forge_backend::create_intent::CreateIntentRequest, + ) -> Pin< + Box< + dyn Future< + Output = Result< + schema_forge_backend::create_intent::CreateIntentReceipt, + schema_forge_backend::create_intent::CreateIntentError, + >, + > + Send + + Sync + + 'a, + >, + > { + Box::pin(SyncFuture::new(EntityStore::create_intent(self, request))) + } + fn get_versioned<'a>( &'a self, schema: &'a SchemaName, diff --git a/crates/schema-forge-acton/tests/create_intents.rs b/crates/schema-forge-acton/tests/create_intents.rs new file mode 100644 index 0000000..5a91f74 --- /dev/null +++ b/crates/schema-forge-acton/tests/create_intents.rs @@ -0,0 +1,549 @@ +//! HTTP record revision contracts for unsupported and explicitly disposable +//! PostgreSQL storage, including authorization before token/capability errors. +use acton_service::{ + config::Config, middleware::Claims, prelude::ActorHandleInterface, + service_builder::ServiceBuilder, +}; +use axum::{ + body::Body, + http::{Request, StatusCode}, + Router, +}; +use http_body_util::BodyExt; +use schema_forge_acton::{ + config::SchemaForgeConfig, + messages::{InitForge, ReplyChannel}, + routes::forge_routes, + state::{DynEntityStore, DynForgeBackend}, + storage::StorageRegistry, + ForgeActor, +}; +use schema_forge_backend::entity::Entity; +use schema_forge_core::types::{ + Annotation, DynamicValue, EntityId, FieldAnnotation, FieldDefinition, FieldName, FieldType, + SchemaDefinition, SchemaId, SchemaName, TextConstraints, +}; +use schema_forge_surrealdb::SurrealBackend; +use std::{ + collections::{BTreeMap, HashMap}, + sync::Arc, + time::Duration, +}; +use tokio::sync::oneshot; +use tower::ServiceExt; + +async fn fixture(owner: &str, roles: &[&str]) -> (Router, String) { + let backend = Arc::new( + SurrealBackend::connect_memory("conditional", "conditional") + .await + .unwrap(), + ); + fixture_with_backend(backend, owner, roles, false).await +} + +async fn fixture_with_backend( + backend: Arc, + owner: &str, + roles: &[&str], + prepare_revisions: bool, +) -> (Router, String) { + let schema = SchemaDefinition::new( + SchemaId::new(), + SchemaName::new("Note").unwrap(), + vec![ + FieldDefinition::new( + FieldName::new("title").unwrap(), + FieldType::Text(TextConstraints::unconstrained()), + ), + FieldDefinition::with_annotations( + FieldName::new("owner").unwrap(), + FieldType::Text(TextConstraints::unconstrained()), + vec![], + vec![FieldAnnotation::Owner], + ), + FieldDefinition::with_annotations( + FieldName::new("restricted").unwrap(), + FieldType::Text(TextConstraints::unconstrained()), + vec![], + vec![FieldAnnotation::FieldAccess { + read: vec!["editor".into()], + write: vec!["manager".into()], + }], + ), + ], + vec![Annotation::Access { + read: vec!["editor".into()], + write: vec!["editor".into()], + delete: vec!["editor".into()], + cross_tenant_read: vec![], + }], + ) + .unwrap(); + let plan = schema_forge_core::migration::DiffEngine::create_new(&schema); + backend + .apply_migration(&schema.name, &plan.steps) + .await + .unwrap(); + backend.store_schema_metadata(&schema).await.unwrap(); + if prepare_revisions { + backend + .prepare_record_revisions(&schema.name) + .await + .unwrap(); + } + let entity = Entity::with_id( + EntityId::new("note"), + schema.name.clone(), + BTreeMap::from([ + ("title".into(), DynamicValue::Text("Original".into())), + ("owner".into(), DynamicValue::Text(owner.into())), + ("restricted".into(), DynamicValue::Text("Restricted".into())), + ]), + ); + DynEntityStore::create(backend.as_ref(), &entity) + .await + .unwrap(); + let service = ServiceBuilder::new() + .with_config(Config::::default()) + .with_actor::() + .build(); + let (tx, rx) = oneshot::channel(); + service + .state() + .actor::() + .unwrap() + .send(InitForge { + registry: HashMap::from([("Note".into(), schema)]), + backend, + tenant_config: None, + record_access_policy: None, + hook_dispatcher: None, + storage_registry: StorageRegistry::default(), + policy_store: None, + custom_policies_dir: None, + reply: ReplyChannel::new(tx), + }) + .await; + tokio::time::timeout(Duration::from_secs(5), rx) + .await + .unwrap() + .unwrap(); + let caller = Claims { + sub: "user:editor".into(), + roles: roles.iter().map(|role| (*role).into()).collect(), + perms: vec![], + exp: 9_999_999_999, + iat: None, + jti: None, + iss: None, + aud: None, + email: None, + username: None, + custom: HashMap::new(), + }; + let app = forge_routes() + .layer(axum::middleware::from_fn( + move |mut req: axum::extract::Request, next: axum::middleware::Next| { + let caller = caller.clone(); + async move { + req.extensions_mut().insert(caller); + next.run(req).await + } + }, + )) + .with_state(service.state().clone()); + (app, format!("/schemas/Note/entities/{}", entity.id)) +} + +async fn request( + app: &Router, + path: &str, + method: &str, + condition: Option<&str>, + fields: serde_json::Value, +) -> (StatusCode, axum::http::HeaderMap, serde_json::Value) { + request_in_tenant(app, path, method, condition, fields, None).await +} + +async fn request_in_tenant( + app: &Router, + path: &str, + method: &str, + condition: Option<&str>, + fields: serde_json::Value, + tenant: Option<&str>, +) -> (StatusCode, axum::http::HeaderMap, serde_json::Value) { + let mut request = Request::builder() + .method(method) + .uri(path) + .header("content-type", "application/json"); + if let Some(condition) = condition { + request = request.header("create-intent", condition); + } + if let Some(tenant) = tenant { + request = request.header("x-active-tenant", tenant); + } + let response = app + .clone() + .oneshot( + request + .body(Body::from( + serde_json::json!({"fields": fields}).to_string(), + )) + .unwrap(), + ) + .await + .unwrap(); + let status = response.status(); + let headers = response.headers().clone(); + let bytes = response.into_body().collect().await.unwrap().to_bytes(); + ( + status, + headers, + serde_json::from_slice(&bytes).unwrap_or(serde_json::Value::Null), + ) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn create_intents_fail_closed_for_unsupported_and_denied_input() { + let (app, _) = fixture("editor", &["editor"]).await; + let (status, _, body) = request( + &app, + "/schemas/Note/create-intents", + "POST", + None, + serde_json::json!({"title":"new"}), + ) + .await; + assert_eq!(status, StatusCode::CONFLICT, "{body}"); + assert_eq!(body["reason"], "create_intent_unsupported"); + let (status, _, body) = request( + &app, + "/schemas/Note/create-intents", + "POST", + None, + serde_json::json!({"restricted":"denied"}), + ) + .await; + assert_eq!(status, StatusCode::FORBIDDEN, "{body}"); + let (app, _) = fixture("editor", &["other"]).await; + let (status, _, body) = request( + &app, + "/schemas/Note/create-intents", + "POST", + None, + serde_json::json!({"title":"new"}), + ) + .await; + assert_eq!(status, StatusCode::FORBIDDEN, "{body}"); +} + +#[cfg(feature = "postgres")] +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[ignore = "requires a scoped disposable PostgreSQL URL and SCHEMAFORGE_TEST_POSTGRES_DISPOSABLE=1"] +async fn postgres_http_create_reconciliation() { + assert_eq!( + std::env::var("SCHEMAFORGE_TEST_POSTGRES_DISPOSABLE").as_deref(), + Ok("1") + ); + let url = std::env::var("SCHEMAFORGE_TEST_POSTGRES_URL").unwrap(); + let backend = Arc::new( + schema_forge_postgres::PgBackend::connect(&url) + .await + .unwrap_or_else(|_| panic!("disposable test database connection failed")), + ); + let (app, _) = fixture_with_backend(backend.clone(), "editor", &["editor"], true).await; + let fields = serde_json::json!({"title":"New record"}); + let (status, _, receipt) = request( + &app, + "/schemas/Note/create-intents", + "POST", + None, + fields.clone(), + ) + .await; + assert_eq!(status, StatusCode::CREATED, "{receipt}"); + assert_eq!(receipt["state"], "pending"); + let id = receipt["id"].as_str().unwrap(); + let receipt_path = format!("/schemas/Note/create-intents/{id}"); + let (status, _, body) = request(&app, &receipt_path, "GET", None, serde_json::json!({})).await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body["state"], "pending"); + let (status, headers, entity) = request( + &app, + "/schemas/Note/entities", + "POST", + Some(id), + fields.clone(), + ) + .await; + assert_eq!(status, StatusCode::CREATED, "{entity}"); + assert_eq!(headers["create-intent"], id); + assert!(headers.contains_key("entity-revision")); + let original_revision = headers["entity-revision"].clone(); + let (status, headers, replay) = request( + &app, + "/schemas/Note/entities", + "POST", + Some(id), + fields.clone(), + ) + .await; + assert_eq!(status, StatusCode::OK, "{replay}"); + assert_eq!(entity, replay); + assert_eq!(headers["entity-revision"], original_revision); + let (status, _, body) = request( + &app, + "/schemas/Note/entities", + "POST", + Some(id), + serde_json::json!({"title":"changed"}), + ) + .await; + assert_eq!(status, StatusCode::CONFLICT, "{body}"); + assert_eq!(body["reason"], "create_intent_content_conflict"); + let entity_id = EntityId::parse(entity["id"].as_str().unwrap()).unwrap(); + let change = Entity::with_id( + entity_id.clone(), + SchemaName::new("Note").unwrap(), + BTreeMap::from([("owner".into(), DynamicValue::Text("other".into()))]), + ); + DynEntityStore::update(backend.as_ref(), &change) + .await + .unwrap(); + let (status, _, body) = request(&app, &receipt_path, "GET", None, serde_json::json!({})).await; + assert_eq!(status, StatusCode::FORBIDDEN, "{body}"); + let (status, _, body) = request( + &app, + "/schemas/Note/entities", + "POST", + Some(id), + fields.clone(), + ) + .await; + assert_eq!(status, StatusCode::FORBIDDEN, "{body}"); + DynEntityStore::delete(backend.as_ref(), &change.schema, &entity_id) + .await + .unwrap(); + let (status, _, body) = request(&app, &receipt_path, "GET", None, serde_json::json!({})).await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body["state"], "committed_unavailable"); + assert!(body["entity_id"].is_null()); + let (status, _, body) = request(&app, "/schemas/Note/entities", "POST", Some(id), fields).await; + assert_eq!(status, StatusCode::CONFLICT, "{body}"); + assert_eq!(body["reason"], "create_result_unavailable"); +} + +#[cfg(feature = "postgres")] +#[ignore = "requires a scoped disposable PostgreSQL namespace"] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn postgres_create_intents_honor_selected_membership() { + use schema_forge_acton::middleware::tenant_scope::{middleware, TenantScopeState}; + + let schema = SchemaDefinition::new( + SchemaId::new(), + SchemaName::new("TenantNote").unwrap(), + vec![FieldDefinition::new( + FieldName::new("title").unwrap(), + FieldType::Text(TextConstraints::unconstrained()), + )], + vec![ + Annotation::Access { + read: vec!["editor".into()], + write: vec!["editor".into()], + delete: vec!["editor".into()], + cross_tenant_read: vec![], + }, + Annotation::Tenant(schema_forge_core::types::TenantKind::Child { + parent: SchemaName::new("Organization").unwrap(), + }), + ], + ) + .unwrap(); + let organization = SchemaDefinition::new( + SchemaId::new(), + SchemaName::new("Organization").unwrap(), + vec![FieldDefinition::new( + FieldName::new("name").unwrap(), + FieldType::Text(TextConstraints::unconstrained()), + )], + vec![Annotation::Tenant( + schema_forge_core::types::TenantKind::Root, + )], + ) + .unwrap(); + assert_eq!( + std::env::var("SCHEMAFORGE_TEST_POSTGRES_DISPOSABLE").as_deref(), + Ok("1") + ); + let url = std::env::var("SCHEMAFORGE_TEST_POSTGRES_URL").unwrap(); + let backend = Arc::new( + schema_forge_postgres::PgBackend::connect(&url) + .await + .unwrap_or_else(|_| panic!("disposable test database connection failed")), + ); + let plan = schema_forge_core::migration::DiffEngine::create_new(&schema); + schema_forge_acton::DynSchemaBackend::apply_migration( + backend.as_ref(), + &schema.name, + &plan.steps, + ) + .await + .unwrap(); + schema_forge_acton::DynSchemaBackend::store_schema_metadata(backend.as_ref(), &schema) + .await + .unwrap(); + let entity = Entity::with_id( + EntityId::new("tenantnote"), + schema.name.clone(), + BTreeMap::from([ + ( + "title".into(), + DynamicValue::Text("Program alpha record".into()), + ), + ( + "_tenant".into(), + DynamicValue::Text("organization_alpha".into()), + ), + ]), + ); + DynEntityStore::create(backend.as_ref(), &entity) + .await + .unwrap(); + let tenant_config = schema_forge_backend::tenant::TenantConfig::from_schemas(&[ + organization.clone(), + schema.clone(), + ]) + .unwrap(); + let scope = TenantScopeState { + entity_store: backend.clone(), + tenant_config: Arc::new(Some(tenant_config.clone())), + }; + let service = ServiceBuilder::new() + .with_config(Config::::default()) + .with_actor::() + .build(); + let (tx, rx) = oneshot::channel(); + service + .state() + .actor::() + .unwrap() + .send(InitForge { + registry: HashMap::from([ + ("TenantNote".into(), schema), + ("Organization".into(), organization), + ]), + backend: backend.clone(), + tenant_config: Some(tenant_config), + record_access_policy: None, + hook_dispatcher: None, + storage_registry: StorageRegistry::default(), + policy_store: None, + custom_policies_dir: None, + reply: ReplyChannel::new(tx), + }) + .await; + tokio::time::timeout(Duration::from_secs(5), rx) + .await + .unwrap() + .unwrap(); + let caller = Claims { + sub: "user:editor".into(), + roles: vec!["editor".into()], + perms: vec![], + exp: 9_999_999_999, + iat: None, + jti: None, + iss: None, + aud: None, + email: None, + username: None, + custom: HashMap::from([( + "tenant_chain".into(), + serde_json::json!([ + {"schema":"Organization","entity_id":"organization_alpha"}, + {"schema":"Organization","entity_id":"organization_beta"}, + ]), + )]), + }; + let app = forge_routes() + .layer(axum::middleware::from_fn_with_state(scope, middleware)) + .layer(axum::middleware::from_fn( + move |mut req: axum::extract::Request, next: axum::middleware::Next| { + let caller = caller.clone(); + async move { + req.extensions_mut().insert(caller); + next.run(req).await + } + }, + )) + .with_state(service.state().clone()); + let alpha = Some("Organization:organization_alpha"); + let beta = Some("Organization:organization_beta"); + let fields = serde_json::json!({"title":"Scoped create"}); + let (status, _, receipt) = request_in_tenant( + &app, + "/schemas/TenantNote/create-intents", + "POST", + None, + fields.clone(), + alpha, + ) + .await; + assert_eq!(status, StatusCode::CREATED, "{receipt}"); + let id = receipt["id"].as_str().unwrap(); + let receipt_path = format!("/schemas/TenantNote/create-intents/{id}"); + for (path, method) in [ + (&receipt_path[..], "GET"), + ("/schemas/TenantNote/entities", "POST"), + ] { + let (status, _, body) = + request_in_tenant(&app, path, method, Some(id), fields.clone(), beta).await; + assert_eq!(status, StatusCode::CONFLICT, "{body}"); + assert_eq!(body["reason"], "create_intent_unavailable"); + } + for tenant in [None, Some("Organization:organization_unknown")] { + let (status, _, body) = request_in_tenant( + &app, + "/schemas/TenantNote/entities", + "POST", + Some(id), + fields.clone(), + tenant, + ) + .await; + assert!(status.is_client_error(), "{body}"); + } + let (status, _, entity) = request_in_tenant( + &app, + "/schemas/TenantNote/entities", + "POST", + Some(id), + fields.clone(), + alpha, + ) + .await; + assert_eq!(status, StatusCode::CREATED, "{entity}"); + let (status, _, body) = request_in_tenant( + &app, + &receipt_path, + "GET", + None, + serde_json::json!({}), + beta, + ) + .await; + assert_eq!(status, StatusCode::CONFLICT, "{body}"); + assert_eq!(body["reason"], "create_intent_unavailable"); + let (status, _, body) = request_in_tenant( + &app, + &receipt_path, + "GET", + None, + serde_json::json!({}), + alpha, + ) + .await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body["state"], "committed"); +} diff --git a/crates/schema-forge-backend/Cargo.toml b/crates/schema-forge-backend/Cargo.toml index 7e7ec0e..e8b176d 100644 --- a/crates/schema-forge-backend/Cargo.toml +++ b/crates/schema-forge-backend/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "schema-forge-backend" -version = "0.14.0" +version = "0.15.0" edition = "2021" [dependencies] diff --git a/crates/schema-forge-backend/src/create_intent.rs b/crates/schema-forge-backend/src/create_intent.rs new file mode 100644 index 0000000..5c63aee --- /dev/null +++ b/crates/schema-forge-backend/src/create_intent.rs @@ -0,0 +1,173 @@ +//! Optional bounded create reconciliation. Receipts never replace authorization. +use crate::{BackendError, Entity}; +use chrono::{DateTime, Utc}; +use schema_forge_core::types::{EntityId, SchemaDefinition}; +use std::{fmt, str::FromStr}; + +/// Admission window for a reserved v1 intent. +pub const ADMISSION_SECONDS: i64 = 900; +/// Recovery window, including after the created record is renamed or deleted. +pub const RECOVERY_SECONDS: i64 = 86_400; + +/// Server-generated identifier for one create attempt. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct CreateIntentId(EntityId); +impl CreateIntentId { + /// Generate an intent ID; clients cannot reserve a chosen identifier. + pub fn fresh() -> Self { + Self(EntityId::new("createintent")) + } + /// Opaque transport representation. + pub fn as_str(&self) -> &str { + self.0.as_str() + } +} +impl FromStr for CreateIntentId { + type Err = CreateIntentError; + fn from_str(value: &str) -> Result { + let id = EntityId::parse(value).map_err(|_| CreateIntentError::Invalid)?; + if id.prefix() != "createintent" || id.as_str() != value { + return Err(CreateIntentError::Invalid); + } + Ok(Self(id)) + } +} +/// Authenticated scope and server schema snapshot, never supplied by the client. +#[derive(Debug, Clone)] +pub struct CreateIntentScope { + /// Stable normalized principal identity. + pub principal: String, + /// Selected tenant schema and identifier; empty only for an unscoped schema. + pub tenant: String, + /// Registered schema identity and definition at this request. + pub schema: SchemaDefinition, +} +/// Fingerprint of canonical submitted JSON, excluding server-generated values. +#[derive(Clone, PartialEq, Eq)] +pub struct CreateFingerprint(String); +impl CreateFingerprint { + /// Validate a v1 SHA-256 hexadecimal fingerprint. + pub fn parse(value: String) -> Result { + if value.len() != 64 + || !value + .bytes() + .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b)) + { + return Err(CreateIntentError::Invalid); + } + Ok(Self(value)) + } + /// Internal storage representation. + pub fn as_str(&self) -> &str { + &self.0 + } +} +impl fmt::Debug for CreateFingerprint { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str("CreateFingerprint([opaque])") + } +} +/// Optional backend operations sharing one receipt contract. +#[derive(Debug, Clone)] +pub enum CreateIntentRequest { + /// Reserve without creating an entity. + Reserve { + scope: CreateIntentScope, + fingerprint: CreateFingerprint, + }, + /// Read a scoped receipt. This does not authorize reading its entity. + Read { + scope: CreateIntentScope, + id: CreateIntentId, + }, + /// Atomically insert entity, revision and committed receipt, or reconcile. + Commit { + scope: CreateIntentScope, + id: CreateIntentId, + fingerprint: CreateFingerprint, + entity: Entity, + }, +} +/// A retained receipt. Entity data must be fetched with current authorization. +#[derive(Debug, Clone)] +pub struct CreateIntentReceipt { + /// Reserved intent identity. + pub id: CreateIntentId, + /// Deadline for admitting an initial commit. + pub expires_at: DateTime, + /// Deadline for recovery of this receipt. + pub recover_until: DateTime, + /// Original created entity, even after deletion; absent while pending. + pub entity_id: Option, + /// True only for the transaction that inserted the entity. + pub created: bool, + /// Bound input fingerprint for trusted route-level preflight comparisons. + pub fingerprint: CreateFingerprint, + /// Whether the uncommitted schema snapshot still matches this request. + pub definition_matches: bool, +} +/// Fail-closed errors for the optional protocol. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum CreateIntentError { + /// Adapter does not implement this protocol. + Unsupported, + /// Malformed protocol input. + Invalid, + /// Unknown, expired or differently scoped intent; never create on absence. + Unavailable, + /// Reusing an intent with changed submitted content. + ContentConflict, + /// Schema changed since an uncommitted reservation. + SchemaChanged, + /// Database failure, including a business uniqueness conflict. + Backend(BackendError), +} +impl fmt::Display for CreateIntentError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(match self { + Self::Unsupported => "create reconciliation unsupported", + Self::Invalid => "invalid create intent", + Self::Unavailable => "create intent unavailable", + Self::ContentConflict => "create intent content changed", + Self::SchemaChanged => "create intent schema changed", + Self::Backend(_) => "create intent storage failed", + }) + } +} +impl std::error::Error for CreateIntentError {} +impl From for CreateIntentError { + fn from(value: BackendError) -> Self { + Self::Backend(value) + } +} + +#[cfg(test)] +mod tests { + use super::*; + #[test] + fn identifiers_are_canonical_server_namespaced_values() { + let id = CreateIntentId::fresh(); + assert_eq!(id.as_str().parse::().unwrap(), id); + for value in [ + "".into(), + "garbage".into(), + format!(" {}", id.as_str()), + format!("{},{}", id.as_str(), id.as_str()), + EntityId::new("note").to_string(), + ] { + assert!(value.parse::().is_err()); + } + } + #[test] + fn fingerprints_require_exact_lowercase_sha256_shape() { + assert!(CreateFingerprint::parse("a".repeat(64)).is_ok()); + for value in [ + "a".repeat(63), + "A".repeat(64), + "g".repeat(64), + "a".repeat(65), + ] { + assert!(CreateFingerprint::parse(value).is_err()); + } + } +} diff --git a/crates/schema-forge-backend/src/lib.rs b/crates/schema-forge-backend/src/lib.rs index c2192cd..f5cee03 100644 --- a/crates/schema-forge-backend/src/lib.rs +++ b/crates/schema-forge-backend/src/lib.rs @@ -22,3 +22,6 @@ pub use user_store::{AuthStore, ForgeUser}; /// Optional atomic record mutation types. pub mod conditional; + +/// Optional durable create reconciliation types. +pub mod create_intent; diff --git a/crates/schema-forge-backend/src/traits.rs b/crates/schema-forge-backend/src/traits.rs index 97473a0..90127db 100644 --- a/crates/schema-forge-backend/src/traits.rs +++ b/crates/schema-forge-backend/src/traits.rs @@ -74,6 +74,19 @@ pub trait SchemaBackend: Send + Sync { /// - Creating, reading, updating, and deleting entities /// - Executing queries with filters, sorting, and pagination pub trait EntityStore: Send + Sync { + /// Optional durable create reconciliation; unsupported adapters must refuse. + fn create_intent( + &self, + _request: &crate::create_intent::CreateIntentRequest, + ) -> impl Future< + Output = Result< + crate::create_intent::CreateIntentReceipt, + crate::create_intent::CreateIntentError, + >, + > + Send { + async { Err(crate::create_intent::CreateIntentError::Unsupported) } + } + /// Read an entity and opaque revision from a single consistent snapshot. /// Unsupported backends must not synthesize revisions from ordinary reads. fn get_versioned( diff --git a/crates/schema-forge-cli/Cargo.toml b/crates/schema-forge-cli/Cargo.toml index 0fb7ed6..530679f 100644 --- a/crates/schema-forge-cli/Cargo.toml +++ b/crates/schema-forge-cli/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "schema-forge-cli" -version = "0.38.3" +version = "0.39.1" edition = "2021" [[bin]] diff --git a/crates/schema-forge-mssql/Cargo.toml b/crates/schema-forge-mssql/Cargo.toml index 3669d58..e1e3b62 100644 --- a/crates/schema-forge-mssql/Cargo.toml +++ b/crates/schema-forge-mssql/Cargo.toml @@ -7,7 +7,7 @@ edition = "2021" acton-service = { version = "0.39.0", features = ["mssql", "crypto-aws-lc-rs"] } bb8 = "0.9.1" bb8-tiberius = "0.16.0" -schema-forge-backend = { version = "0.14.0", path = "../schema-forge-backend" } +schema-forge-backend = { version = "0.15.0", path = "../schema-forge-backend" } schema-forge-core = { version = "0.16.0", path = "../schema-forge-core" } serde_json = "1.0.151" tiberius = "0.12.3" diff --git a/crates/schema-forge-postgres/Cargo.toml b/crates/schema-forge-postgres/Cargo.toml index e7e0cb2..20c06f3 100644 --- a/crates/schema-forge-postgres/Cargo.toml +++ b/crates/schema-forge-postgres/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "schema-forge-postgres" -version = "0.9.1" +version = "0.10.0" edition = "2021" [dependencies] diff --git a/crates/schema-forge-postgres/src/backend.rs b/crates/schema-forge-postgres/src/backend.rs index d7d8a55..148a0df 100644 --- a/crates/schema-forge-postgres/src/backend.rs +++ b/crates/schema-forge-postgres/src/backend.rs @@ -36,7 +36,7 @@ const PG_UNIQUE_VIOLATION: &str = "23505"; /// recovered from the constraint name (`uq_{table}_{field}`) when present, /// and otherwise falls back to the literal constraint identifier so the /// client still has *something* actionable to display. -fn map_write_error(err: sqlx::Error, schema: &str, context: &str) -> BackendError { +pub(crate) fn map_write_error(err: sqlx::Error, schema: &str, context: &str) -> BackendError { if let sqlx::Error::Database(ref db_err) = err { if db_err.code().as_deref() == Some(PG_UNIQUE_VIOLATION) { let constraint = db_err.constraint().unwrap_or(""); @@ -187,9 +187,26 @@ impl PgBackend { reason: e.to_string(), })?; self.ensure_revision_tables().await?; + self.ensure_create_intents().await?; Ok(()) } + pub(crate) async fn insert_with_revision( + connection: &mut sqlx::PgConnection, + entity: &Entity, + schema: Option<&SchemaDefinition>, + ) -> Result { + let (sql, args) = Self::build_insert(entity, schema)?; + let row = sqlx::query_with(&sql, args) + .persistent(false) + .fetch_one(&mut *connection) + .await + .map_err(|e| map_write_error(e, entity.schema.as_str(), "create entity"))?; + let entity = row_to_entity(&row, &entity.schema, schema)?; + let revision = Self::advance_revision(connection, &entity).await?; + Ok(schema_forge_backend::conditional::VersionedEntity { entity, revision }) + } + /// Build a parameterized INSERT statement and arguments for an entity. /// /// When `schema_def` is provided, per-column `FieldType` context is passed @@ -574,21 +591,26 @@ impl SchemaBackend for PgBackend { } impl EntityStore for PgBackend { + async fn create_intent( + &self, + request: &schema_forge_backend::create_intent::CreateIntentRequest, + ) -> Result< + schema_forge_backend::create_intent::CreateIntentReceipt, + schema_forge_backend::create_intent::CreateIntentError, + > { + self.process_create_intent(request).await + } + async fn create(&self, entity: &Entity) -> Result { let schema_def = self.load_schema_metadata(&entity.schema).await?; - let (sql, args) = Self::build_insert(entity, schema_def.as_ref())?; let mut tx = self .pool .begin() .await .map_err(|e| map_write_error(e, entity.schema.as_str(), "begin create"))?; - let row: PgRow = sqlx::query_with(&sql, args) - .persistent(false) - .fetch_one(&mut *tx) - .await - .map_err(|e| map_write_error(e, entity.schema.as_str(), "failed to create entity"))?; - let created = row_to_entity(&row, &entity.schema, schema_def.as_ref())?; - Self::advance_revision(&mut tx, &created).await?; + let created = Self::insert_with_revision(&mut tx, entity, schema_def.as_ref()) + .await? + .entity; tx.commit() .await .map_err(|e| map_write_error(e, entity.schema.as_str(), "commit create"))?; diff --git a/crates/schema-forge-postgres/src/create_intent.rs b/crates/schema-forge-postgres/src/create_intent.rs new file mode 100644 index 0000000..e5f2d50 --- /dev/null +++ b/crates/schema-forge-postgres/src/create_intent.rs @@ -0,0 +1,145 @@ +//! Intent commitment shares the entity/revision transaction. Missing IDs never create. +use crate::backend::PgBackend; +use schema_forge_backend::{ + create_intent::{ + CreateFingerprint, CreateIntentError as Error, CreateIntentId, CreateIntentReceipt, + CreateIntentRequest, CreateIntentScope, ADMISSION_SECONDS, RECOVERY_SECONDS, + }, + BackendError, +}; +use schema_forge_core::types::EntityId; +use sqlx::{postgres::PgRow, Row}; + +fn storage(error: sqlx::Error) -> Error { + BackendError::QueryError { + message: format!("create receipt storage failed: {error}"), + } + .into() +} +fn receipt(row: &PgRow, created: bool) -> Result { + let id: String = row.try_get("id").map_err(storage)?; + let entity: Option = row.try_get("entity_id").map_err(storage)?; + Ok(CreateIntentReceipt { + id: id.parse()?, + expires_at: row.try_get("expires_at").map_err(storage)?, + recover_until: row.try_get("recover_until").map_err(storage)?, + fingerprint: CreateFingerprint::parse(row.try_get("fingerprint").map_err(storage)?)?, + definition_matches: true, + entity_id: entity + .map(|id| EntityId::parse(&id).map_err(|_| Error::Invalid)) + .transpose()?, + created, + }) +} +impl PgBackend { + pub(crate) async fn ensure_create_intents(&self) -> Result<(), BackendError> { + sqlx::query("CREATE TABLE IF NOT EXISTS _schema_create_intents (id TEXT PRIMARY KEY, principal TEXT NOT NULL, tenant TEXT NOT NULL, schema_name TEXT NOT NULL, schema_id TEXT NOT NULL, definition JSONB NOT NULL, fingerprint TEXT NOT NULL, expires_at TIMESTAMPTZ NOT NULL, recover_until TIMESTAMPTZ NOT NULL, entity_id TEXT)").execute(self.pool()).await.map_err(|e| BackendError::MigrationFailed { step: "create intent receipts".into(), reason: e.to_string() })?; + sqlx::query("CREATE INDEX IF NOT EXISTS _schema_create_intents_expiry ON _schema_create_intents (recover_until)").execute(self.pool()).await.map_err(|e| BackendError::MigrationFailed { step: "create intent expiry index".into(), reason: e.to_string() })?; + Ok(()) + } + + async fn check_intent_schema( + connection: &mut sqlx::PgConnection, + scope: &CreateIntentScope, + ) -> Result<(), Error> { + let definition: Option = + sqlx::query_scalar("SELECT definition FROM _schema_metadata WHERE name = $1 FOR SHARE") + .bind(scope.schema.name.as_str()) + .fetch_optional(connection) + .await + .map_err(storage)?; + let expected = serde_json::to_value(&scope.schema).map_err(|_| Error::Invalid)?; + if definition.as_ref() != Some(&expected) { + return Err(Error::SchemaChanged); + } + Ok(()) + } + + pub(crate) async fn process_create_intent( + &self, + request: &CreateIntentRequest, + ) -> Result { + let mut tx = self.pool().begin().await.map_err(storage)?; + sqlx::query("SET LOCAL lock_timeout = '3s'") + .execute(&mut *tx) + .await + .map_err(storage)?; + sqlx::query("SET LOCAL statement_timeout = '4s'") + .execute(&mut *tx) + .await + .map_err(storage)?; + let result = match request { + CreateIntentRequest::Reserve { scope, fingerprint } => { + Self::check_intent_schema(&mut tx, scope).await?; + // No user may choose a reservation ID, including after pruning. + let id = CreateIntentId::fresh(); + sqlx::query("DELETE FROM _schema_create_intents WHERE id IN (SELECT id FROM _schema_create_intents WHERE recover_until <= clock_timestamp() LIMIT 100 FOR UPDATE SKIP LOCKED)").execute(&mut *tx).await.map_err(storage)?; + let row = sqlx::query("INSERT INTO _schema_create_intents (id, principal, tenant, schema_name, schema_id, definition, fingerprint, expires_at, recover_until) VALUES ($1,$2,$3,$4,$5,$6,$7,clock_timestamp()+make_interval(secs=>$8),clock_timestamp()+make_interval(secs=>$9)) RETURNING *") + .bind(id.as_str()).bind(&scope.principal).bind(&scope.tenant).bind(scope.schema.name.as_str()).bind(scope.schema.id.as_str()).bind(serde_json::to_value(&scope.schema).map_err(|_| Error::Invalid)?).bind(fingerprint.as_str()).bind(ADMISSION_SECONDS as f64).bind(RECOVERY_SECONDS as f64) + .fetch_one(&mut *tx).await.map_err(storage)?; + receipt(&row, false)? + } + CreateIntentRequest::Read { scope, id } + | CreateIntentRequest::Commit { scope, id, .. } => { + let row = sqlx::query("SELECT * FROM _schema_create_intents WHERE id=$1 AND principal=$2 AND tenant=$3 AND schema_name=$4 AND schema_id=$5 AND recover_until > clock_timestamp() FOR UPDATE") + .bind(id.as_str()).bind(&scope.principal).bind(&scope.tenant).bind(scope.schema.name.as_str()).bind(scope.schema.id.as_str()).fetch_optional(&mut *tx).await.map_err(storage)?.ok_or(Error::Unavailable)?; + let mut outcome = receipt(&row, false)?; + let now: chrono::DateTime = + sqlx::query_scalar("SELECT clock_timestamp()") + .fetch_one(&mut *tx) + .await + .map_err(storage)?; + if outcome.recover_until <= now { + return Err(Error::Unavailable); + } + outcome.definition_matches = row + .try_get::("definition") + .map_err(storage)? + == serde_json::to_value(&scope.schema).map_err(|_| Error::Invalid)?; + if let CreateIntentRequest::Commit { + fingerprint, + entity, + .. + } = request + { + let stored: String = row.try_get("fingerprint").map_err(storage)?; + if stored != fingerprint.as_str() { + return Err(Error::ContentConflict); + } + if outcome.entity_id.is_none() { + if outcome.expires_at <= now { + return Err(Error::Unavailable); + } + let definition: serde_json::Value = + row.try_get("definition").map_err(storage)?; + if definition + != serde_json::to_value(&scope.schema).map_err(|_| Error::Invalid)? + { + return Err(Error::SchemaChanged); + } + Self::check_intent_schema(&mut tx, scope).await?; + if entity.schema != scope.schema.name { + return Err(Error::Invalid); + } + let created = + Self::insert_with_revision(&mut tx, entity, Some(&scope.schema)) + .await?; + sqlx::query("UPDATE _schema_create_intents SET entity_id=$2 WHERE id=$1") + .bind(id.as_str()) + .bind(created.entity.id.as_str()) + .execute(&mut *tx) + .await + .map_err(storage)?; + outcome.entity_id = Some(created.entity.id); + outcome.created = true; + } + } else if outcome.entity_id.is_none() && outcome.expires_at <= now { + return Err(Error::Unavailable); + } + outcome + } + }; + tx.commit().await.map_err(storage)?; + Ok(result) + } +} diff --git a/crates/schema-forge-postgres/src/lib.rs b/crates/schema-forge-postgres/src/lib.rs index 929ed51..ad081a1 100644 --- a/crates/schema-forge-postgres/src/lib.rs +++ b/crates/schema-forge-postgres/src/lib.rs @@ -6,3 +6,5 @@ pub mod value; pub use backend::PgBackend; mod conditional; + +mod create_intent; diff --git a/crates/schema-forge-postgres/tests/create_intent.rs b/crates/schema-forge-postgres/tests/create_intent.rs new file mode 100644 index 0000000..b3c0959 --- /dev/null +++ b/crates/schema-forge-postgres/tests/create_intent.rs @@ -0,0 +1,303 @@ +//! Opt-in live PostgreSQL concurrency checks, isolated in a disposable namespace. + +use std::{collections::BTreeMap, sync::Arc}; + +use schema_forge_backend::{ + create_intent::{ + CreateFingerprint, CreateIntentError, CreateIntentId, CreateIntentRequest, + CreateIntentScope, + }, + Entity, EntityStore, SchemaBackend, +}; +use schema_forge_core::{ + migration::DiffEngine, + types::{ + DynamicValue, EntityId, FieldDefinition, FieldName, FieldType, SchemaDefinition, SchemaId, + SchemaName, TextConstraints, + }, +}; +use schema_forge_postgres::PgBackend; +use sqlx::postgres::PgPoolOptions; +use tokio::sync::Barrier; + +fn definition() -> SchemaDefinition { + SchemaDefinition::new( + SchemaId::new(), + SchemaName::new("Note").unwrap(), + ["name", "detail"] + .map(|name| { + FieldDefinition::new( + FieldName::new(name).unwrap(), + FieldType::Text(TextConstraints::unconstrained()), + ) + }) + .to_vec(), + vec![], + ) + .unwrap() +} + +fn patch(entity: &Entity, name: &str) -> Entity { + Entity::with_id( + entity.id.clone(), + entity.schema.clone(), + BTreeMap::from([("name".into(), DynamicValue::Text(name.into()))]), + ) +} + +#[tokio::test] +#[ignore = "requires SCHEMAFORGE_TEST_POSTGRES_URL with CREATE SCHEMA privilege"] +async fn atomic_create_receipts() { + let url = std::env::var("SCHEMAFORGE_TEST_POSTGRES_URL").expect("test PostgreSQL URL required"); + let admin = PgPoolOptions::new() + .max_connections(1) + .connect(&url) + .await + .unwrap_or_else(|_| panic!("disposable test database connection failed")); + let namespace = EntityId::new("intenttest").to_string(); + sqlx::query(&format!("CREATE SCHEMA \"{namespace}\"")) + .execute(&admin) + .await + .unwrap(); + let scope = namespace.clone(); + let pool = PgPoolOptions::new() + .max_connections(4) + .after_connect(move |connection, _| { + let scope = scope.clone(); + Box::pin(async move { + sqlx::query("SELECT set_config('search_path', $1, false)") + .bind(scope) + .execute(connection) + .await?; + Ok(()) + }) + }) + .connect(&url) + .await + .unwrap(); + let backend = Arc::new(PgBackend::from_pool(pool.clone()).await.unwrap()); + // Catch assertion panics in the child so cleanup still happens. + let result = tokio::spawn(async move { exercise(backend).await }).await; + pool.close().await; + sqlx::query(&format!("DROP SCHEMA \"{namespace}\" CASCADE")) + .execute(&admin) + .await + .unwrap(); + admin.close().await; + result.unwrap(); +} + +async fn exercise(backend: Arc) { + let mut schema = definition(); + schema.fields[0] + .modifiers + .push(schema_forge_core::types::FieldModifier::Unique); + backend + .apply_migration(&schema.name, &DiffEngine::create_new(&schema).steps) + .await + .unwrap(); + backend.store_schema_metadata(&schema).await.unwrap(); + backend + .prepare_record_revisions(&schema.name) + .await + .unwrap(); + let scope = CreateIntentScope { + principal: "alice".into(), + tenant: "Organization:one".into(), + schema: schema.clone(), + }; + let fingerprint = CreateFingerprint::parse("a".repeat(64)).unwrap(); + let reserve = CreateIntentRequest::Reserve { + scope: scope.clone(), + fingerprint: fingerprint.clone(), + }; + let receipt = backend.create_intent(&reserve).await.unwrap(); + let entity = Entity::new( + schema.name.clone(), + BTreeMap::from([ + ("name".into(), DynamicValue::Text("original".into())), + ("detail".into(), DynamicValue::Text("value".into())), + ]), + ); + let commit = CreateIntentRequest::Commit { + scope: scope.clone(), + id: receipt.id.clone(), + fingerprint: fingerprint.clone(), + entity: entity.clone(), + }; + let read = CreateIntentRequest::Read { + scope: scope.clone(), + id: receipt.id.clone(), + }; + assert!(backend + .create_intent(&read) + .await + .unwrap() + .entity_id + .is_none()); + for changed_scope in [ + CreateIntentScope { + principal: "bob".into(), + ..scope.clone() + }, + CreateIntentScope { + tenant: "Organization:two".into(), + ..scope.clone() + }, + ] { + assert!(matches!( + backend + .create_intent(&CreateIntentRequest::Read { + scope: changed_scope.clone(), + id: receipt.id.clone() + }) + .await, + Err(CreateIntentError::Unavailable) + )); + assert!(matches!( + backend + .create_intent(&CreateIntentRequest::Commit { + scope: changed_scope, + id: receipt.id.clone(), + fingerprint: fingerprint.clone(), + entity: entity.clone() + }) + .await, + Err(CreateIntentError::Unavailable) + )); + } + let barrier = Arc::new(Barrier::new(2)); + let mut tasks = vec![]; + for _ in 0..2 { + let backend = backend.clone(); + let barrier = barrier.clone(); + let commit = commit.clone(); + tasks.push(tokio::spawn(async move { + barrier.wait().await; + backend.create_intent(&commit).await.unwrap() + })); + } + let a = tasks.remove(0).await.unwrap(); + let b = tasks.remove(0).await.unwrap(); + assert_eq!(usize::from(a.created) + usize::from(b.created), 1); + assert_eq!(a.entity_id, b.entity_id); + let baseline = backend + .get_versioned(&schema.name, &entity.id) + .await + .unwrap(); + assert!(!backend.create_intent(&commit).await.unwrap().created); + assert_eq!( + backend + .get_versioned(&schema.name, &entity.id) + .await + .unwrap() + .revision, + baseline.revision + ); + assert!(matches!( + backend + .create_intent(&CreateIntentRequest::Commit { + scope: scope.clone(), + id: receipt.id.clone(), + fingerprint: CreateFingerprint::parse("b".repeat(64)).unwrap(), + entity: entity.clone() + }) + .await, + Err(CreateIntentError::ContentConflict) + )); + backend.update(&patch(&entity, "renamed")).await.unwrap(); + assert_eq!( + backend.create_intent(&read).await.unwrap().entity_id, + Some(entity.id.clone()) + ); + backend.delete(&schema.name, &entity.id).await.unwrap(); + assert_eq!( + backend.create_intent(&commit).await.unwrap().entity_id, + Some(entity.id.clone()) + ); + assert!(backend.get(&schema.name, &entity.id).await.is_err()); + + // Failed insertion rolls back receipt commitment and its revision together. + backend.create(&entity).await.unwrap(); + let existing = Entity::new(entity.schema.clone(), entity.fields.clone()); + let pending = backend.create_intent(&reserve).await.unwrap(); + assert!(matches!( + backend + .create_intent(&CreateIntentRequest::Commit { + scope: scope.clone(), + id: pending.id.clone(), + fingerprint: fingerprint.clone(), + entity: existing + }) + .await, + Err(CreateIntentError::Backend( + schema_forge_backend::BackendError::UniqueViolation { .. } + )) + )); + assert!(backend + .create_intent(&CreateIntentRequest::Read { + scope: scope.clone(), + id: pending.id.clone() + }) + .await + .unwrap() + .entity_id + .is_none()); + + sqlx::query("UPDATE _schema_create_intents SET expires_at=clock_timestamp()-interval '1 second' WHERE id=$1").bind(pending.id.as_str()).execute(backend.pool()).await.unwrap(); + assert!(matches!( + backend + .create_intent(&CreateIntentRequest::Commit { + scope: scope.clone(), + id: pending.id, + fingerprint: fingerprint.clone(), + entity: entity.clone() + }) + .await, + Err(CreateIntentError::Unavailable) + )); + assert!(matches!( + backend + .create_intent(&CreateIntentRequest::Commit { + scope: scope.clone(), + id: CreateIntentId::fresh(), + fingerprint: fingerprint.clone(), + entity: entity.clone() + }) + .await, + Err(CreateIntentError::Unavailable) + )); + sqlx::query("UPDATE _schema_create_intents SET recover_until=clock_timestamp()-interval '1 second' WHERE id=$1").bind(receipt.id.as_str()).execute(backend.pool()).await.unwrap(); + assert!(matches!( + backend.create_intent(&commit).await, + Err(CreateIntentError::Unavailable) + )); + backend.create_intent(&reserve).await.unwrap(); // Prunes the expired receipt. + assert!(matches!( + backend.create_intent(&commit).await, + Err(CreateIntentError::Unavailable) + )); + let pending = backend.create_intent(&reserve).await.unwrap(); + let mut changed = scope.clone(); + changed + .schema + .annotations + .push(schema_forge_core::types::Annotation::Version { + version: schema_forge_core::types::SchemaVersion::new(2).unwrap(), + }); + backend + .store_schema_metadata(&changed.schema) + .await + .unwrap(); + assert!(matches!( + backend + .create_intent(&CreateIntentRequest::Commit { + scope: changed, + id: pending.id, + fingerprint, + entity + }) + .await, + Err(CreateIntentError::SchemaChanged) + )); +} diff --git a/docs/create-reconciliation.md b/docs/create-reconciliation.md new file mode 100644 index 0000000..bfa0135 --- /dev/null +++ b/docs/create-reconciliation.md @@ -0,0 +1,58 @@ +# Create reconciliation v1 + +This opt-in PostgreSQL protocol binds one submitted create request to a server-reserved intent. The entity, its initial revision, and the committed receipt share one database transaction. Ordinary entity creation remains available without an intent. + +Only authenticated requests to hook-free schemas are supported initially. Configured schema hooks or enabled global webhooks refuse reservation and pending commitment. Receipt recovery remains available when hooks are subsequently configured. This protocol does not guarantee delivery or exactly-once execution of external effects. + +## Requests and responses + +All paths below are relative to the existing SchemaForge API mount. Use the same authentication and selected `X-Active-Tenant` as ordinary entity requests. The server binds the intent to the normalized principal, selected tenant, and registered schema identity. Changing any part of this scope cannot recover or commit the intent. Current authorization is evaluated on every request. + +1. `POST /schemas/{schema}/create-intents` with the ordinary `{ "fields": { ... } }` entity body reserves an intent without inserting an entity. A successful response is HTTP 201 with a receipt: + + ```json + { + "id": "createintent_", + "state": "pending", + "expires_at": "2026-01-01T00:15:00Z", + "recover_until": "2026-01-02T00:00:00Z", + "entity_id": null + } + ``` + +2. `POST /schemas/{schema}/entities` with the same body and exactly one `Create-Intent` header commits or reconciles the intent. Treat the identifier as opaque: send the exact returned value, without quotes, whitespace, lists, or additional header instances. Malformed identifiers are HTTP 400. Successful first commitment is HTTP 201; successful reconciliation is HTTP 200. Both return the ordinary entity response and `Create-Intent` response header. The returned entity is its currently authorized representation, including subsequent changes, not a historical response. `Entity-Revision` is present when revision reads have been prepared for the schema. + +3. `GET /schemas/{schema}/create-intents/{id}` returns HTTP 200 and the same receipt structure. States are `pending` with a null entity ID, `committed` with the original currently readable entity ID, or `committed_unavailable` with a null entity ID when the committed record has been deleted. The latter is an outcome, not a successful usable entity. Fetch a committed entity normally to obtain current data and its revision. + +The `fields` fingerprint ignores object-key order recursively. Array order, omitted fields versus explicit null, and parsed integer versus floating-point values remain significant. Server defaults, computed fields, owner injection, and tenant injection are not part of the submitted fingerprint. Unknown or hidden input fields are rejected. Changed submitted content requires a separate explicit create decision; it cannot reuse an intent. + +## Deadlines and uncertain responses + +V1 reserves a 900-second admission window and an 86,400-second recovery window, both measured from server reservation time. The returned timestamps are authoritative. Initial commitment must be admitted while holding the receipt lock strictly before `expires_at`. Lock waits are followed by a fresh server-time check. A transaction admitted before expiry can complete after it. A committed receipt can be recovered strictly before `recover_until`. Expired pending intents, expired recovery windows, unknown IDs, pruned receipts, and differently scoped IDs are refused. Absence never authorizes insertion. + +A lost reservation response may be retried by reserving again: reservation never creates an entity, and orphan pending receipts expire. Once an intent is used for entity creation, preserve that identifier for recovery. + +A pending receipt is a snapshot; an in-flight commit can finish after the read. Retry the same body with the same intent rather than reserving another. A connection failure, timeout, or HTTP 503 can occur after commitment, so it never proves that no record was created. Failure of the current result read can likewise follow a successful commitment. Reconcile the same intent. If the recovery window expires without a known outcome, stop automatic retries and require an explicit operator or user decision; never silently reserve a replacement. + +Validation and business uniqueness failures do not commit an entity and leave the intent pending until its admission deadline. They are not terminal cached outcomes. A later identical request can succeed if the relevant external state changes. Altering the body produces a content conflict. Schema-definition changes refuse an uncommitted intent. Completed receipts retain the original record identity through renames and deletion and never recreate a deleted record. + +## Errors and authorization + +Protocol conflicts use the existing HTTP 409 envelope with `error: "conflict"` and a `reason`: + +| Reason | Meaning | +| --- | --- | +| `create_intent_unsupported` | Backend or configured side-effect path does not support this protocol. | +| `create_intent_unavailable` | Unknown, expired, pruned, or differently scoped intent. No replacement create is attempted. | +| `create_intent_content_conflict` | Submitted fields differ from the reserved input. | +| `create_intent_schema_changed` | An uncommitted reservation's schema definition is no longer current. | +| `unique_violation` | A business uniqueness constraint rejected the insertion; constraint and hidden field names are not disclosed. | +| `create_result_unavailable` | Commitment exists, but POST reconciliation cannot return the deleted result. It was not recreated. | + +Normal validation errors, authentication errors, and authorization denials retain their existing envelopes and statuses. Intent commitment returns the redacted conflict envelope above for business uniqueness errors; ordinary creation without an intent retains its existing `error: "unique_violation"` envelope. HTTP 400 malformed intent errors use `error: "invalid_query"`. HTTP 503 uses the existing backend-unavailable envelope. Do not classify an outcome from human-readable messages. + +Reservation, commitment, and receipt lookup require current create permission. Returning a committed result additionally uses ordinary current record and field read authorization. Denied reads return their authorization error, never `committed` or `committed_unavailable` success. A former owner or tenant member cannot replay historical authorization. Receipt storage contains a content fingerprint and schema snapshot, not an entity response or credentials. Live-name collision rules remain schema/business constraints independent of receipt identity. + +## Operational limits + +Receipts are stored in a reserved internal PostgreSQL table. Expired receipts are removed in bounded batches on reservation. This contract assumes SchemaForge transactional writers and does not cover direct SQL changes to internal tables, restoring partial database state, or external hook effects. Revisions use the existing record-revision infrastructure. Capability metadata advertises `create_reconciliation.protocol = "create-intent-v1"`, PostgreSQL availability, the request header, and the two v1 windows; individual schema support still depends on its hooks and the deployment configuration.