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

arete_server/
lib.rs
