boxology-runtime 0.1.1

use std::future::{Future, poll_fn, ready};
use std::pin::Pin;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::{Arc, LazyLock, Mutex};
use std::task::{Context, Poll, Waker};
use std::thread::{self, ThreadId};
use std::time::{Duration, Instant};

use boxology_contract::{
    BoxHandle, BoxId, CallContext, CallError, Caller, CancelToken, CapabilityDescriptor,
    CapabilityId, CapabilityName, CapabilityShape, ContractDescriptor, ContractError,
    ContractRevision, ContractType, ContractValue, Deadline, DecodeError, DecodeErrorKind, Detail,
    EncodeError, ErasedCallError, ErasedCallTarget, ErasedTarget, ExposureLevel, Idempotency,
    ImplementationDescriptor, ImportDescriptor, SlotValue, TraceContext, TypeDescriptor,
    VariantDescriptor, VariantPayload,
};
#[cfg(feature = "test-support")]
use boxology_runtime::{AssemblyError, test_support::StubTransport};
use boxology_runtime::{Composition, CompositionBuilder, ImportHandle, Imports};

type ErasedFuture<'a> =
    Pin<Box<dyn Future<Output = Result<SlotValue, ErasedCallError>> + Send + 'a>>;

const DOMAIN_TAG: &str = "failing_domain";
const ORDER: Ordering = Ordering::SeqCst;

#[derive(Debug, Clone, PartialEq, Eq)]
struct FailingDomain;

type TypedResult = Result<f32, CallError<FailingDomain>>;

struct TypedBoxHandle(Arc<dyn ErasedCallTarget>);

impl BoxHandle for TypedBoxHandle {
    fn from_erased(target: Arc<dyn ErasedCallTarget>) -> Self {
        Self(target)
    }
}

impl TypedBoxHandle {
    async fn call(&self, input: f32) -> Result<f32, ErasedCallError> {
        let output = self
            .0
            .call(
                &capability(),
                context(None, CancelToken::new()),
                input.encode().unwrap(),
            )
            .await?;
        Ok(f32::decode(&output).unwrap())
    }
}

impl ContractType for FailingDomain {
    fn encode_value(&self) -> Result<ContractValue, EncodeError> {
        let payload = f32::NAN.encode()?;
        Ok(ContractValue::enum_value(self.error_tag(), payload))
    }

    fn decode_value(_value: &ContractValue) -> Result<Self, DecodeError> {
        Err(DecodeError::new(DecodeErrorKind::KindMismatch))
    }
}

impl ContractError for FailingDomain {
    fn error_tag(&self) -> &str {
        DOMAIN_TAG
    }
}

#[derive(Clone)]
struct GeneratedHandle {
    import: ImportHandle,
    capability: CapabilityId,
}

impl GeneratedHandle {
    async fn call(&self, context: CallContext, input: f32) -> TypedResult {
        let input = input
            .encode()
            .map_err(|error| CallError::ContractViolation(conversion_detail(error)))?;
        match self.import.call(&self.capability, context, input).await {
            Ok(output) => f32::decode(&output)
                .map_err(|error| CallError::InvalidResponse(conversion_detail(error))),
            Err(error) => Err(error.into_typed(error_descriptor())),
        }
    }
}

struct GeneratedAdapter {
    capability: CapabilityId,
    service: Service,
}

impl ErasedTarget for GeneratedAdapter {
    fn call<'a>(
        &'a self,
        capability: &'a CapabilityId,
        context: CallContext,
        input: SlotValue,
    ) -> ErasedFuture<'a> {
        self.service.state.target_calls.fetch_add(1, ORDER);
        assert_eq!(capability, &self.capability);
        let input = match f32::decode(&input) {
            Ok(input) => input,
            Err(error) => {
                return Box::pin(ready(Err(ErasedCallError::ContractViolation(
                    conversion_detail(error),
                ))));
            }
        };
        Box::pin(async move {
            match self.service.call(context, input).await {
                Ok(output) => output
                    .encode()
                    .map_err(|error| ErasedCallError::InvalidResponse(conversion_detail(error))),
                Err(error) => Err(ErasedCallError::from_domain(&error)),
            }
        })
    }
}

struct InertConsumer;

impl ErasedTarget for InertConsumer {
    fn call<'a>(
        &'a self,
        _capability: &'a CapabilityId,
        _context: CallContext,
        _input: SlotValue,
    ) -> ErasedFuture<'a> {
        panic!("inert consumer has no capabilities")
    }
}

#[derive(Debug, Clone, Copy)]
enum Mode {
    Echo,
    InvalidOutput,
    DomainError,
    Panic,
    ObserveCancellation,
    WaitPastDeadline(ThreadId),
}

struct State {
    mode: Mutex<Mode>,
    target_calls: AtomicUsize,
    calls: AtomicUsize,
    polls: AtomicUsize,
    cancellation_seen: AtomicBool,
}

impl State {
    fn new() -> Self {
        Self {
            mode: Mutex::new(Mode::Echo),
            target_calls: AtomicUsize::new(0),
            calls: AtomicUsize::new(0),
            polls: AtomicUsize::new(0),
            cancellation_seen: AtomicBool::new(false),
        }
    }

    fn prepare(&self, mode: Mode) {
        *self.mode.lock().unwrap() = mode;
        self.target_calls.store(0, ORDER);
        self.calls.store(0, ORDER);
        self.polls.store(0, ORDER);
        self.cancellation_seen.store(false, ORDER);
    }

    fn counts(&self) -> (usize, usize) {
        (self.target_calls.load(ORDER), self.calls.load(ORDER))
    }
}

struct Service {
    state: Arc<State>,
}

impl Service {
    async fn call(&self, context: CallContext, input: f32) -> Result<f32, FailingDomain> {
        self.state.calls.fetch_add(1, ORDER);
        let mode = *self.state.mode.lock().unwrap();
        match mode {
            Mode::Echo => Ok(input),
            Mode::InvalidOutput => Ok(f32::NAN),
            Mode::DomainError => Err(FailingDomain),
            Mode::Panic => panic!("provider poll panic"),
            Mode::ObserveCancellation => {
                let cancelled = context.cancellation().is_cancelled();
                self.state.cancellation_seen.store(cancelled, ORDER);
                Ok(input)
            }
            Mode::WaitPastDeadline(calling_thread) => {
                let deadline = context.deadline().expect("test supplies a deadline");
                poll_fn(|context| {
                    assert_eq!(thread::current().id(), calling_thread);
                    self.state.polls.fetch_add(1, ORDER);
                    if deadline.remaining().is_zero() {
                        Poll::Ready(Ok(input))
                    } else {
                        context.waker().wake_by_ref();
                        Poll::Pending
                    }
                })
                .await
            }
        }
    }
}

type Unstarted = (CompositionBuilder, GeneratedHandle, Arc<State>);

struct Assembled {
    handle: GeneratedHandle,
    state: Arc<State>,
    _composition: Composition,
}

fn build() -> Unstarted {
    build_with_provider(implementation("provider", true, false))
}

fn build_with_provider(provider_descriptor: ImplementationDescriptor) -> Unstarted {
    let state = Arc::new(State::new());
    let provider = box_id("provider");
    let capability = capability();
    let mut captured = None;
    let mut builder = CompositionBuilder::new();
    let consumer = implementation("consumer", false, true);
    let consumer = builder.register(consumer, |imports: Imports| {
        captured = Some(GeneratedHandle {
            import: imports.handle(&provider).unwrap().clone(),
            capability: capability.clone(),
        });
        InertConsumer
    });
    assert!(captured.is_some(), "consumer factory did not run inline");
    let provider = builder.register(provider_descriptor, |_| GeneratedAdapter {
        capability: capability.clone(),
        service: Service {
            state: Arc::clone(&state),
        },
    });
    assert_eq!(consumer.id(), &box_id("consumer"));
    assert_eq!(provider.id(), &box_id("provider"));
    builder.connect(&consumer, &provider);
    (builder, captured.unwrap(), state)
}

fn assemble() -> Assembled {
    let (builder, handle, state) = build();
    Assembled {
        handle,
        state,
        _composition: builder.start().unwrap(),
    }
}

fn implementation(box_name: &str, provides: bool, imports: bool) -> ImplementationDescriptor {
    implementation_with_shape(box_name, provides, imports, CapabilityShape::Unary)
}

fn implementation_with_shape(
    box_name: &str,
    provides: bool,
    imports: bool,
    shape: CapabilityShape,
) -> ImplementationDescriptor {
    let revision = ContractRevision::new("r1").unwrap();
    let capabilities = provides.then(|| {
        CapabilityDescriptor::new(
            capability(),
            TypeDescriptor::f32(),
            TypeDescriptor::f32(),
            error_descriptor().clone(),
            shape,
            ExposureLevel::CodeOnly,
            Idempotency::None,
            None,
        )
    });
    let contract = Box::leak(Box::new(
        ContractDescriptor::new(box_id(box_name), capabilities, revision.clone()).unwrap(),
    ));
    let imports = imports
        .then(|| ImportDescriptor::new(box_id("provider"), revision, [capability()]).unwrap());
    ImplementationDescriptor::new(contract, imports).unwrap()
}

static ERROR_DESCRIPTOR: LazyLock<TypeDescriptor> = LazyLock::new(|| {
    TypeDescriptor::enumeration([VariantDescriptor::new(
        DOMAIN_TAG,
        VariantPayload::Value(TypeDescriptor::f32()),
        None,
    )])
    .unwrap()
});

fn error_descriptor() -> &'static TypeDescriptor {
    &ERROR_DESCRIPTOR
}

fn box_id(value: &str) -> BoxId {
    BoxId::new(value).unwrap()
}

fn capability() -> CapabilityId {
    CapabilityId::new(box_id("provider"), CapabilityName::new("compute").unwrap())
}

fn conversion_detail(error: impl std::fmt::Display) -> Detail {
    Detail::new("test_conversion").with_message(error.to_string())
}

fn context(deadline: Option<Deadline>, cancellation: CancelToken) -> CallContext {
    let trace = TraceContext::empty();
    CallContext::new(Caller::Anonymous, deadline, cancellation, trace, None)
}

fn block_on<F: Future>(future: F) -> F::Output {
    let mut future = std::pin::pin!(future);
    let mut context = Context::from_waker(Waker::noop());
    loop {
        if let Poll::Ready(output) = future.as_mut().poll(&mut context) {
            return output;
        }
    }
}

fn invoke(handle: &GeneratedHandle, deadline: Option<Deadline>, input: f32) -> TypedResult {
    block_on(handle.call(context(deadline, CancelToken::new()), input))
}

fn assert_send<T: Send>(value: T) -> T {
    value
}

fn assert_send_sync_static<T: Send + Sync + 'static>() {}

#[test]
fn start_is_the_only_authorization_and_typed_success_selects_the_provider() {
    assert_send_sync_static::<GeneratedAdapter>();
    let (builder, handle, state) = build();
    let before_start = block_on(assert_send(
        handle.call(context(None, CancelToken::new()), 7.25),
    ));
    assert_eq!(
        before_start,
        Err(CallError::Unavailable(Detail::new("unsealed_import")))
    );
    assert_eq!(state.counts(), (0, 0));

    let assembled = Assembled {
        handle,
        state,
        _composition: builder.start().unwrap(),
    };
    assert_eq!(invoke(&assembled.handle, None, 7.25), Ok(7.25));
    assert_eq!(assembled.state.counts(), (1, 1));
}

#[cfg(feature = "test-support")]
#[test]
fn local_and_stub_paths_reach_the_same_provider_after_start() {
    let (mut builder, handle, state) = build();
    let stub = Arc::new(StubTransport::new());
    builder.expose(
        box_id("provider"),
        capability(),
        stub.clone(),
        ExposureLevel::CodeOnly,
    );
    assert!(stub.runtime().is_none());
    assert_eq!(
        invoke(&handle, None, 7.25),
        Err(CallError::Unavailable(Detail::new("unsealed_import")))
    );
    assert_eq!(state.counts(), (0, 0));

    let composition = builder.start().unwrap();
    let runtime = stub.runtime().expect("stub runtime was not retained");
    assert!(runtime.is_active());
    assert_eq!(runtime.exposures().len(), 1);
    let exposure = &runtime.exposures()[0];
    assert_eq!(exposure.descriptor().id(), &capability());
    assert_eq!(exposure.level(), ExposureLevel::CodeOnly);
    assert_eq!(invoke(&handle, None, 7.25), Ok(7.25));
    let output = block_on(exposure.dispatch(
        context(None, CancelToken::new()),
        7.25_f32.encode().unwrap(),
    ))
    .unwrap();
    assert_eq!(f32::decode(&output).unwrap(), 7.25);
    assert_eq!(state.counts(), (2, 2));
    drop(composition);
}

#[test]
fn box_like_registration_exposure_and_local_calls_hide_descriptor_plumbing() {
    let state = Arc::new(State::new());
    let mut builder = CompositionBuilder::new();
    let provider = builder.register(implementation("provider", true, false), |_| {
        GeneratedAdapter {
            capability: capability(),
            service: Service {
                state: state.clone(),
            },
        }
    });
    let handle: TypedBoxHandle = builder.handle(&provider);
    let _composition = builder.start().unwrap();

    assert_eq!(block_on(handle.call(41.0)), Ok(41.0));
    assert_eq!(state.counts(), (1, 1));
}

#[cfg(feature = "test-support")]
#[test]
fn reserved_shape_is_constructible_but_rejected_before_stub_start() {
    let provider =
        implementation_with_shape("provider", true, false, CapabilityShape::ServerStreaming);
    let descriptor = &provider.contract().capabilities()[0];
    assert_eq!(descriptor.id(), &capability());
    assert_eq!(descriptor.shape(), CapabilityShape::ServerStreaming);
    let (mut builder, _handle, _state) = build_with_provider(provider);
    let stub = Arc::new(StubTransport::new());
    builder.expose(
        box_id("provider"),
        capability(),
        stub.clone(),
        ExposureLevel::CodeOnly,
    );
    let expected = AssemblyError::TransportConformanceFailed {
        capability: capability(),
        detail: Detail::new("unsupported_interaction_shape")
            .with_message("stub transport supports unary capabilities only"),
    };
    let validated = builder.validate().unwrap_err();
    assert_eq!(validated.errors(), &[expected]);
    assert_eq!(builder.validate().unwrap_err(), validated);
    assert_eq!(
        validated.to_string(),
        "transport conformance failed for capability provider.compute: unsupported_interaction_shape: stub transport supports unary capabilities only"
    );
    let started = builder.start().err().expect("reserved shape started");
    assert_eq!(started, validated);
    assert!(stub.runtime().is_none());
}

#[test]
fn violations_deadlines_invalid_results_domain_errors_and_panics_are_ordered() {
    let assembled = assemble();

    assert!(matches!(
        invoke(&assembled.handle, None, f32::NAN),
        Err(CallError::ContractViolation(_))
    ));
    assert_eq!(assembled.state.counts(), (0, 0));

    let expired = Deadline::at(Instant::now());
    assert!(expired.remaining().is_zero());
    let expired_result = invoke(&assembled.handle, Some(expired), 1.0);
    assert_eq!(expired_result, Err(CallError::Deadline));
    assert_eq!(assembled.state.counts(), (0, 0));

    assembled.state.prepare(Mode::InvalidOutput);
    assert!(matches!(
        invoke(&assembled.handle, None, 2.0),
        Err(CallError::InvalidResponse(_))
    ));
    assert_eq!(assembled.state.counts(), (1, 1));

    assembled.state.prepare(Mode::DomainError);
    assert!(matches!(
        invoke(&assembled.handle, None, 3.0),
        Err(CallError::InvalidResponse(detail)) if detail.code() == "domain_error_encode"
    ));
    assert_eq!(assembled.state.counts(), (1, 1));

    assembled.state.prepare(Mode::Panic);
    assert!(matches!(
        invoke(&assembled.handle, None, 4.0),
        Err(CallError::Internal(detail)) if detail.code() == "panic"
    ));
    assert_eq!(assembled.state.counts(), (1, 1));
}

#[test]
fn cancellation_is_advisory_and_observable_by_the_provider() {
    let assembled = assemble();
    assembled.state.prepare(Mode::ObserveCancellation);
    let cancellation = CancelToken::new();
    cancellation.cancel();

    let output = block_on(assembled.handle.call(context(None, cancellation), 5.0));
    assert_eq!(output, Ok(5.0));
    assert!(assembled.state.cancellation_seen.load(ORDER));
    assert_eq!(assembled.state.counts(), (1, 1));
}

#[test]
fn provider_polling_is_inline_and_a_positive_deadline_is_not_a_mid_call_timer() {
    let assembled = assemble();
    assembled
        .state
        .prepare(Mode::WaitPastDeadline(thread::current().id()));
    let deadline = Deadline::at(Instant::now() + Duration::from_millis(20));
    assert!(!deadline.remaining().is_zero());

    assert_eq!(invoke(&assembled.handle, Some(deadline), 6.0), Ok(6.0));
    assert!(deadline.remaining().is_zero());
    assert!(assembled.state.polls.load(ORDER) > 1);
    assert_eq!(assembled.state.counts(), (1, 1));
}