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

arete_server/
lib.rs
