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

arete_server/
lib.rs
