arete_server/
lib.rs

1//! # arete-server
2//!
3//! WebSocket server and projection handlers for Arete streaming pipelines.
4//!
5//! This crate provides a builder API for creating Arete servers that:
6//!
7//! - Process Solana blockchain data via Yellowstone gRPC
8//! - Transform data using the Arete VM
9//! - Stream entity updates over WebSockets to connected clients
10//! - Support multiple streaming modes (State, List, Append)
11//!
12//! ## Quick Start
13//!
14//! ```rust,ignore
15//! use arete_server::{Server, Spec};
16//!
17//! #[tokio::main]
18//! async fn main() -> anyhow::Result<()> {
19//!     Server::builder()
20//!         .spec(my_spec())
21//!         .websocket()
22//!         .bind("[::]:8877".parse()?)
23//!         .health_monitoring()
24//!         .start()
25//!         .await
26//! }
27//! ```
28//!
29//! ## Feature Flags
30//!
31//! - `otel` - OpenTelemetry integration for metrics and distributed tracing
32
33pub mod account_policy;
34pub mod bus;
35pub mod cache;
36pub mod compression;
37pub mod config;
38pub mod health;
39mod http;
40pub mod http_health;
41pub mod http_server;
42pub mod materialized_view;
43#[cfg(feature = "otel")]
44pub mod metrics;
45pub mod mutation_batch;
46pub mod program_runtime;
47pub mod projector;
48pub mod runtime;
49pub mod snapshot;
50pub mod sorted_cache;
51pub mod telemetry;
52pub mod view;
53pub mod websocket;
54
55pub use arete_auth::{
56    AsyncVerifier, KeyLoader, Limits, SolanaGatewayAuthorization, SolanaGatewayAuthorizationError,
57    SolanaGatewayScope, TargetKind, TokenVerifier, VerifyingKey, SCOPE_READ,
58    SCOPE_TRANSACTION_INSPECT, SCOPE_TRANSACTION_SEND, SOLANA_GATEWAY_AUDIENCE,
59};
60pub use bus::{BusManager, BusMessage};
61pub use cache::{EntityCache, EntityCacheConfig};
62pub use config::{
63    HealthConfig, HttpHealthConfig, HttpServerConfig, ReconnectionConfig, RuntimePlan,
64    ServerConfig, TransactionConfig, WebSocketConfig, YellowstoneConfig,
65};
66pub use health::{HealthMonitor, SlotTracker, StreamStatus};
67pub use http_health::HttpHealthServer;
68pub use http_server::HttpServer;
69pub use materialized_view::{MaterializedView, MaterializedViewRegistry, ViewEffect};
70#[cfg(feature = "otel")]
71pub use metrics::Metrics;
72pub use mutation_batch::{EventContext, MutationBatch, SlotContext};
73pub use program_runtime::{
74    IdlContentHash, NormalizedIdlHash, ProgramAccountReaderFn, ProgramReleaseHash,
75    ProgramRuntimeCatalog, ProgramRuntimeDefinition, ProgramSpecHash,
76};
77pub use projector::Projector;
78pub use runtime::Runtime;
79pub use snapshot::{SnapshotConfig, SnapshotService};
80pub use telemetry::{init as init_telemetry, TelemetryConfig};
81#[cfg(feature = "otel")]
82pub use telemetry::{init_with_otel, TelemetryGuard};
83pub use view::{Delivery, Filters, Projection, ViewIndex, ViewSpec};
84pub use websocket::{
85    AllowAllAuthPlugin, AuthContext, AuthDecision, AuthDeny, AuthErrorDetails, ChannelUsageEmitter,
86    ClientInfo, ClientManager, ConnectionAuthRequest, ErrorResponse, Frame, HttpUsageEmitter, Mode,
87    RateLimitConfig, RateLimitResult, RateLimiterConfig, RefreshAuthRequest, RefreshAuthResponse,
88    RetryPolicy, SignedSessionAuthPlugin, SnapshotOptions, SocketIssueMessage,
89    StaticTokenAuthPlugin, Subscription, SubscriptionQuery, WebSocketAuthPlugin,
90    WebSocketRateLimiter, WebSocketServer, WebSocketUsageBatch, WebSocketUsageEmitter,
91    WebSocketUsageEnvelope, WebSocketUsageEvent,
92};
93
94use anyhow::Result;
95use arete_interpreter::ast::ViewDef;
96use std::net::SocketAddr;
97use std::sync::Arc;
98
99/// Type alias for a parser setup function.
100pub type ParserSetupFn = Arc<
101    dyn Fn(
102            tokio::sync::mpsc::Sender<MutationBatch>,
103            Option<HealthMonitor>,
104            ReconnectionConfig,
105        ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<()>> + Send>>
106        + Send
107        + Sync,
108>;
109
110/// Specification for a Arete server
111/// Contains bytecode, parsers, and program information
112pub struct Spec {
113    pub bytecode: arete_interpreter::compiler::MultiEntityBytecode,
114    pub program_ids: Vec<String>,
115    pub parser_setup: Option<ParserSetupFn>,
116    pub program_runtime_definitions: Vec<ProgramRuntimeDefinition>,
117    pub entity_specs: Vec<arete_interpreter::ast::SerializableStreamSpec>,
118    pub views: Vec<ViewDef>,
119}
120
121impl Spec {
122    pub fn new(
123        bytecode: arete_interpreter::compiler::MultiEntityBytecode,
124        program_id: impl Into<String>,
125    ) -> Self {
126        Self {
127            bytecode,
128            program_ids: vec![program_id.into()],
129            parser_setup: None,
130            program_runtime_definitions: Vec::new(),
131            entity_specs: Vec::new(),
132            views: Vec::new(),
133        }
134    }
135
136    pub fn with_parser_setup(mut self, setup_fn: ParserSetupFn) -> Self {
137        self.parser_setup = Some(setup_fn);
138        self
139    }
140
141    pub fn with_program_runtime_definitions(
142        mut self,
143        definitions: Vec<ProgramRuntimeDefinition>,
144    ) -> Self {
145        for definition in &definitions {
146            if !self.program_ids.contains(&definition.program_id) {
147                self.program_ids.push(definition.program_id.clone());
148            }
149        }
150        self.program_runtime_definitions = definitions;
151        self
152    }
153
154    pub fn with_entity_specs(
155        mut self,
156        entity_specs: Vec<arete_interpreter::ast::SerializableStreamSpec>,
157    ) -> Self {
158        self.entity_specs = entity_specs;
159        self
160    }
161
162    pub fn with_views(mut self, views: Vec<ViewDef>) -> Self {
163        self.views = views;
164        self
165    }
166}
167
168/// Main server interface with fluent builder API
169pub struct Server;
170
171impl Server {
172    /// Create a new server builder
173    pub fn builder() -> ServerBuilder {
174        ServerBuilder::new()
175    }
176
177    /// Build a standalone HTTP gateway without stack or live-stream capabilities.
178    pub fn solana_gateway(target_id: impl Into<String>) -> SolanaGatewayBuilder {
179        SolanaGatewayBuilder::new(target_id.into())
180    }
181}
182
183/// Constrained builder for health, chain reads, and fixed transaction routes.
184///
185/// The final runtime plan is fixed when `build` or `start` is called. This
186/// builder intentionally has no Spec, WebSocket, stack-query, program-read, or
187/// live-runtime configuration surface.
188pub struct SolanaGatewayBuilder {
189    inner: ServerBuilder,
190}
191
192impl SolanaGatewayBuilder {
193    fn new(target_id: String) -> Self {
194        let mut inner = ServerBuilder::new();
195        inner.config.http_health = Some(HttpHealthConfig::default());
196        inner.config.runtime_plan = RuntimePlan::solana_gateway();
197        inner.config.solana_gateway_target_id = Some(target_id);
198        Self { inner }
199    }
200
201    /// Set the gateway HTTP bind address.
202    pub fn bind(mut self, addr: impl Into<SocketAddr>) -> Self {
203        self.inner.config.http_health = Some(HttpHealthConfig::new(addr));
204        self
205    }
206
207    /// Set the auth plugin for chain and transaction requests.
208    pub fn auth_plugin(mut self, plugin: Arc<dyn WebSocketAuthPlugin>) -> Self {
209        self.inner.http_auth_plugin = Some(plugin);
210        self
211    }
212
213    /// Configure the existing fixed `/transactions/v1/*` handlers.
214    pub fn transactions_config(mut self, config: TransactionConfig) -> Self {
215        self.inner.config.transactions = Some(config);
216        self
217    }
218
219    fn finalize(mut self) -> Result<ServerBuilder> {
220        if self
221            .inner
222            .config
223            .solana_gateway_target_id
224            .as_deref()
225            .is_none_or(|target_id| target_id.trim().is_empty())
226        {
227            anyhow::bail!("the Solana gateway target ID must not be empty");
228        }
229        if let Some(config) = self.inner.config.transactions.as_ref() {
230            config.validate()?;
231            if !config.enabled {
232                anyhow::bail!("Solana gateway transaction configuration must be enabled");
233            }
234        }
235        self.inner.config.runtime_plan = RuntimePlan::solana_gateway();
236        Ok(self.inner)
237    }
238
239    /// Build the reusable runtime without starting it.
240    pub fn build(self) -> Result<Runtime> {
241        self.finalize()?.build()
242    }
243
244    /// Start the gateway and wait for shutdown.
245    pub async fn start(self) -> Result<()> {
246        self.finalize()?.start().await
247    }
248}
249
250/// Builder for configuring and creating a Arete server
251pub struct ServerBuilder {
252    spec: Option<Spec>,
253    views: Option<ViewIndex>,
254    materialized_views: Option<MaterializedViewRegistry>,
255    config: ServerConfig,
256    websocket_auth_plugin: Option<Arc<dyn WebSocketAuthPlugin>>,
257    http_auth_plugin: Option<Arc<dyn WebSocketAuthPlugin>>,
258    websocket_usage_emitter: Option<Arc<dyn WebSocketUsageEmitter>>,
259    websocket_max_clients: Option<usize>,
260    websocket_rate_limit_config: Option<crate::websocket::client_manager::RateLimitConfig>,
261    #[cfg(feature = "otel")]
262    metrics: Option<Arc<Metrics>>,
263}
264
265impl ServerBuilder {
266    fn new() -> Self {
267        Self {
268            spec: None,
269            views: None,
270            materialized_views: None,
271            config: ServerConfig::new(),
272            websocket_auth_plugin: None,
273            http_auth_plugin: None,
274            websocket_usage_emitter: None,
275            websocket_max_clients: None,
276            websocket_rate_limit_config: None,
277            #[cfg(feature = "otel")]
278            metrics: None,
279        }
280    }
281
282    /// Set the specification (bytecode, parsers, program_ids)
283    pub fn spec(mut self, spec: Spec) -> Self {
284        self.spec = Some(spec);
285        self
286    }
287
288    /// Set custom view index
289    pub fn views(mut self, views: ViewIndex) -> Self {
290        self.views = Some(views);
291        self
292    }
293
294    /// Enable metrics collection (requires 'otel' feature)
295    #[cfg(feature = "otel")]
296    pub fn metrics(mut self, metrics: Metrics) -> Self {
297        self.metrics = Some(Arc::new(metrics));
298        self
299    }
300
301    /// Enable WebSocket server with default configuration
302    pub fn websocket(mut self) -> Self {
303        self.config.websocket = Some(WebSocketConfig::default());
304        self.config.runtime_plan.websocket = true;
305        self.config.runtime_plan.live_runtime = true;
306        self
307    }
308
309    /// Configure WebSocket server
310    pub fn websocket_config(mut self, config: WebSocketConfig) -> Self {
311        self.config.websocket = Some(config);
312        self.config.runtime_plan.websocket = true;
313        self.config.runtime_plan.live_runtime = true;
314        self
315    }
316
317    /// Set a WebSocket auth plugin used to authorize inbound connections.
318    pub fn websocket_auth_plugin(mut self, plugin: Arc<dyn WebSocketAuthPlugin>) -> Self {
319        self.websocket_auth_plugin = Some(plugin);
320        self
321    }
322
323    /// Set an HTTP auth plugin used to authorize inbound read requests.
324    pub fn http_auth_plugin(mut self, plugin: Arc<dyn WebSocketAuthPlugin>) -> Self {
325        self.http_auth_plugin = Some(plugin);
326        self
327    }
328
329    /// Set an async usage emitter for billing-grade websocket usage events.
330    pub fn websocket_usage_emitter(mut self, emitter: Arc<dyn WebSocketUsageEmitter>) -> Self {
331        self.websocket_usage_emitter = Some(emitter);
332        self
333    }
334
335    /// Set the maximum number of concurrent WebSocket clients.
336    pub fn websocket_max_clients(mut self, max_clients: usize) -> Self {
337        self.websocket_max_clients = Some(max_clients);
338        self
339    }
340
341    /// Configure rate limiting for WebSocket connections.
342    ///
343    /// This sets global rate limits such as maximum connections per IP,
344    /// timeouts, and rate windows. Per-subject limits are controlled
345    /// via AuthContext.Limits from the authentication token.
346    pub fn websocket_rate_limit_config(
347        mut self,
348        config: crate::websocket::client_manager::RateLimitConfig,
349    ) -> Self {
350        self.websocket_rate_limit_config = Some(config);
351        self
352    }
353
354    /// Set the bind address for WebSocket server
355    pub fn bind(mut self, addr: impl Into<SocketAddr>) -> Self {
356        if let Some(ws_config) = &mut self.config.websocket {
357            ws_config.bind_address = addr.into();
358        } else {
359            self.config.websocket = Some(WebSocketConfig::new(addr.into()));
360        }
361        self.config.runtime_plan.websocket = true;
362        self.config.runtime_plan.live_runtime = true;
363        self
364    }
365
366    /// Configure Yellowstone gRPC connection
367    pub fn yellowstone(mut self, config: YellowstoneConfig) -> Self {
368        self.config.yellowstone = Some(config);
369        self.config.runtime_plan.live_runtime = true;
370        self
371    }
372
373    /// Enable health monitoring with default configuration
374    pub fn health_monitoring(mut self) -> Self {
375        self.config.health = Some(HealthConfig::default());
376        self.config.runtime_plan.health = true;
377        self
378    }
379
380    /// Configure health monitoring
381    pub fn health_config(mut self, config: HealthConfig) -> Self {
382        self.config.health = Some(config);
383        self.config.runtime_plan.health = true;
384        self
385    }
386
387    /// Configure state snapshots (restart recovery). Without this call the
388    /// server falls back to `SnapshotConfig::from_env()` (`ARETE_SNAPSHOT_*`).
389    pub fn snapshots(mut self, config: crate::snapshot::SnapshotConfig) -> Self {
390        self.config.snapshots = Some(config);
391        self
392    }
393
394    /// Enable reconnection with default configuration
395    pub fn reconnection(mut self) -> Self {
396        self.config.reconnection = Some(ReconnectionConfig::default());
397        self
398    }
399
400    /// Configure reconnection behavior
401    pub fn reconnection_config(mut self, config: ReconnectionConfig) -> Self {
402        self.config.reconnection = Some(config);
403        self
404    }
405
406    /// Enable the HTTP server with default configuration (port 8081).
407    ///
408    /// This serves health endpoints plus stack-scoped HTTP reads.
409    pub fn http(mut self) -> Self {
410        self.config.http_health = Some(HttpHealthConfig::default());
411        self.config.runtime_plan.health = true;
412        self.config.runtime_plan.chain_reads = true;
413        self.config.runtime_plan.program_reads = true;
414        self.config.runtime_plan.stack_queries = true;
415        self
416    }
417
418    /// Configure the HTTP server.
419    pub fn http_config(mut self, config: crate::http_server::HttpServerConfig) -> Self {
420        self.config.http_health = Some(config);
421        self.config.runtime_plan.health = true;
422        self.config.runtime_plan.chain_reads = true;
423        self.config.runtime_plan.program_reads = true;
424        self.config.runtime_plan.stack_queries = true;
425        self
426    }
427
428    /// Configure and explicitly enable the fixed transaction HTTP routes.
429    pub fn transactions_config(mut self, config: TransactionConfig) -> Self {
430        self.config.runtime_plan.transactions = config.enabled;
431        self.config.transactions = Some(config);
432        self
433    }
434
435    /// Replace the inferred capability set with an explicit runtime plan.
436    pub fn runtime_plan(mut self, plan: RuntimePlan) -> Self {
437        self.config.runtime_plan = plan;
438        self
439    }
440
441    /// Enable only health and release-pinned program reads over HTTP.
442    pub fn program_reads(mut self) -> Self {
443        if self.config.http_health.is_none() {
444            self.config.http_health = Some(HttpHealthConfig::default());
445        }
446        self.config.runtime_plan.health = true;
447        self.config.runtime_plan.program_reads = true;
448        self
449    }
450
451    /// Enable program reads bound to one exact hosted program-read target.
452    pub fn program_read_binding(mut self, target_id: impl Into<String>) -> Self {
453        if self.config.http_health.is_none() {
454            self.config.http_health = Some(HttpHealthConfig::default());
455        }
456        self.config.runtime_plan.health = true;
457        self.config.runtime_plan.program_reads = true;
458        self.config.program_read_binding_target_id = Some(target_id.into());
459        self
460    }
461
462    pub fn chain_reads(mut self) -> Self {
463        if self.config.http_health.is_none() {
464            self.config.http_health = Some(HttpHealthConfig::default());
465        }
466        self.config.runtime_plan.chain_reads = true;
467        self
468    }
469
470    pub fn stack_queries(mut self) -> Self {
471        if self.config.http_health.is_none() {
472            self.config.http_health = Some(HttpHealthConfig::default());
473        }
474        self.config.runtime_plan.stack_queries = true;
475        self
476    }
477
478    pub fn live_runtime(mut self) -> Self {
479        self.config.runtime_plan.live_runtime = true;
480        self
481    }
482
483    /// Set the bind address for the HTTP server.
484    pub fn http_bind(mut self, addr: impl Into<SocketAddr>) -> Self {
485        if let Some(http_config) = &mut self.config.http_health {
486            http_config.bind_address = addr.into();
487        } else {
488            self.config.http_health = Some(HttpHealthConfig::new(addr.into()));
489        }
490        self.config.runtime_plan.health = true;
491        self.config.runtime_plan.chain_reads = true;
492        self.config.runtime_plan.program_reads = true;
493        self.config.runtime_plan.stack_queries = true;
494        self
495    }
496
497    /// Enable HTTP health server with default configuration (port 8081)
498    pub fn http_health(self) -> Self {
499        self.http()
500    }
501
502    /// Configure HTTP health server
503    pub fn http_health_config(self, config: HttpHealthConfig) -> Self {
504        self.http_config(config)
505    }
506
507    /// Set the bind address for HTTP health server
508    pub fn health_bind(self, addr: impl Into<SocketAddr>) -> Self {
509        self.http_bind(addr)
510    }
511
512    pub async fn start(self) -> Result<()> {
513        let (view_index, materialized_registry) =
514            Self::build_view_index_and_registry(self.views, self.materialized_views, &self.spec);
515
516        #[cfg(feature = "otel")]
517        let mut runtime = Runtime::new(self.config, view_index, self.metrics);
518        #[cfg(not(feature = "otel"))]
519        let mut runtime = Runtime::new(self.config, view_index);
520
521        if let Some(plugin) = self.websocket_auth_plugin {
522            runtime = runtime.with_websocket_auth_plugin(plugin);
523        }
524
525        if let Some(plugin) = self.http_auth_plugin {
526            runtime = runtime.with_http_auth_plugin(plugin);
527        }
528
529        if let Some(emitter) = self.websocket_usage_emitter {
530            runtime = runtime.with_websocket_usage_emitter(emitter);
531        }
532
533        if let Some(max_clients) = self.websocket_max_clients {
534            runtime = runtime.with_websocket_max_clients(max_clients);
535        }
536
537        if let Some(rate_limit_config) = self.websocket_rate_limit_config {
538            runtime = runtime.with_websocket_rate_limit_config(rate_limit_config);
539        }
540
541        if let Some(registry) = materialized_registry {
542            runtime = runtime.with_materialized_views(registry);
543        }
544
545        if let Some(spec) = self.spec {
546            runtime = runtime.with_spec(spec)?;
547        }
548
549        runtime.run().await
550    }
551
552    fn build_view_index_and_registry(
553        views: Option<ViewIndex>,
554        materialized_views: Option<MaterializedViewRegistry>,
555        spec: &Option<Spec>,
556    ) -> (ViewIndex, Option<MaterializedViewRegistry>) {
557        let mut index = views.unwrap_or_default();
558        let mut registry = materialized_views;
559
560        if let Some(ref spec) = spec {
561            let entity_wire_formats = spec
562                .entity_specs
563                .iter()
564                .map(|entity_spec| {
565                    (
566                        entity_spec.state_name.clone(),
567                        ViewSpec::wire_format_from_entity_spec(entity_spec),
568                    )
569                })
570                .collect::<std::collections::HashMap<_, _>>();
571
572            for entity_name in spec.bytecode.entities.keys() {
573                let wire_format = entity_wire_formats
574                    .get(entity_name)
575                    .cloned()
576                    .unwrap_or_default();
577                index.add_spec(ViewSpec {
578                    id: format!("{}/list", entity_name),
579                    export: entity_name.clone(),
580                    mode: Mode::List,
581                    wire_format: wire_format.clone(),
582                    projection: Projection::all(),
583                    filters: Filters::all(),
584                    delivery: Delivery::default(),
585                    pipeline: None,
586                    source_view: None,
587                });
588
589                index.add_spec(ViewSpec {
590                    id: format!("{}/state", entity_name),
591                    export: entity_name.clone(),
592                    mode: Mode::State,
593                    wire_format: wire_format.clone(),
594                    projection: Projection::all(),
595                    filters: Filters::all(),
596                    delivery: Delivery::default(),
597                    pipeline: None,
598                    source_view: None,
599                });
600
601                index.add_spec(ViewSpec {
602                    id: format!("{}/append", entity_name),
603                    export: entity_name.clone(),
604                    mode: Mode::Append,
605                    wire_format,
606                    projection: Projection::all(),
607                    filters: Filters::all(),
608                    delivery: Delivery::default(),
609                    pipeline: None,
610                    source_view: None,
611                });
612            }
613
614            if !spec.views.is_empty() {
615                let reg = registry.get_or_insert_with(MaterializedViewRegistry::new);
616
617                for view_def in &spec.views {
618                    let export = match &view_def.source {
619                        arete_interpreter::ast::ViewSource::Entity { name } => name.clone(),
620                        arete_interpreter::ast::ViewSource::View { id } => {
621                            id.split('/').next().unwrap_or(id).to_string()
622                        }
623                    };
624
625                    // The compiler includes the canonical entity list/state views in
626                    // `spec.views`. They are already registered above as native views
627                    // backed directly by EntityCache. Registering them again through
628                    // `from_view_def` marks them as derived and replaces the native
629                    // by-id entry, so subscribers read an unpopulated derived cache
630                    // while the projector continues writing to the native cache.
631                    if Self::is_canonical_entity_view(view_def, &export) {
632                        tracing::debug!(
633                            view_id = %view_def.id,
634                            "Keeping canonical entity view backed by the native cache"
635                        );
636                        continue;
637                    }
638
639                    let wire_format = entity_wire_formats
640                        .get(&export)
641                        .cloned()
642                        .unwrap_or_default();
643                    let view_spec = ViewSpec::from_view_def(view_def, &export, wire_format);
644                    let pipeline = view_spec.pipeline.clone().unwrap_or_default();
645                    let source_id = view_spec.source_view.clone().unwrap_or_default();
646                    tracing::debug!(
647                        view_id = %view_def.id,
648                        source = %source_id,
649                        "Registering derived view"
650                    );
651
652                    index.add_spec(view_spec);
653
654                    let materialized =
655                        MaterializedView::new(view_def.id.clone(), source_id, pipeline);
656                    reg.register(materialized);
657                }
658            }
659        }
660
661        (index, registry)
662    }
663
664    fn is_canonical_entity_view(view_def: &ViewDef, export: &str) -> bool {
665        use arete_interpreter::ast::{ViewOutput, ViewSource};
666
667        let ViewSource::Entity { name } = &view_def.source else {
668            return false;
669        };
670        if name != export || !view_def.pipeline.is_empty() {
671            return false;
672        }
673
674        match &view_def.output {
675            ViewOutput::Collection => view_def.id == format!("{export}/list"),
676            ViewOutput::Single | ViewOutput::Keyed { .. } => {
677                view_def.id == format!("{export}/state")
678            }
679        }
680    }
681
682    pub fn build(self) -> Result<Runtime> {
683        let (view_index, materialized_registry) =
684            Self::build_view_index_and_registry(self.views, self.materialized_views, &self.spec);
685
686        #[cfg(feature = "otel")]
687        let mut runtime = Runtime::new(self.config, view_index, self.metrics);
688        #[cfg(not(feature = "otel"))]
689        let mut runtime = Runtime::new(self.config, view_index);
690
691        if let Some(plugin) = self.websocket_auth_plugin {
692            runtime = runtime.with_websocket_auth_plugin(plugin);
693        }
694
695        if let Some(plugin) = self.http_auth_plugin {
696            runtime = runtime.with_http_auth_plugin(plugin);
697        }
698
699        if let Some(max_clients) = self.websocket_max_clients {
700            runtime = runtime.with_websocket_max_clients(max_clients);
701        }
702
703        if let Some(registry) = materialized_registry {
704            runtime = runtime.with_materialized_views(registry);
705        }
706
707        if let Some(spec) = self.spec {
708            runtime = runtime.with_spec(spec)?;
709        }
710        Ok(runtime)
711    }
712}
713
714#[cfg(test)]
715mod tests {
716    use super::*;
717
718    #[test]
719    fn test_builder_pattern() {
720        let _builder = Server::builder()
721            .websocket()
722            .bind("[::]:8877".parse::<SocketAddr>().unwrap());
723    }
724
725    #[test]
726    fn test_spec_creation() {
727        let bytecode = arete_interpreter::compiler::MultiEntityBytecode::new().build();
728        let spec = Spec::new(bytecode, "test_program");
729        assert_eq!(
730            spec.program_ids.first().map(String::as_str),
731            Some("test_program")
732        );
733    }
734
735    #[test]
736    fn http_without_websocket_has_a_read_only_runtime_plan() {
737        let builder = Server::builder().http();
738        assert!(builder.config.runtime_plan.program_reads);
739        assert!(!builder.config.runtime_plan.live_runtime_enabled());
740    }
741
742    #[test]
743    fn program_read_binding_configures_the_exact_auth_target() {
744        let builder = Server::builder().program_read_binding("binding-1");
745
746        assert!(builder.config.runtime_plan.program_reads);
747        assert_eq!(
748            builder.config.program_read_binding_target_id.as_deref(),
749            Some("binding-1")
750        );
751    }
752
753    #[test]
754    fn websocket_and_http_preserve_all_in_one_runtime_behavior() {
755        let builder = Server::builder().websocket().http();
756        assert!(builder.config.runtime_plan.websocket);
757        assert!(builder.config.runtime_plan.live_runtime_enabled());
758    }
759
760    #[test]
761    fn explicit_hosted_plan_disables_program_reads_after_http_helpers() {
762        let plan = RuntimePlan {
763            health: true,
764            chain_reads: true,
765            program_reads: false,
766            stack_queries: true,
767            transactions: true,
768            websocket: true,
769            live_runtime: true,
770        };
771        let builder = Server::builder()
772            .websocket()
773            .http_health()
774            .health_bind("[::]:8081".parse::<SocketAddr>().unwrap())
775            .runtime_plan(plan);
776
777        assert_eq!(builder.config.runtime_plan, plan);
778    }
779
780    #[test]
781    fn solana_gateway_builder_excludes_stack_and_live_capabilities() {
782        let builder = Server::solana_gateway("gateway-us-east-1");
783
784        assert_eq!(
785            builder.inner.config.runtime_plan,
786            RuntimePlan::solana_gateway()
787        );
788        assert!(builder.inner.config.http_health.is_some());
789        assert_eq!(
790            builder.inner.config.solana_gateway_target_id.as_deref(),
791            Some("gateway-us-east-1")
792        );
793        assert!(builder.inner.config.websocket.is_none());
794        assert!(builder.inner.config.yellowstone.is_none());
795        assert!(builder.inner.spec.is_none());
796        assert!(builder.inner.views.is_none());
797        assert!(builder.inner.materialized_views.is_none());
798        assert!(!builder.inner.config.runtime_plan.websocket);
799        assert!(!builder.inner.config.runtime_plan.live_runtime_enabled());
800        assert!(!builder.inner.config.runtime_plan.stack_queries);
801        assert!(!builder.inner.config.runtime_plan.program_reads);
802    }
803
804    #[test]
805    fn solana_gateway_builder_rejects_invalid_gateway_configuration() {
806        assert!(Server::solana_gateway("").build().is_err());
807        assert!(Server::solana_gateway("gateway-us-east-1")
808            .transactions_config(TransactionConfig::default())
809            .build()
810            .is_err());
811    }
812
813    #[test]
814    fn builder_rejects_mismatched_program_release_definitions() {
815        let bytecode = arete_interpreter::compiler::MultiEntityBytecode::new().build();
816        let definition = ProgramRuntimeDefinition {
817            program_id: "Program111".to_string(),
818            program_spec_hash: ProgramSpecHash::from_digest([1; 32]),
819            idl_content_hash: IdlContentHash::from_digest([2; 32]),
820            normalized_idl_hash: NormalizedIdlHash::from_digest([3; 32]),
821            program_release_hash: ProgramReleaseHash::from_digest([4; 32]),
822            account_reader: Arc::new(|_, _| Ok(serde_json::Value::Null)),
823        };
824        let spec =
825            Spec::new(bytecode, "Program111").with_program_runtime_definitions(vec![definition]);
826
827        assert!(Server::builder().spec(spec).build().is_err());
828    }
829
830    #[test]
831    fn canonical_entity_views_are_not_reclassified_as_derived_views() {
832        use arete_interpreter::ast::{IdentitySpec, TypedStreamSpec, ViewDef, ViewSource};
833
834        let list = ViewDef::list("OreBoard");
835        let state = ViewDef::state("OreBoard", &["id", "address"]);
836        assert!(ServerBuilder::is_canonical_entity_view(&list, "OreBoard"));
837        assert!(ServerBuilder::is_canonical_entity_view(&state, "OreBoard"));
838
839        let mut named_view = ViewDef::list("OreBoard");
840        named_view.id = "OreBoard/all".to_string();
841        assert!(!ServerBuilder::is_canonical_entity_view(
842            &named_view,
843            "OreBoard"
844        ));
845
846        let mut chained_view = ViewDef::list("OreBoard");
847        chained_view.source = ViewSource::View {
848            id: "OreBoard/list".to_string(),
849        };
850        assert!(!ServerBuilder::is_canonical_entity_view(
851            &chained_view,
852            "OreBoard"
853        ));
854
855        let entity_spec = TypedStreamSpec::<serde_json::Value>::new(
856            "OreBoard".to_string(),
857            IdentitySpec {
858                primary_keys: vec!["id.address".to_string()],
859                lookup_indexes: Vec::new(),
860            },
861            Vec::new(),
862        );
863        let bytecode = arete_interpreter::compiler::MultiEntityBytecode::new()
864            .add_entity("OreBoard".to_string(), entity_spec, 1)
865            .build();
866        let spec = Some(Spec::new(bytecode, "Program111").with_views(vec![list, state]));
867
868        let (index, _) = ServerBuilder::build_view_index_and_registry(None, None, &spec);
869        assert!(!index
870            .get_view("OreBoard/list")
871            .expect("list view should exist")
872            .is_derived());
873        assert!(!index
874            .get_view("OreBoard/state")
875            .expect("state view should exist")
876            .is_derived());
877        assert_eq!(index.by_export("OreBoard").len(), 3);
878    }
879}
arete_server/lib.rs

arete_server/
lib.rs
