next_napi_bindings/next_api/

endpoint.rs

use std::{ops::Deref, sync::Arc};

use anyhow::Result;
use futures_util::TryFutureExt;
use napi::{JsFunction, bindgen_prelude::External};
use napi_derive::napi;
use next_api::{
    operation::OptionEndpoint,
    paths::AssetPath,
    route::{
        Endpoint, EndpointOutputPaths, endpoint_client_changed_operation,
        endpoint_server_changed_operation, endpoint_write_to_disk_operation,
    },
};
use tracing::Instrument;
use turbo_rcstr::RcStr;
use turbo_tasks::{
    Completion, Effects, OperationVc, ReadRef, Vc, read_strongly_consistent_and_apply_effects,
};
use turbopack_core::issue::{IssueFilter, PlainIssue};

use crate::next_api::utils::{
    DetachedVc, NapiIssue, RootTask, TurbopackResult, strongly_consistent_catch_collectables,
    subscribe,
};

#[napi(object)]
#[derive(Default)]
pub struct NapiEndpointConfig {}

#[napi(object)]
#[derive(Default)]
pub struct NapiAssetPath {
    pub path: RcStr,
    pub content_hash: RcStr,
}

impl From<AssetPath> for NapiAssetPath {
    fn from(asset_path: AssetPath) -> Self {
        Self {
            path: asset_path.path,
            content_hash: asset_path.content_hash,
        }
    }
}

#[napi(object)]
#[derive(Default)]
pub struct NapiWrittenEndpoint {
    pub r#type: String,
    pub entry_path: Option<String>,
    pub client_paths: Vec<String>,
    pub server_paths: Vec<NapiAssetPath>,
    pub config: NapiEndpointConfig,
}

impl From<Option<EndpointOutputPaths>> for NapiWrittenEndpoint {
    fn from(written_endpoint: Option<EndpointOutputPaths>) -> Self {
        match written_endpoint {
            Some(EndpointOutputPaths::NodeJs {
                server_entry_path,
                server_paths,
                client_paths,
            }) => Self {
                r#type: "nodejs".to_string(),
                entry_path: Some(server_entry_path.into_owned()),
                client_paths: client_paths.into_iter().map(From::from).collect(),
                server_paths: server_paths.into_iter().map(From::from).collect(),
                ..Default::default()
            },
            Some(EndpointOutputPaths::Edge {
                server_paths,
                client_paths,
            }) => Self {
                r#type: "edge".to_string(),
                client_paths: client_paths.into_iter().map(From::from).collect(),
                server_paths: server_paths.into_iter().map(From::from).collect(),
                ..Default::default()
            },
            Some(EndpointOutputPaths::NotFound) | None => Self {
                r#type: "none".to_string(),
                ..Default::default()
            },
        }
    }
}

// NOTE(alexkirsz) We go through an extra layer of indirection here because of
// two factors:
// 1. rustc currently has a bug where using a dyn trait as a type argument to
//    some async functions (in this case `endpoint_write_to_disk`) can cause
//    higher-ranked lifetime errors. See https://github.com/rust-lang/rust/issues/102211
// 2. the type_complexity clippy lint.
pub struct ExternalEndpoint(pub DetachedVc<OptionEndpoint>);

impl Deref for ExternalEndpoint {
    type Target = DetachedVc<OptionEndpoint>;

    fn deref(&self) -> &Self::Target {
        &self.0
    }
}

/// Build an `IssueFilter` by reading the project from the endpoint's
/// `OperationVc<OptionEndpoint>` and extracting ignore rules from its config.
///
/// If the upstream endpoint operation fails to resolve (e.g. because the build
/// graph cannot be evaluated transiently — for example during a mid-session
/// `node_modules` reshuffle), this falls back to a default filter rather than
/// propagating the error.  In this scenario we believe the caller will already be observing the
/// same error
async fn issue_filter_from_endpoint(
    endpoint_op: OperationVc<OptionEndpoint>,
) -> ReadRef<IssueFilter> {
    if let Ok(ep_option) = endpoint_op.connect().await
        && let Some(ep) = &*ep_option
        && let Ok(filter) = ep.project().issue_filter().await
    {
        filter
    } else {
        ReadRef::new_owned(IssueFilter::warnings_and_foreign_errors())
    }
}

#[turbo_tasks::value(serialization = "skip")]
struct WrittenEndpointWithIssues {
    written: Option<ReadRef<EndpointOutputPaths>>,
    issues: Arc<Vec<ReadRef<PlainIssue>>>,
    effects: Arc<Effects>,
}

#[turbo_tasks::function(operation, root)]
async fn get_written_endpoint_with_issues_operation(
    endpoint_op: OperationVc<OptionEndpoint>,
) -> Result<Vc<WrittenEndpointWithIssues>> {
    let write_to_disk_op = endpoint_write_to_disk_operation(endpoint_op);
    let filter = issue_filter_from_endpoint(endpoint_op).await;
    let (written, issues, effects) =
        strongly_consistent_catch_collectables(write_to_disk_op, &filter).await?;
    Ok(WrittenEndpointWithIssues {
        written,
        issues,
        effects,
    }
    .cell())
}

#[tracing::instrument(level = "info", name = "write endpoint to disk", skip_all)]
#[napi]
pub async fn endpoint_write_to_disk(
    #[napi(ts_arg_type = "{ __napiType: \"Endpoint\" }")] endpoint: External<ExternalEndpoint>,
) -> napi::Result<TurbopackResult<NapiWrittenEndpoint>> {
    let ctx = endpoint.turbopack_ctx();
    let endpoint_op = ***endpoint;
    let (written, issues) = endpoint
        .turbopack_ctx()
        .turbo_tasks()
        .run(async move {
            let written_entrypoint_with_issues_op =
                get_written_endpoint_with_issues_operation(endpoint_op);
            let read = read_strongly_consistent_and_apply_effects(
                written_entrypoint_with_issues_op,
                |v| &v.effects,
            )
            .await?;
            let WrittenEndpointWithIssues {
                written, issues, ..
            } = &*read;

            Ok((written.clone(), issues.clone()))
        })
        .or_else(|e| ctx.throw_turbopack_internal_result(&e.into()))
        .await?;
    Ok(TurbopackResult {
        result: NapiWrittenEndpoint::from(written.map(ReadRef::into_owned)),
        issues: issues.iter().map(|i| NapiIssue::from(&**i)).collect(),
    })
}

#[tracing::instrument(level = "info", name = "get server-side endpoint changes", skip_all)]
#[napi(ts_return_type = "{ __napiType: \"RootTask\" }")]
pub fn endpoint_server_changed_subscribe(
    #[napi(ts_arg_type = "{ __napiType: \"Endpoint\" }")] endpoint: External<ExternalEndpoint>,
    issues: bool,
    func: JsFunction,
) -> napi::Result<External<RootTask>> {
    let turbopack_ctx = endpoint.turbopack_ctx().clone();
    let endpoint = ***endpoint;
    subscribe(
        turbopack_ctx,
        func,
        move || {
            async move {
                let issues_and_diags_op = subscribe_issues_and_diags_operation(endpoint, issues);
                let result =
                    read_strongly_consistent_and_apply_effects(issues_and_diags_op, |v| &v.effects)
                        .await?;
                Ok(result)
            }
            .instrument(tracing::info_span!("server changes subscription"))
        },
        |ctx| {
            let EndpointIssuesAndDiags {
                changed: _,
                issues,
                effects: _,
            } = &*ctx.value;

            Ok(vec![TurbopackResult {
                result: (),
                issues: issues.iter().map(|i| NapiIssue::from(&**i)).collect(),
            }])
        },
    )
}

#[turbo_tasks::value(shared, serialization = "skip", eq = "manual")]
struct EndpointIssuesAndDiags {
    changed: Option<ReadRef<Completion>>,
    issues: Arc<Vec<ReadRef<PlainIssue>>>,
    effects: Arc<Effects>,
}

impl PartialEq for EndpointIssuesAndDiags {
    fn eq(&self, other: &Self) -> bool {
        (match (&self.changed, &other.changed) {
            (Some(a), Some(b)) => ReadRef::ptr_eq(a, b),
            (None, None) => true,
            (None, Some(_)) | (Some(_), None) => false,
        }) && self.issues == other.issues
    }
}

impl Eq for EndpointIssuesAndDiags {}

#[turbo_tasks::function(operation, root)]
async fn subscribe_issues_and_diags_operation(
    endpoint_op: OperationVc<OptionEndpoint>,
    should_include_issues: bool,
) -> Result<Vc<EndpointIssuesAndDiags>> {
    let changed_op = endpoint_server_changed_operation(endpoint_op);

    // Use catch-collectables in both branches so transient build-graph errors
    // (e.g. missing `node_modules/next` during a concurrent install) surface as
    // Issues rather than killing the subscription with a `TurbopackInternalError`.
    // When `should_include_issues` is false the caller doesn't need the Issue
    // payload, but we still need the catch path to avoid the FATAL.
    let filter = issue_filter_from_endpoint(endpoint_op).await;
    let (changed_value, issues, effects) =
        strongly_consistent_catch_collectables(changed_op, &filter).await?;
    Ok(EndpointIssuesAndDiags {
        changed: changed_value,
        issues: if should_include_issues {
            issues
        } else {
            Arc::new(vec![])
        },
        effects,
    }
    .cell())
}

#[tracing::instrument(level = "info", name = "get client-side endpoint changes", skip_all)]
#[napi(ts_return_type = "{ __napiType: \"RootTask\" }")]
pub fn endpoint_client_changed_subscribe(
    #[napi(ts_arg_type = "{ __napiType: \"Endpoint\" }")] endpoint: External<ExternalEndpoint>,
    func: JsFunction,
) -> napi::Result<External<RootTask>> {
    let turbopack_ctx = endpoint.turbopack_ctx().clone();
    let endpoint_op = ***endpoint;
    subscribe(
        turbopack_ctx,
        func,
        move || {
            async move {
                let changed_op = endpoint_client_changed_operation(endpoint_op);
                // We don't capture issues and diagnostics here since we don't want to be
                // notified when they change.  We also want errors to propagate so we don't use
                // strongly_consistent_catch_collectibles either.
                //
                // This must be a *read*, not just a resolve, because we need the root task created
                // by `subscribe` to re-run when the `Completion`'s value changes (via equality),
                // even if the cell id doesn't change.
                //
                let _ = changed_op.read_strongly_consistent().await?;
                Ok(())
            }
            .instrument(tracing::info_span!("client changes subscription"))
        },
        |_| {
            Ok(vec![TurbopackResult {
                result: (),
                issues: vec![],
            }])
        },
    )
}