use chio_core::economic_continuity::{
economic_effect_slot_from_head, EconomicContentV1, EconomicEffectSlotV1, EconomicEffectStateV1,
EconomicEffectTerminalV1, EconomicRequestReplayV1, EconomicResourceHeadV1,
EconomicResourceKeyV1, EconomicStateAnchorError, EconomicStateBatchV1,
EconomicStateTransitionV1, EconomicTerminalResultV1, EconomicTransitionAuthorizationV1,
EconomicTransitionProofVerifier, VerifiedEconomicStateView, CHIO_ECONOMIC_RESOURCE_HEAD_SCHEMA,
};
use serde::Serialize;
mod cancellation;
mod dispatch;
mod prepared;
pub use cancellation::*;
pub use dispatch::*;
pub use prepared::*;
use super::validation::digest;
use super::{
derive_channel_service_dispatch_idempotency_key, verify_channel_lifecycle_snapshot,
ChannelError, ChannelEscrowReservationStatusV1, ChannelEscrowReservationViewV1,
ChannelLifecycleStatusV1, ChannelLifecycleViewV1, VerifiedAdmittedChannelReservationV1,
VerifiedChannelOpenConsentV1, VerifiedChannelReceiptBindingV1, VerifiedChannelStateV1,
VerifiedChannelTerminalOutcomeCommitmentV1, CHANNEL_ESCROW_RESERVATION_RESOURCE_FAMILY,
CHANNEL_LIFECYCLE_RESOURCE_FAMILY, CHANNEL_SERVICE_DISPATCH_EFFECT_KIND,
};
const CHANNEL_TERMINAL_TRANSITION_PROOF_SCHEMA: &str = "chio.channel.terminal-transition-proof.v1";
const CHANNEL_TERMINAL_TRANSITION_PROOF_DOMAIN: &[u8] =
b"chio.channel.terminal-transition-proof.digest.v1\0";
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
#[serde(rename_all = "camelCase")]
struct ChannelTerminalTransitionProofV1 {
schema: String,
open_intent_digest: String,
open_digest: String,
prior_state_digest: String,
reservation_digest: String,
operation_id: String,
request: chio_core::economic_continuity::EconomicRequestBindingV1,
provider: chio_core::economic_continuity::EconomicEffectTargetV1,
receipt_id: String,
receipt_digest: String,
receipt_authority_digest: String,
next_state_digest: String,
obligation_atom_id: Option<String>,
obligation_atom_digest: Option<String>,
outcome_id: String,
outcome_digest: String,
source_checkpoint_digest: String,
source_channel_head_digest: String,
source_escrow_head_digest: String,
source_effect_head_digest: String,
terminal_channel_head_digest: String,
terminal_escrow_head_digest: String,
terminal_effect_head_digest: String,
issued_at: u64,
}
impl ChannelTerminalTransitionProofV1 {
fn digest(&self) -> Result<String, ChannelError> {
digest(CHANNEL_TERMINAL_TRANSITION_PROOF_DOMAIN, self)
}
}
#[derive(Debug, Clone)]
pub struct ChannelLifecycleProjectionV1 {
current: VerifiedEconomicStateView,
proof_digest: String,
transitions: Vec<EconomicStateTransitionV1>,
effect_slots: Vec<EconomicEffectSlotV1>,
request_replays: Vec<EconomicRequestReplayV1>,
operation_id: String,
issued_at: u64,
not_after_unix_ms: Option<u64>,
}
impl ChannelLifecycleProjectionV1 {
#[must_use]
pub fn proof_digest(&self) -> &str {
&self.proof_digest
}
#[must_use]
pub fn transitions(&self) -> &[EconomicStateTransitionV1] {
&self.transitions
}
#[must_use]
pub fn effect_slots(&self) -> &[EconomicEffectSlotV1] {
&self.effect_slots
}
#[must_use]
pub fn request_replays(&self) -> &[EconomicRequestReplayV1] {
&self.request_replays
}
#[must_use]
pub fn operation_id(&self) -> Option<&str> {
Some(&self.operation_id)
}
#[must_use]
pub const fn not_after_unix_ms(&self) -> Option<u64> {
self.not_after_unix_ms
}
}
pub fn compose_channel_terminal_transition(
open: &VerifiedChannelOpenConsentV1,
reservation: &VerifiedAdmittedChannelReservationV1,
next_state: &VerifiedChannelStateV1,
receipt: &VerifiedChannelReceiptBindingV1,
outcome: &VerifiedChannelTerminalOutcomeCommitmentV1,
current: &VerifiedEconomicStateView,
issued_at: u64,
) -> Result<ChannelLifecycleProjectionV1, ChannelError> {
let terminal_result = outcome.terminal_result();
terminal_result
.validate()
.map_err(|_| ChannelError::AuthorityVerification)?;
let body = &reservation.artifact().body;
let obligation_time_is_ordered = receipt.obligation_atom().is_none_or(|atom| {
receipt.receipt_timestamp_unix_ms() <= atom.created_at_unix_ms()
&& outcome.terminalized_at_unix_ms() <= atom.created_at_unix_ms()
&& atom.created_at_unix_ms() <= issued_at
});
if issued_at < current.view().observed_at
|| issued_at < outcome.terminalized_at_unix_ms()
|| !obligation_time_is_ordered
{
return Err(ChannelError::AuthorityVerification);
}
let open_intent_digest = open.intent().digest()?;
let open_digest = open.artifact().digest()?;
let prior_state_digest = body.prior_state_digest.clone();
let reservation_digest = reservation.artifact().digest()?;
let next_state_digest = next_state.digest()?;
let outcome_body = &outcome.artifact().body;
let scope_id = &open.intent().body.settlement_authority_scope_id;
let channel_key = EconomicResourceKeyV1 {
resource_family: CHANNEL_LIFECYCLE_RESOURCE_FAMILY.to_owned(),
scope_id: scope_id.clone(),
resource_id: body.channel_id.clone(),
};
let escrow_key = EconomicResourceKeyV1 {
resource_family: CHANNEL_ESCROW_RESERVATION_RESOURCE_FAMILY.to_owned(),
scope_id: scope_id.clone(),
resource_id: body.channel_id.clone(),
};
let snapshot = verify_channel_lifecycle_snapshot(current, scope_id, &body.channel_id)?;
let admitted = reservation.snapshot();
if snapshot.channel_head() != admitted.channel_head()
|| snapshot.escrow_head() != admitted.escrow_head()
|| current.view().checkpoint_sequence <= admitted.checkpoint_sequence()
|| current.view().checkpoint_digest == admitted.checkpoint_digest()
|| current.view().observed_at < admitted.observed_at_unix_ms()
{
return Err(ChannelError::AuthorityVerification);
}
let current_channel_head = snapshot.channel_head();
let current_escrow_head = snapshot.escrow_head();
let effect_key = reservation.ready_effect().resource_head_key();
let current_effect_head = current
.view()
.head(&effect_key)
.ok_or(ChannelError::AuthorityVerification)?;
let dispatch_effect = economic_effect_slot_from_head(current_effect_head)
.map_err(|_| ChannelError::AuthorityVerification)?;
let expected_idempotency_key = derive_channel_service_dispatch_idempotency_key(
&body.operation_id,
&body.reservation_id,
body.next_sequence,
)?;
reservation
.ready_effect()
.validate_successor(&dispatch_effect)
.map_err(|_| ChannelError::AuthorityVerification)?;
let next = next_state.body();
if body.channel_id != open.artifact().body.channel_id
|| body.open_digest != open_digest
|| body.prior_state_digest != prior_state_digest
|| body.next_sequence != next.seq
|| receipt.channel_id() != body.channel_id
|| receipt.open_digest() != open_digest
|| receipt.reservation_digest() != reservation_digest
|| receipt.sequence() != body.next_sequence
|| outcome_body.operation_id != body.operation_id
|| outcome_body.reservation_id != body.reservation_id
|| outcome_body.reservation_digest != reservation_digest
|| outcome_body.receipt_id != receipt.receipt_id()
|| outcome_body.receipt_digest != receipt.receipt_digest()
|| next.channel_id != body.channel_id
|| next.prev_state_digest.as_deref() != Some(prior_state_digest.as_str())
|| next.receipt_id.as_deref() != Some(receipt.receipt_id())
|| next.receipt_digest.as_deref() != Some(receipt.receipt_digest())
|| next.receipt_authority_digest.as_deref() != Some(receipt.receipt_authority_digest())
|| next.obligation_atom_digest.as_deref() != receipt.obligation_atom_digest()
|| next.reservation_digest.as_deref() != Some(reservation_digest.as_str())
|| next.actual_charge.as_ref() != Some(receipt.actual_charge())
|| dispatch_effect.operation_id != body.operation_id
|| dispatch_effect.request.request_id != body.request_id
|| dispatch_effect.effect_kind != CHANNEL_SERVICE_DISPATCH_EFFECT_KIND
|| dispatch_effect.resource_key != channel_key
|| dispatch_effect.resource_head_digest != snapshot.channel_head_digest()
|| dispatch_effect.parameters_digest != reservation_digest
|| dispatch_effect.idempotency_key != expected_idempotency_key
|| dispatch_effect.state != EconomicEffectStateV1::DispatchCommitted
|| dispatch_effect.terminal.is_some()
|| dispatch_effect.frost.is_some()
|| dispatch_effect.resource_head_key() != effect_key
|| current_effect_head.resource_key != effect_key
|| current_effect_head.head_version != 2
|| current_effect_head.resource_version != 2
|| current_effect_head.lifecycle_fence != 2
|| current_effect_head.lifecycle_state != "dispatch_committed"
|| current_effect_head.operation_id.as_deref() != Some(body.operation_id.as_str())
|| current_effect_head.effect_idempotency_key.as_deref()
!= Some(expected_idempotency_key.as_str())
|| current_effect_head.frost.is_some()
|| current_effect_head.terminal_result.is_some()
|| current_effect_head.predecessor_digest.as_deref()
!= Some(reservation.ready_effect_head_digest())
|| current_effect_head.trusted_clock_high_water < admitted.observed_at_unix_ms()
|| current_effect_head.trusted_clock_high_water > current.view().observed_at
{
return Err(ChannelError::AuthorityVerification);
}
let lifecycle = snapshot.lifecycle();
let escrow = snapshot.escrow();
let state_version = lifecycle
.state_version
.checked_add(1)
.ok_or(ChannelError::ArithmeticOverflow)?;
let escrow_version = escrow
.version
.checked_add(1)
.ok_or(ChannelError::ArithmeticOverflow)?;
let lifecycle_fence = lifecycle
.lifecycle_fence
.checked_add(1)
.ok_or(ChannelError::ArithmeticOverflow)?;
let terminal_lifecycle = ChannelLifecycleViewV1 {
schema: lifecycle.schema.clone(),
channel_id: lifecycle.channel_id.clone(),
status: ChannelLifecycleStatusV1::Open,
latest_state_digest: next_state_digest.clone(),
latest_sequence: next.seq,
state_version,
lifecycle_fence,
pending_close_body_digest: None,
admitted_dispute_digest: lifecycle.admitted_dispute_digest.clone(),
live_reservation_id: None,
operation_id: None,
};
let terminal_escrow = ChannelEscrowReservationViewV1 {
schema: escrow.schema.clone(),
channel_id: escrow.channel_id.clone(),
open_digest: escrow.open_digest.clone(),
escrow_reference: escrow.escrow_reference.clone(),
status: ChannelEscrowReservationStatusV1::Open,
version: escrow_version,
lifecycle_fence,
pending_close_body_digest: None,
};
terminal_lifecycle.validate()?;
terminal_escrow.validate()?;
let next_channel_head = successor_head(
current_channel_head,
&terminal_lifecycle,
SuccessorHeadBinding {
resource_version: state_version,
lifecycle_fence,
lifecycle_state: "open",
operation_id: None,
effect_idempotency_key: None,
terminal_result: None,
},
issued_at,
)?;
let next_escrow_head = successor_head(
current_escrow_head,
&terminal_escrow,
SuccessorHeadBinding {
resource_version: escrow_version,
lifecycle_fence,
lifecycle_state: "open",
operation_id: None,
effect_idempotency_key: None,
terminal_result: None,
},
issued_at,
)?;
let mut completed_effect = dispatch_effect.clone();
completed_effect.state = EconomicEffectStateV1::Completed;
completed_effect.terminal = Some(EconomicEffectTerminalV1::Completed {
result_id: terminal_result.result_id.clone(),
result_digest: terminal_result.result_digest.clone(),
result: terminal_result.result.clone(),
});
dispatch_effect
.validate_successor(&completed_effect)
.map_err(|_| ChannelError::AuthorityVerification)?;
let effect_resource_version = current_effect_head
.resource_version
.checked_add(1)
.ok_or(ChannelError::ArithmeticOverflow)?;
let effect_lifecycle_fence = current_effect_head
.lifecycle_fence
.checked_add(1)
.ok_or(ChannelError::ArithmeticOverflow)?;
let next_effect_head = successor_head(
current_effect_head,
&completed_effect,
SuccessorHeadBinding {
resource_version: effect_resource_version,
lifecycle_fence: effect_lifecycle_fence,
lifecycle_state: "completed",
operation_id: Some(body.operation_id.clone()),
effect_idempotency_key: Some(expected_idempotency_key),
terminal_result: Some(terminal_result.clone()),
},
issued_at,
)?;
let source_channel_head_digest = head_digest(current_channel_head)?;
let source_escrow_head_digest = head_digest(current_escrow_head)?;
let source_effect_head_digest = head_digest(current_effect_head)?;
let proof = ChannelTerminalTransitionProofV1 {
schema: CHANNEL_TERMINAL_TRANSITION_PROOF_SCHEMA.to_owned(),
open_intent_digest,
open_digest,
prior_state_digest,
reservation_digest,
operation_id: body.operation_id.clone(),
request: dispatch_effect.request.clone(),
provider: dispatch_effect.target.clone(),
receipt_id: receipt.receipt_id().to_owned(),
receipt_digest: receipt.receipt_digest().to_owned(),
receipt_authority_digest: receipt.receipt_authority_digest().to_owned(),
next_state_digest,
obligation_atom_id: receipt.obligation_atom_id().map(str::to_owned),
obligation_atom_digest: receipt.obligation_atom_digest().map(str::to_owned),
outcome_id: terminal_result.result_id.clone(),
outcome_digest: terminal_result.result_digest.clone(),
source_checkpoint_digest: current.view().checkpoint_digest.clone(),
source_channel_head_digest: source_channel_head_digest.clone(),
source_escrow_head_digest: source_escrow_head_digest.clone(),
source_effect_head_digest: source_effect_head_digest.clone(),
terminal_channel_head_digest: head_digest(&next_channel_head)?,
terminal_escrow_head_digest: head_digest(&next_escrow_head)?,
terminal_effect_head_digest: head_digest(&next_effect_head)?,
issued_at,
};
let proof_digest = proof.digest()?;
let mut transitions = vec![
transition(
channel_key,
source_channel_head_digest,
next_channel_head,
&proof_digest,
),
transition(
escrow_key,
source_escrow_head_digest,
next_escrow_head,
&proof_digest,
),
transition(
effect_key,
source_effect_head_digest,
next_effect_head,
&proof_digest,
),
];
transitions.sort_by(|left, right| left.resource_key.cmp(&right.resource_key));
Ok(ChannelLifecycleProjectionV1 {
current: current.clone(),
proof_digest,
transitions,
effect_slots: Vec::new(),
request_replays: Vec::new(),
operation_id: body.operation_id.clone(),
issued_at,
not_after_unix_ms: None,
})
}
#[derive(Debug, Clone)]
pub struct ChannelLifecycleBatchVerifier {
projection: ChannelLifecycleProjectionV1,
}
impl ChannelLifecycleBatchVerifier {
#[must_use]
pub const fn new(projection: ChannelLifecycleProjectionV1) -> Self {
Self { projection }
}
}
impl EconomicTransitionProofVerifier for ChannelLifecycleBatchVerifier {
fn verify_transition(
&self,
_current: Option<&EconomicResourceHeadV1>,
transition: &EconomicStateTransitionV1,
) -> Result<EconomicTransitionAuthorizationV1, EconomicStateAnchorError> {
Err(EconomicStateAnchorError::TransitionProofRejected(
transition.resource_key.clone(),
))
}
fn verify_batch(
&self,
current: &VerifiedEconomicStateView,
batch: &EconomicStateBatchV1,
) -> Result<Vec<EconomicTransitionAuthorizationV1>, EconomicStateAnchorError> {
let rejected_key = batch
.transitions
.first()
.ok_or(EconomicStateAnchorError::InvalidView(
"channel batch has no transition",
))?
.resource_key
.clone();
let rejected = || EconomicStateAnchorError::TransitionProofRejected(rejected_key.clone());
let expected_sequence = current
.view()
.checkpoint_sequence
.checked_add(1)
.ok_or_else(rejected)?;
if current.view() != self.projection.current.view()
|| batch.anchor_id != current.view().anchor_id
|| batch.namespace != current.view().namespace
|| batch.checkpoint_sequence != expected_sequence
|| batch.previous_checkpoint_digest.as_deref()
!= Some(current.view().checkpoint_digest.as_str())
|| batch.transitions != self.projection.transitions
|| batch.effect_slots != self.projection.effect_slots
|| batch.request_replays != self.projection.request_replays
|| batch.operation_id.as_deref() != Some(self.projection.operation_id.as_str())
|| batch.issued_at != self.projection.issued_at
{
return Err(rejected());
}
Ok(vec![
EconomicTransitionAuthorizationV1::Direct;
batch.transitions.len()
])
}
}
fn transition(
resource_key: EconomicResourceKeyV1,
expected_head_digest: String,
next_head: EconomicResourceHeadV1,
proof_digest: &str,
) -> EconomicStateTransitionV1 {
EconomicStateTransitionV1 {
resource_key,
expected_head_digest: Some(expected_head_digest),
next_head,
transition_proof_digest: proof_digest.to_owned(),
prepared_effect: None,
}
}
struct SuccessorHeadBinding {
resource_version: u64,
lifecycle_fence: u64,
lifecycle_state: &'static str,
operation_id: Option<String>,
effect_idempotency_key: Option<String>,
terminal_result: Option<EconomicTerminalResultV1>,
}
fn successor_head<T: Serialize>(
current: &EconomicResourceHeadV1,
state: &T,
binding: SuccessorHeadBinding,
issued_at: u64,
) -> Result<EconomicResourceHeadV1, ChannelError> {
let state = EconomicContentV1::Inline {
value: serde_json::to_value(state)
.map_err(|error| ChannelError::Canonicalization(error.to_string()))?,
};
let head = EconomicResourceHeadV1 {
schema: CHIO_ECONOMIC_RESOURCE_HEAD_SCHEMA.to_owned(),
anchor_id: current.anchor_id.clone(),
namespace: current.namespace.clone(),
resource_key: current.resource_key.clone(),
head_version: current
.head_version
.checked_add(1)
.ok_or(ChannelError::ArithmeticOverflow)?,
resource_version: binding.resource_version,
lifecycle_fence: binding.lifecycle_fence,
lifecycle_state: binding.lifecycle_state.to_owned(),
state_digest: state
.digest()
.map_err(|_| ChannelError::AuthorityVerification)?,
state,
operation_id: binding.operation_id,
effect_idempotency_key: binding.effect_idempotency_key,
frost: None,
terminal_result: binding.terminal_result,
trusted_clock_high_water: issued_at,
predecessor_digest: Some(head_digest(current)?),
};
head.validate()
.map_err(|_| ChannelError::AuthorityVerification)?;
Ok(head)
}
fn head_digest(head: &EconomicResourceHeadV1) -> Result<String, ChannelError> {
head.digest()
.map_err(|_| ChannelError::AuthorityVerification)
}