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

arete_server/
lib.rs
