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

arete_server/
lib.rs
