wip-rs/crates/wip-public-api-example/tests/scoped_http.rs

//! External-style consumer: a transactional provider, real TCP HTTP, and Client.
use std::collections::BTreeMap;
use std::io::{Read, Write};
use std::net::{TcpListener, TcpStream};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Duration;

use wip_client::{
    CallOutcome, Client, ClientLimits, Completion, ObservationState, SecurityContext,
};
use wip_host::{
    ExecutionContext, InterfaceRegistry, ObjectProvider, ProviderResult, PublishedObject,
    RegisteredInterface, Scope, ScopeObservation, Validator, WipObject,
};
use wip_http::{http, *};
use wip_interface as wip;
use wip_protocol::{InterfaceReference, ObjectObservation, ProtocolErrorCode, ProtocolInteraction};

#[wip::interface]
pub trait Status {
    fn status(&self) -> wip::OperationResult<String>;
    fn uncertain(&self) -> wip::OperationResult<String>;
}

struct Service {
    name: String,
    reference: Option<String>,
    validator: Validator,
    calls: Arc<AtomicUsize>,
}
impl WipObject for Service {
    fn name(&self) -> &str {
        &self.name
    }
    fn reference(&self) -> Option<&str> {
        self.reference.as_deref()
    }
    fn validator(&self) -> Option<&Validator> {
        Some(&self.validator)
    }
}
impl Status for Service {
    fn status(&self) -> wip::OperationResult<String> {
        self.calls.fetch_add(1, Ordering::SeqCst);
        Ok(self.name.clone())
    }
    fn uncertain(&self) -> wip::OperationResult<String> {
        self.calls.fetch_add(1, Ordering::SeqCst);
        Err(wip::HostError::internal(
            "commit acknowledgement lost",
            wip::EffectGuarantee::Unknown,
        ))
    }
}
struct Publication {
    scope: Scope,
    api: RegisteredInterface<StatusDefinition>,
    reference: Option<String>,
}
struct State {
    registry: InterfaceRegistry,
    root: Publication,
    service: Option<Publication>,
    calls: Arc<AtomicUsize>,
}
impl State {
    fn new(reference: Option<&str>) -> Self {
        let mut registry = InterfaceRegistry::new();
        let scope = Scope::new("/").unwrap();
        let root = Publication {
            api: registry
                .register(&scope, "status", StatusDefinition, None)
                .unwrap(),
            scope,
            reference: None,
        };
        let mut state = Self {
            registry,
            root,
            service: None,
            calls: Arc::default(),
        };
        state.republish(reference);
        state
    }
    fn remove(&mut self) {
        if let Some(old) = self.service.take() {
            self.registry.remove_scope(&old.scope);
        }
    }
    fn republish(&mut self, reference: Option<&str>) {
        self.remove();
        let scope = Scope::new("/service").unwrap();
        self.service = Some(Publication {
            api: self
                .registry
                .register(
                    &scope,
                    "status",
                    StatusDefinition,
                    Some(Validator::new(b"api-v1")),
                )
                .unwrap(),
            scope,
            reference: reference.map(str::to_owned),
        });
    }
}

// The HTTP adapter holds the State lock across all provider lookups, preconditions
// and dispatch. No separate scope prelookup can race a replacement.
struct Provider<'a> {
    state: &'a State,
    visible: bool,
}
impl ObjectProvider for Provider<'_> {
    type Object = Service;
    fn with_object<R>(
        &self,
        _: &ExecutionContext,
        path: &str,
        operation: impl FnOnce(&PublishedObject<Service>) -> R,
    ) -> ProviderResult<R> {
        let publication = match path {
            "/" => &self.state.root,
            "/service" if self.visible => match &self.state.service {
                Some(value) => value,
                None => return ProviderResult::NotFound,
            },
            _ => return ProviderResult::NotFound,
        };
        let object = Service {
            name: if path == "/" { "" } else { "service" }.into(),
            reference: publication.reference.clone(),
            validator: Validator::new(b"object-v1"),
            calls: self.state.calls.clone(),
        };
        let current = ScopeObservation::new(&publication.scope, &object);
        let mut object = PublishedObject::new(object)
            .bind_scope(current)
            .expose(&publication.api)
            .unwrap();
        if path == "/service" {
            let root_object = Service {
                name: "".into(),
                reference: None,
                validator: Validator::new(b"object-v1"),
                calls: self.state.calls.clone(),
            };
            object = object
                .bind_scope(ScopeObservation::new(&self.state.root.scope, &root_object))
                .expose(&self.state.root.api)
                .unwrap();
        }
        ProviderResult::Found(operation(&object))
    }
}
fn limits() -> Limits {
    Limits::new(65536, 65536, 64).unwrap()
}
fn reference(scope: &str) -> InterfaceReference {
    InterfaceReference {
        scope: scope.into(),
        name: "status".into(),
    }
}

fn handle(state: &State, path: &str, body: &[u8], visible: bool) -> http::Response<Vec<u8>> {
    let provider = Provider { state, visible };
    match path {
        "/wip/v1/observe" => {
            let request = decode_observe_request(body, limits()).unwrap();
            match provider.observe_object(&request.path) {
                Ok(object) => encode_observe_response(
                    &request,
                    &ObjectObservation {
                        object,
                        children: None,
                    },
                    limits(),
                )
                .unwrap(),
                Err(error) => encode_protocol_error_response(
                    ProtocolInteraction::Observe,
                    &error.protocol_error(),
                    limits(),
                )
                .unwrap(),
            }
        }
        "/wip/v1/fetch_interface" => {
            let request = decode_fetch_interface_request(body, limits()).unwrap();
            match provider.fetch_interface(&state.registry, &request.interface) {
                Ok(observation) => encode_fetch_interface_response(
                    &request,
                    &observation.protocol_response().unwrap(),
                    limits(),
                )
                .unwrap(),
                Err(error) => encode_protocol_error_response(
                    ProtocolInteraction::FetchInterface,
                    &error,
                    limits(),
                )
                .unwrap(),
            }
        }
        "/wip/v1/call_operation" => {
            let metadata = decode_call_operation_metadata(body, limits()).unwrap();
            let decoded = Mutex::new(None);
            let result = provider.call_operation_with_context_and_decoder(
                &ExecutionContext::default(),
                metadata.target.clone(),
                metadata.interface.clone(),
                metadata.operation.clone(),
                |descriptor| {
                    let request = metadata.decode_request(body, descriptor, limits())?;
                    let arguments = request.arguments.clone();
                    *decoded.lock().unwrap() = Some((request, descriptor.clone()));
                    Ok::<_, CodecError>(arguments)
                },
            );
            match result {
                Ok(result) => {
                    let (request, descriptor) = decoded.into_inner().unwrap().unwrap();
                    encode_call_operation_response(&request, &descriptor, &result, limits())
                        .unwrap()
                }
                Err(error) => encode_protocol_error_response(
                    ProtocolInteraction::CallOperation,
                    &error.protocol_error(),
                    limits(),
                )
                .unwrap(),
            }
        }
        _ => panic!("unexpected route {path}"),
    }
}

// Small bounded test-only HTTP/1.1 listener. Production serving is deliberately
// not an example dependency; both directions still cross an actual TCP socket.
fn exchange(
    state: &Arc<Mutex<State>>,
    request: http::Request<Vec<u8>>,
    visible: bool,
) -> http::Response<Vec<u8>> {
    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
    let address = listener.local_addr().unwrap();
    let state = state.clone();
    let server = thread::spawn(move || {
        let (mut stream, _) = listener.accept().unwrap();
        stream
            .set_read_timeout(Some(Duration::from_secs(5)))
            .unwrap();
        let mut raw = Vec::new();
        let (split, length) = loop {
            let mut chunk = [0; 1024];
            let count = stream.read(&mut chunk).unwrap();
            assert!(count > 0);
            raw.extend_from_slice(&chunk[..count]);
            assert!(raw.len() <= 70000);
            if let Some(split) = raw.windows(4).position(|part| part == b"\r\n\r\n") {
                let head = std::str::from_utf8(&raw[..split]).unwrap();
                let length = head
                    .lines()
                    .find_map(|line| line.strip_prefix("Content-Length: "))
                    .unwrap()
                    .parse::<usize>()
                    .unwrap();
                if raw.len() >= split + 4 + length {
                    break (split, length);
                }
            }
        };
        let path = std::str::from_utf8(&raw[..split])
            .unwrap()
            .lines()
            .next()
            .unwrap()
            .split_whitespace()
            .nth(1)
            .unwrap();
        let response = handle(
            &state.lock().unwrap(),
            path,
            &raw[split + 4..split + 4 + length],
            visible,
        );
        write!(stream, "HTTP/1.1 {} OK\r\nConnection: close\r\nContent-Length: {}\r\nContent-Type: application/json\r\n\r\n", response.status().as_u16(), response.body().len()).unwrap();
        stream.write_all(response.body()).unwrap();
    });
    let mut stream = TcpStream::connect(address).unwrap();
    stream
        .set_read_timeout(Some(Duration::from_secs(5)))
        .unwrap();
    write!(stream, "POST {} HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n", request.uri().path(), request.body().len()).unwrap();
    stream.write_all(request.body()).unwrap();
    let mut raw = Vec::new();
    stream.read_to_end(&mut raw).unwrap();
    server.join().unwrap();
    let split = raw.windows(4).position(|part| part == b"\r\n\r\n").unwrap();
    let status = std::str::from_utf8(&raw[..split])
        .unwrap()
        .split_whitespace()
        .nth(1)
        .unwrap()
        .parse::<u16>()
        .unwrap();
    http::Response::builder()
        .status(status)
        .header("content-type", "application/json")
        .body(raw[split + 4..].to_vec())
        .unwrap()
}

fn setup() -> (Client, wip_client::SessionId) {
    let mut client = Client::new(ClientLimits::new(1, 16, 16, 16, 16).unwrap(), limits());
    let session = client
        .open_session("http://localhost/wip", SecurityContext::new("alice"))
        .unwrap();
    (client, session)
}
fn observe(
    client: &mut Client,
    session: &wip_client::SessionId,
    state: &Arc<Mutex<State>>,
    path: &str,
) {
    let request = client.refresh_object(session, path).unwrap();
    let response = exchange(state, request.request, true);
    client.complete(request.id, response).unwrap();
}
fn fetch(
    client: &mut Client,
    session: &wip_client::SessionId,
    state: &Arc<Mutex<State>>,
    reference: &InterfaceReference,
) {
    let request = client
        .prepare_interface(session, reference.clone())
        .unwrap();
    let body = std::str::from_utf8(request.request.body()).unwrap();
    assert!(body.contains("\"scope\""));
    assert!(body.contains("\"name\""));
    let response = exchange(state, request.request, true);
    assert!(matches!(
        client.complete(request.id, response).unwrap(),
        Completion::Interface(_)
    ));
}

#[test]
fn real_http_root_self_same_name_and_optional_scope_refs() {
    for scope_ref in [Some("instance-a"), None] {
        let state = Arc::new(Mutex::new(State::new(scope_ref)));
        let (mut client, session) = setup();
        observe(&mut client, &session, &state, "/");
        let root = client.object(&session, "/").unwrap();
        assert_eq!(root.object.unwrap().name, "");
        observe(&mut client, &session, &state, "/service");
        for scope in ["/", "/service"] {
            let selected = reference(scope);
            fetch(&mut client, &session, &state, &selected);
            let observation = client.interface(&session, &selected).unwrap().clone();
            assert_eq!(
                observation.scope_ref.as_deref(),
                if scope == "/" { None } else { scope_ref }
            );
            let request = client
                .prepare_call(&session, "/service", &selected, "status", BTreeMap::new())
                .unwrap();
            let logical = decode_call_operation_request(
                request.request.body(),
                observation.descriptor.as_ref().unwrap(),
                limits(),
            )
            .unwrap();
            assert_eq!(logical.interface.reference, selected);
            assert_eq!(logical.interface.scope_ref, observation.scope_ref);
            let response = exchange(&state, request.request, true);
            assert!(matches!(
                client.complete(request.id, response).unwrap(),
                Completion::Call(record) if matches!(record.outcome, CallOutcome::Success(_))
            ));
        }
        assert_eq!(state.lock().unwrap().calls.load(Ordering::SeqCst), 2);
    }
}

#[test]
fn stale_scope_republication_and_ref_less_replacement_require_explicit_calls() {
    let selected = reference("/service");
    let state = Arc::new(Mutex::new(State::new(Some("instance-a"))));
    let (mut client, session) = setup();
    observe(&mut client, &session, &state, "/service");
    fetch(&mut client, &session, &state, &selected);
    let request = client
        .prepare_call(&session, "/service", &selected, "status", BTreeMap::new())
        .unwrap();
    state.lock().unwrap().republish(Some("instance-b"));
    let response = exchange(&state, request.request, true);
    assert_eq!(response.status(), http::StatusCode::CONFLICT);
    assert!(matches!(client.complete(request.id, response).unwrap(),
        Completion::Call(record) if matches!(record.outcome,
            CallOutcome::ProtocolFailure(ref error) if error.code == ProtocolErrorCode::InterfaceMismatch)));
    assert_eq!(state.lock().unwrap().calls.load(Ordering::SeqCst), 0);
    assert_ne!(
        client.interface(&session, &selected).unwrap().state,
        ObservationState::Fresh
    );
    assert!(
        client
            .prepare_call(&session, "/service", &selected, "status", BTreeMap::new())
            .is_err()
    );
    observe(&mut client, &session, &state, "/service");
    fetch(&mut client, &session, &state, &selected);
    assert_eq!(
        client
            .interface(&session, &selected)
            .unwrap()
            .scope_ref
            .as_deref(),
        Some("instance-b")
    );
    assert_eq!(
        state.lock().unwrap().calls.load(Ordering::SeqCst),
        0,
        "re-observation is not execution"
    );

    state.lock().unwrap().remove();
    observe(&mut client, &session, &state, "/service");
    assert!(client.interface(&session, &selected).is_none());
    state.lock().unwrap().republish(None);
    observe(&mut client, &session, &state, "/service");
    fetch(&mut client, &session, &state, &selected);
    let request = client
        .prepare_call(&session, "/service", &selected, "status", BTreeMap::new())
        .unwrap();
    // Ref-less replacement is intentionally not identity-trackable. Current
    // membership, live publication and both validators still apply at dispatch.
    state.lock().unwrap().republish(None);
    let response = exchange(&state, request.request, true);
    client.complete(request.id, response).unwrap();
    assert_eq!(state.lock().unwrap().calls.load(Ordering::SeqCst), 1);
}

#[test]
fn real_http_unknown_outcome_is_never_automatically_retried() {
    let state = Arc::new(Mutex::new(State::new(Some("instance-a"))));
    let (mut client, session) = setup();
    observe(&mut client, &session, &state, "/service");
    let selected = reference("/service");
    fetch(&mut client, &session, &state, &selected);
    let request = client
        .prepare_call(
            &session,
            "/service",
            &selected,
            "uncertain",
            BTreeMap::new(),
        )
        .unwrap();
    let response = exchange(&state, request.request, true);
    let completion = client.complete(request.id, response).unwrap();
    assert!(
        matches!(completion, Completion::Call(record) if matches!(record.outcome, CallOutcome::Unknown { .. }))
    );
    assert_eq!(client.in_flight_len(), 0);
    assert_eq!(state.lock().unwrap().calls.load(Ordering::SeqCst), 1);
    observe(&mut client, &session, &state, "/service");
    assert_eq!(state.lock().unwrap().calls.load(Ordering::SeqCst), 1);
}

#[test]
fn principal_visibility_and_validator_precedence_cross_real_http() {
    let state = Arc::new(Mutex::new(State::new(Some("instance-a"))));
    let endpoint = Endpoint::parse("http://localhost/wip").unwrap();
    let selected = reference("/service");
    let fetch = wip_protocol::FetchInterfaceRequest {
        interface: selected.clone(),
    };
    let response = exchange(
        &state,
        encode_fetch_interface_request(&endpoint, &fetch, limits()).unwrap(),
        false,
    );
    assert!(
        matches!(decode_fetch_interface_response(&fetch, &response, limits()).unwrap(), DecodedResponse::ProtocolFailure { error, .. } if error.code == ProtocolErrorCode::InterfaceNotFound)
    );
    let descriptor = StatusDefinition::descriptor();
    let mut call = wip_protocol::CallOperationRequest {
        target: wip_protocol::Target {
            path: "/service".into(),
            validator: Some(b"stale-object".to_vec()),
        },
        interface: wip_protocol::InterfaceTarget {
            reference: selected,
            scope_ref: Some("stale-scope".into()),
            validator: Some(b"stale-api".to_vec()),
        },
        operation: "status".into(),
        arguments: BTreeMap::new(),
    };
    for (visible, object_validator, scope_ref, interface_validator, expected) in [
        (
            false,
            b"stale-object".as_slice(),
            Some("stale-scope"),
            Some(b"stale-api".as_slice()),
            ProtocolErrorCode::NotFound,
        ),
        (
            true,
            b"stale-object".as_slice(),
            Some("stale-scope"),
            Some(b"stale-api".as_slice()),
            ProtocolErrorCode::ValidatorMismatch,
        ),
        (
            true,
            b"object-v1".as_slice(),
            Some("stale-scope"),
            Some(b"stale-api".as_slice()),
            ProtocolErrorCode::InterfaceMismatch,
        ),
        (
            true,
            b"object-v1".as_slice(),
            Some("instance-a"),
            None,
            ProtocolErrorCode::InterfaceValidatorRequired,
        ),
        (
            true,
            b"object-v1".as_slice(),
            None,
            Some(b"stale-api".as_slice()),
            ProtocolErrorCode::InterfaceValidatorMismatch,
        ),
    ] {
        call.target.validator = Some(object_validator.to_vec());
        call.interface.scope_ref = scope_ref.map(str::to_owned);
        call.interface.validator = interface_validator.map(<[u8]>::to_vec);
        let response = exchange(
            &state,
            encode_call_operation_request(&endpoint, &call, &descriptor, limits()).unwrap(),
            visible,
        );
        assert!(
            matches!(decode_call_operation_response(&call, &descriptor, &response, limits()).unwrap(), DecodedResponse::ProtocolFailure { error, .. } if error.code == expected)
        );
    }
    assert_eq!(state.lock().unwrap().calls.load(Ordering::SeqCst), 0);
}