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

arete_server/
lib.rs
