haste_operation_executor/providers/deno_embedded/

mod.rs

use deno_core::cppgc::GcCell;
use deno_core::error::ModuleLoaderError;

pub mod pool;

use deno_core::{
    Extension, GarbageCollected, ModuleLoadOptions, ModuleLoadReferrer, ModuleLoader, ModuleType,
    OpState, PollEventLoopOptions, op2, resolve_import, serde_json, v8,
};
// main.rs
use deno_core::error::AnyError;
use haste_fhir_client::FHIRClient;
use haste_fhir_client::url::ParsedParameters;
use haste_fhir_model::r4::generated::resources::{Bundle, Parameters, Resource, ResourceType};
use haste_fhir_operation_error::OperationOutcomeError;
use json_patch::Patch;
use std::cell::RefCell;
use std::collections::HashMap;
use std::rc::Rc;
use std::sync::{Arc, LazyLock};
use std::time::Duration;
use tokio::sync::Mutex;

use deno_ast::{MediaType, ModuleSpecifier};
use deno_ast::{ParseParams, SourceMapOption};
use deno_core::ModuleLoadResponse;
use deno_error::JsErrorBox;

pub use pool::DenoPool;

use crate::structs::PluginCodeType;

/// Hard ceiling on how long a single custom-operation script may run before
/// its isolate is forcibly terminated.
///
/// V8 execution is fully synchronous from Rust's point of view: a runaway
/// script (e.g. an infinite loop) never yields back to the async executor,
/// so `tokio::time::timeout` cannot interrupt it. Instead an
/// [`ExecutionWatchdog`] on a separate OS thread calls
/// `IsolateHandle::terminate_execution`, which V8 supports specifically for
/// this purpose.
const EXECUTION_TIMEOUT: Duration = Duration::from_secs(10);

/// Hard ceiling on the V8 heap a single custom-operation script may grow to
/// before its isolate is forcibly terminated. Without this, one tenant's
/// script can exhaust memory shared by every other tenant's executions on
/// the same process.
const MAX_HEAP_BYTES: usize = 128 * 1024 * 1024;

/// Custom-operation source rarely changes between invocations of the same
/// operation, so cache the transpiled output keyed by (declared media type,
/// raw source). This is a pure function of its inputs -- caching it cannot
/// leak state between runs, it only avoids redundant TypeScript parsing on
/// every call. Bounded to avoid unbounded growth if many distinct scripts
/// are ever seen.
const TRANSPILE_CACHE_CAPACITY: usize = 256;

type TranspileCacheKey = (PluginCodeType, String);

static TRANSPILE_CACHE: LazyLock<
    std::sync::Mutex<HashMap<TranspileCacheKey, (ModuleType, String)>>,
> = LazyLock::new(|| std::sync::Mutex::new(HashMap::new()));

impl From<PluginCodeType> for MediaType {
    fn from(value: PluginCodeType) -> Self {
        match value {
            PluginCodeType::JavaScript => MediaType::JavaScript,
            PluginCodeType::TypeScript => MediaType::TypeScript,
        }
    }
}

// Load the snapshot generated by build.rs:
static RUNTIME_SNAPSHOT: &[u8] = include_bytes!(concat!(env!("OUT_DIR"), "/RUNJS_SNAPSHOT.bin"));

fn transpile_code_to_js(
    module_specifier: &ModuleSpecifier,
    media_type: MediaType,
    code: &str,
) -> Result<(ModuleType, String), JsErrorBox> {
    let (module_type, should_transpile) = match media_type {
        MediaType::JavaScript | MediaType::Mjs | MediaType::Cjs => (ModuleType::JavaScript, false),
        MediaType::Jsx
        | MediaType::TypeScript
        | MediaType::Mts
        | MediaType::Cts
        | MediaType::Dts
        | MediaType::Dmts
        | MediaType::Dcts
        | MediaType::Tsx => (ModuleType::JavaScript, true),
        MediaType::Json => (ModuleType::Json, false),
        _ => {
            return Err(JsErrorBox::generic(format!(
                "Unknown extension {media_type:?}",
            )));
        }
    };
    if should_transpile {
        let parsed = deno_ast::parse_module(ParseParams {
            specifier: module_specifier.clone(),
            text: code.into(),
            media_type,
            capture_tokens: false,
            scope_analysis: false,
            maybe_syntax: None,
        })
        .map_err(JsErrorBox::from_err)?;
        let res = parsed
            .transpile(
                &deno_ast::TranspileOptions {
                    imports_not_used_as_values: deno_ast::ImportsNotUsedAsValues::Remove,
                    decorators: deno_ast::DecoratorsTranspileOption::Ecma,
                    ..Default::default()
                },
                &deno_ast::TranspileModuleOptions { module_kind: None },
                &deno_ast::EmitOptions {
                    source_map: SourceMapOption::Separate,
                    inline_sources: true,
                    ..Default::default()
                },
            )
            .map_err(JsErrorBox::from_err)?;
        let res = res.into_source();
        // let source_map = res.source_map.unwrap().into_bytes();

        Ok((module_type, res.text))
    } else {
        Ok((module_type, code.to_string()))
    }
}

/// Resolves import specifiers but never actually loads anything from disk or
/// the network.
struct TsModuleLoader;
impl ModuleLoader for TsModuleLoader {
    fn resolve(
        &self,
        specifier: &str,
        referrer: &str,
        _kind: deno_core::ResolutionKind,
    ) -> Result<deno_core::ModuleSpecifier, ModuleLoaderError> {
        resolve_import(specifier, referrer).map_err(JsErrorBox::from_err)
    }

    fn load(
        &self,
        _module_specifier: &ModuleSpecifier,
        _maybe_referrer: Option<&ModuleLoadReferrer>,
        _options: ModuleLoadOptions,
    ) -> ModuleLoadResponse {
        ModuleLoadResponse::Sync(Err(JsErrorBox::generic(
            "Imports are not supported in custom operation scripts.",
        )))
    }
}

/// Transpiles `code` to JS, reusing a cached result when the same
/// `(media type, source)` pair has already been transpiled.
fn cached_transpile_to_js(
    plugin_code_type: PluginCodeType,
    code: &str,
) -> Result<(ModuleType, String), JsErrorBox> {
    let key = (plugin_code_type, code.to_string());

    if let Some(cached) = TRANSPILE_CACHE
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
        .get(&key)
    {
        return Ok(cached.clone());
    }

    let user_module_specifier = ModuleSpecifier::parse("memo://user.ts").unwrap();
    let result = transpile_code_to_js(&user_module_specifier, plugin_code_type.into(), code)?;

    let mut cache = TRANSPILE_CACHE
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner);
    if cache.len() >= TRANSPILE_CACHE_CAPACITY {
        cache.clear();
    }
    cache.insert(key, result.clone());

    Ok(result)
}

/// Runs a callback on a dedicated OS thread that forcibly terminates a V8
/// isolate if it doesn't finish within a deadline.
///
/// V8's `run_event_loop` blocks the calling thread for as long as the script
/// keeps the event loop busy; a script stuck in an infinite loop never polls
/// again, so nothing on that same thread (including a `tokio::time::sleep`)
/// can ever run to interrupt it. `IsolateHandle::terminate_execution` is
/// V8's supported mechanism for interrupting a busy isolate from another
/// thread, so the watchdog has to live on one.
struct ExecutionWatchdog {
    stop_tx: Option<std::sync::mpsc::Sender<()>>,
    thread: Option<std::thread::JoinHandle<()>>,
}

impl ExecutionWatchdog {
    fn start(isolate_handle: v8::IsolateHandle, timeout: Duration) -> Self {
        let (stop_tx, stop_rx) = std::sync::mpsc::channel::<()>();

        let thread = std::thread::spawn(move || {
            if matches!(
                stop_rx.recv_timeout(timeout),
                Err(std::sync::mpsc::RecvTimeoutError::Timeout)
            ) {
                isolate_handle.terminate_execution();
            }
        });

        Self {
            stop_tx: Some(stop_tx),
            thread: Some(thread),
        }
    }
}

impl Drop for ExecutionWatchdog {
    fn drop(&mut self) {
        if let Some(stop_tx) = self.stop_tx.take() {
            let _ = stop_tx.send(());
        }
        if let Some(thread) = self.thread.take() {
            let _ = thread.join();
        }
    }
}

struct JSRuntimeState<CTX, Client: FHIRClient<CTX, OperationOutcomeError>> {
    fhir_client: Arc<Client>,
    ctx: CTX,
    return_value: Option<serde_json::Value>,
    input: Rc<serde_json::Value>,
}

#[repr(C)]
pub struct InteropObject {
    value: GcCell<f64>,
}

unsafe impl GarbageCollected for InteropObject {
    fn trace(&self, _visitor: &mut v8::cppgc::Visitor) {}
    fn get_name(&self) -> &'static std::ffi::CStr {
        c"InteropObject"
    }
}

#[op2]
impl InteropObject {
    #[constructor]
    #[cppgc]
    fn new(value: f64) -> InteropObject {
        InteropObject {
            value: GcCell::new(value),
        }
    }

    #[getter]
    fn value(&self, isolate: &v8::Isolate) -> f64 {
        *self.value.get(isolate)
    }

    #[setter]
    fn value(&self, isolate: &mut v8::Isolate, value: f64) {
        self.value.set(isolate, value);
    }

    // #[fast]
    // fn double_value(&self, isolate: &v8::Isolate) -> f64 {
    //     *self.value.get(isolate) * 2.0
    // }

    #[static_method]
    #[cppgc]
    fn create(value: f64) -> InteropObject {
        InteropObject {
            value: GcCell::new(value),
        }
    }
}

/// Fetches the shared per-invocation state that every op needs to reach the
/// tenant's `FHIRClient` and request context.
fn runtime_state<CTX: Clone + 'static, Client: FHIRClient<CTX, OperationOutcomeError> + 'static>(
    state: &Rc<RefCell<OpState>>,
) -> Arc<Mutex<JSRuntimeState<CTX, Client>>> {
    state
        .borrow()
        .borrow::<Arc<Mutex<JSRuntimeState<CTX, Client>>>>()
        .clone()
}

fn parse_resource_type(resource_type: String) -> Result<ResourceType, JsErrorBox> {
    ResourceType::try_from(resource_type)
        .map_err(|_| JsErrorBox::type_error("Invalid resource type"))
}

/// Parses a FHIR search query string (e.g. `"name=Smith&_count=10"`, with or
/// without a leading `?`) into the parameter list `FHIRClient` expects.
fn parse_query(query: &str) -> Result<ParsedParameters, JsErrorBox> {
    ParsedParameters::try_from(query)
        .map_err(|error| JsErrorBox::type_error(format!("Invalid search parameters: {error}")))
}

fn parse_body<T: serde::de::DeserializeOwned>(
    value: serde_json::Value,
    what: &str,
) -> Result<T, JsErrorBox> {
    serde_json::from_value(value)
        .map_err(|error| JsErrorBox::type_error(format!("Invalid {what}: {error}")))
}

fn to_json<T: serde::Serialize>(value: &T) -> Result<serde_json::Value, JsErrorBox> {
    serde_json::to_value(value).map_err(|_| JsErrorBox::type_error("Failed to serialize response"))
}

/// Logs the underlying `OperationOutcomeError` and surfaces its diagnostics
/// to the script -- these errors are already meant to be shown to FHIR API
/// callers, so passing the message through gives script authors an
/// actionable reason their operation failed.
fn map_fhir_error(operation: &'static str) -> impl FnOnce(OperationOutcomeError) -> JsErrorBox {
    move |error| {
        tracing::error!(error = ?error, operation, "FHIR operation failed in custom operation script");
        JsErrorBox::type_error(error.to_string())
    }
}

#[op2]
#[serde]
pub async fn fhir_capabilities<
    CTX: Clone + 'static,
    Client: FHIRClient<CTX, OperationOutcomeError> + 'static,
>(
    state: Rc<RefCell<OpState>>,
) -> Result<serde_json::Value, JsErrorBox> {
    let app_state = runtime_state::<CTX, Client>(&state);
    let app_state = app_state.lock().await;

    let capabilities = app_state
        .fhir_client
        .capabilities(app_state.ctx.clone())
        .await
        .map_err(map_fhir_error("capabilities"))?;

    to_json(&capabilities)
}

#[op2]
#[serde]
pub async fn fhir_search_type<
    CTX: Clone + 'static,
    Client: FHIRClient<CTX, OperationOutcomeError> + 'static,
>(
    state: Rc<RefCell<OpState>>,
    #[string] resource_type: String,
    #[string] query: String,
) -> Result<serde_json::Value, JsErrorBox> {
    let resource_type = parse_resource_type(resource_type)?;
    let query = parse_query(&query)?;
    let app_state = runtime_state::<CTX, Client>(&state);
    let app_state = app_state.lock().await;

    let bundle = app_state
        .fhir_client
        .search_type(app_state.ctx.clone(), resource_type, query)
        .await
        .map_err(map_fhir_error("searchType"))?;

    to_json(&bundle)
}

#[op2]
#[serde]
pub async fn fhir_search_system<
    CTX: Clone + 'static,
    Client: FHIRClient<CTX, OperationOutcomeError> + 'static,
>(
    state: Rc<RefCell<OpState>>,
    #[string] query: String,
) -> Result<serde_json::Value, JsErrorBox> {
    let query = parse_query(&query)?;
    let app_state = runtime_state::<CTX, Client>(&state);
    let app_state = app_state.lock().await;

    let bundle = app_state
        .fhir_client
        .search_system(app_state.ctx.clone(), query)
        .await
        .map_err(map_fhir_error("searchSystem"))?;

    to_json(&bundle)
}

#[op2]
#[serde]
pub async fn fhir_create<
    CTX: Clone + 'static,
    Client: FHIRClient<CTX, OperationOutcomeError> + 'static,
>(
    state: Rc<RefCell<OpState>>,
    #[string] resource_type: String,
    #[serde] resource: serde_json::Value,
) -> Result<serde_json::Value, JsErrorBox> {
    let resource_type = parse_resource_type(resource_type)?;
    let resource: Resource = parse_body(resource, "resource")?;
    let app_state = runtime_state::<CTX, Client>(&state);
    let app_state = app_state.lock().await;

    let created = app_state
        .fhir_client
        .create(app_state.ctx.clone(), resource_type, resource)
        .await
        .map_err(map_fhir_error("create"))?;

    to_json(&created)
}

#[op2]
#[serde]
pub async fn fhir_update<
    CTX: Clone + 'static,
    Client: FHIRClient<CTX, OperationOutcomeError> + 'static,
>(
    state: Rc<RefCell<OpState>>,
    #[string] resource_type: String,
    #[string] id: String,
    #[serde] resource: serde_json::Value,
) -> Result<serde_json::Value, JsErrorBox> {
    let resource_type = parse_resource_type(resource_type)?;
    let resource: Resource = parse_body(resource, "resource")?;
    let app_state = runtime_state::<CTX, Client>(&state);
    let app_state = app_state.lock().await;

    let updated = app_state
        .fhir_client
        .update(app_state.ctx.clone(), resource_type, id, resource)
        .await
        .map_err(map_fhir_error("update"))?;

    to_json(&updated)
}

#[op2]
#[serde]
pub async fn fhir_conditional_update<
    CTX: Clone + 'static,
    Client: FHIRClient<CTX, OperationOutcomeError> + 'static,
>(
    state: Rc<RefCell<OpState>>,
    #[string] resource_type: String,
    #[string] query: String,
    #[serde] resource: serde_json::Value,
) -> Result<serde_json::Value, JsErrorBox> {
    let resource_type = parse_resource_type(resource_type)?;
    let query = parse_query(&query)?;
    let resource: Resource = parse_body(resource, "resource")?;
    let app_state = runtime_state::<CTX, Client>(&state);
    let app_state = app_state.lock().await;

    let updated = app_state
        .fhir_client
        .conditional_update(app_state.ctx.clone(), resource_type, query, resource)
        .await
        .map_err(map_fhir_error("conditionalUpdate"))?;

    to_json(&updated)
}

#[op2]
#[serde]
pub async fn fhir_patch<
    CTX: Clone + 'static,
    Client: FHIRClient<CTX, OperationOutcomeError> + 'static,
>(
    state: Rc<RefCell<OpState>>,
    #[string] resource_type: String,
    #[string] id: String,
    #[serde] patch: serde_json::Value,
) -> Result<serde_json::Value, JsErrorBox> {
    let resource_type = parse_resource_type(resource_type)?;
    let patch: Patch = parse_body(patch, "patch")?;
    let app_state = runtime_state::<CTX, Client>(&state);
    let app_state = app_state.lock().await;

    let patched = app_state
        .fhir_client
        .patch(app_state.ctx.clone(), resource_type, id, patch)
        .await
        .map_err(map_fhir_error("patch"))?;

    to_json(&patched)
}

#[op2]
#[serde]
pub async fn fhir_read<
    CTX: Clone + 'static,
    Client: FHIRClient<CTX, OperationOutcomeError> + 'static,
>(
    state: Rc<RefCell<OpState>>,
    #[string] resource_type: String,
    #[string] id: String,
) -> Result<serde_json::Value, JsErrorBox> {
    let resource_type = parse_resource_type(resource_type)?;
    let app_state = runtime_state::<CTX, Client>(&state);
    let app_state = app_state.lock().await;

    let resource = app_state
        .fhir_client
        .read(app_state.ctx.clone(), resource_type, id)
        .await
        .map_err(map_fhir_error("read"))?;

    match resource {
        Some(resource) => to_json(&resource),
        None => Ok(serde_json::Value::Null),
    }
}

#[op2]
#[serde]
pub async fn fhir_vread<
    CTX: Clone + 'static,
    Client: FHIRClient<CTX, OperationOutcomeError> + 'static,
>(
    state: Rc<RefCell<OpState>>,
    #[string] resource_type: String,
    #[string] id: String,
    #[string] version_id: String,
) -> Result<serde_json::Value, JsErrorBox> {
    let resource_type = parse_resource_type(resource_type)?;
    let app_state = runtime_state::<CTX, Client>(&state);
    let app_state = app_state.lock().await;

    let resource = app_state
        .fhir_client
        .vread(app_state.ctx.clone(), resource_type, id, version_id)
        .await
        .map_err(map_fhir_error("vread"))?;

    match resource {
        Some(resource) => to_json(&resource),
        None => Ok(serde_json::Value::Null),
    }
}

#[op2]
pub async fn fhir_delete_instance<
    CTX: Clone + 'static,
    Client: FHIRClient<CTX, OperationOutcomeError> + 'static,
>(
    state: Rc<RefCell<OpState>>,
    #[string] resource_type: String,
    #[string] id: String,
) -> Result<(), JsErrorBox> {
    let resource_type = parse_resource_type(resource_type)?;
    let app_state = runtime_state::<CTX, Client>(&state);
    let app_state = app_state.lock().await;

    app_state
        .fhir_client
        .delete_instance(app_state.ctx.clone(), resource_type, id)
        .await
        .map_err(map_fhir_error("deleteInstance"))
}

#[op2]
pub async fn fhir_delete_type<
    CTX: Clone + 'static,
    Client: FHIRClient<CTX, OperationOutcomeError> + 'static,
>(
    state: Rc<RefCell<OpState>>,
    #[string] resource_type: String,
    #[string] query: String,
) -> Result<(), JsErrorBox> {
    let resource_type = parse_resource_type(resource_type)?;
    let query = parse_query(&query)?;
    let app_state = runtime_state::<CTX, Client>(&state);
    let app_state = app_state.lock().await;

    app_state
        .fhir_client
        .delete_type(app_state.ctx.clone(), resource_type, query)
        .await
        .map_err(map_fhir_error("deleteType"))
}

#[op2]
pub async fn fhir_delete_system<
    CTX: Clone + 'static,
    Client: FHIRClient<CTX, OperationOutcomeError> + 'static,
>(
    state: Rc<RefCell<OpState>>,
    #[string] query: String,
) -> Result<(), JsErrorBox> {
    let query = parse_query(&query)?;
    let app_state = runtime_state::<CTX, Client>(&state);
    let app_state = app_state.lock().await;

    app_state
        .fhir_client
        .delete_system(app_state.ctx.clone(), query)
        .await
        .map_err(map_fhir_error("deleteSystem"))
}

#[op2]
#[serde]
pub async fn fhir_history_instance<
    CTX: Clone + 'static,
    Client: FHIRClient<CTX, OperationOutcomeError> + 'static,
>(
    state: Rc<RefCell<OpState>>,
    #[string] resource_type: String,
    #[string] id: String,
    #[string] query: String,
) -> Result<serde_json::Value, JsErrorBox> {
    let resource_type = parse_resource_type(resource_type)?;
    let query = parse_query(&query)?;
    let app_state = runtime_state::<CTX, Client>(&state);
    let app_state = app_state.lock().await;

    let bundle = app_state
        .fhir_client
        .history_instance(app_state.ctx.clone(), resource_type, id, query)
        .await
        .map_err(map_fhir_error("historyInstance"))?;

    to_json(&bundle)
}

#[op2]
#[serde]
pub async fn fhir_history_type<
    CTX: Clone + 'static,
    Client: FHIRClient<CTX, OperationOutcomeError> + 'static,
>(
    state: Rc<RefCell<OpState>>,
    #[string] resource_type: String,
    #[string] query: String,
) -> Result<serde_json::Value, JsErrorBox> {
    let resource_type = parse_resource_type(resource_type)?;
    let query = parse_query(&query)?;
    let app_state = runtime_state::<CTX, Client>(&state);
    let app_state = app_state.lock().await;

    let bundle = app_state
        .fhir_client
        .history_type(app_state.ctx.clone(), resource_type, query)
        .await
        .map_err(map_fhir_error("historyType"))?;

    to_json(&bundle)
}

#[op2]
#[serde]
pub async fn fhir_history_system<
    CTX: Clone + 'static,
    Client: FHIRClient<CTX, OperationOutcomeError> + 'static,
>(
    state: Rc<RefCell<OpState>>,
    #[string] query: String,
) -> Result<serde_json::Value, JsErrorBox> {
    let query = parse_query(&query)?;
    let app_state = runtime_state::<CTX, Client>(&state);
    let app_state = app_state.lock().await;

    let bundle = app_state
        .fhir_client
        .history_system(app_state.ctx.clone(), query)
        .await
        .map_err(map_fhir_error("historySystem"))?;

    to_json(&bundle)
}

#[op2]
#[serde]
pub async fn fhir_invoke_instance<
    CTX: Clone + 'static,
    Client: FHIRClient<CTX, OperationOutcomeError> + 'static,
>(
    state: Rc<RefCell<OpState>>,
    #[string] resource_type: String,
    #[string] id: String,
    #[string] operation: String,
    #[serde] parameters: serde_json::Value,
) -> Result<serde_json::Value, JsErrorBox> {
    let resource_type = parse_resource_type(resource_type)?;
    let parameters: Parameters = parse_body(parameters, "parameters")?;
    let app_state = runtime_state::<CTX, Client>(&state);
    let app_state = app_state.lock().await;

    let result = app_state
        .fhir_client
        .invoke_instance(
            app_state.ctx.clone(),
            resource_type,
            id,
            operation,
            parameters,
        )
        .await
        .map_err(map_fhir_error("invokeInstance"))?;

    to_json(&result)
}

#[op2]
#[serde]
pub async fn fhir_invoke_type<
    CTX: Clone + 'static,
    Client: FHIRClient<CTX, OperationOutcomeError> + 'static,
>(
    state: Rc<RefCell<OpState>>,
    #[string] resource_type: String,
    #[string] operation: String,
    #[serde] parameters: serde_json::Value,
) -> Result<serde_json::Value, JsErrorBox> {
    let resource_type = parse_resource_type(resource_type)?;
    let parameters: Parameters = parse_body(parameters, "parameters")?;
    let app_state = runtime_state::<CTX, Client>(&state);
    let app_state = app_state.lock().await;

    let result = app_state
        .fhir_client
        .invoke_type(app_state.ctx.clone(), resource_type, operation, parameters)
        .await
        .map_err(map_fhir_error("invokeType"))?;

    to_json(&result)
}

#[op2]
#[serde]
pub async fn fhir_invoke_system<
    CTX: Clone + 'static,
    Client: FHIRClient<CTX, OperationOutcomeError> + 'static,
>(
    state: Rc<RefCell<OpState>>,
    #[string] operation: String,
    #[serde] parameters: serde_json::Value,
) -> Result<serde_json::Value, JsErrorBox> {
    let parameters: Parameters = parse_body(parameters, "parameters")?;
    let app_state = runtime_state::<CTX, Client>(&state);
    let app_state = app_state.lock().await;

    let result = app_state
        .fhir_client
        .invoke_system(app_state.ctx.clone(), operation, parameters)
        .await
        .map_err(map_fhir_error("invokeSystem"))?;

    to_json(&result)
}

#[op2]
#[serde]
pub async fn fhir_transaction<
    CTX: Clone + 'static,
    Client: FHIRClient<CTX, OperationOutcomeError> + 'static,
>(
    state: Rc<RefCell<OpState>>,
    #[serde] bundle: serde_json::Value,
) -> Result<serde_json::Value, JsErrorBox> {
    let bundle: Bundle = parse_body(bundle, "bundle")?;
    let app_state = runtime_state::<CTX, Client>(&state);
    let app_state = app_state.lock().await;

    let result = app_state
        .fhir_client
        .transaction(app_state.ctx.clone(), bundle)
        .await
        .map_err(map_fhir_error("transaction"))?;

    to_json(&result)
}

#[op2]
#[serde]
pub async fn fhir_batch<
    CTX: Clone + 'static,
    Client: FHIRClient<CTX, OperationOutcomeError> + 'static,
>(
    state: Rc<RefCell<OpState>>,
    #[serde] bundle: serde_json::Value,
) -> Result<serde_json::Value, JsErrorBox> {
    let bundle: Bundle = parse_body(bundle, "bundle")?;
    let app_state = runtime_state::<CTX, Client>(&state);
    let app_state = app_state.lock().await;

    let result = app_state
        .fhir_client
        .batch(app_state.ctx.clone(), bundle)
        .await
        .map_err(map_fhir_error("batch"))?;

    to_json(&result)
}

#[op2]
#[serde]
pub async fn set_return_value<
    CTX: Clone + 'static,
    Client: FHIRClient<CTX, OperationOutcomeError> + 'static,
>(
    state: Rc<RefCell<OpState>>,
    #[serde] value: serde_json::Value,
) -> Result<(), JsErrorBox> {
    let app_state = runtime_state::<CTX, Client>(&state);
    let mut app_state = app_state.lock().await;

    app_state.return_value = Some(value);
    Ok(())
}

#[op2]
#[serde]
pub async fn get_input_value<
    CTX: Clone + 'static,
    Client: FHIRClient<CTX, OperationOutcomeError> + 'static,
>(
    state: Rc<RefCell<OpState>>,
) -> Result<serde_json::Value, JsErrorBox> {
    let app_state = runtime_state::<CTX, Client>(&state);
    let app_state = app_state.lock().await;

    Ok(serde_json::json!({
        "request": app_state.input.clone(),
    }))
}

/// Builds a fresh, never-yet-used V8 isolate wired up with the ops a
/// custom-operation script needs, but does not run anything against it.
fn build_deno_runtime<
    CTX: Clone + 'static,
    Client: FHIRClient<CTX, OperationOutcomeError> + 'static,
>() -> deno_core::JsRuntime {
    let runjs = Extension {
        name: "runjs",
        ops: std::borrow::Cow::Owned(vec![
            fhir_capabilities::<CTX, Client>(),
            fhir_search_type::<CTX, Client>(),
            fhir_search_system::<CTX, Client>(),
            fhir_create::<CTX, Client>(),
            fhir_update::<CTX, Client>(),
            fhir_conditional_update::<CTX, Client>(),
            fhir_patch::<CTX, Client>(),
            fhir_read::<CTX, Client>(),
            fhir_vread::<CTX, Client>(),
            fhir_delete_instance::<CTX, Client>(),
            fhir_delete_type::<CTX, Client>(),
            fhir_delete_system::<CTX, Client>(),
            fhir_history_instance::<CTX, Client>(),
            fhir_history_type::<CTX, Client>(),
            fhir_history_system::<CTX, Client>(),
            fhir_invoke_instance::<CTX, Client>(),
            fhir_invoke_type::<CTX, Client>(),
            fhir_invoke_system::<CTX, Client>(),
            fhir_transaction::<CTX, Client>(),
            fhir_batch::<CTX, Client>(),
            set_return_value::<CTX, Client>(),
            get_input_value::<CTX, Client>(),
        ]),
        ..Default::default()
    };

    let create_params = v8::Isolate::create_params().heap_limits(0, MAX_HEAP_BYTES);

    deno_core::JsRuntime::new(deno_core::RuntimeOptions {
        module_loader: Some(Rc::new(TsModuleLoader)),
        extensions: vec![runjs],
        startup_snapshot: Some(RUNTIME_SNAPSHOT),
        create_params: Some(create_params),
        ..Default::default()
    })
}

async fn run_code<
    CTX: Clone + 'static,
    Client: FHIRClient<CTX, OperationOutcomeError> + 'static,
>(
    mut deno_runtime: deno_core::JsRuntime,
    ctx: CTX,
    client: Arc<Client>,
    media_type: PluginCodeType,
    code: &str,
    input: serde_json::Value,
    timeout: Duration,
) -> Result<Option<serde_json::Value>, AnyError> {
    let isolate_handle = deno_runtime.v8_isolate().thread_safe_handle();

    {
        let terminate_handle = isolate_handle.clone();
        deno_runtime.add_near_heap_limit_callback(move |current_limit, _initial_limit| {
            // Forcefully stop the script and give V8 enough headroom to unwind
            // cleanly instead of retriggering this callback in a tight loop.
            terminate_handle.terminate_execution();
            current_limit * 2
        });
    }

    // Guards the whole execution below: if the script doesn't finish within
    // `timeout`, this forcibly terminates it from another thread.
    let _watchdog = ExecutionWatchdog::start(isolate_handle, timeout);

    let js_runtime_state = Arc::new(Mutex::new(JSRuntimeState {
        fhir_client: client,
        ctx,
        return_value: None,
        input: Rc::new(input),
    }));

    {
        let op_state = deno_runtime.op_state();
        let mut op_state = op_state.borrow_mut();
        op_state.put(js_runtime_state.clone());
    }

    let user_module_specifier = ModuleSpecifier::parse("memo://user.ts").unwrap();

    let (_module_type, js_code) = cached_transpile_to_js(media_type, code)?;

    let user_mod_id = deno_runtime
        .load_side_es_module_from_code(&user_module_specifier, js_code)
        .await?;

    let main_mod_id = deno_runtime
        .load_main_es_module_from_code(
            &ModuleSpecifier::parse("memo://main.ts").unwrap(),
            "import userFunction from 'memo://user.ts'; _internal_.setReturnValue(await userFunction(await _internal_.getInputValue()));"
                .to_string(),
        )
        .await?;

    // let mod_id = deno_runtime.load_main_es_module(&main_module).await?;
    let user_module_load = deno_runtime.mod_evaluate(user_mod_id);
    let main_module_load = deno_runtime.mod_evaluate(main_mod_id);

    deno_runtime
        .run_event_loop(PollEventLoopOptions::default())
        .await?;

    user_module_load.await?;
    main_module_load.await?;
    // Clean up the JSRuntimeState from the op state
    // Allows unwrapping RC in next call to have owned_values.
    {
        let op_state = deno_runtime.op_state();
        let mut op_state = op_state.borrow_mut();

        op_state.take::<Arc<Mutex<JSRuntimeState<CTX, Client>>>>();
    }

    let owned_runetime_state = Arc::try_unwrap(js_runtime_state)
        .map_err(|_| deno_error::JsErrorBox::type_error("Failed to unwrap JSRuntimeState"))?
        .into_inner();

    Ok(owned_runetime_state.return_value)
}

#[cfg(test)]
pub(crate) mod tests {
    use super::*;
    use deno_core::serde_json::json;
    use haste_fhir_client::request::{FHIRRequest, FHIRResponse};
    use haste_fhir_client::url::ParsedParameters;
    use haste_fhir_model::r4::generated::resources::{
        Bundle, CapabilityStatement, Parameters, Resource,
    };
    use haste_fhir_model::r4::generated::terminology::IssueType;
    use json_patch::Patch;
    use std::time::Instant;

    #[derive(Clone)]
    pub(crate) struct TestCtx;

    /// A `FHIRClient` double that isn't expected to be called by any of these
    /// tests -- they only exercise the sandboxing behavior of `run_code`
    /// itself, not the `fhir.read` op.
    pub(crate) struct MockClient;

    impl FHIRClient<TestCtx, OperationOutcomeError> for MockClient {
        async fn request(
            &self,
            _ctx: TestCtx,
            _request: FHIRRequest,
        ) -> Result<FHIRResponse, OperationOutcomeError> {
            unimplemented!("not used by these tests")
        }

        async fn capabilities(
            &self,
            _ctx: TestCtx,
        ) -> Result<CapabilityStatement, OperationOutcomeError> {
            Ok(CapabilityStatement::default())
        }

        async fn search_system(
            &self,
            _ctx: TestCtx,
            _parameters: ParsedParameters,
        ) -> Result<Bundle, OperationOutcomeError> {
            unimplemented!("not used by these tests")
        }

        async fn search_type(
            &self,
            _ctx: TestCtx,
            _resource_type: ResourceType,
            _parameters: ParsedParameters,
        ) -> Result<Bundle, OperationOutcomeError> {
            Ok(Bundle::default())
        }

        async fn create(
            &self,
            _ctx: TestCtx,
            _resource_type: ResourceType,
            resource: Resource,
        ) -> Result<Resource, OperationOutcomeError> {
            Ok(resource)
        }

        async fn update(
            &self,
            _ctx: TestCtx,
            _resource_type: ResourceType,
            _id: String,
            _resource: Resource,
        ) -> Result<Resource, OperationOutcomeError> {
            unimplemented!("not used by these tests")
        }

        async fn conditional_update(
            &self,
            _ctx: TestCtx,
            _resource_type: ResourceType,
            _parameters: ParsedParameters,
            _resource: Resource,
        ) -> Result<Resource, OperationOutcomeError> {
            unimplemented!("not used by these tests")
        }

        async fn patch(
            &self,
            _ctx: TestCtx,
            _resource_type: ResourceType,
            _id: String,
            _patches: Patch,
        ) -> Result<Resource, OperationOutcomeError> {
            unimplemented!("not used by these tests")
        }

        async fn read(
            &self,
            _ctx: TestCtx,
            _resource_type: ResourceType,
            _id: String,
        ) -> Result<Option<Resource>, OperationOutcomeError> {
            Ok(None)
        }

        async fn vread(
            &self,
            _ctx: TestCtx,
            _resource_type: ResourceType,
            _id: String,
            _version_id: String,
        ) -> Result<Option<Resource>, OperationOutcomeError> {
            unimplemented!("not used by these tests")
        }

        async fn delete_instance(
            &self,
            _ctx: TestCtx,
            _resource_type: ResourceType,
            _id: String,
        ) -> Result<(), OperationOutcomeError> {
            Ok(())
        }

        async fn delete_type(
            &self,
            _ctx: TestCtx,
            _resource_type: ResourceType,
            _parameters: ParsedParameters,
        ) -> Result<(), OperationOutcomeError> {
            unimplemented!("not used by these tests")
        }

        async fn delete_system(
            &self,
            _ctx: TestCtx,
            _parameters: ParsedParameters,
        ) -> Result<(), OperationOutcomeError> {
            unimplemented!("not used by these tests")
        }

        async fn history_system(
            &self,
            _ctx: TestCtx,
            _parameters: ParsedParameters,
        ) -> Result<Bundle, OperationOutcomeError> {
            unimplemented!("not used by these tests")
        }

        async fn history_type(
            &self,
            _ctx: TestCtx,
            _resource_type: ResourceType,
            _parameters: ParsedParameters,
        ) -> Result<Bundle, OperationOutcomeError> {
            unimplemented!("not used by these tests")
        }

        async fn history_instance(
            &self,
            _ctx: TestCtx,
            _resource_type: ResourceType,
            _id: String,
            _parameters: ParsedParameters,
        ) -> Result<Bundle, OperationOutcomeError> {
            unimplemented!("not used by these tests")
        }

        async fn invoke_instance(
            &self,
            _ctx: TestCtx,
            _resource_type: ResourceType,
            _id: String,
            _operation: String,
            _parameters: Parameters,
        ) -> Result<Resource, OperationOutcomeError> {
            unimplemented!("not used by these tests")
        }

        async fn invoke_type(
            &self,
            _ctx: TestCtx,
            _resource_type: ResourceType,
            _operation: String,
            _parameters: Parameters,
        ) -> Result<Resource, OperationOutcomeError> {
            unimplemented!("not used by these tests")
        }

        async fn invoke_system(
            &self,
            _ctx: TestCtx,
            operation: String,
            _parameters: Parameters,
        ) -> Result<Resource, OperationOutcomeError> {
            serde_json::from_value(json!({ "resourceType": "Parameters", "id": operation }))
                .map_err(|error| {
                    OperationOutcomeError::error(IssueType::invalid(), error.to_string())
                })
        }

        async fn transaction(
            &self,
            _ctx: TestCtx,
            _bundle: Bundle,
        ) -> Result<Bundle, OperationOutcomeError> {
            unimplemented!("not used by these tests")
        }

        async fn batch(
            &self,
            _ctx: TestCtx,
            _bundle: Bundle,
        ) -> Result<Bundle, OperationOutcomeError> {
            unimplemented!("not used by these tests")
        }
    }

    #[tokio::test]
    async fn happy_path_javascript_returns_value() {
        let deno_runtime = build_deno_runtime::<TestCtx, MockClient>();
        let result = run_code(
            deno_runtime,
            TestCtx,
            Arc::new(MockClient),
            PluginCodeType::JavaScript,
            "export default async function () { return { answer: 42 }; }",
            json!({}),
            Duration::from_secs(5),
        )
        .await
        .expect("script should run successfully");

        assert_eq!(result, Some(json!({ "answer": 42 })));
    }

    #[tokio::test]
    async fn happy_path_typescript_returns_value() {
        let deno_runtime = build_deno_runtime::<TestCtx, MockClient>();
        let result = run_code(
            deno_runtime,
            TestCtx,
            Arc::new(MockClient),
            PluginCodeType::TypeScript,
            "export default async function (): Promise<{ answer: number }> { return { answer: 42 }; }",
            json!({}),
            Duration::from_secs(5),
        )
        .await
        .expect("script should run successfully");

        assert_eq!(result, Some(json!({ "answer": 42 })));
    }

    #[tokio::test]
    async fn fhir_create_op_round_trips_through_the_client() {
        let deno_runtime = build_deno_runtime::<TestCtx, MockClient>();
        let result = run_code(
            deno_runtime,
            TestCtx,
            Arc::new(MockClient),
            PluginCodeType::JavaScript,
            "export default async function () { return await fhir.create('Patient', { resourceType: 'Patient', id: 'example' }); }",
            json!({}),
            Duration::from_secs(5),
        )
        .await
        .expect("script should call fhir.create");

        assert_eq!(
            result,
            Some(json!({ "resourceType": "Patient", "id": "example" }))
        );
    }

    #[tokio::test]
    async fn fhir_search_type_op_returns_a_bundle() {
        let deno_runtime = build_deno_runtime::<TestCtx, MockClient>();
        let result = run_code(
            deno_runtime,
            TestCtx,
            Arc::new(MockClient),
            PluginCodeType::JavaScript,
            "export default async function () { return await fhir.searchType('Patient', { name: 'Smith', _count: 10 }); }",
            json!({}),
            Duration::from_secs(5),
        )
        .await
        .expect("script should call fhir.searchType");

        assert_eq!(result, Some(json!({ "resourceType": "Bundle" })));
    }

    #[tokio::test]
    async fn fhir_delete_instance_op_resolves() {
        let deno_runtime = build_deno_runtime::<TestCtx, MockClient>();
        let result = run_code(
            deno_runtime,
            TestCtx,
            Arc::new(MockClient),
            PluginCodeType::JavaScript,
            "export default async function () { await fhir.deleteInstance('Patient', 'example'); return { deleted: true }; }",
            json!({}),
            Duration::from_secs(5),
        )
        .await
        .expect("script should call fhir.deleteInstance");

        assert_eq!(result, Some(json!({ "deleted": true })));
    }

    #[tokio::test]
    async fn fhir_invoke_system_op_defaults_missing_parameters() {
        let deno_runtime = build_deno_runtime::<TestCtx, MockClient>();
        let result = run_code(
            deno_runtime,
            TestCtx,
            Arc::new(MockClient),
            PluginCodeType::JavaScript,
            "export default async function () { return await fhir.invokeSystem('everything'); }",
            json!({}),
            Duration::from_secs(5),
        )
        .await
        .expect("script should call fhir.invokeSystem");

        assert_eq!(
            result,
            Some(json!({ "resourceType": "Parameters", "id": "everything" }))
        );
    }

    #[tokio::test]
    async fn fhir_capabilities_op_returns_the_capability_statement() {
        let deno_runtime = build_deno_runtime::<TestCtx, MockClient>();
        let result = run_code(
            deno_runtime,
            TestCtx,
            Arc::new(MockClient),
            PluginCodeType::JavaScript,
            "export default async function () { return await fhir.capabilities(); }",
            json!({}),
            Duration::from_secs(5),
        )
        .await
        .expect("script should call fhir.capabilities");

        assert_eq!(
            result,
            Some(json!({ "resourceType": "CapabilityStatement" }))
        );
    }

    #[tokio::test]
    async fn imports_are_rejected() {
        // Regression test: this loader used to read arbitrary files off the
        // host filesystem for any `import` a tenant's script happened to
        // contain. It must now be rejected outright.
        let deno_runtime = build_deno_runtime::<TestCtx, MockClient>();
        let result = run_code(
            deno_runtime,
            TestCtx,
            Arc::new(MockClient),
            PluginCodeType::JavaScript,
            "import secret from 'file:///etc/passwd'; export default async function () { return secret; }",
            json!({}),
            Duration::from_secs(5),
        )
        .await;

        assert!(
            result.is_err(),
            "importing an arbitrary file must never succeed"
        );
    }

    #[tokio::test]
    async fn runaway_script_is_terminated_within_timeout() {
        let timeout = Duration::from_millis(200);
        let started = Instant::now();

        let deno_runtime = build_deno_runtime::<TestCtx, MockClient>();
        let result = run_code(
            deno_runtime,
            TestCtx,
            Arc::new(MockClient),
            PluginCodeType::JavaScript,
            "export default async function () { while (true) {} }",
            json!({}),
            timeout,
        )
        .await;

        assert!(result.is_err(), "runaway script must be terminated");
        assert!(
            started.elapsed() < timeout * 5,
            "termination took far longer than the configured timeout: {:?}",
            started.elapsed()
        );
    }
}