saddle-framework 0.3.15

use std::sync::{Arc, Mutex};

use saddle_core::CallContext;
use saddle_db::internal::{
    ManagedOptionalRow, ManagedWriteResult, QueryOptionalOperationProof, ScopeDatabaseError,
    StaticQueryOptionalOperation, StaticWriteOperation, WriteOperationProof,
};
use saddle_observability::EventContext;
use saddle_runtime::profusegw::{
    ProfuseGwManagedDispatch, ProfuseGwPostDatabaseRequestTerminal,
    finish_profusegw_after_database, finish_profusegw_without_database,
};
use tokio::sync::Notify;

/// Generated named ABI conversion; only generated code may implement this
/// internal surface. It carries no runtime execution or admission authority.
#[doc(hidden)]
pub trait GeneratedNamedQuery: StaticQueryOptionalOperation {
    type NamedParameters: Send;
    type NamedRow: Send;
    const REQUIRES_TRANSACTION: bool;
    fn parameters(value: Self::NamedParameters) -> Self::Parameters;
    fn row(value: ManagedOptionalRow<Self::Row>) -> Option<Self::NamedRow>;
}

#[doc(hidden)]
pub trait GeneratedNamedWrite: StaticWriteOperation {
    type NamedParameters: Send;
    fn parameters(value: Self::NamedParameters) -> Self::Parameters;
}

/// Generated conversion only, not a new execution/decoder authority.
#[doc(hidden)]
pub trait GeneratedOwnedQuery: saddle_db::internal::StaticOwnedQuery {
    type NamedParameters: Send;
    type NamedRow: From<Self::Row> + Send;
    fn parameters(value: Self::NamedParameters) -> Self::Parameters;
    fn restore_parameters(value: Self::Parameters) -> Self::NamedParameters;
    fn validate_parameters(
        _: &Self::NamedParameters,
        _: &DatabaseTarget,
    ) -> Result<(), saddle_db::internal::TimestampTargetError> {
        Ok(())
    }
}
#[doc(hidden)]
pub trait GeneratedOwnedWrite: saddle_db::internal::StaticOwnedWrite {
    type NamedParameters: Send;
    fn parameters(value: Self::NamedParameters) -> Self::Parameters;
    fn restore_parameters(value: Self::Parameters) -> Self::NamedParameters;
    fn validate_parameters(
        _: &Self::NamedParameters,
        _: &DatabaseTarget,
    ) -> Result<(), saddle_db::internal::TimestampTargetError> {
        Ok(())
    }
}

/// Internal borrowed target access, not a range constructor or execution owner.
#[doc(hidden)]
pub struct DatabaseTarget(
    Option<Arc<saddle_db::internal::StartupManagedDatabaseProcessCapability>>,
);
impl DatabaseTarget {
    pub fn timestamp<const F: usize>(
        &self,
        value: &saddle_db::internal::DbTimestamp<F>,
    ) -> Result<(), saddle_db::internal::TimestampTargetError> {
        self.0
            .as_ref()
            .ok_or(saddle_db::internal::TimestampTargetError::ProcessClosed)?
            .native_timestamp_range()?
            .validate(value)
    }
}

/// A source-rejected argument is returned intact for an explicit caller decision.
pub enum QueryFailure<P> {
    Parameters {
        parameters: P,
    },
    Target {
        parameters: P,
        error: saddle_db::internal::TimestampTargetError,
    },
    Database(ScopeDatabaseError),
}
impl<P> std::fmt::Debug for QueryFailure<P> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Parameters { .. } => f.write_str("Parameters"),
            Self::Target { error, .. } => f.debug_tuple("Target").field(error).finish(),
            Self::Database(e) => f.debug_tuple("Database").field(e).finish(),
        }
    }
}
impl<P> QueryFailure<P> {
    /// Explicitly abandon any returned arguments and retain the technical failure.
    pub fn into_failure(self) -> ScopeDatabaseError {
        match self {
            Self::Parameters { .. } => ScopeDatabaseError::Resource,
            Self::Target { error, .. } => match error {
                saddle_db::internal::TimestampTargetError::OutOfRange => {
                    ScopeDatabaseError::InvalidValue
                }
                saddle_db::internal::TimestampTargetError::UnsupportedTarget
                | saddle_db::internal::TimestampTargetError::ProcessClosed => {
                    ScopeDatabaseError::Unavailable
                }
            },
            Self::Database(e) => e,
        }
    }
}
fn named_failure<O: GeneratedOwnedQuery>(
    error: saddle_db::internal::OwnedExecutionError<O::Parameters>,
) -> QueryFailure<O::NamedParameters> {
    match error {
        saddle_db::internal::OwnedExecutionError::Parameters { parameters, .. } => {
            QueryFailure::Parameters {
                parameters: O::restore_parameters(parameters),
            }
        }
        saddle_db::internal::OwnedExecutionError::Database(e) => QueryFailure::Database(e),
    }
}

#[doc(hidden)]
pub async fn transaction_rows<O: GeneratedOwnedQuery>(
    session: &mut TransactionSession<'_, '_>,
    parameters: O::NamedParameters,
) -> Result<crate::database::Rows<O::Row, O::NamedRow>, QueryFailure<O::NamedParameters>> {
    let proof =
        saddle_db::internal::OwnedQueryProof::<O>::bind().map_err(QueryFailure::Database)?;
    session
        .owned_rows(proof.invocation(O::parameters(parameters)))
        .await
        .map(crate::database_rows::Rows::from_database)
        .map_err(named_failure::<O>)
}
#[doc(hidden)]
pub async fn transaction_optional<O: GeneratedOwnedQuery>(
    session: &mut TransactionSession<'_, '_>,
    parameters: O::NamedParameters,
) -> Result<Option<O::NamedRow>, QueryFailure<O::NamedParameters>> {
    let proof =
        saddle_db::internal::OwnedQueryProof::<O>::bind().map_err(QueryFailure::Database)?;
    session
        .owned_optional(proof.invocation(O::parameters(parameters)))
        .await
        .map(|row| row.map(O::NamedRow::from))
        .map_err(named_failure::<O>)
}
#[doc(hidden)]
pub async fn transaction_probe<O: GeneratedOwnedQuery>(
    session: &mut TransactionSession<'_, '_>,
    parameters: O::NamedParameters,
) -> Result<crate::database::MatchCardinality<O::NamedRow>, QueryFailure<O::NamedParameters>> {
    use crate::database::MatchCardinality;
    use saddle_db::internal::OwnedDbProbe;
    let proof =
        saddle_db::internal::OwnedQueryProof::<O>::bind().map_err(QueryFailure::Database)?;
    session
        .owned_probe(proof.invocation(O::parameters(parameters)))
        .await
        .map(|row| match row {
            OwnedDbProbe::Zero => MatchCardinality::Zero,
            OwnedDbProbe::One(row) => MatchCardinality::One(O::NamedRow::from(row)),
            OwnedDbProbe::Many => MatchCardinality::Many,
        })
        .map_err(named_failure::<O>)
}

#[doc(hidden)]
pub trait GeneratedPageQuery: GeneratedOwnedQuery {
    const IS_OFFSET: bool;
    type Key: Copy + Send + AsRef<[u64]> + AsMut<[u64]>;
    type Cursor: crate::database_rows::CursorWire;
    fn empty_key() -> Self::Key;
    fn filter_fingerprint(value: &Self::NamedParameters) -> [u8; 32];
    fn page_parameters<'a>(
        value: &'a mut Self::NamedParameters,
        after: Option<Self::Key>,
        request: &'a mut DatabaseRequest,
    ) -> std::pin::Pin<
        Box<
            dyn std::future::Future<
                    Output = Result<
                        Self::Parameters,
                        saddle_db::internal::ParameterConstructionError,
                    >,
                > + Send
                + 'a,
        >,
    >;
    fn page_key(row: &Self::Row) -> Self::Key;
}
#[doc(hidden)]
pub trait GeneratedOffsetPageQuery: GeneratedPageQuery {}

enum RequestTerminal {
    Untouched(ProfuseGwManagedDispatch),
    InFlight,
    PostDatabase(ProfuseGwPostDatabaseRequestTerminal),
    Serial(SerialScope),
    Finished,
}

struct RequestState {
    terminal: Mutex<RequestTerminal>,
    cancel: Arc<Notify>,
    completed: Notify,
}

struct AbortOnLostRequestOwner(bool);
type SerialScope =
    saddle_runtime::profusegw::ProfuseGwSerialScope<tokio::sync::futures::OwnedNotified>;
#[doc(hidden)]
pub type TransactionSession<'session, 'request> = saddle_db::internal::DatabaseTransactionSession<
    'session,
    'request,
    tokio::sync::futures::OwnedNotified,
>;

impl Drop for AbortOnLostRequestOwner {
    fn drop(&mut self) {
        if self.0 {
            std::process::abort();
        }
    }
}

/// Framework-owned request database carrier. Generated capabilities retain
/// only a private reference to this state; business code cannot construct or
/// recover Runtime or Database owners from it.
#[doc(hidden)]
pub struct DatabaseRequest {
    process: Option<Arc<saddle_db::internal::StartupManagedDatabaseProcessCapability>>,
    context: CallContext,
    event: EventContext,
    state: Arc<RequestState>,
}

#[doc(hidden)]
pub struct DatabaseRequestCompletion {
    state: Arc<RequestState>,
}

#[doc(hidden)]
pub struct ResponseEncodingContext {
    state: Arc<RequestState>,
}
impl ResponseEncodingContext {
    pub async fn encode<F>(self, action: F) -> saddle_core::Result<Vec<u8>>
    where
        F: FnOnce() -> saddle_core::Result<Vec<u8>> + Send,
    {
        let scope = {
            let mut terminal = self.state.terminal.lock().unwrap();
            match std::mem::replace(&mut *terminal, RequestTerminal::InFlight) {
                RequestTerminal::Serial(scope) => Some(scope),
                other => {
                    *terminal = other;
                    None
                }
            }
        };
        let encode = async move { action() };
        if let Some(scope) = scope {
            let mut guard = ConstructionScope {
                scope: Some(scope),
                state: self.state,
            };
            guard
                .scope
                .as_mut()
                .unwrap()
                .supervise_between(encode)
                .await
                .map_err(|_| crate::process::process_dispatch_error())?
        } else {
            encode.await
        }
    }
}

// Construction never enters SQL. Restore the same scope even when the borrowing
// future is dropped by the outer request deadline/disconnect select.
struct ConstructionScope {
    scope: Option<SerialScope>,
    state: Arc<RequestState>,
}
impl Drop for ConstructionScope {
    fn drop(&mut self) {
        if let Some(scope) = self.scope.take() {
            *self.state.terminal.lock().unwrap() = RequestTerminal::Serial(scope);
            self.state.completed.notify_one();
        }
    }
}
macro_rules! parameter_constructor {
    ($name:ident,$input:ty,$output:ty,$factory:ident,$convert:expr) => {
        pub async fn $name(
            &mut self,
            source: $input,
        ) -> Result<$output, saddle_db::internal::ParameterConstructionError> {
            use saddle_db::internal::{DatabaseParameterFactory, ParameterConstructionError};
            let mut scope = self.construction_scope()?;
            let construction = scope
                .scope
                .as_mut()
                .unwrap()
                .parameter_construction()
                .map_err(ParameterConstructionError::Scope)?;
            DatabaseParameterFactory::new(&construction)
                .$factory(source)
                .await
                .map($convert)
        }
    };
}
impl DatabaseRequest {
    pub fn parameter_factory(&mut self) -> crate::database_uid::ParameterFactory<'_> {
        crate::database_uid::ParameterFactory { request: self }
    }
    pub fn response_encoding_context(&self) -> ResponseEncodingContext {
        ResponseEncodingContext {
            state: Arc::clone(&self.state),
        }
    }
    fn construction_scope(
        &mut self,
    ) -> Result<ConstructionScope, saddle_db::internal::ParameterConstructionError> {
        use saddle_db::internal::ParameterConstructionError;
        let mut terminal = self.state.terminal.lock().unwrap();
        let scope = match std::mem::replace(&mut *terminal, RequestTerminal::InFlight) {
            RequestTerminal::Untouched(dispatch) => {
                dispatch.into_parameter_scope(Arc::clone(&self.state.cancel).notified_owned())
            }
            RequestTerminal::Serial(scope) => scope,
            other => {
                *terminal = other;
                return Err(ParameterConstructionError::Scope(
                    saddle_runtime::profusegw::ProfuseGwScopeFailure::ScopeAlreadyEntered,
                ));
            }
        };
        Ok(ConstructionScope {
            scope: Some(scope),
            state: Arc::clone(&self.state),
        })
    }
    parameter_constructor!(parameter_text, &str, crate::database::Text, text, |v| v);
    parameter_constructor!(parameter_json, &str, crate::database::Json, json, |v| v);
    parameter_constructor!(
        parameter_i8s,
        &[i8],
        crate::database::List<i8>,
        i8s,
        Into::into
    );
    parameter_constructor!(
        parameter_optional_i8s,
        &[Option<i8>],
        crate::database::List<Option<i8>>,
        optional_i8s,
        Into::into
    );
    parameter_constructor!(
        parameter_i16s,
        &[i16],
        crate::database::List<i16>,
        i16s,
        Into::into
    );
    parameter_constructor!(
        parameter_optional_i16s,
        &[Option<i16>],
        crate::database::List<Option<i16>>,
        optional_i16s,
        Into::into
    );
    parameter_constructor!(
        parameter_i32s,
        &[i32],
        crate::database::List<i32>,
        i32s,
        Into::into
    );
    parameter_constructor!(
        parameter_optional_i32s,
        &[Option<i32>],
        crate::database::List<Option<i32>>,
        optional_i32s,
        Into::into
    );
    parameter_constructor!(
        parameter_i64s,
        &[i64],
        crate::database::List<i64>,
        i64s,
        Into::into
    );
    parameter_constructor!(
        parameter_optional_i64s,
        &[Option<i64>],
        crate::database::List<Option<i64>>,
        optional_i64s,
        Into::into
    );
    parameter_constructor!(
        parameter_u8s,
        &[u8],
        crate::database::List<u8>,
        u8s,
        Into::into
    );
    parameter_constructor!(
        parameter_optional_u8s,
        &[Option<u8>],
        crate::database::List<Option<u8>>,
        optional_u8s,
        Into::into
    );
    parameter_constructor!(
        parameter_u16s,
        &[u16],
        crate::database::List<u16>,
        u16s,
        Into::into
    );
    parameter_constructor!(
        parameter_optional_u16s,
        &[Option<u16>],
        crate::database::List<Option<u16>>,
        optional_u16s,
        Into::into
    );
    parameter_constructor!(
        parameter_u32s,
        &[u32],
        crate::database::List<u32>,
        u32s,
        Into::into
    );
    parameter_constructor!(
        parameter_optional_u32s,
        &[Option<u32>],
        crate::database::List<Option<u32>>,
        optional_u32s,
        Into::into
    );
    parameter_constructor!(
        parameter_u64s,
        &[u64],
        crate::database::List<u64>,
        u64s,
        Into::into
    );
    parameter_constructor!(
        parameter_optional_u64s,
        &[Option<u64>],
        crate::database::List<Option<u64>>,
        optional_u64s,
        Into::into
    );
    parameter_constructor!(
        parameter_bools,
        &[bool],
        crate::database::List<bool>,
        bools,
        Into::into
    );
    parameter_constructor!(
        parameter_optional_bools,
        &[Option<bool>],
        crate::database::List<Option<bool>>,
        optional_bools,
        Into::into
    );
    parameter_constructor!(parameter_bytes, &[u8], crate::database::Bytes, bytes, |v| v);
    parameter_constructor!(
        parameter_texts,
        &[Option<&str>],
        crate::database::List<Option<crate::database::Text>>,
        texts,
        Into::into
    );
    parameter_constructor!(
        parameter_byte_list,
        &[Option<&[u8]>],
        crate::database::List<Option<crate::database::Bytes>>,
        byte_list,
        Into::into
    );
}

struct QueryTask<O: StaticQueryOptionalOperation> {
    request: SerialScope,
    process: Arc<saddle_db::internal::StartupManagedDatabaseProcessCapability>,
    context: CallContext,
    event: EventContext,
    state: Arc<RequestState>,
    proof: QueryOptionalOperationProof<O>,
    parameters: O::Parameters,
    sender: tokio::sync::oneshot::Sender<Result<ManagedOptionalRow<O::Row>, ScopeDatabaseError>>,
}

// Each kind keeps its actual future/result layout and original serial owner.
// A dropped caller cannot drop the physical finalizer task before resume.
macro_rules! owned_query_task {
    (@type $op:ident) => { saddle_db::internal::OwnedQueryInvocation<$op> };
    (@type $op:ident, $size:ident) => { saddle_db::internal::OwnedPageInvocation<$op> };
    (@invocation $proof:ident, $parameters:ident) => { Ok::<_, ScopeDatabaseError>($proof.invocation($parameters)) };
    (@invocation $proof:ident, $parameters:ident, $size:ident) => { match $size.1 { Some(offset)=>$proof.page_offset_invocation($parameters,$size.0,offset), None=>$proof.page_invocation($parameters,$size.0) } };
    ($task:ident, $method:ident, $process_method:ident, $result:ty, $layout:ident $(, $size:ident)?) => {
        struct $task<O: saddle_db::internal::StaticOwnedQuery> {
            request: SerialScope,
            process: Arc<saddle_db::internal::StartupManagedDatabaseProcessCapability>,
            context: CallContext,
            event: EventContext,
            state: Arc<RequestState>,
            invocation: owned_query_task!(@type O $(,$size)?),
            sender: tokio::sync::oneshot::Sender<Result<$result, saddle_db::internal::OwnedExecutionError<O::Parameters>>>,
        }
        impl<O: saddle_db::internal::StaticOwnedQuery> $task<O> {
            async fn run(self) {
                let mut owner_guard = AbortOnLostRequestOwner(true);
                let suspended = self
                    .process
                    .$process_method(self.request, &self.context, self.event, self.invocation)
                    .await;
                DatabaseRequest::publish_ordinary(suspended, &self.state, self.sender).await;
                owner_guard.0 = false;
            }
        }
        pub(crate) fn $layout<O: saddle_db::internal::StaticOwnedQuery, C>()
        -> [std::alloc::Layout; 4] {
            [
                future_layout($task::<O>::run),
                std::alloc::Layout::new::<Result<$result, saddle_db::internal::OwnedExecutionError<O::Parameters>>>(),
                std::alloc::Layout::new::<RequestState>(),
                std::alloc::Layout::new::<C>(),
            ]
        }
        impl DatabaseRequest {
            pub async fn $method<O: saddle_db::internal::StaticOwnedQuery>(
                &mut self,
                parameters: O::Parameters,
                $($size: (usize, Option<u64>),)?
            ) -> Result<$result, saddle_db::internal::OwnedExecutionError<O::Parameters>> {
                let proof = saddle_db::internal::OwnedQueryProof::<O>::bind()?;
                let invocation = owned_query_task!(@invocation proof, parameters $(,$size)?)?;
                let process = self
                    .process
                    .as_ref()
                    .cloned()
                    .ok_or(ScopeDatabaseError::State)?;
                let request = self.begin_serial()?;
                let (sender, receiver) = tokio::sync::oneshot::channel();
                tokio::spawn(
                    $task {
                        request,
                        process,
                        context: self.context.clone(),
                        event: self.event.clone(),
                        state: Arc::clone(&self.state),
                        invocation,
                        sender,
                    }
                    .run(),
                );
                receiver.await.unwrap_or_else(|_| std::process::abort())
            }
        }
    };
}
owned_query_task!(
    OwnedPageTask,
    owned_page,
    owned_page_scope,
    saddle_db::internal::OwnedDbPage<O::Row>,
    owned_page_memory_layout,
    page_size
);
owned_query_task!(
    OwnedRowsTask,
    owned_rows,
    owned_rows_scope,
    saddle_db::internal::OwnedDbRows<O::Row>,
    owned_rows_memory_layout
);
owned_query_task!(
    OwnedProbeTask,
    owned_probe,
    owned_probe_scope,
    saddle_db::internal::OwnedDbProbe<O::Row>,
    owned_probe_memory_layout
);
owned_query_task!(
    OwnedOptionalTask,
    owned_optional,
    owned_optional_scope,
    Option<O::Row>,
    owned_optional_memory_layout
);

struct OwnedWriteTask<O: saddle_db::internal::StaticOwnedWrite> {
    request: SerialScope,
    process: Arc<saddle_db::internal::StartupManagedDatabaseProcessCapability>,
    context: CallContext,
    event: EventContext,
    state: Arc<RequestState>,
    invocation: saddle_db::internal::OwnedWriteInvocation<O>,
    sender: tokio::sync::oneshot::Sender<
        Result<ManagedWriteResult, saddle_db::internal::OwnedExecutionError<O::Parameters>>,
    >,
}
impl<O: saddle_db::internal::StaticOwnedWrite> OwnedWriteTask<O> {
    async fn run(self) {
        let mut guard = AbortOnLostRequestOwner(true);
        let suspended = self
            .process
            .owned_write_scope(self.request, &self.context, self.event, self.invocation)
            .await;
        DatabaseRequest::publish_ordinary(suspended, &self.state, self.sender).await;
        guard.0 = false;
    }
}
pub(crate) fn owned_write_memory_layout<O: saddle_db::internal::StaticOwnedWrite, C>()
-> [std::alloc::Layout; 4] {
    [
        future_layout(OwnedWriteTask::<O>::run),
        std::alloc::Layout::new::<
            Result<ManagedWriteResult, saddle_db::internal::OwnedExecutionError<O::Parameters>>,
        >(),
        std::alloc::Layout::new::<RequestState>(),
        std::alloc::Layout::new::<C>(),
    ]
}
#[doc(hidden)]
pub fn named_write_failure<O: GeneratedOwnedWrite>(
    error: saddle_db::internal::OwnedExecutionError<O::Parameters>,
) -> QueryFailure<O::NamedParameters> {
    match error {
        saddle_db::internal::OwnedExecutionError::Parameters { parameters, .. } => {
            QueryFailure::Parameters {
                parameters: O::restore_parameters(parameters),
            }
        }
        saddle_db::internal::OwnedExecutionError::Database(e) => QueryFailure::Database(e),
    }
}
impl DatabaseRequest {
    pub async fn named_write<O: GeneratedOwnedWrite>(
        &mut self,
        parameters: O::NamedParameters,
    ) -> Result<ManagedWriteResult, QueryFailure<O::NamedParameters>> {
        if let Err(error) = O::validate_parameters(&parameters, &self.target()) {
            return Err(QueryFailure::Target { parameters, error });
        }
        let proof =
            saddle_db::internal::OwnedWriteProof::<O>::bind().map_err(QueryFailure::Database)?;
        let process = self
            .process
            .as_ref()
            .cloned()
            .ok_or(QueryFailure::Database(ScopeDatabaseError::State))?;
        let request = self.begin_serial().map_err(QueryFailure::Database)?;
        let (sender, receiver) = tokio::sync::oneshot::channel();
        tokio::spawn(
            OwnedWriteTask {
                request,
                process,
                context: self.context.clone(),
                event: self.event.clone(),
                state: Arc::clone(&self.state),
                invocation: proof.invocation(O::parameters(parameters)),
                sender,
            }
            .run(),
        );
        receiver
            .await
            .unwrap_or_else(|_| std::process::abort())
            .map_err(named_write_failure::<O>)
    }
}

impl<O: StaticQueryOptionalOperation> QueryTask<O> {
    async fn run(self) {
        let Self {
            request,
            context,
            event,
            state,
            process,
            proof,
            parameters,
            sender,
        } = self;
        let mut owner_guard = AbortOnLostRequestOwner(true);
        let suspended = process
            .query_scope(request, &context, event, proof.invocation(parameters))
            .await;
        DatabaseRequest::publish_ordinary(suspended, &state, sender).await;
        owner_guard.0 = false;
    }
}

struct WriteTask<O: StaticWriteOperation> {
    request: SerialScope,
    process: Arc<saddle_db::internal::StartupManagedDatabaseProcessCapability>,
    context: CallContext,
    event: EventContext,
    state: Arc<RequestState>,
    proof: WriteOperationProof<O>,
    parameters: O::Parameters,
    sender: tokio::sync::oneshot::Sender<Result<ManagedWriteResult, ScopeDatabaseError>>,
}

impl<O: StaticWriteOperation> WriteTask<O> {
    async fn run(self) {
        let Self {
            request,
            context,
            event,
            state,
            process,
            proof,
            parameters,
            sender,
        } = self;
        let mut owner_guard = AbortOnLostRequestOwner(true);
        let suspended = process
            .write_scope(request, &context, event, proof.invocation(parameters))
            .await;
        DatabaseRequest::publish_ordinary(suspended, &state, sender).await;
        owner_guard.0 = false;
    }
}

fn future_layout<A, F: std::future::Future>(_: fn(A) -> F) -> std::alloc::Layout {
    // Type inspection only: no fake values, allocation, construction or poll.
    std::alloc::Layout::new::<F>()
}

pub(crate) fn query_memory_layout<O: StaticQueryOptionalOperation, C>() -> [std::alloc::Layout; 4] {
    [
        future_layout(QueryTask::<O>::run),
        std::alloc::Layout::new::<Result<ManagedOptionalRow<O::Row>, ScopeDatabaseError>>(),
        std::alloc::Layout::new::<RequestState>(),
        std::alloc::Layout::new::<(Result<ManagedOptionalRow<O::Row>, ScopeDatabaseError>, C)>(),
    ]
}

pub(crate) fn write_memory_layout<O: StaticWriteOperation, C>() -> [std::alloc::Layout; 4] {
    [
        future_layout(WriteTask::<O>::run),
        std::alloc::Layout::new::<Result<ManagedWriteResult, ScopeDatabaseError>>(),
        std::alloc::Layout::new::<RequestState>(),
        std::alloc::Layout::new::<(Result<ManagedWriteResult, ScopeDatabaseError>, C)>(),
    ]
}

/// Known persistent framework storage for a borrowed transaction route.
/// Not a bound on the generic business callback, spawned Future, result T/E,
/// executor headers or driver allocations. In particular, do not substitute
/// the unrelated standalone QueryTask/WriteTask Future as scope evidence.
pub(crate) fn combined_scope_layout<C>(
    ordinary: [std::alloc::Layout; 4],
) -> [std::alloc::Layout; 4] {
    let scoped = scoped_memory_layout::<C>();
    std::array::from_fn(|index| {
        ordinary[index]
            .extend(scoped[index])
            .expect("concrete execution layouts fit")
            .0
            .pad_to_align()
    })
}

pub(crate) fn scoped_memory_layout<C>() -> [std::alloc::Layout; 4] {
    [
        std::alloc::Layout::new::<SerialScope>(),
        std::alloc::Layout::new::<RequestState>(),
        std::alloc::Layout::new::<DatabaseRequest>(),
        std::alloc::Layout::new::<C>(),
    ]
}

impl DatabaseRequest {
    pub fn target(&self) -> DatabaseTarget {
        DatabaseTarget(self.process.clone())
    }
    pub async fn named_page<O: GeneratedPageQuery>(
        &mut self,
        mut request: crate::database_rows::PageRequest<O>,
    ) -> Result<crate::database_rows::Page<O>, QueryFailure<crate::database_rows::PageRequest<O>>>
    {
        if let Err(error) = O::validate_parameters(&request.parameters, &self.target()) {
            return Err(QueryFailure::Target {
                parameters: request,
                error,
            });
        }
        let parameters =
            match O::page_parameters(&mut request.parameters, request.after, self).await {
                Ok(parameters) => parameters,
                Err(_) => {
                    return Err(QueryFailure::Parameters {
                        parameters: request,
                    });
                }
            };
        let page = match self
            .owned_page::<O>(parameters, (request.size, request.offset))
            .await
        {
            Ok(page) => page,
            Err(saddle_db::internal::OwnedExecutionError::Parameters { .. }) => {
                return Err(QueryFailure::Parameters {
                    parameters: request,
                });
            }
            Err(saddle_db::internal::OwnedExecutionError::Database(e)) => {
                return Err(QueryFailure::Database(e));
            }
        };
        let next = if page.has_more() {
            page.rows().last().map(O::page_key)
        } else {
            None
        };
        let (rows, _) = page.into_parts();
        Ok(crate::database_rows::Page {
            rows: crate::database_rows::Rows::from_database(rows),
            next: next.map(|key| crate::database_rows::PageRequest {
                after: Some(key),
                offset: request.offset.map(|offset| offset + request.size as u64),
                ..request
            }),
        })
    }
    pub async fn named_rows<O: GeneratedOwnedQuery>(
        &mut self,
        parameters: O::NamedParameters,
    ) -> Result<crate::database::Rows<O::Row, O::NamedRow>, QueryFailure<O::NamedParameters>> {
        if let Err(error) = O::validate_parameters(&parameters, &self.target()) {
            return Err(QueryFailure::Target { parameters, error });
        }
        self.owned_rows::<O>(O::parameters(parameters))
            .await
            .map(crate::database_rows::Rows::from_database)
            .map_err(named_failure::<O>)
    }

    pub async fn named_optional<O: GeneratedOwnedQuery>(
        &mut self,
        parameters: O::NamedParameters,
    ) -> Result<Option<O::NamedRow>, QueryFailure<O::NamedParameters>> {
        if let Err(error) = O::validate_parameters(&parameters, &self.target()) {
            return Err(QueryFailure::Target { parameters, error });
        }
        self.owned_optional::<O>(O::parameters(parameters))
            .await
            .map(|row| row.map(O::NamedRow::from))
            .map_err(named_failure::<O>)
    }

    pub async fn named_probe<O: GeneratedOwnedQuery>(
        &mut self,
        parameters: O::NamedParameters,
    ) -> Result<crate::database::MatchCardinality<O::NamedRow>, QueryFailure<O::NamedParameters>>
    {
        use crate::database::MatchCardinality;
        use saddle_db::internal::OwnedDbProbe;
        if let Err(error) = O::validate_parameters(&parameters, &self.target()) {
            return Err(QueryFailure::Target { parameters, error });
        }
        self.owned_probe::<O>(O::parameters(parameters))
            .await
            .map(|result| match result {
                OwnedDbProbe::Zero => MatchCardinality::Zero,
                OwnedDbProbe::One(row) => MatchCardinality::One(O::NamedRow::from(row)),
                OwnedDbProbe::Many => MatchCardinality::Many,
            })
            .map_err(named_failure::<O>)
    }

    pub(crate) fn supervise_external<T: Send + 'static>(
        &self,
        future: std::pin::Pin<
            Box<
                dyn std::future::Future<Output = crate::programming::ExternalFunctionResult<T>>
                    + Send,
            >,
        >,
    ) -> std::pin::Pin<
        Box<dyn std::future::Future<Output = crate::programming::ExternalFunctionResult<T>> + Send>,
    > {
        let state = Arc::clone(&self.state);
        Box::pin(async move {
            use crate::programming::{
                ExecutionCertainty, ExternalFunctionResult, TechnicalFailure, TechnicalFailureCode,
            };
            let stopped = |code, certainty| {
                ExternalFunctionResult::TechnicalFailure(TechnicalFailure::from_framework(
                    code, certainty,
                ))
            };
            let scope = {
                let mut terminal = state.terminal.lock().unwrap();
                match std::mem::replace(&mut *terminal, RequestTerminal::InFlight) {
                    RequestTerminal::Serial(scope) => Some(scope),
                    other => {
                        let untouched = matches!(other, RequestTerminal::Untouched(_));
                        *terminal = other;
                        if !untouched {
                            return stopped(
                                TechnicalFailureCode::InternalFailure,
                                ExecutionCertainty::NotExecuted,
                            );
                        }
                        None
                    }
                }
            };
            let Some(mut scope) = scope else {
                return future.await;
            };
            let (sender, receiver) = tokio::sync::oneshot::channel();
            tokio::spawn(async move {
                let mut guard = AbortOnLostRequestOwner(true);
                let result = scope.supervise_between(future).await;
                *state.terminal.lock().unwrap() = RequestTerminal::Serial(scope);
                state.completed.notify_one();
                let _ = sender.send(result);
                guard.0 = false;
            });
            match receiver.await {
                Ok(Ok(result)) => result,
                Ok(Err(failure)) => {
                    use saddle_runtime::profusegw::{ProfuseGwScopeFailure, ProfuseGwScopeStop};
                    let code = match failure {
                        ProfuseGwScopeFailure::Stopped(ProfuseGwScopeStop::TimedOut) => {
                            TechnicalFailureCode::DeadlineExceeded
                        }
                        ProfuseGwScopeFailure::Stopped(ProfuseGwScopeStop::Cancelled) => {
                            TechnicalFailureCode::DependencyUnavailable
                        }
                        ProfuseGwScopeFailure::Panicked
                        | ProfuseGwScopeFailure::AlreadySupervised
                        | ProfuseGwScopeFailure::ScopeAlreadyEntered => {
                            TechnicalFailureCode::InternalFailure
                        }
                    };
                    stopped(code, ExecutionCertainty::MayHaveExecuted)
                }
                Err(_) => std::process::abort(),
            }
        })
    }

    pub async fn scoped_transaction<T, E, B>(
        &mut self,
        isolation: saddle_db::internal::TransactionIsolation,
        body: B,
    ) -> saddle_db::internal::ScopeTransactionOutcome<T, E>
    where
        T: Send + 'static,
        E: Send + 'static,
        B: for<'tx, 'session, 'request> FnOnce(
                &'tx mut TransactionSession<'session, 'request>,
            )
                -> saddle_db::internal::ScopeTransactionFuture<
                'tx,
                T,
                E,
            > + Send
            + 'static,
    {
        use saddle_db::internal::{
            ScopeDatabaseError, ScopeTransactionAbort, ScopeTransactionOutcome,
        };
        let rejected = || {
            ScopeTransactionOutcome::Rejected(ScopeTransactionAbort::Technical(
                ScopeDatabaseError::Unavailable,
            ))
        };
        let Some(process) = self.process.as_ref().cloned() else {
            return rejected();
        };
        let scope = {
            let mut state = self.state.terminal.lock().unwrap();
            match std::mem::replace(&mut *state, RequestTerminal::InFlight) {
                RequestTerminal::Untouched(dispatch) => dispatch
                    .into_database_request()
                    .into_serial_scope(Arc::clone(&self.state.cancel).notified_owned()),
                RequestTerminal::Serial(scope) => scope,
                other => {
                    *state = other;
                    return rejected();
                }
            }
        };
        let state = Arc::clone(&self.state);
        let context = self.context.clone();
        let event = self.event.clone();
        let (sender, receiver) = tokio::sync::oneshot::channel();
        // Request-loop cancellation never aborts this owning finalizer task.
        tokio::spawn(async move {
            let mut guard = AbortOnLostRequestOwner(true);
            let mut suspended = process
                .transaction_scope_with_isolation(scope, &context, event, isolation, body)
                .await;
            let (terminal, result) = match suspended.resume().await {
                Ok((next, result)) => (RequestTerminal::Serial(next), result),
                Err(_) => {
                    let (result, terminal) = suspended
                        .into_response_parts()
                        .unwrap_or_else(|_| std::process::abort());
                    (RequestTerminal::PostDatabase(terminal), result)
                }
            };
            *state.terminal.lock().unwrap() = terminal;
            state.completed.notify_one();
            let _ = sender.send(result.outcome);
            guard.0 = false;
        });
        receiver.await.unwrap_or_else(|_| std::process::abort())
    }

    pub(crate) fn new(
        process: Option<Arc<saddle_db::internal::StartupManagedDatabaseProcessCapability>>,
        dispatch: ProfuseGwManagedDispatch,
        context: CallContext,
        event: EventContext,
    ) -> (Self, DatabaseRequestCompletion) {
        let state = Arc::new(RequestState {
            terminal: Mutex::new(RequestTerminal::Untouched(dispatch)),
            cancel: Arc::new(Notify::new()),
            completed: Notify::new(),
        });
        (
            Self {
                process,
                context,
                event,
                state: Arc::clone(&state),
            },
            DatabaseRequestCompletion { state },
        )
    }

    fn begin_serial(&mut self) -> Result<SerialScope, ScopeDatabaseError> {
        if self.process.is_none() {
            return Err(ScopeDatabaseError::Unavailable);
        }
        let mut terminal = self.state.terminal.lock().unwrap();
        match std::mem::replace(&mut *terminal, RequestTerminal::InFlight) {
            RequestTerminal::Untouched(dispatch) => Ok(dispatch
                .into_database_request()
                .into_serial_scope(Arc::clone(&self.state.cancel).notified_owned())),
            RequestTerminal::Serial(scope) => Ok(scope),
            other => {
                *terminal = other;
                Err(ScopeDatabaseError::State)
            }
        }
    }

    async fn publish_ordinary<T, E>(
        mut suspended: saddle_runtime::profusegw::ProfuseGwSuspendedScope<
            tokio::sync::futures::OwnedNotified,
            Result<T, E>,
        >,
        state: &RequestState,
        sender: tokio::sync::oneshot::Sender<Result<T, E>>,
    ) {
        let (terminal, value) = match suspended.resume().await {
            Ok((scope, value)) => (RequestTerminal::Serial(scope), value),
            Err(_) => {
                let (value, terminal) = suspended
                    .into_response_parts()
                    .unwrap_or_else(|_| std::process::abort());
                (RequestTerminal::PostDatabase(terminal), value)
            }
        };
        *state.terminal.lock().unwrap() = terminal;
        state.completed.notify_one();
        let _ = sender.send(value);
    }

    pub async fn query_optional<O>(
        &mut self,
        parameters: O::Parameters,
    ) -> Result<ManagedOptionalRow<O::Row>, ScopeDatabaseError>
    where
        O: StaticQueryOptionalOperation,
    {
        let proof =
            QueryOptionalOperationProof::<O>::bind().map_err(|_| ScopeDatabaseError::Mapping)?;
        let request = self.begin_serial()?;
        let process = self
            .process
            .as_ref()
            .cloned()
            .unwrap_or_else(|| std::process::abort());
        let context = self.context.clone();
        let event = self.event.clone();
        let state = Arc::clone(&self.state);
        let (sender, receiver) = tokio::sync::oneshot::channel();
        tokio::spawn(
            QueryTask {
                request,
                context,
                event,
                state,
                process,
                proof,
                parameters,
                sender,
            }
            .run(),
        );
        receiver.await.unwrap_or_else(|_| std::process::abort())
    }

    pub async fn write<O>(
        &mut self,
        parameters: O::Parameters,
    ) -> Result<ManagedWriteResult, ScopeDatabaseError>
    where
        O: StaticWriteOperation,
    {
        let proof = WriteOperationProof::<O>::bind().map_err(|_| ScopeDatabaseError::Mapping)?;
        let request = self.begin_serial()?;
        let process = self
            .process
            .as_ref()
            .cloned()
            .unwrap_or_else(|| std::process::abort());
        let context = self.context.clone();
        let event = self.event.clone();
        let state = Arc::clone(&self.state);
        let (sender, receiver) = tokio::sync::oneshot::channel();
        tokio::spawn(
            WriteTask {
                request,
                context,
                event,
                state,
                process,
                proof,
                parameters,
                sender,
            }
            .run(),
        );
        receiver.await.unwrap_or_else(|_| std::process::abort())
    }
}

impl DatabaseRequestCompletion {
    pub(crate) async fn finish(self, cancelled: bool) {
        if cancelled {
            self.state.cancel.notify_one();
        }
        loop {
            let terminal = {
                let mut state = self.state.terminal.lock().unwrap();
                match std::mem::replace(&mut *state, RequestTerminal::Finished) {
                    RequestTerminal::InFlight => {
                        *state = RequestTerminal::InFlight;
                        None
                    }
                    terminal => Some(terminal),
                }
            };
            match terminal {
                Some(RequestTerminal::Untouched(dispatch)) => {
                    if cancelled {
                        dispatch.cancel();
                    } else if finish_profusegw_without_database(dispatch, ()).is_err() {
                        std::process::abort();
                    }
                    return;
                }
                Some(RequestTerminal::PostDatabase(terminal)) => {
                    finish_profusegw_after_database(terminal);
                    return;
                }
                Some(RequestTerminal::Serial(scope)) => {
                    if scope.finish_unentered_response(()).is_err() {
                        std::process::abort();
                    }
                    return;
                }
                Some(RequestTerminal::Finished) => return,
                Some(RequestTerminal::InFlight) | None => self.state.completed.notified().await,
            }
        }
    }
}