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

arete_server/
lib.rs
