#[cfg(feature = "sql")]
use crate::db::codec::{
finalize_hash_sha256, new_hash_sha256_prefixed, write_hash_len_u32, write_hash_u32,
};
#[cfg(feature = "sql")]
use crate::db::data::CanonicalSlotReader;
#[cfg(test)]
use crate::db::data::PersistedRow;
#[cfg(feature = "sql")]
use crate::db::data::persisted_row::types::FieldSlot;
use crate::{
db::{
data::{
CanonicalRow, RawRow, StructuralRowContract,
encode_accepted_value_ref_for_accepted_field_contract,
encode_canonical_value_for_accepted_field_contract,
persisted_row::{
contract::{
RETIRED_SLOT_PLACEHOLDER_PAYLOAD,
canonical_row_from_runtime_value_source_with_accepted_contract,
emit_raw_row_from_slot_payloads,
},
reader::StructuralSlotReader,
types::{
AcceptedInsertPolicyRequest, AcceptedMutationFieldWriteIntent,
AcceptedMutationIntentPatch, SlotReader,
},
},
},
schema::{
AcceptedFieldPersistenceContract, AcceptedInsertOmissionPolicy,
AcceptedRowDecodeContract,
authored_projection::{AcceptedAuthoredFieldProjection, AuthoredFieldAdmissionError},
enum_catalog::{ValueAdmissionBudget, ValueAdmissionError},
},
},
error::InternalError,
model::field::{FieldInsertGeneration, FieldWriteManagement},
sanitize::SanitizeWriteContext,
traits::AuthoredFieldProjection,
types::Ulid,
value::{InputValue, Value},
};
#[cfg(feature = "sql")]
use sha2::Digest;
use std::borrow::Cow;
#[cfg(feature = "sql")]
const ACCEPTED_FIXED_UPDATE_PATCH_FINGERPRINT_DOMAIN: &[u8] =
b"icydb.accepted-fixed-update-patch.v1";
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(in crate::db) enum AcceptedFieldWriteProvenance {
Authored,
ResolvedDefault(AcceptedInsertPolicyRequest),
ResolvedNull(AcceptedInsertPolicyRequest),
InsertGenerated(AcceptedInsertPolicyRequest),
InsertManaged(AcceptedInsertPolicyRequest),
UpdateManaged,
Preserved,
PreservedReplacementIdentity,
HistoricalFill,
}
impl AcceptedFieldWriteProvenance {
#[must_use]
pub(in crate::db) const fn sanitizer_may_transform(self) -> bool {
matches!(self, Self::Authored)
}
}
pub(in crate::db) struct ResolvedAcceptedMutationRow {
row: CanonicalRow,
provenance: Vec<Option<AcceptedFieldWriteProvenance>>,
}
#[derive(Clone, Debug, Eq, PartialEq)]
#[cfg(feature = "sql")]
pub(in crate::db) struct AcceptedFixedUpdateField {
slot: FieldSlot,
payload: Vec<u8>,
}
#[cfg(feature = "sql")]
impl AcceptedFixedUpdateField {
#[must_use]
pub(in crate::db) const fn slot(&self) -> FieldSlot {
self.slot
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
#[cfg(feature = "sql")]
pub(in crate::db) struct AcceptedFixedUpdatePatch {
fields: Vec<AcceptedFixedUpdateField>,
fingerprint: [u8; 32],
}
#[cfg(feature = "sql")]
impl AcceptedFixedUpdatePatch {
pub(in crate::db) fn from_update_intent(
entity_path: &'static str,
accepted_decode_contract: AcceptedRowDecodeContract,
patch: &AcceptedMutationIntentPatch,
) -> Result<Self, InternalError> {
let contract = StructuralRowContract::from_accepted_decode_contract(
entity_path,
accepted_decode_contract,
);
let mut intents = vec![None; contract.field_count()];
for entry in patch.entries() {
let slot = entry.slot().index();
let _ = contract.required_accepted_field_contract(slot)?;
intents[slot] = Some(entry.intent().clone());
}
let mut fields = Vec::with_capacity(patch.entries().len());
for (slot, intent) in intents.into_iter().enumerate() {
let Some(intent) = intent else {
continue;
};
let payload = match intent {
AcceptedMutationFieldWriteIntent::Authored(input) => {
encode_authored_value_for_accepted_field_contract(
contract.required_accepted_field_persistence_contract(slot)?,
input,
)?
}
AcceptedMutationFieldWriteIntent::Resolve(
AcceptedInsertPolicyRequest::ExplicitUpdateDefault,
) => resolve_explicit_update_default(&contract, slot)?.0,
AcceptedMutationFieldWriteIntent::PreservedReplacementIdentity(_)
| AcceptedMutationFieldWriteIntent::Resolve(
AcceptedInsertPolicyRequest::OmittedInsert
| AcceptedInsertPolicyRequest::ExplicitInsertDefault,
) => return Err(InternalError::executor_invariant()),
};
fields.push(AcceptedFixedUpdateField {
slot: FieldSlot::from_validated_index(slot),
payload,
});
}
if fields.is_empty() {
return Err(InternalError::executor_invariant());
}
let mut hasher = new_hash_sha256_prefixed(ACCEPTED_FIXED_UPDATE_PATCH_FINGERPRINT_DOMAIN);
write_hash_len_u32(&mut hasher, fields.len());
for field in &fields {
write_hash_u32(
&mut hasher,
u32::try_from(field.slot.index())
.map_err(|_| InternalError::executor_invariant())?,
);
write_hash_len_u32(&mut hasher, field.payload.len());
hasher.update(field.payload.as_slice());
}
Ok(Self {
fields,
fingerprint: finalize_hash_sha256(hasher),
})
}
#[must_use]
pub(in crate::db) const fn fields(&self) -> &[AcceptedFixedUpdateField] {
self.fields.as_slice()
}
pub(in crate::db) fn is_satisfied_by(
&self,
row: &dyn CanonicalSlotReader,
) -> Result<bool, InternalError> {
for field in &self.fields {
if row.required_bytes(field.slot.index())? != field.payload.as_slice() {
return Ok(false);
}
}
Ok(true)
}
#[must_use]
pub(in crate::db) const fn fingerprint(&self) -> [u8; 32] {
self.fingerprint
}
}
impl ResolvedAcceptedMutationRow {
#[must_use]
const fn new(row: CanonicalRow, provenance: Vec<Option<AcceptedFieldWriteProvenance>>) -> Self {
Self { row, provenance }
}
#[must_use]
pub(in crate::db) fn into_parts(
self,
) -> (CanonicalRow, Vec<Option<AcceptedFieldWriteProvenance>>) {
(self.row, self.provenance)
}
}
#[cfg(test)]
pub(in crate::db) fn canonical_row_from_entity_for_model_proposal_for_test<E>(
entity: &E,
) -> Result<CanonicalRow, InternalError>
where
E: PersistedRow,
{
let accepted_decode_contract =
AcceptedRowDecodeContract::from_model_proposal_for_test(E::MODEL);
canonical_row_from_entity_with_accepted_contract(
E::MODEL.path(),
accepted_decode_contract,
entity,
)
}
#[cfg(test)]
pub(in crate::db) fn canonical_row_from_entity_with_accepted_contract<E>(
entity_path: &'static str,
accepted_decode_contract: AcceptedRowDecodeContract,
entity: &E,
) -> Result<CanonicalRow, InternalError>
where
E: AuthoredFieldProjection,
{
canonical_row_from_entity_with_optional_resolved_row(
entity_path,
accepted_decode_contract,
entity,
None,
)
}
pub(in crate::db) fn canonical_row_from_resolved_entity_with_accepted_contract<E>(
entity_path: &'static str,
accepted_decode_contract: AcceptedRowDecodeContract,
entity: &E,
resolved_row: &RawRow,
) -> Result<CanonicalRow, InternalError>
where
E: AuthoredFieldProjection,
{
canonical_row_from_entity_with_optional_resolved_row(
entity_path,
accepted_decode_contract,
entity,
Some(resolved_row),
)
}
fn canonical_row_from_entity_with_optional_resolved_row<E>(
entity_path: &'static str,
accepted_decode_contract: AcceptedRowDecodeContract,
entity: &E,
resolved_row: Option<&RawRow>,
) -> Result<CanonicalRow, InternalError>
where
E: AuthoredFieldProjection,
{
let authored = AcceptedAuthoredFieldProjection::new(&accepted_decode_contract);
let contract = StructuralRowContract::from_accepted_decode_contract(
entity_path,
accepted_decode_contract.clone(),
);
let resolved = resolved_row
.map(|row| {
StructuralSlotReader::from_raw_row_with_validated_borrowed_contract(row, &contract)
})
.transpose()?;
let mut slot_payloads = Vec::with_capacity(contract.field_count());
for slot in 0..contract.field_count() {
if !contract.has_active_field_slot(slot) {
slot_payloads.push(RETIRED_SLOT_PLACEHOLDER_PAYLOAD.to_vec());
continue;
}
let field = accepted_decode_contract
.field_for_slot(slot)
.ok_or_else(InternalError::persisted_row_encode_internal)?;
if field.generated() {
let mut budget = ValueAdmissionBudget::standard();
slot_payloads.push(
authored
.encode_field(entity, slot, &mut budget)
.map_err(authored_field_admission_error)?,
);
continue;
}
if let Some(resolved) = resolved.as_ref() {
let payload = resolved
.get_bytes(slot)
.ok_or_else(InternalError::persisted_row_encode_internal)?;
slot_payloads.push(payload.to_vec());
continue;
}
let value = contract.insert_omission_value(slot)?;
let encoding = contract.required_accepted_field_persistence_contract(slot)?;
slot_payloads.push(encode_canonical_value_for_accepted_field_contract(
encoding, &value,
)?);
}
emit_raw_row_from_slot_payloads(
contract.current_layout_version(),
contract.field_count(),
slot_payloads.as_slice(),
)
}
fn authored_field_admission_error(error: AuthoredFieldAdmissionError) -> InternalError {
match error {
AuthoredFieldAdmissionError::Admission(error) => value_admission_error(error),
AuthoredFieldAdmissionError::FieldNotGenerated { .. }
| AuthoredFieldAdmissionError::MissingAuthoredValue { .. }
| AuthoredFieldAdmissionError::MissingFieldContract { .. } => {
InternalError::executor_invariant()
}
AuthoredFieldAdmissionError::PersistenceEncoding { .. } => {
InternalError::persisted_row_encode_internal()
}
}
}
fn value_admission_error(error: ValueAdmissionError) -> InternalError {
match error {
ValueAdmissionError::InvalidAcceptedContract
| ValueAdmissionError::MissingSchemaRevision
| ValueAdmissionError::UnknownCompositeType => InternalError::executor_invariant(),
ValueAdmissionError::DepthExceeded
| ValueAdmissionError::SizeExceeded
| ValueAdmissionError::TypeMismatch
| ValueAdmissionError::ScalarConstraint
| ValueAdmissionError::EnumPathMismatch
| ValueAdmissionError::EnumTypeMismatch
| ValueAdmissionError::UnknownEnumType
| ValueAdmissionError::UnknownEnumVariant
| ValueAdmissionError::EnumBodyMismatch
| ValueAdmissionError::CompositeShapeMismatch
| ValueAdmissionError::CompositeFieldMismatch
| ValueAdmissionError::DuplicateSetItem
| ValueAdmissionError::DuplicateMapKey => InternalError::executor_unsupported(),
}
}
pub(in crate::db) fn canonical_row_from_structural_slot_reader_with_accepted_contract(
row_fields: &StructuralSlotReader<'_>,
) -> Result<CanonicalRow, InternalError> {
canonical_row_from_runtime_value_source_with_accepted_contract(row_fields.contract(), |slot| {
structural_slot_reader_value(row_fields, slot)
})
}
pub(in crate::db) fn canonical_row_from_raw_row_with_structural_contract(
raw_row: &RawRow,
contract: &StructuralRowContract,
) -> Result<CanonicalRow, InternalError> {
let row_fields =
StructuralSlotReader::from_raw_row_with_validated_borrowed_contract(raw_row, contract)?;
canonical_row_from_structural_slot_reader_with_accepted_contract(&row_fields)
}
pub(in crate::db) fn canonical_row_from_raw_row_with_accepted_decode_contract(
entity_path: &'static str,
accepted_decode_contract: AcceptedRowDecodeContract,
raw_row: &RawRow,
) -> Result<CanonicalRow, InternalError> {
let contract =
StructuralRowContract::from_accepted_decode_contract(entity_path, accepted_decode_contract);
canonical_row_from_raw_row_with_structural_contract(raw_row, &contract)
}
pub(in crate::db) const fn canonical_row_from_stored_raw_row(raw_row: RawRow) -> CanonicalRow {
CanonicalRow::from_canonical_raw_row(raw_row)
}
fn encode_authored_value_for_accepted_field_contract(
encoding: AcceptedFieldPersistenceContract<'_>,
input: InputValue,
) -> Result<Vec<u8>, InternalError> {
let field = encoding.field();
let mut budget = ValueAdmissionBudget::standard();
encoding
.admission_contract()
.with_normalized(input, &mut budget, |accepted| {
encode_accepted_value_ref_for_accepted_field_contract(field, &accepted)
})
.map_err(value_admission_error)?
.map_err(|_| InternalError::persisted_row_field_encode_internal(field.field_name()))
}
fn resolve_insert_active_slot(
entity_path: &'static str,
contract: &StructuralRowContract,
slot: usize,
intent: Option<AcceptedMutationFieldWriteIntent>,
write_context: SanitizeWriteContext,
) -> Result<(Vec<u8>, AcceptedFieldWriteProvenance), InternalError> {
let field = contract.required_accepted_field_contract(slot)?;
let write_policy = field.write_policy();
let request = match intent {
Some(AcceptedMutationFieldWriteIntent::Authored(input)) => {
if write_policy.insert_generation().is_some()
|| write_policy.write_management().is_some()
{
return Err(InternalError::mutation_database_owned_field_explicit(
entity_path,
field.field_name(),
));
}
let encoding = contract.required_accepted_field_persistence_contract(slot)?;
let payload = encode_authored_value_for_accepted_field_contract(encoding, input)?;
return Ok((payload, AcceptedFieldWriteProvenance::Authored));
}
Some(AcceptedMutationFieldWriteIntent::PreservedReplacementIdentity(input)) => {
if !contract.primary_key_slot_indices().contains(&slot)
|| (write_policy.insert_generation().is_none()
&& write_policy.write_management().is_none())
{
return Err(InternalError::executor_invariant());
}
let encoding = contract.required_accepted_field_persistence_contract(slot)?;
let payload = encode_authored_value_for_accepted_field_contract(encoding, input)?;
return Ok((
payload,
AcceptedFieldWriteProvenance::PreservedReplacementIdentity,
));
}
Some(AcceptedMutationFieldWriteIntent::Resolve(
AcceptedInsertPolicyRequest::ExplicitInsertDefault,
)) => AcceptedInsertPolicyRequest::ExplicitInsertDefault,
Some(AcceptedMutationFieldWriteIntent::Resolve(
AcceptedInsertPolicyRequest::OmittedInsert
| AcceptedInsertPolicyRequest::ExplicitUpdateDefault,
)) => return Err(InternalError::executor_invariant()),
None => AcceptedInsertPolicyRequest::OmittedInsert,
};
if let Some(generation) = write_policy.insert_generation() {
let value = accepted_insert_generated_value(generation, write_context);
let encoding = contract.required_accepted_field_persistence_contract(slot)?;
let payload = encode_canonical_value_for_accepted_field_contract(encoding, &value)?;
return Ok((
payload,
AcceptedFieldWriteProvenance::InsertGenerated(request),
));
}
if let Some(management) = write_policy.write_management() {
let value = accepted_insert_managed_value(management, write_context);
let encoding = contract.required_accepted_field_persistence_contract(slot)?;
let payload = encode_canonical_value_for_accepted_field_contract(encoding, &value)?;
return Ok((
payload,
AcceptedFieldWriteProvenance::InsertManaged(request),
));
}
let provenance = match field.insert_omission_policy() {
AcceptedInsertOmissionPolicy::NullIfMissing => {
AcceptedFieldWriteProvenance::ResolvedNull(request)
}
AcceptedInsertOmissionPolicy::DefaultIfMissing => {
AcceptedFieldWriteProvenance::ResolvedDefault(request)
}
AcceptedInsertOmissionPolicy::Required => {
if matches!(request, AcceptedInsertPolicyRequest::ExplicitInsertDefault) {
return Err(InternalError::query_sql_write_boundary(
icydb_diagnostic_code::SqlWriteBoundaryCode::InsertDefaultRequiredField,
));
}
return Err(InternalError::mutation_required_field_missing(
entity_path,
field.field_name(),
));
}
};
Ok((contract.insert_omission_payload(slot)?, provenance))
}
pub(in crate::db) fn resolve_insert_structural_patch_with_accepted_contract(
entity_path: &'static str,
accepted_decode_contract: AcceptedRowDecodeContract,
patch: &AcceptedMutationIntentPatch,
write_context: SanitizeWriteContext,
) -> Result<ResolvedAcceptedMutationRow, InternalError> {
let contract =
StructuralRowContract::from_accepted_decode_contract(entity_path, accepted_decode_contract);
let mut payloads = vec![None; contract.field_count()];
let mut provenance = vec![None; contract.field_count()];
let mut intents = vec![None; contract.field_count()];
for entry in patch.entries() {
let slot = entry.slot().index();
let _ = contract.required_accepted_field_contract(slot)?;
intents[slot] = Some(entry.intent().clone());
}
for slot in 0..contract.field_count() {
if !contract.has_active_field_slot(slot) {
payloads[slot] = Some(RETIRED_SLOT_PLACEHOLDER_PAYLOAD.to_vec());
continue;
}
let (payload, source) = resolve_insert_active_slot(
entity_path,
&contract,
slot,
intents[slot].take(),
write_context,
)?;
payloads[slot] = Some(payload);
provenance[slot] = Some(source);
}
let slot_payloads = payloads
.into_iter()
.map(|payload| payload.ok_or_else(InternalError::persisted_row_encode_internal))
.collect::<Result<Vec<_>, _>>()?;
let row = emit_raw_row_from_slot_payloads(
contract.current_layout_version(),
contract.field_count(),
slot_payloads.as_slice(),
)?;
Ok(ResolvedAcceptedMutationRow::new(row, provenance))
}
pub(in crate::db) fn resolve_update_structural_patch_with_accepted_contract(
entity_path: &'static str,
accepted_decode_contract: AcceptedRowDecodeContract,
raw_row: &RawRow,
patch: &AcceptedMutationIntentPatch,
write_context: SanitizeWriteContext,
) -> Result<ResolvedAcceptedMutationRow, InternalError> {
let contract =
StructuralRowContract::from_accepted_decode_contract(entity_path, accepted_decode_contract);
let baseline =
StructuralSlotReader::from_raw_row_with_validated_borrowed_contract(raw_row, &contract)?;
let mut payloads = vec![None; contract.field_count()];
let mut provenance = vec![None; contract.field_count()];
let mut intents = vec![None; contract.field_count()];
for entry in patch.entries() {
let slot = entry.slot().index();
let _ = contract.required_accepted_field_contract(slot)?;
intents[slot] = Some(entry.intent().clone());
}
for slot in 0..contract.field_count() {
if payloads[slot].is_some() {
continue;
}
if !contract.has_active_field_slot(slot) {
payloads[slot] = Some(RETIRED_SLOT_PLACEHOLDER_PAYLOAD.to_vec());
continue;
}
let field = contract.required_accepted_field_contract(slot)?;
let write_policy = field.write_policy();
match intents[slot].take() {
Some(AcceptedMutationFieldWriteIntent::Authored(input)) => {
if write_policy.insert_generation().is_some()
|| write_policy.write_management().is_some()
{
return Err(InternalError::mutation_database_owned_field_explicit(
entity_path,
field.field_name(),
));
}
let encoding = contract.required_accepted_field_persistence_contract(slot)?;
payloads[slot] = Some(encode_authored_value_for_accepted_field_contract(
encoding, input,
)?);
provenance[slot] = Some(AcceptedFieldWriteProvenance::Authored);
continue;
}
Some(AcceptedMutationFieldWriteIntent::PreservedReplacementIdentity(_)) => {
return Err(InternalError::executor_invariant());
}
Some(AcceptedMutationFieldWriteIntent::Resolve(
AcceptedInsertPolicyRequest::ExplicitUpdateDefault,
)) => {
let (payload, resolved_provenance) =
resolve_explicit_update_default(&contract, slot)?;
payloads[slot] = Some(payload);
provenance[slot] = Some(resolved_provenance);
continue;
}
Some(AcceptedMutationFieldWriteIntent::Resolve(
AcceptedInsertPolicyRequest::OmittedInsert
| AcceptedInsertPolicyRequest::ExplicitInsertDefault,
)) => return Err(InternalError::executor_invariant()),
None => {}
}
if matches!(
write_policy.write_management(),
Some(FieldWriteManagement::UpdatedAt)
) {
let value = Value::Timestamp(write_context.now());
let encoding = contract.required_accepted_field_persistence_contract(slot)?;
payloads[slot] = Some(encode_canonical_value_for_accepted_field_contract(
encoding, &value,
)?);
provenance[slot] = Some(AcceptedFieldWriteProvenance::UpdateManaged);
continue;
}
let value = baseline.required_cached_value(slot)?;
let encoding = contract.required_accepted_field_persistence_contract(slot)?;
payloads[slot] = Some(encode_canonical_value_for_accepted_field_contract(
encoding, value,
)?);
provenance[slot] = Some(if baseline.get_bytes(slot).is_some() {
AcceptedFieldWriteProvenance::Preserved
} else {
AcceptedFieldWriteProvenance::HistoricalFill
});
}
let slot_payloads = payloads
.into_iter()
.map(|payload| payload.ok_or_else(InternalError::persisted_row_encode_internal))
.collect::<Result<Vec<_>, _>>()?;
let row = emit_raw_row_from_slot_payloads(
contract.current_layout_version(),
contract.field_count(),
slot_payloads.as_slice(),
)?;
Ok(ResolvedAcceptedMutationRow::new(row, provenance))
}
fn resolve_explicit_update_default(
contract: &StructuralRowContract,
slot: usize,
) -> Result<(Vec<u8>, AcceptedFieldWriteProvenance), InternalError> {
let field = contract.required_accepted_field_contract(slot)?;
let write_policy = field.write_policy();
if write_policy.insert_generation().is_some() || write_policy.write_management().is_some() {
return Err(InternalError::query_sql_write_boundary(
icydb_diagnostic_code::SqlWriteBoundaryCode::UpdateDefaultDatabaseOwnedField,
));
}
let provenance = match field.insert_omission_policy() {
AcceptedInsertOmissionPolicy::NullIfMissing => AcceptedFieldWriteProvenance::ResolvedNull(
AcceptedInsertPolicyRequest::ExplicitUpdateDefault,
),
AcceptedInsertOmissionPolicy::DefaultIfMissing => {
AcceptedFieldWriteProvenance::ResolvedDefault(
AcceptedInsertPolicyRequest::ExplicitUpdateDefault,
)
}
AcceptedInsertOmissionPolicy::Required => {
return Err(InternalError::query_sql_write_boundary(
icydb_diagnostic_code::SqlWriteBoundaryCode::UpdateDefaultRequiredField,
));
}
};
Ok((contract.insert_omission_payload(slot)?, provenance))
}
fn accepted_insert_generated_value(
generation: FieldInsertGeneration,
write_context: SanitizeWriteContext,
) -> Value {
match generation {
FieldInsertGeneration::Ulid => Value::Ulid(Ulid::generate()),
FieldInsertGeneration::Timestamp => Value::Timestamp(write_context.now()),
}
}
const fn accepted_insert_managed_value(
management: FieldWriteManagement,
write_context: SanitizeWriteContext,
) -> Value {
match management {
FieldWriteManagement::CreatedAt | FieldWriteManagement::UpdatedAt => {
Value::Timestamp(write_context.now())
}
}
}
fn structural_slot_reader_value<'a>(
row_fields: &'a StructuralSlotReader<'_>,
slot: usize,
) -> Result<Cow<'a, Value>, InternalError> {
row_fields
.required_cached_value(slot)
.map(Cow::Borrowed)
.map_err(|_| InternalError::persisted_row_encode_internal())
}