arete_server/

lib.rs

//! # arete-server
//!
//! WebSocket server and projection handlers for Arete streaming pipelines.
//!
//! This crate provides a builder API for creating Arete servers that:
//!
//! - Process Solana blockchain data via Yellowstone gRPC
//! - Transform data using the Arete VM
//! - Stream entity updates over WebSockets to connected clients
//! - Support multiple streaming modes (State, List, Append)
//!
//! ## Quick Start
//!
//! ```rust,ignore
//! use arete_server::{Server, Spec};
//!
//! #[tokio::main]
//! async fn main() -> anyhow::Result<()> {
//!     Server::builder()
//!         .spec(my_spec())
//!         .websocket()
//!         .bind("[::]:8877".parse()?)
//!         .health_monitoring()
//!         .start()
//!         .await
//! }
//! ```
//!
//! ## Feature Flags
//!
//! - `otel` - OpenTelemetry integration for metrics and distributed tracing

pub mod account_policy;
pub mod bus;
pub mod cache;
pub mod compression;
pub mod config;
pub mod health;
mod http;
pub mod http_health;
pub mod http_server;
pub mod journal;
pub mod materialized_view;
#[cfg(feature = "otel")]
pub mod metrics;
pub mod mutation_batch;
pub mod program_runtime;
pub mod projector;
pub mod row_scope;
pub mod runtime;
pub mod shared_entity;
pub mod snapshot;
pub mod solana_gateway_usage;
pub mod sorted_cache;
pub mod telemetry;
pub mod view;
pub mod websocket;

pub use arete_auth::{
    AsyncVerifier, KeyLoader, Limits, SolanaGatewayAuthorization, SolanaGatewayAuthorizationError,
    SolanaGatewayScope, TargetKind, TokenVerifier, VerifyingKey, SCOPE_FORK_READ, SCOPE_READ,
    SCOPE_TRANSACTION_INSPECT, SCOPE_TRANSACTION_SEND, SOLANA_GATEWAY_AUDIENCE,
};
pub use bus::{BusManager, BusMessage, StateUpdate};
pub use cache::{CacheWrite, EntityCache, EntityCacheConfig, PatchOrigin};
pub use config::{
    HealthConfig, HttpHealthConfig, HttpLimitsConfig, HttpServerConfig, ReconnectionConfig,
    RuntimePlan, ServerConfig, TransactionConfig, WebSocketConfig, WebSocketDeliveryConfig,
    WebSocketLimitsConfig, YellowstoneConfig,
};
pub use health::{HealthMonitor, SlotTracker, StreamStatus};
pub use http_health::HttpHealthServer;
pub use http_server::HttpServer;
pub use journal::{EventJournal, JournalConfig, ReplayWindow};
pub use materialized_view::{MaterializedView, MaterializedViewRegistry, ViewEffect};
#[cfg(feature = "otel")]
pub use metrics::Metrics;
pub use mutation_batch::{EventContext, MutationBatch, SlotContext, SlotIndexDomain};
pub use program_runtime::{
    IdlContentHash, NormalizedIdlHash, ProgramAccountReaderFn, ProgramReleaseHash,
    ProgramRuntimeCatalog, ProgramRuntimeDefinition, ProgramSpecHash,
};
pub use projector::{EntityResync, Projector};
pub use row_scope::{RowScope, ScopeField, DEFAULT_MAX_SCOPE_PARTITIONS};
pub use runtime::{ConnectionServer, Runtime, RuntimeHandle};
pub use shared_entity::{EntityFields, SharedEntity};
pub use snapshot::{SnapshotConfig, SnapshotService};
pub use solana_gateway_usage::{
    SolanaGatewayUsageObservation, SolanaGatewayUsageObserver, SolanaGatewayUsageSurface,
};
pub use telemetry::{init as init_telemetry, TelemetryConfig};
#[cfg(feature = "otel")]
pub use telemetry::{init_with_otel, TelemetryGuard};
pub use view::{Delivery, Filters, Projection, ViewIndex, ViewSpec};
pub use websocket::{
    AllowAllAuthPlugin, AuthContext, AuthDecision, AuthDeny, AuthErrorDetails, ChannelUsageEmitter,
    ClientInfo, ClientManager, ConnectionAuthRequest, ErrorResponse, Frame, HttpUsageEmitter, Mode,
    RateLimitConfig, RateLimitResult, RateLimiterConfig, RefreshAuthRequest, RefreshAuthResponse,
    RetryPolicy, SignedSessionAuthPlugin, SnapshotOptions, SocketIssueMessage,
    StaticTokenAuthPlugin, Subscription, SubscriptionQuery, WebSocketAuthPlugin,
    WebSocketRateLimiter, WebSocketServer, WebSocketUsageBatch, WebSocketUsageEmitter,
    WebSocketUsageEnvelope, WebSocketUsageEvent,
};
pub use websocket::{WebSocketAdmissionProvider, WebSocketConnectionPermit};

use anyhow::Result;
use arete_interpreter::ast::ViewDef;
use std::net::SocketAddr;
use std::sync::Arc;

/// Type alias for a parser setup function.
pub type ParserSetupFn = Arc<
    dyn Fn(
            tokio::sync::mpsc::Sender<MutationBatch>,
            Option<HealthMonitor>,
            ReconnectionConfig,
        ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<()>> + Send>>
        + Send
        + Sync,
>;

/// Specification for a Arete server
/// Contains bytecode, parsers, and program information
pub struct Spec {
    pub bytecode: arete_interpreter::compiler::MultiEntityBytecode,
    pub program_ids: Vec<String>,
    pub parser_setup: Option<ParserSetupFn>,
    pub program_runtime_definitions: Vec<ProgramRuntimeDefinition>,
    pub entity_specs: Vec<arete_interpreter::ast::SerializableStreamSpec>,
    pub views: Vec<ViewDef>,
}

impl Spec {
    pub fn new(
        bytecode: arete_interpreter::compiler::MultiEntityBytecode,
        program_id: impl Into<String>,
    ) -> Self {
        Self {
            bytecode,
            program_ids: vec![program_id.into()],
            parser_setup: None,
            program_runtime_definitions: Vec::new(),
            entity_specs: Vec::new(),
            views: Vec::new(),
        }
    }

    pub fn with_parser_setup(mut self, setup_fn: ParserSetupFn) -> Self {
        self.parser_setup = Some(setup_fn);
        self
    }

    pub fn with_program_runtime_definitions(
        mut self,
        definitions: Vec<ProgramRuntimeDefinition>,
    ) -> Self {
        for definition in &definitions {
            if !self.program_ids.contains(&definition.program_id) {
                self.program_ids.push(definition.program_id.clone());
            }
        }
        self.program_runtime_definitions = definitions;
        self
    }

    pub fn with_entity_specs(
        mut self,
        entity_specs: Vec<arete_interpreter::ast::SerializableStreamSpec>,
    ) -> Self {
        self.entity_specs = entity_specs;
        self
    }

    pub fn with_views(mut self, views: Vec<ViewDef>) -> Self {
        self.views = views;
        self
    }
}

/// Main server interface with fluent builder API
pub mod token_discovery;

pub struct Server;

impl Server {
    /// Create a new server builder
    pub fn builder() -> ServerBuilder {
        ServerBuilder::new()
    }

    /// Build a standalone HTTP gateway without stack or live-stream capabilities.
    pub fn solana_gateway(target_id: impl Into<String>) -> SolanaGatewayBuilder {
        SolanaGatewayBuilder::new(target_id.into())
    }
}

/// Constrained builder for health, chain reads, and fixed transaction routes.
///
/// The final runtime plan is fixed when `build` or `start` is called. This
/// builder intentionally has no Spec, WebSocket, stack-query, program-read, or
/// live-runtime configuration surface.
pub struct SolanaGatewayBuilder {
    inner: ServerBuilder,
}

impl SolanaGatewayBuilder {
    fn new(target_id: String) -> Self {
        let mut inner = ServerBuilder::new();
        inner.config.http_health = Some(HttpHealthConfig::default());
        inner.config.runtime_plan = RuntimePlan::solana_gateway();
        inner.config.solana_gateway_target_id = Some(target_id);
        Self { inner }
    }

    /// Set the gateway HTTP bind address.
    pub fn bind(mut self, addr: impl Into<SocketAddr>) -> Self {
        self.inner.config.http_health = Some(HttpHealthConfig::new(addr));
        self
    }

    /// Set the auth plugin for chain and transaction requests.
    pub fn auth_plugin(mut self, plugin: Arc<dyn WebSocketAuthPlugin>) -> Self {
        self.inner.http_auth_plugin = Some(plugin);
        self
    }

    /// Configure HTTP body, header and read RPC limits.
    pub fn http_limits(mut self, config: HttpLimitsConfig) -> Self {
        self.inner.config.http_limits = Some(config);
        self
    }

    /// Configure the existing fixed `/transactions/v1/*` handlers.
    pub fn transactions_config(mut self, config: TransactionConfig) -> Self {
        self.inner.config.transactions = Some(config);
        self
    }

    /// Observe completed chain and transaction operations.
    pub fn owner_token_accounts_provider(
        mut self,
        provider: Arc<dyn crate::token_discovery::OwnerTokenAccountsProvider>,
    ) -> Self {
        self.inner.token_accounts_provider = Some(provider);
        self
    }

    pub fn usage_observer(mut self, observer: Arc<dyn SolanaGatewayUsageObserver>) -> Self {
        self.inner.solana_gateway_usage_observer = Some(observer);
        self
    }

    fn finalize(mut self) -> Result<ServerBuilder> {
        if self
            .inner
            .config
            .solana_gateway_target_id
            .as_deref()
            .is_none_or(|target_id| target_id.trim().is_empty())
        {
            anyhow::bail!("the Solana gateway target ID must not be empty");
        }
        if let Some(config) = self.inner.config.transactions.as_ref() {
            config.validate()?;
            if !config.enabled {
                anyhow::bail!("Solana gateway transaction configuration must be enabled");
            }
        }
        self.inner.config.runtime_plan = RuntimePlan::solana_gateway();
        Ok(self.inner)
    }

    /// Build the reusable runtime without starting it.
    pub fn build(self) -> Result<Runtime> {
        self.finalize()?.build()
    }

    /// Start the gateway and wait for shutdown.
    pub async fn start(self) -> Result<()> {
        self.finalize()?.start().await
    }
}

/// Builder for configuring and creating a Arete server
pub struct ServerBuilder {
    spec: Option<Spec>,
    views: Option<ViewIndex>,
    materialized_views: Option<MaterializedViewRegistry>,
    config: ServerConfig,
    websocket_auth_plugin: Option<Arc<dyn WebSocketAuthPlugin>>,
    http_auth_plugin: Option<Arc<dyn WebSocketAuthPlugin>>,
    websocket_usage_emitter: Option<Arc<dyn WebSocketUsageEmitter>>,
    token_accounts_provider: Option<Arc<dyn crate::token_discovery::OwnerTokenAccountsProvider>>,
    solana_gateway_usage_observer: Option<Arc<dyn SolanaGatewayUsageObserver>>,
    websocket_max_clients: Option<usize>,
    websocket_rate_limit_config: Option<crate::websocket::client_manager::RateLimitConfig>,
    websocket_admission_provider: Option<Arc<dyn WebSocketAdmissionProvider>>,
    #[cfg(feature = "otel")]
    metrics: Option<Arc<Metrics>>,
}

impl ServerBuilder {
    fn new() -> Self {
        Self {
            spec: None,
            views: None,
            materialized_views: None,
            config: ServerConfig::new(),
            websocket_auth_plugin: None,
            http_auth_plugin: None,
            websocket_usage_emitter: None,
            token_accounts_provider: None,
            solana_gateway_usage_observer: None,
            websocket_max_clients: None,
            websocket_rate_limit_config: None,
            websocket_admission_provider: None,
            #[cfg(feature = "otel")]
            metrics: None,
        }
    }

    /// Set the specification (bytecode, parsers, program_ids)
    pub fn spec(mut self, spec: Spec) -> Self {
        self.spec = Some(spec);
        self
    }

    /// Set custom view index
    pub fn views(mut self, views: ViewIndex) -> Self {
        self.views = Some(views);
        self
    }

    /// Enable metrics collection (requires 'otel' feature)
    #[cfg(feature = "otel")]
    pub fn metrics(mut self, metrics: Metrics) -> Self {
        self.metrics = Some(Arc::new(metrics));
        self
    }

    /// Enable WebSocket server with default configuration
    pub fn websocket(mut self) -> Self {
        self.config.websocket = Some(WebSocketConfig::default());
        self.config.runtime_plan.websocket = true;
        self.config.runtime_plan.live_runtime = true;
        self
    }

    /// Configure WebSocket server
    pub fn websocket_config(mut self, config: WebSocketConfig) -> Self {
        self.config.websocket = Some(config);
        self.config.runtime_plan.websocket = true;
        self.config.runtime_plan.live_runtime = true;
        self
    }

    /// Set a WebSocket auth plugin used to authorize inbound connections.
    pub fn websocket_auth_plugin(mut self, plugin: Arc<dyn WebSocketAuthPlugin>) -> Self {
        self.websocket_auth_plugin = Some(plugin);
        self
    }

    /// Set an HTTP auth plugin used to authorize inbound read requests.
    pub fn http_auth_plugin(mut self, plugin: Arc<dyn WebSocketAuthPlugin>) -> Self {
        self.http_auth_plugin = Some(plugin);
        self
    }

    /// Set an async usage emitter for billing-grade websocket usage events.
    pub fn owner_token_accounts_provider(
        mut self,
        provider: Arc<dyn crate::token_discovery::OwnerTokenAccountsProvider>,
    ) -> Self {
        self.token_accounts_provider = Some(provider);
        self
    }

    pub fn websocket_usage_emitter(mut self, emitter: Arc<dyn WebSocketUsageEmitter>) -> Self {
        self.websocket_usage_emitter = Some(emitter);
        self
    }

    /// Set the maximum number of concurrent WebSocket clients.
    pub fn websocket_max_clients(mut self, max_clients: usize) -> Self {
        self.websocket_max_clients = Some(max_clients);
        self
    }

    /// Configure rate limiting for WebSocket connections.
    ///
    /// This sets global rate limits such as maximum connections per IP,
    /// timeouts, and rate windows. Per-subject limits are controlled
    /// via AuthContext.Limits from the authentication token.
    pub fn websocket_rate_limit_config(
        mut self,
        config: crate::websocket::client_manager::RateLimitConfig,
    ) -> Self {
        self.websocket_rate_limit_config = Some(config);
        self
    }

    /// Install an embedder-owned provider for cross-runtime admission.
    pub fn websocket_admission_provider(
        mut self,
        provider: Arc<dyn WebSocketAdmissionProvider>,
    ) -> Self {
        self.websocket_admission_provider = Some(provider);
        self
    }

    /// Configure list-view buffering and latest-state collection delivery.
    pub fn websocket_delivery_config(mut self, config: WebSocketDeliveryConfig) -> Self {
        self.config.websocket_delivery = Some(config);
        self.config.runtime_plan.live_runtime = true;
        self
    }

    /// Configure WebSocket message, frame and handshake limits.
    pub fn websocket_limits(mut self, config: WebSocketLimitsConfig) -> Self {
        self.config.websocket_limits = Some(config);
        self
    }

    /// Configure HTTP body, header and read RPC limits.
    pub fn http_limits(mut self, config: HttpLimitsConfig) -> Self {
        self.config.http_limits = Some(config);
        self
    }

    /// Set the bind address for WebSocket server
    pub fn bind(mut self, addr: impl Into<SocketAddr>) -> Self {
        if let Some(ws_config) = &mut self.config.websocket {
            ws_config.bind_address = addr.into();
        } else {
            self.config.websocket = Some(WebSocketConfig::new(addr.into()));
        }
        self.config.runtime_plan.websocket = true;
        self.config.runtime_plan.live_runtime = true;
        self
    }

    /// Configure Yellowstone gRPC connection
    pub fn yellowstone(mut self, config: YellowstoneConfig) -> Self {
        self.config.yellowstone = Some(config);
        self.config.runtime_plan.live_runtime = true;
        self
    }

    /// Enable health monitoring with default configuration
    pub fn health_monitoring(mut self) -> Self {
        self.config.health = Some(HealthConfig::default());
        self.config.runtime_plan.health = true;
        self
    }

    /// Configure health monitoring
    pub fn health_config(mut self, config: HealthConfig) -> Self {
        self.config.health = Some(config);
        self.config.runtime_plan.health = true;
        self
    }

    /// Configure state snapshots (restart recovery). Without this call the
    /// server falls back to `SnapshotConfig::from_env()` (`ARETE_SNAPSHOT_*`).
    pub fn snapshots(mut self, config: crate::snapshot::SnapshotConfig) -> Self {
        self.config.snapshots = Some(config);
        self
    }

    /// Retain published events for replayable append subscriptions on this
    /// runtime, overriding `ARETE_JOURNAL_*` for this instance only.
    pub fn journal(mut self, config: crate::journal::JournalConfig) -> Self {
        self.config.journal = Some(config);
        self
    }

    /// Enable reconnection with default configuration
    pub fn reconnection(mut self) -> Self {
        self.config.reconnection = Some(ReconnectionConfig::default());
        self
    }

    /// Configure reconnection behavior
    pub fn reconnection_config(mut self, config: ReconnectionConfig) -> Self {
        self.config.reconnection = Some(config);
        self
    }

    /// Enable the HTTP server with default configuration (port 8081).
    ///
    /// This serves health endpoints plus stack-scoped HTTP reads.
    pub fn http(mut self) -> Self {
        self.config.http_health = Some(HttpHealthConfig::default());
        self.config.runtime_plan.health = true;
        self.config.runtime_plan.chain_reads = true;
        self.config.runtime_plan.program_reads = true;
        self.config.runtime_plan.stack_queries = true;
        self
    }

    /// Configure the HTTP server.
    pub fn http_config(mut self, config: crate::http_server::HttpServerConfig) -> Self {
        self.config.http_health = Some(config);
        self.config.runtime_plan.health = true;
        self.config.runtime_plan.chain_reads = true;
        self.config.runtime_plan.program_reads = true;
        self.config.runtime_plan.stack_queries = true;
        self
    }

    /// Configure and explicitly enable the fixed transaction HTTP routes.
    pub fn transactions_config(mut self, config: TransactionConfig) -> Self {
        self.config.runtime_plan.transactions = config.enabled;
        self.config.transactions = Some(config);
        self
    }

    /// Replace the inferred capability set with an explicit runtime plan.
    pub fn runtime_plan(mut self, plan: RuntimePlan) -> Self {
        self.config.runtime_plan = plan;
        self
    }

    /// Enable only health and release-pinned program reads over HTTP.
    pub fn program_reads(mut self) -> Self {
        if self.config.http_health.is_none() {
            self.config.http_health = Some(HttpHealthConfig::default());
        }
        self.config.runtime_plan.health = true;
        self.config.runtime_plan.program_reads = true;
        self
    }

    /// Enable program reads bound to one exact hosted program-read target.
    pub fn program_read_binding(mut self, target_id: impl Into<String>) -> Self {
        if self.config.http_health.is_none() {
            self.config.http_health = Some(HttpHealthConfig::default());
        }
        self.config.runtime_plan.health = true;
        self.config.runtime_plan.program_reads = true;
        self.config.program_read_binding_target_id = Some(target_id.into());
        self
    }

    pub fn chain_reads(mut self) -> Self {
        if self.config.http_health.is_none() {
            self.config.http_health = Some(HttpHealthConfig::default());
        }
        self.config.runtime_plan.chain_reads = true;
        self
    }

    pub fn stack_queries(mut self) -> Self {
        if self.config.http_health.is_none() {
            self.config.http_health = Some(HttpHealthConfig::default());
        }
        self.config.runtime_plan.stack_queries = true;
        self
    }

    pub fn live_runtime(mut self) -> Self {
        self.config.runtime_plan.live_runtime = true;
        self
    }

    /// Set the bind address for the HTTP server.
    pub fn http_bind(mut self, addr: impl Into<SocketAddr>) -> Self {
        if let Some(http_config) = &mut self.config.http_health {
            http_config.bind_address = addr.into();
        } else {
            self.config.http_health = Some(HttpHealthConfig::new(addr.into()));
        }
        self.config.runtime_plan.health = true;
        self.config.runtime_plan.chain_reads = true;
        self.config.runtime_plan.program_reads = true;
        self.config.runtime_plan.stack_queries = true;
        self
    }

    /// Enable HTTP health server with default configuration (port 8081)
    pub fn http_health(self) -> Self {
        self.http()
    }

    /// Configure HTTP health server
    pub fn http_health_config(self, config: HttpHealthConfig) -> Self {
        self.http_config(config)
    }

    /// Set the bind address for HTTP health server
    pub fn health_bind(self, addr: impl Into<SocketAddr>) -> Self {
        self.http_bind(addr)
    }

    /// Build, start and block until shutdown. See [`Runtime::run`].
    pub async fn start(self) -> Result<()> {
        self.build()?.run().await
    }

    /// Build the [`Runtime`] without starting it.
    ///
    /// For callers that embed the server: `build` then [`Runtime::spawn`]
    /// gives a [`runtime::RuntimeHandle`] that serves connections the caller
    /// accepted and can be shut down on demand.
    pub fn build(self) -> Result<Runtime> {
        let (view_index, materialized_registry) =
            Self::build_view_index_and_registry(self.views, self.materialized_views, &self.spec);

        view_index.validate_scope_projections()?;
        #[cfg(feature = "otel")]
        let mut runtime = Runtime::new(self.config, view_index, self.metrics);
        #[cfg(not(feature = "otel"))]
        let mut runtime = Runtime::new(self.config, view_index);

        if let Some(plugin) = self.websocket_auth_plugin {
            runtime = runtime.with_websocket_auth_plugin(plugin);
        }

        if let Some(plugin) = self.http_auth_plugin {
            runtime = runtime.with_http_auth_plugin(plugin);
        }

        if let Some(emitter) = self.websocket_usage_emitter {
            runtime = runtime.with_websocket_usage_emitter(emitter);
        }

        if let Some(provider) = self.token_accounts_provider {
            runtime = runtime.with_owner_token_accounts_provider(provider);
        }

        if let Some(observer) = self.solana_gateway_usage_observer {
            runtime = runtime.with_solana_gateway_usage_observer(observer);
        }

        if let Some(max_clients) = self.websocket_max_clients {
            runtime = runtime.with_websocket_max_clients(max_clients);
        }

        if let Some(rate_limit_config) = self.websocket_rate_limit_config {
            runtime = runtime.with_websocket_rate_limit_config(rate_limit_config);
        }

        if let Some(provider) = self.websocket_admission_provider {
            runtime = runtime.with_websocket_admission_provider(provider);
        }

        if let Some(registry) = materialized_registry {
            runtime = runtime.with_materialized_views(registry);
        }

        if let Some(spec) = self.spec {
            runtime = runtime.with_spec(spec)?;
        }

        Ok(runtime)
    }

    fn build_view_index_and_registry(
        views: Option<ViewIndex>,
        materialized_views: Option<MaterializedViewRegistry>,
        spec: &Option<Spec>,
    ) -> (ViewIndex, Option<MaterializedViewRegistry>) {
        let mut index = views.unwrap_or_default();
        let mut registry = materialized_views;

        if let Some(ref spec) = spec {
            let entity_wire_formats = spec
                .entity_specs
                .iter()
                .map(|entity_spec| {
                    (
                        entity_spec.state_name.clone(),
                        ViewSpec::wire_format_from_entity_spec(entity_spec),
                    )
                })
                .collect::<std::collections::HashMap<_, _>>();

            // Row scoping comes with the entity spec, so a scoped entity is
            // never served without it.
            for entity_spec in &spec.entity_specs {
                if let Some(path) = entity_spec.scope_field.as_deref() {
                    index.set_entity_scope(
                        entity_spec.state_name.clone(),
                        row_scope::ScopeField::new(path),
                    );
                }
            }

            for entity_name in spec.bytecode.entities.keys() {
                let wire_format = entity_wire_formats
                    .get(entity_name)
                    .cloned()
                    .unwrap_or_default();
                index.add_spec(ViewSpec {
                    id: format!("{}/list", entity_name),
                    export: entity_name.clone(),
                    mode: Mode::List,
                    wire_format: wire_format.clone(),
                    projection: Projection::all(),
                    filters: Filters::all(),
                    delivery: Delivery::default(),
                    pipeline: None,
                    source_view: None,
                });

                index.add_spec(ViewSpec {
                    id: format!("{}/state", entity_name),
                    export: entity_name.clone(),
                    mode: Mode::State,
                    wire_format: wire_format.clone(),
                    projection: Projection::all(),
                    filters: Filters::all(),
                    delivery: Delivery::default(),
                    pipeline: None,
                    source_view: None,
                });

                index.add_spec(ViewSpec {
                    id: format!("{}/append", entity_name),
                    export: entity_name.clone(),
                    mode: Mode::Append,
                    wire_format,
                    projection: Projection::all(),
                    filters: Filters::all(),
                    delivery: Delivery::default(),
                    pipeline: None,
                    source_view: None,
                });
            }

            if !spec.views.is_empty() {
                let reg = registry.get_or_insert_with(MaterializedViewRegistry::new);

                for view_def in &spec.views {
                    let export = match &view_def.source {
                        arete_interpreter::ast::ViewSource::Entity { name } => name.clone(),
                        arete_interpreter::ast::ViewSource::View { id } => {
                            id.split('/').next().unwrap_or(id).to_string()
                        }
                    };

                    // The compiler includes the canonical entity list/state views in
                    // `spec.views`. They are already registered above as native views
                    // backed directly by EntityCache. Registering them again through
                    // `from_view_def` marks them as derived and replaces the native
                    // by-id entry, so subscribers read an unpopulated derived cache
                    // while the projector continues writing to the native cache.
                    if Self::is_canonical_entity_view(view_def, &export) {
                        tracing::debug!(
                            view_id = %view_def.id,
                            "Keeping canonical entity view backed by the native cache"
                        );
                        continue;
                    }

                    let wire_format = entity_wire_formats
                        .get(&export)
                        .cloned()
                        .unwrap_or_default();
                    let view_spec = ViewSpec::from_view_def(view_def, &export, wire_format);
                    let pipeline = view_spec.pipeline.clone().unwrap_or_default();
                    let source_id = view_spec.source_view.clone().unwrap_or_default();
                    tracing::debug!(
                        view_id = %view_def.id,
                        source = %source_id,
                        "Registering derived view"
                    );

                    index.add_spec(view_spec);

                    let materialized =
                        MaterializedView::new(view_def.id.clone(), source_id, pipeline);
                    reg.register(materialized);
                }
            }
        }

        (index, registry)
    }

    fn is_canonical_entity_view(view_def: &ViewDef, export: &str) -> bool {
        use arete_interpreter::ast::{ViewOutput, ViewSource};

        let ViewSource::Entity { name } = &view_def.source else {
            return false;
        };
        if name != export || !view_def.pipeline.is_empty() {
            return false;
        }

        match &view_def.output {
            ViewOutput::Collection => view_def.id == format!("{export}/list"),
            ViewOutput::Single | ViewOutput::Keyed { .. } => {
                view_def.id == format!("{export}/state")
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_builder_pattern() {
        let _builder = Server::builder()
            .websocket()
            .bind("[::]:8877".parse::<SocketAddr>().unwrap());
    }

    #[test]
    fn test_spec_creation() {
        let bytecode = arete_interpreter::compiler::MultiEntityBytecode::new().build();
        let spec = Spec::new(bytecode, "test_program");
        assert_eq!(
            spec.program_ids.first().map(String::as_str),
            Some("test_program")
        );
    }

    #[test]
    fn http_without_websocket_has_a_read_only_runtime_plan() {
        let builder = Server::builder().http();
        assert!(builder.config.runtime_plan.program_reads);
        assert!(!builder.config.runtime_plan.live_runtime_enabled());
    }

    #[test]
    fn program_read_binding_configures_the_exact_auth_target() {
        let builder = Server::builder().program_read_binding("binding-1");

        assert!(builder.config.runtime_plan.program_reads);
        assert_eq!(
            builder.config.program_read_binding_target_id.as_deref(),
            Some("binding-1")
        );
    }

    #[test]
    fn websocket_and_http_preserve_all_in_one_runtime_behavior() {
        let builder = Server::builder().websocket().http();
        assert!(builder.config.runtime_plan.websocket);
        assert!(builder.config.runtime_plan.live_runtime_enabled());
    }

    #[test]
    fn explicit_hosted_plan_disables_program_reads_after_http_helpers() {
        let plan = RuntimePlan {
            health: true,
            chain_reads: true,
            program_reads: false,
            stack_queries: true,
            transactions: true,
            websocket: true,
            live_runtime: true,
        };
        let builder = Server::builder()
            .websocket()
            .http_health()
            .health_bind("[::]:8081".parse::<SocketAddr>().unwrap())
            .runtime_plan(plan);

        assert_eq!(builder.config.runtime_plan, plan);
    }

    #[test]
    fn solana_gateway_builder_excludes_stack_and_live_capabilities() {
        let builder = Server::solana_gateway("gateway-us-east-1");

        assert_eq!(
            builder.inner.config.runtime_plan,
            RuntimePlan::solana_gateway()
        );
        assert!(builder.inner.config.http_health.is_some());
        assert_eq!(
            builder.inner.config.solana_gateway_target_id.as_deref(),
            Some("gateway-us-east-1")
        );
        assert!(builder.inner.config.websocket.is_none());
        assert!(builder.inner.config.yellowstone.is_none());
        assert!(builder.inner.spec.is_none());
        assert!(builder.inner.views.is_none());
        assert!(builder.inner.materialized_views.is_none());
        assert!(builder.inner.solana_gateway_usage_observer.is_none());
        assert!(!builder.inner.config.runtime_plan.websocket);
        assert!(!builder.inner.config.runtime_plan.live_runtime_enabled());
        assert!(!builder.inner.config.runtime_plan.stack_queries);
        assert!(!builder.inner.config.runtime_plan.program_reads);
    }

    #[test]
    fn solana_gateway_builder_rejects_invalid_gateway_configuration() {
        assert!(Server::solana_gateway("").build().is_err());
        assert!(Server::solana_gateway("gateway-us-east-1")
            .transactions_config(TransactionConfig::default())
            .build()
            .is_err());
    }

    #[test]
    fn builder_rejects_mismatched_program_release_definitions() {
        let bytecode = arete_interpreter::compiler::MultiEntityBytecode::new().build();
        let definition = ProgramRuntimeDefinition {
            program_id: "Program111".to_string(),
            program_spec_hash: ProgramSpecHash::from_digest([1; 32]),
            idl_content_hash: IdlContentHash::from_digest([2; 32]),
            normalized_idl_hash: NormalizedIdlHash::from_digest([3; 32]),
            program_release_hash: ProgramReleaseHash::from_digest([4; 32]),
            account_reader: Arc::new(|_, _| Ok(serde_json::Value::Null)),
        };
        let spec =
            Spec::new(bytecode, "Program111").with_program_runtime_definitions(vec![definition]);

        assert!(Server::builder().spec(spec).build().is_err());
    }

    #[test]
    fn canonical_entity_views_are_not_reclassified_as_derived_views() {
        use arete_interpreter::ast::{IdentitySpec, TypedStreamSpec, ViewDef, ViewSource};

        let list = ViewDef::list("OreBoard");
        let state = ViewDef::state("OreBoard", &["id", "address"]);
        assert!(ServerBuilder::is_canonical_entity_view(&list, "OreBoard"));
        assert!(ServerBuilder::is_canonical_entity_view(&state, "OreBoard"));

        let mut named_view = ViewDef::list("OreBoard");
        named_view.id = "OreBoard/all".to_string();
        assert!(!ServerBuilder::is_canonical_entity_view(
            &named_view,
            "OreBoard"
        ));

        let mut chained_view = ViewDef::list("OreBoard");
        chained_view.source = ViewSource::View {
            id: "OreBoard/list".to_string(),
        };
        assert!(!ServerBuilder::is_canonical_entity_view(
            &chained_view,
            "OreBoard"
        ));

        let entity_spec = TypedStreamSpec::<serde_json::Value>::new(
            "OreBoard".to_string(),
            IdentitySpec {
                primary_keys: vec!["id.address".to_string()],
                lookup_indexes: Vec::new(),
            },
            Vec::new(),
        );
        let bytecode = arete_interpreter::compiler::MultiEntityBytecode::new()
            .add_entity("OreBoard".to_string(), entity_spec, 1)
            .build();
        let spec = Some(Spec::new(bytecode, "Program111").with_views(vec![list, state]));

        let (index, _) = ServerBuilder::build_view_index_and_registry(None, None, &spec);
        assert!(!index
            .get_view("OreBoard/list")
            .expect("list view should exist")
            .is_derived());
        assert!(!index
            .get_view("OreBoard/state")
            .expect("state view should exist")
            .is_derived());
        assert_eq!(index.by_export("OreBoard").len(), 3);
    }

    #[test]
    fn an_entity_specs_scope_field_scopes_every_view_of_the_entity() {
        use arete_interpreter::ast::{
            FieldPath, IdentitySpec, SortOrder, TypedStreamSpec, ViewDef, ViewOutput, ViewSource,
            ViewTransform,
        };

        let identity = IdentitySpec {
            primary_keys: vec!["id.address".to_string()],
            lookup_indexes: Vec::new(),
        };
        let scoped = TypedStreamSpec::<serde_json::Value>::new(
            "Round".to_string(),
            identity.clone(),
            Vec::new(),
        );
        let plain =
            TypedStreamSpec::<serde_json::Value>::new("Board".to_string(), identity, Vec::new());
        let mut scoped_spec = scoped.to_serializable();
        scoped_spec.scope_field = Some("meta.scope".to_string());
        let plain_spec = plain.to_serializable();
        let bytecode = arete_interpreter::compiler::MultiEntityBytecode::new()
            .add_entity("Round".to_string(), scoped, 1)
            .add_entity("Board".to_string(), plain, 1)
            .build();
        // A derived view whose id does not name its entity.
        let top = ViewDef {
            id: "Leaders/top".to_string(),
            source: ViewSource::Entity {
                name: "Round".to_string(),
            },
            pipeline: vec![ViewTransform::Sort {
                key: FieldPath::new(&["amount"]),
                order: SortOrder::Desc,
            }],
            output: ViewOutput::Collection,
        };
        let spec = Some(
            Spec::new(bytecode, "Program111")
                .with_entity_specs(vec![scoped_spec, plain_spec])
                .with_views(vec![top]),
        );

        let (index, _) = ServerBuilder::build_view_index_and_registry(None, None, &spec);
        for view in ["Round/list", "Round/state", "Round/append", "Leaders/top"] {
            assert_eq!(
                index
                    .scope_field_for_view(view)
                    .map(|field| field.path().to_vec()),
                Some(vec!["meta".to_string(), "scope".to_string()]),
                "{view}"
            );
        }
        assert!(index.scope_field_for_view("Board/list").is_none());
        // A custom projection cannot strip the scope, including when its
        // field is nested. Reject it at build time, before serving any data.
        let mut custom = index.get_view("Round/list").unwrap().clone();
        custom.id = "Round/custom".into();
        custom.projection.fields = Some(vec!["amount".into()]);
        let mut invalid = ViewIndex::new();
        invalid.set_entity_scope("Round", row_scope::ScopeField::new("meta.scope"));
        invalid.add_spec(custom.clone());
        assert!(ServerBuilder::new().views(invalid).build().is_err());
        custom.projection.fields = Some(vec!["amount".into(), "meta".into()]);
        let mut valid = ViewIndex::new();
        valid.set_entity_scope("Round", row_scope::ScopeField::new("meta.scope"));
        valid.add_spec(custom);
        assert!(ServerBuilder::new().views(valid).build().is_ok());
        let caches = index.sorted_caches();
        let caches = caches.try_read().unwrap();
        assert!(caches.get("Leaders/top").unwrap().is_scope_partitioned());
    }
}