Skip to main content

flatbed/
lib.rs

1/// FlatBuffers utilities and route registry for the flatbed framework.
2///
3/// This library provides helper functions and utilities for working with
4/// FlatBuffers, plus the compile-time route registry that powers the
5/// `#[route]` macro.
6///
7/// # Plain Struct API
8///
9/// The primary API uses plain Rust structs. Handlers work with regular structs:
10///
11/// ```rust,ignore
12/// #[route("/ping")]
13/// fn handle_ping(req: PingRequest) -> Result<PingResponse, Error> {
14///     Ok(PingResponse {
15///         message: Some(format!("pong: {}", req.message.unwrap_or_default())),
16///         timestamp: req.timestamp,
17///         success: true,
18///     })
19/// }
20/// ```
21///
22/// FlatBuffer serialization is handled automatically by the framework.
23use flatbuffers::FlatBufferBuilder;
24use std::borrow::Cow;
25use std::collections::HashMap;
26use std::fmt;
27use std::ops::Deref;
28#[cfg(feature = "telemetry")]
29use std::sync::Arc;
30use std::sync::LazyLock;
31
32#[cfg(feature = "nats")]
33pub mod nats;
34#[cfg(feature = "nats")]
35pub use nats::{run_stream_worker, HasJetStream, HasNatsClient, NatsResult, StreamWorker};
36
37// Re-exported through `nats` rather than declared public here, so the managed
38// connector has one canonical path.
39#[cfg(feature = "nats")]
40mod nats_connect;
41
42#[cfg(feature = "nats")]
43pub mod nats_route;
44#[cfg(feature = "nats")]
45pub use flatbed_macros::nats_route;
46#[cfg(feature = "nats")]
47pub use nats_route::{
48    get_nats_routes, run_nats_route, validate_nats_routes, NatsEncoding, NatsRouteConflict,
49    NatsRouteInfo,
50};
51
52#[cfg(feature = "nats")]
53pub mod nats_request;
54#[cfg(feature = "nats")]
55pub use nats_request::{NatsRequest, NatsRequestError, NatsRequestExt};
56
57// NATS JetStream KV cache worker (enabled with "nats" feature)
58#[cfg(feature = "nats")]
59pub mod kv;
60#[cfg(feature = "nats")]
61pub use kv::{run_kv_worker, KvWorker};
62
63#[cfg(feature = "nats")]
64#[doc(hidden)]
65pub use async_nats;
66
67// Re-export futures for use in generated macro code + by the k8s
68// reconciler / watcher runtime executors.
69#[cfg(any(feature = "nats", feature = "k8s"))]
70#[doc(hidden)]
71pub use futures;
72
73// Kubernetes reconciler support. The `k8s` module itself only needs
74// the kube client; the NATS-bound `KubeReconciler` trait and its
75// runtime executor are further gated behind `feature = "nats"`
76// inside the module.
77#[cfg(feature = "k8s")]
78pub mod k8s;
79#[cfg(feature = "k8s")]
80pub use k8s::{
81    run_kube_native_reconciler, run_kube_watcher, wait_for_follower, wait_for_follower_loss,
82    wait_for_leadership, wait_for_leadership_loss, HasKubeClient, HasLeaderElection,
83    KubeNativeReconciler, KubeWatcher, ReconcileError,
84};
85#[cfg(all(feature = "nats", feature = "k8s"))]
86pub use k8s::{run_kube_reconciler, KubeReconciler};
87
88// Generic context wrapper (available when any service feature is enabled)
89#[cfg(any(feature = "nats", feature = "k8s"))]
90mod context;
91#[cfg(any(feature = "nats", feature = "k8s"))]
92pub use context::FlatbedContext;
93
94// Re-export kube for use in generated macro code
95#[cfg(feature = "k8s")]
96#[doc(hidden)]
97pub use kube;
98
99// Hyper integration module
100pub mod hyper;
101
102pub mod readiness;
103pub use readiness::{Readiness, ReadinessGate};
104
105// Worker supervision
106mod supervisor;
107pub use supervisor::{worker_states, RestartPolicy, WorkerState};
108
109// Telemetry module (enabled with "telemetry" feature)
110#[cfg(feature = "telemetry")]
111pub mod telemetry;
112#[cfg(feature = "telemetry")]
113pub use telemetry::{TelemetryConfig, TelemetryError, TelemetryService};
114
115// Re-export dependencies so consumers don't need them directly
116// The #[route] macro and generated types use these
117pub use flatbuffers;
118pub use serde;
119pub use serde_json;
120#[cfg(feature = "openapi")]
121pub use utoipa;
122
123// Re-export include_dir: the `embed` form of `static_route!` expands to its
124// macro, so a consumer never names the crate.
125#[doc(hidden)]
126pub use include_dir;
127
128// Re-export tokio for use in generated main macro code
129#[doc(hidden)]
130pub use tokio;
131
132/// Boxed, Send future alias for trait return types.
133///
134/// Eliminates the need for `use std::future::Future; use std::pin::Pin;`
135/// in every file that implements a Flatbed trait.
136pub type BoxFuture<T> = std::pin::Pin<Box<dyn std::future::Future<Output = T> + Send>>;
137
138// Re-export http types so users don't need to add the dependency
139pub use http::{HeaderMap, HeaderName, HeaderValue, Method, StatusCode};
140
141// Re-export uuid for use in generated code (request ID generation)
142#[doc(hidden)]
143pub use uuid;
144
145/// A guard invoked before a matched `#[route]` handler runs.
146///
147/// Returning `Err` short-circuits the request with the error's status,
148/// code, and message; the handler never runs. See
149/// [`FlatbedConfig::before_request`].
150pub type BeforeRequestHook =
151    std::sync::Arc<dyn Fn(&RequestParts) -> Result<(), FlatbedRouteError> + Send + Sync>;
152
153// ============================================================================
154// OpenAPI Support Types
155// ============================================================================
156
157/// Configuration for flatbed routes and optional telemetry integration.
158///
159/// This replaces the old `ApiInfo` struct and adds support for automatic
160/// health endpoint registration when telemetry is enabled.
161///
162/// # Example
163///
164/// ```rust,ignore
165/// use flatbed::{FlatbedConfig, configure_flatbed_routes};
166///
167/// // Basic configuration
168/// let config = FlatbedConfig::new("My API")
169///     .description("API description")
170///     .external_url("https://api.example.com");
171///
172/// // With telemetry (registers /healthz, /readyz, /metrics automatically)
173/// let config = FlatbedConfig::new("My API")
174///     .with_telemetry(telemetry_service);
175/// ```
176#[derive(Clone)]
177pub struct FlatbedConfig {
178    /// API title for OpenAPI spec
179    pub title: &'static str,
180    /// API description for OpenAPI spec
181    pub description: Option<&'static str>,
182    /// External URL for OpenAPI spec (e.g., public endpoint through Envoy/TLS)
183    pub external_url: Option<String>,
184    /// Host address to bind to (default: "0.0.0.0")
185    pub host: String,
186    /// Port to bind to (default: 8080)
187    pub port: u16,
188    /// Splash message printed on server startup (None = no splash)
189    pub splash: Option<String>,
190    /// Guard invoked before a matched route handler runs (see
191    /// [`FlatbedConfig::before_request`]).
192    pub before_request: Option<BeforeRequestHook>,
193    /// Runtime dependencies readiness answers on, on top of the boot latch.
194    /// The boot function receives this config by value, so it can register a
195    /// gate for anything it connects (see [`Readiness::gate`]).
196    pub readiness: Readiness,
197    /// Telemetry service for health endpoints (when telemetry feature is enabled)
198    #[cfg(feature = "telemetry")]
199    pub telemetry: Option<Arc<dyn TelemetryService>>,
200}
201
202impl Default for FlatbedConfig {
203    fn default() -> Self {
204        Self {
205            title: "API",
206            description: None,
207            external_url: None,
208            host: "0.0.0.0".to_string(),
209            port: 8080,
210            splash: None,
211            before_request: None,
212            readiness: Readiness::new(),
213            #[cfg(feature = "telemetry")]
214            telemetry: None,
215        }
216    }
217}
218
219impl fmt::Debug for FlatbedConfig {
220    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
221        let mut debug = f.debug_struct("FlatbedConfig");
222        debug
223            .field("title", &self.title)
224            .field("description", &self.description)
225            .field("external_url", &self.external_url)
226            .field("host", &self.host)
227            .field("port", &self.port)
228            .field("splash", &self.splash.as_ref().map(|_| "<splash>"))
229            .field(
230                "before_request",
231                &self.before_request.as_ref().map(|_| "<hook>"),
232            )
233            .field("readiness", &self.readiness);
234        #[cfg(feature = "telemetry")]
235        debug.field(
236            "telemetry",
237            &self.telemetry.as_ref().map(|_| "<TelemetryService>"),
238        );
239        debug.finish()
240    }
241}
242
243impl FlatbedConfig {
244    /// Create a new FlatbedConfig with the given API title.
245    pub fn new(title: &'static str) -> Self {
246        Self {
247            title,
248            ..Self::default()
249        }
250    }
251
252    /// Set the bind host address (default: "0.0.0.0").
253    pub fn host(mut self, host: &str) -> Self {
254        self.host = host.to_string();
255        self
256    }
257
258    /// Set the bind port (default: 8080).
259    pub fn port(mut self, port: u16) -> Self {
260        self.port = port;
261        self
262    }
263
264    /// Set the API description for OpenAPI spec.
265    pub fn description(mut self, desc: &'static str) -> Self {
266        self.description = Some(desc);
267        self
268    }
269
270    /// Set a splash page returned at `GET /`.
271    ///
272    /// The splash is served as a plain text response when accessing the root URL.
273    /// Use this to display API information, available endpoints, or ASCII art.
274    ///
275    /// # Example
276    ///
277    /// ```rust,ignore
278    /// let config = FlatbedConfig::new("My API")
279    ///     .host("0.0.0.0")
280    ///     .port(8080)
281    ///     .splash("My API Server\nEndpoints:\n  GET /healthz");
282    /// ```
283    pub fn splash(mut self, splash: impl Into<String>) -> Self {
284        self.splash = Some(splash.into());
285        self
286    }
287
288    /// Register a guard invoked before every matched `#[route]` handler runs.
289    ///
290    /// Returning `Err` short-circuits the request with the error's status,
291    /// code, and message — the handler never runs. Built-in endpoints
292    /// (`/healthz`, `/readyz`, `/metrics`, `/openapi.json`, `/schema.bfbs`)
293    /// and `static_route!` mounts are not matched routes and skip the guard.
294    ///
295    /// # Example
296    ///
297    /// ```rust,ignore
298    /// let config = FlatbedConfig::new("My API").before_request(|req| match req
299    ///     .header("authorization")
300    /// {
301    ///     Some("Bearer secret") => Ok(()),
302    ///     _ => Err(FlatbedRouteError::unauthorized("missing or invalid bearer token")),
303    /// });
304    /// ```
305    pub fn before_request<F>(mut self, hook: F) -> Self
306    where
307        F: Fn(&RequestParts) -> Result<(), FlatbedRouteError> + Send + Sync + 'static,
308    {
309        self.before_request = Some(std::sync::Arc::new(hook));
310        self
311    }
312
313    /// Answer readiness on an existing [`Readiness`] registry rather than the
314    /// empty one every config starts with.
315    ///
316    /// Only needed when gates are registered before the config exists; a boot
317    /// function can reach the default registry through the config it receives.
318    ///
319    /// # Example
320    ///
321    /// ```rust,ignore
322    /// let readiness = Readiness::new();
323    /// let nats_gate = readiness.gate("nats");
324    /// let config = FlatbedConfig::new("My API").readiness(readiness);
325    /// ```
326    pub fn readiness(mut self, readiness: Readiness) -> Self {
327        self.readiness = readiness;
328        self
329    }
330
331    /// Set the external URL for OpenAPI spec (e.g., public endpoint through Envoy/TLS).
332    pub fn external_url(mut self, url: impl Into<String>) -> Self {
333        self.external_url = Some(url.into());
334        self
335    }
336
337    /// Enable telemetry with automatic health endpoint registration.
338    ///
339    /// When telemetry is enabled, the following endpoints are automatically registered:
340    /// - `GET /healthz` - Health check endpoint (returns "OK")
341    /// - `GET /readyz` - Readiness probe endpoint (returns "Ready")
342    /// - `GET /metrics` - Prometheus metrics endpoint
343    #[cfg(feature = "telemetry")]
344    pub fn with_telemetry(mut self, telemetry: Arc<dyn TelemetryService>) -> Self {
345        self.telemetry = Some(telemetry);
346        self
347    }
348
349    /// Get the bind address as a SocketAddr string.
350    pub fn bind_address(&self) -> String {
351        format!("{}:{}", self.host, self.port)
352    }
353}
354
355// ============================================================================
356// Flatbed Server Runner
357// ============================================================================
358
359/// Main entry point for running a Flatbed server with two-phase boot pattern.
360///
361/// The server starts immediately and begins accepting connections. The framework
362/// automatically manages health and readiness probes during the boot lifecycle:
363///
364/// 1. Server starts listening — `/healthz` returns 200 (alive)
365/// 2. Boot function runs — `/readyz` returns 503, user routes return 503
366/// 3. Boot completes, context is stored — `/readyz` returns 200, routes accept traffic
367/// 4. Workers are spawned after the ready signal
368///
369/// Step 3's latch is one-shot, but readiness is not: a boot function that
370/// registers a [`Readiness`] gate for a runtime dependency keeps driving
371/// `/readyz` for the life of the process, and `/readyz` returns 503 again
372/// whenever a gate reports its dependency unusable.
373///
374/// If a worker fails, `/healthz` returns 503 to trigger Kubernetes restart.
375///
376/// # Context type
377///
378/// The context type `C` can be any `Clone + Send + Sync + 'static` type.
379/// You have two options:
380///
381/// - **Custom struct**: Define your own struct and implement any required
382///   traits (`HasJetStream`, `HasKubeClient`, `HasLeaderElection`) manually.
383/// - **[`FlatbedContext<C>`]**: Use the provided wrapper which owns framework
384///   clients and auto-implements the traits based on enabled features.
385///   Your application data is accessed transparently via `Deref`.
386///
387/// # Example: custom context
388///
389/// ```rust,ignore
390/// use flatbed::{Flatbed, FlatbedConfig};
391///
392/// #[derive(Clone)]
393/// struct AppContext {
394///     db: DatabasePool,
395/// }
396///
397/// Flatbed::run(config, |_| async move {
398///     Ok(AppContext { db: DatabasePool::connect().await? })
399/// }).await?;
400/// ```
401///
402/// # Example: FlatbedContext wrapper
403///
404/// ```rust,ignore
405/// use flatbed::{Flatbed, FlatbedConfig, FlatbedContext};
406///
407/// #[derive(Clone)]
408/// struct AppData { pub namespace: String }
409/// type AppContext = FlatbedContext<AppData>;
410///
411/// Flatbed::run(config, |_| async move {
412///     let ctx = FlatbedContext::builder(AppData { namespace: "default".into() })
413///         .nats_client(nats_client)
414///         .jetstream(jetstream)
415///         .kube_client(kube_client)
416///         .leader_election(is_leader_tx, is_leader_rx, ha_mode)
417///         .build();
418///     Ok(ctx)
419/// }).await?;
420/// ```
421pub struct Flatbed;
422
423impl Flatbed {
424    /// Run the server with a boot function for two-phase initialization.
425    ///
426    /// The boot function receives a clone of the [`FlatbedConfig`] (for access to
427    /// host, port, etc.) and returns the application context. The framework handles
428    /// all probe signaling automatically:
429    ///
430    /// - `/healthz` → 200 as soon as the server starts listening
431    /// - Boot function runs (probes: healthz=200, ready=503)
432    /// - Context is stored, then `/readyz` → 200 and workers are spawned
433    ///
434    /// # Type Parameters
435    ///
436    /// - `C`: The application context type. Must be `Clone + Send + Sync + 'static`.
437    /// - `F`: The boot function type.
438    /// - `Fut`: The future returned by the boot function.
439    ///
440    /// # Arguments
441    ///
442    /// - `config`: Server configuration including host, port, and optional telemetry.
443    /// - `boot`: Async function that receives a [`FlatbedConfig`] clone and returns
444    ///   the application context.
445    ///
446    /// # Errors
447    ///
448    /// Returns an error if:
449    /// - Route validation fails (duplicate paths with different types)
450    /// - The bind address is invalid
451    /// - The server fails to start
452    /// - The boot function returns an error
453    pub async fn run<C, F, Fut>(config: FlatbedConfig, boot: F) -> Result<(), Error>
454    where
455        C: Clone + Send + Sync + 'static,
456        F: FnOnce(FlatbedConfig) -> Fut + Send,
457        Fut: std::future::Future<Output = Result<C, Box<dyn std::error::Error + Send + Sync>>>
458            + Send,
459    {
460        use std::sync::Arc;
461        use tokio::sync::{watch, RwLock};
462
463        // Validate routes (fails fast on conflicts)
464        crate::validate_routes().map_err(|e| Error::Custom(format!("Route conflict: {}", e)))?;
465        crate::validate_type_registry()
466            .map_err(|e| Error::Custom(format!("Type registry conflict: {}", e)))?;
467        #[cfg(feature = "nats")]
468        crate::validate_nats_routes()
469            .map_err(|e| Error::Custom(format!("NATS subject conflict: {}", e)))?;
470
471        // Build router from inventory-registered routes
472        let router = Arc::new(hyper::build_router());
473
474        // Parse bind address
475        let addr: std::net::SocketAddr = config
476            .bind_address()
477            .parse()
478            .map_err(|e| Error::Custom(format!("Invalid bind address: {}", e)))?;
479
480        // Create watch channels for probes (start as unhealthy/not ready)
481        let (healthz_tx, healthz_rx) = watch::channel(false);
482        let (booted_tx, booted_rx) = watch::channel(false);
483
484        // Create shared context storage (None until boot completes)
485        let context: Arc<RwLock<Option<Arc<C>>>> = Arc::new(RwLock::new(None));
486
487        // Clone config for the boot function before moving original into ServiceContext
488        let boot_config = config.clone();
489
490        // Create ServiceContext with probe receivers
491        let service_ctx = hyper::ServiceContext {
492            router,
493            healthz_rx,
494            booted_rx,
495            context: Arc::clone(&context),
496            config,
497            static_routes: Arc::new(crate::get_static_routes()),
498        };
499
500        // Create and spawn the server
501        let server = hyper::AutoServer::new(addr, service_ctx, healthz_tx.clone());
502        let server_handle = tokio::spawn(async move { server.serve().await });
503
504        // Server is listening — mark as alive immediately
505        // (prevents Kubernetes from killing pod during slow boot)
506        let _ = healthz_tx.send(true);
507
508        // Run boot function (receives config clone, returns context)
509        let ctx = boot(boot_config)
510            .await
511            .map_err(|e| Error::Custom(format!("Boot error: {}", e)))?;
512
513        // Store context FIRST, then signal ready
514        // This fixes the race condition where workers could start before context was set
515        {
516            let mut guard = context.write().await;
517            *guard = Some(Arc::new(ctx));
518        }
519        let _ = booted_tx.send(true);
520
521        // Await server completion
522        server_handle
523            .await
524            .map_err(|e| Error::Custom(format!("Server task error: {}", e)))?
525            .map_err(|e| Error::Custom(format!("Server error: {}", e)))
526    }
527}
528
529/// Schema field info for OpenAPI generation
530#[derive(Clone, Copy, Debug)]
531pub struct FieldInfo {
532    /// Field name as declared in the `.fbs` schema.
533    pub name: &'static str,
534    /// Adapter-language FlatBuffer type (`"uint64"`, `"[Address]"`, `"Severity"`).
535    pub fbs_type: &'static str,
536    /// True for scalars and enums; false for strings, tables and vectors,
537    /// which are optional on the wire.
538    pub required: bool,
539}
540
541/// Schema information
542#[derive(Clone, Copy, Debug)]
543pub struct SchemaInfo {
544    /// Bare type name, used as the OpenAPI component key.
545    pub name: &'static str,
546    /// The type's fields, in declaration order.
547    pub fields: &'static [FieldInfo],
548}
549
550/// A generated FlatBuffer table, registered at compile time so the full set
551/// of types is discoverable at runtime.
552///
553/// One is submitted for *every* generated table — including tables that only
554/// ever appear nested inside another and never as a route body — so the
555/// registry is a complete inventory of the generated types. `name` is the bare
556/// type name (the key routes use to reference their request/response types);
557/// `namespace` is the FlatBuffer namespace (`"v_1"`, `"test"`).
558#[derive(Clone, Copy, Debug)]
559pub struct TypeSchema {
560    pub name: &'static str,
561    pub namespace: &'static str,
562    pub fields: &'static [TypeFieldInfo],
563}
564
565/// A single field of a registered [`TypeSchema`].
566///
567/// `fbs_type` is the rich adapter-language string (`"uint64"`, `"[Address]"`,
568/// `"Severity"`) — enough to recover the exact wire shape. `field_id` is the
569/// FlatBuffer vtable slot; `required` is false for wire-optional fields
570/// (strings, tables, vectors) and true for scalars and enums.
571#[derive(Clone, Copy, Debug)]
572pub struct TypeFieldInfo {
573    pub name: &'static str,
574    pub fbs_type: &'static str,
575    pub field_id: u16,
576    pub required: bool,
577}
578
579/// A generated FlatBuffer enum, registered at compile time alongside
580/// [`TypeSchema`]. `variants` are listed in FlatBuffer value order.
581#[derive(Clone, Copy, Debug)]
582pub struct EnumSchema {
583    pub name: &'static str,
584    pub namespace: &'static str,
585    pub variants: &'static [&'static str],
586}
587
588/// The FlatBuffer binary reflection (`.bfbs`) for a service's schemas, baked in
589/// by generated code so the runtime can serve it at `/schema.bfbs`. The bytes
590/// are the exact `flatc -b --schema` output — complete for the wire codec
591/// (field ids, enum underlying types + values, defaults). One is submitted per
592/// compiled root schema.
593#[derive(Clone, Copy, Debug)]
594pub struct FlatbedSchema {
595    pub bfbs: &'static [u8],
596}
597
598inventory::collect!(TypeSchema);
599inventory::collect!(EnumSchema);
600inventory::collect!(FlatbedSchema);
601
602/// The baked-in FlatBuffer reflection when a schema was compiled into this
603/// service. Defined for the single-root case: flatc's `.bfbs` covers a root's
604/// full include graph, so one submission describes everything.
605/// With multiple root schemas the result is unspecified — `inventory` collection
606/// order gives no defined choice among them.
607#[must_use]
608pub fn get_schema_bfbs() -> Option<&'static [u8]> {
609    inventory::iter::<FlatbedSchema>
610        .into_iter()
611        .next()
612        .map(|s| s.bfbs)
613}
614
615// ============================================================================
616// Trait Definitions for JSON Companion Types
617// ============================================================================
618
619/// Trait for JSON companion types that can convert to FlatBuffer bytes
620///
621/// This trait is implemented by generated JSON companion structs (e.g., `PingRequestJson`)
622/// and provides the interface for serialization and schema information.
623///
624/// # Example
625///
626/// ```rust,ignore
627/// use flatbed::ToFlatBuffer;
628///
629/// // Generated companion type implements this trait
630/// let json_req = PingRequestJson { message: Some("hello".into()), timestamp: 123 };
631/// let bytes = json_req.to_flatbuffer();
632/// ```
633pub trait ToFlatBuffer: serde::de::DeserializeOwned + serde::Serialize {
634    /// Schema fields for OpenAPI generation (as a const)
635    const SCHEMA_FIELDS: &'static [FieldInfo];
636
637    /// Schema name for OpenAPI generation (as a const)
638    const SCHEMA_NAME: &'static str;
639
640    /// Convert this JSON struct to FlatBuffer bytes
641    fn to_flatbuffer(&self) -> Vec<u8>;
642
643    /// Get schema field info for OpenAPI generation (runtime method)
644    fn schema_fields() -> &'static [FieldInfo] {
645        Self::SCHEMA_FIELDS
646    }
647
648    /// Get the schema name for OpenAPI (runtime method)
649    fn schema_name() -> &'static str {
650        Self::SCHEMA_NAME
651    }
652}
653
654/// Implementation of ToFlatBuffer for unit type
655///
656/// This allows `FlatbedError<()>` (errors without details) to work with
657/// the error response builder that requires ToFlatBuffer.
658impl ToFlatBuffer for () {
659    const SCHEMA_FIELDS: &'static [FieldInfo] = &[];
660    const SCHEMA_NAME: &'static str = "Empty";
661
662    fn to_flatbuffer(&self) -> Vec<u8> {
663        Vec::new()
664    }
665}
666
667/// Trait for JSON companion types that can be decoded from FlatBuffer bytes
668///
669/// The mirror of [`ToFlatBuffer`], implemented by the same generated companion
670/// structs. It exists so code generic over a body type — a typed NATS reply,
671/// for one — can decode without naming the concrete type.
672///
673/// # Example
674///
675/// ```rust,ignore
676/// use flatbed::FromFlatBuffer;
677///
678/// let decoded = PingResponse::from_flatbuffer(&bytes)?;
679/// ```
680pub trait FromFlatBuffer: Sized {
681    /// Decode this struct from a FlatBuffer payload.
682    ///
683    /// # Errors
684    ///
685    /// Returns [`flatbuffers::InvalidFlatbuffer`] when the bytes are not a
686    /// valid buffer of this type.
687    fn from_flatbuffer(bytes: &[u8]) -> Result<Self, flatbuffers::InvalidFlatbuffer>;
688}
689
690/// The mirror of [`ToFlatBuffer`]'s unit implementation, so a body-less
691/// response is as decodable as it is encodable.
692impl FromFlatBuffer for () {
693    fn from_flatbuffer(_bytes: &[u8]) -> Result<Self, flatbuffers::InvalidFlatbuffer> {
694        Ok(())
695    }
696}
697
698/// Trait connecting FlatBuffer types to their plain struct companions
699///
700/// This allows the `#[route]` macro to discover the plain struct type
701/// without hardcoding module paths, making flatbed reusable across projects.
702///
703/// # Example
704///
705/// ```rust,ignore
706/// use flatbed::HasJsonCompanion;
707///
708/// // Generated automatically by flatbed_build:
709/// impl<'a> HasJsonCompanion for __fb::v_1::PingRequest<'a> {
710///     type Json = PingRequest;  // Plain struct
711/// }
712/// ```
713pub trait HasJsonCompanion {
714    /// The plain struct companion type for this FlatBuffer type
715    type Json: ToFlatBuffer;
716}
717
718/// Trait connecting plain structs to their FlatBuffer types
719///
720/// This provides bidirectional type mapping for the plain struct API.
721///
722/// # Example
723///
724/// ```rust,ignore
725/// impl HasPlainStruct for PingRequest {
726///     type Plain = PingRequest;
727///     type FlatBuffer<'a> = __fb::v_1::PingRequest<'a>;
728/// }
729/// ```
730pub trait HasPlainStruct {
731    /// The plain struct type (usually Self)
732    type Plain: ToFlatBuffer;
733    /// The FlatBuffer view type
734    type FlatBuffer<'a>;
735}
736
737/// OpenAPI metadata for a route
738#[derive(Clone, Copy, Debug, Default)]
739pub struct OpenApiRouteInfo {
740    pub version: Option<&'static str>, // e.g., "v1", "v2"
741    pub tag: Option<&'static str>,
742    pub summary: Option<&'static str>,
743    pub operation_id: Option<&'static str>,
744    pub deprecated: bool,
745    pub request_schema: Option<SchemaInfo>,
746    pub response_schema: Option<SchemaInfo>,
747}
748
749// Re-export the route macro from flatbed_macros
750pub use flatbed_macros::route;
751
752pub use flatbed_macros::static_route;
753
754// Re-export the main macro from flatbed_macros
755pub use flatbed_macros::main;
756
757// Re-export inventory for use by the macro
758#[doc(hidden)]
759pub use inventory;
760
761// ============================================================================
762// HTTP Request/Response Wrapper Types
763// ============================================================================
764
765/// HTTP request wrapper providing context alongside the deserialized body
766///
767/// The generic parameter `T` is the request body type (flatbed-generated struct).
768/// The optional parameter `C` is the application context type (defaults to `()`).
769///
770/// # Example
771///
772/// ```rust,ignore
773/// // Simple handler without context
774/// #[route("/ping")]
775/// async fn handle_ping(req: Request<PingRequest>) -> Result<Response<PingResponse>, FlatbedError> {
776///     Ok(Response::ok(PingResponse { message: Some("pong".into()), ... }))
777/// }
778///
779/// // Handler with application context
780/// #[route("/users")]
781/// async fn handle_users(req: Request<UserRequest, AppCtx>) -> Result<Response<UserResponse>, FlatbedError> {
782///     let user = req.ctx.db.find_user(&req.body.id).await?;
783///     Ok(Response::ok(UserResponse { name: user.name, ... }))
784/// }
785/// ```
786#[derive(Debug)]
787pub struct Request<T, C = ()> {
788    /// The deserialized request body
789    pub body: T,
790    /// Application context (zero-sized when C = ())
791    pub ctx: C,
792    /// HTTP headers from the request
793    pub headers: HeaderMap,
794    /// HTTP method
795    pub method: Method,
796    /// Request path
797    pub path: String,
798    /// Path parameters (e.g., /users/{id} -> {"id": "123"})
799    pub path_params: HashMap<String, String>,
800    /// Query string parameters
801    pub query_params: HashMap<String, String>,
802    /// Request ID (propagated or generated)
803    pub request_id: String,
804}
805
806impl<T, C> Request<T, C> {
807    /// Get a header value by name (case-insensitive)
808    pub fn header(&self, name: &str) -> Option<&str> {
809        self.headers.get(name).and_then(|v| v.to_str().ok())
810    }
811
812    /// Get a path parameter by name
813    pub fn param(&self, name: &str) -> Option<&str> {
814        self.path_params.get(name).map(|s| s.as_str())
815    }
816
817    /// Get a query parameter by name
818    pub fn query(&self, name: &str) -> Option<&str> {
819        self.query_params.get(name).map(|s| s.as_str())
820    }
821
822    /// Get the request ID
823    pub fn request_id(&self) -> &str {
824        &self.request_id
825    }
826}
827
828impl<T, C> Deref for Request<T, C> {
829    type Target = T;
830    fn deref(&self) -> &T {
831        &self.body
832    }
833}
834
835/// HTTP response wrapper with status code and headers
836///
837/// # Example
838///
839/// ```rust,ignore
840/// // Simple 200 OK response
841/// Response::ok(MyResponse { ... })
842///
843/// // 201 Created
844/// Response::created(MyResponse { ... })
845///
846/// // Custom status and headers
847/// Response::with_status(MyResponse { ... }, StatusCode::ACCEPTED)
848///     .header("X-Custom", "value")
849/// ```
850#[derive(Debug)]
851pub struct Response<T> {
852    /// The response body
853    pub body: T,
854    /// HTTP status code (default: 200 OK)
855    pub status: StatusCode,
856    /// Response headers
857    pub headers: HeaderMap,
858    raw: Option<(Vec<u8>, Cow<'static, str>)>,
859}
860
861impl<T> Response<T> {
862    /// Create a response with 200 OK status
863    pub fn ok(body: T) -> Self {
864        Self {
865            body,
866            status: StatusCode::OK,
867            headers: HeaderMap::new(),
868            raw: None,
869        }
870    }
871
872    /// Create a response with 201 Created status
873    pub fn created(body: T) -> Self {
874        Self {
875            body,
876            status: StatusCode::CREATED,
877            headers: HeaderMap::new(),
878            raw: None,
879        }
880    }
881
882    /// Create a response with 202 Accepted status
883    pub fn accepted(body: T) -> Self {
884        Self {
885            body,
886            status: StatusCode::ACCEPTED,
887            headers: HeaderMap::new(),
888            raw: None,
889        }
890    }
891
892    /// Create a response with 204 No Content status
893    pub fn no_content(body: T) -> Self {
894        Self {
895            body,
896            status: StatusCode::NO_CONTENT,
897            headers: HeaderMap::new(),
898            raw: None,
899        }
900    }
901
902    /// Create a response with custom status code
903    pub fn with_status(body: T, status: StatusCode) -> Self {
904        Self {
905            body,
906            status,
907            headers: HeaderMap::new(),
908            raw: None,
909        }
910    }
911
912    /// Set the HTTP status code (builder pattern)
913    pub fn status(mut self, status: StatusCode) -> Self {
914        self.status = status;
915        self
916    }
917
918    /// Add a header to the response (builder pattern)
919    pub fn header(mut self, key: impl AsRef<str>, value: impl AsRef<str>) -> Self {
920        if let (Ok(name), Ok(val)) = (
921            HeaderName::try_from(key.as_ref()),
922            HeaderValue::try_from(value.as_ref()),
923        ) {
924            self.headers.insert(name, val);
925        }
926        self
927    }
928
929    /// Yields `Some((bytes, content_type))` only for a response constructed as a
930    /// raw response, and `None` for all other constructors.
931    #[doc(hidden)]
932    pub fn take_raw(&mut self) -> Option<(Vec<u8>, Cow<'static, str>)> {
933        self.raw.take()
934    }
935}
936
937impl Response<()> {
938    /// Create a response that emits `bytes` verbatim under `content_type`,
939    /// bypassing JSON/FlatBuffer serialization.
940    ///
941    /// This is the escape hatch for handlers that need to return a body the
942    /// typed path can't express — HTML, CSV, JavaScript, images, or any other
943    /// media type. The `content_type` is written to the `Content-Type` header
944    /// as-is.
945    ///
946    /// ```rust,ignore
947    /// #[route("/report.csv", method = "POST")]
948    /// async fn report(req: Request<ReportQuery>) -> Result<Response<()>, FlatbedRouteError> {
949    ///     let csv = build_report_csv(&req.body);
950    ///     Ok(Response::raw(csv.into_bytes(), "text/csv; charset=utf-8"))
951    /// }
952    /// ```
953    pub fn raw(bytes: impl Into<Vec<u8>>, content_type: impl Into<Cow<'static, str>>) -> Self {
954        Self {
955            body: (),
956            status: StatusCode::OK,
957            headers: HeaderMap::new(),
958            raw: Some((bytes.into(), content_type.into())),
959        }
960    }
961}
962
963// ============================================================================
964// Worker Error Type
965// ============================================================================
966
967/// Simple error type for background workers
968///
969/// Workers use this simpler error type since they don't need HTTP-specific
970/// fields like status codes or headers.
971///
972/// # Example
973///
974/// ```rust,ignore
975/// use flatbed::{worker, FlatbedWorkerError};
976/// use std::sync::Arc;
977///
978/// #[worker]
979/// async fn my_worker(ctx: Arc<AppContext>) -> Result<(), FlatbedWorkerError> {
980///     ctx.do_something().await
981///         .map_err(|e| FlatbedWorkerError::new("TASK_ERROR", e.to_string()))?;
982///     Ok(())
983/// }
984/// ```
985#[derive(Debug, Clone)]
986pub struct FlatbedWorkerError {
987    /// Application-defined error code (e.g., "CACHE_ERROR")
988    pub code: String,
989    /// Human-readable error message
990    pub message: String,
991}
992
993impl FlatbedWorkerError {
994    /// Create a new worker error with code and message
995    pub fn new(code: impl Into<String>, message: impl Into<String>) -> Self {
996        Self {
997            code: code.into(),
998            message: message.into(),
999        }
1000    }
1001}
1002
1003impl fmt::Display for FlatbedWorkerError {
1004    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1005        write!(f, "[{}] {}", self.code, self.message)
1006    }
1007}
1008
1009impl std::error::Error for FlatbedWorkerError {}
1010
1011// ============================================================================
1012// HTTP Route Error Type
1013// ============================================================================
1014
1015/// HTTP-conformant error type with status code, error code, and optional details
1016///
1017/// The generic parameter `D` is the type for structured error details (must be a
1018/// flatbed-generated struct implementing `ToFlatBuffer`). Defaults to `()` for
1019/// errors without structured details.
1020///
1021/// # Example
1022///
1023/// ```rust,ignore
1024/// // Simple error without details
1025/// return Err(FlatbedRouteError::bad_request("Invalid input"));
1026///
1027/// // Error with code
1028/// return Err(FlatbedRouteError::bad_request("Validation failed")
1029///     .code("VALIDATION_ERROR"));
1030///
1031/// // Error with typed details (flatbed-generated struct)
1032/// return Err(FlatbedRouteError::bad_request("Validation failed")
1033///     .code("VALIDATION_ERROR")
1034///     .with_details(ValidationErrorDetails {
1035///         field: Some("email".into()),
1036///         reason: Some("Invalid format".into()),
1037///     }));
1038/// ```
1039#[derive(Debug, Clone)]
1040pub struct FlatbedRouteError<D = ()> {
1041    /// HTTP status code
1042    pub status: StatusCode,
1043    /// Application-defined error code (e.g., "VALIDATION_ERROR")
1044    pub code: String,
1045    /// Human-readable error message
1046    pub message: String,
1047    /// Response headers
1048    pub headers: HeaderMap,
1049    /// Optional structured error details
1050    pub details: Option<D>,
1051}
1052
1053/// Deprecated alias for `FlatbedRouteError`
1054///
1055/// Use `FlatbedRouteError` instead for new code.
1056#[deprecated(since = "0.2.0", note = "Use FlatbedRouteError instead")]
1057pub type FlatbedError<D = ()> = FlatbedRouteError<D>;
1058
1059impl FlatbedRouteError<()> {
1060    /// Create a 400 Bad Request error
1061    pub fn bad_request(message: impl Into<String>) -> Self {
1062        Self::with_status(StatusCode::BAD_REQUEST, message)
1063    }
1064
1065    /// Create a 401 Unauthorized error
1066    pub fn unauthorized(message: impl Into<String>) -> Self {
1067        Self::with_status(StatusCode::UNAUTHORIZED, message)
1068    }
1069
1070    /// Create a 403 Forbidden error
1071    pub fn forbidden(message: impl Into<String>) -> Self {
1072        Self::with_status(StatusCode::FORBIDDEN, message)
1073    }
1074
1075    /// Create a 404 Not Found error
1076    pub fn not_found(message: impl Into<String>) -> Self {
1077        Self::with_status(StatusCode::NOT_FOUND, message)
1078    }
1079
1080    /// Create a 409 Conflict error
1081    pub fn conflict(message: impl Into<String>) -> Self {
1082        Self::with_status(StatusCode::CONFLICT, message)
1083    }
1084
1085    /// Create a 422 Unprocessable Entity error
1086    pub fn unprocessable(message: impl Into<String>) -> Self {
1087        Self::with_status(StatusCode::UNPROCESSABLE_ENTITY, message)
1088    }
1089
1090    /// Create a 500 Internal Server Error
1091    pub fn internal(message: impl Into<String>) -> Self {
1092        Self::with_status(StatusCode::INTERNAL_SERVER_ERROR, message)
1093    }
1094
1095    /// Create an error with custom status code
1096    pub fn with_status(status: StatusCode, message: impl Into<String>) -> Self {
1097        Self {
1098            status,
1099            code: String::new(),
1100            message: message.into(),
1101            headers: HeaderMap::new(),
1102            details: None,
1103        }
1104    }
1105}
1106
1107impl<D> FlatbedRouteError<D> {
1108    /// Set the error code (builder pattern)
1109    pub fn code(mut self, code: impl Into<String>) -> Self {
1110        self.code = code.into();
1111        self
1112    }
1113
1114    /// Add a header to the error response (builder pattern)
1115    pub fn header(mut self, key: impl AsRef<str>, value: impl AsRef<str>) -> Self {
1116        if let (Ok(name), Ok(val)) = (
1117            HeaderName::try_from(key.as_ref()),
1118            HeaderValue::try_from(value.as_ref()),
1119        ) {
1120            self.headers.insert(name, val);
1121        }
1122        self
1123    }
1124
1125    /// Add structured details to the error, changing the type parameter
1126    pub fn with_details<D2>(self, details: D2) -> FlatbedRouteError<D2> {
1127        FlatbedRouteError {
1128            status: self.status,
1129            code: self.code,
1130            message: self.message,
1131            headers: self.headers,
1132            details: Some(details),
1133        }
1134    }
1135
1136    /// Get the HTTP status code
1137    pub fn status_code(&self) -> StatusCode {
1138        self.status
1139    }
1140}
1141
1142impl<D> fmt::Display for FlatbedRouteError<D> {
1143    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1144        if self.code.is_empty() {
1145            write!(f, "{}", self.message)
1146        } else {
1147            write!(f, "[{}] {}", self.code, self.message)
1148        }
1149    }
1150}
1151
1152impl<D: fmt::Debug> std::error::Error for FlatbedRouteError<D> {}
1153
1154/// Parts extracted from an HTTP request (used internally by macro)
1155#[derive(Debug, Clone)]
1156pub struct RequestParts {
1157    /// HTTP headers
1158    pub headers: HeaderMap,
1159    /// HTTP method
1160    pub method: Method,
1161    /// Request path
1162    pub path: String,
1163    /// Path parameters
1164    pub path_params: HashMap<String, String>,
1165    /// Query parameters
1166    pub query_params: HashMap<String, String>,
1167    /// Request ID (propagated or generated)
1168    pub request_id: String,
1169}
1170
1171impl RequestParts {
1172    /// Create new RequestParts with generated request ID
1173    pub fn new(method: Method, path: String) -> Self {
1174        Self {
1175            headers: HeaderMap::new(),
1176            method,
1177            path,
1178            path_params: HashMap::new(),
1179            query_params: HashMap::new(),
1180            request_id: uuid::Uuid::new_v4().to_string(),
1181        }
1182    }
1183
1184    /// Set request ID from header or keep generated one
1185    pub fn with_request_id_from_header(mut self) -> Self {
1186        if let Some(id) = self
1187            .headers
1188            .get("x-request-id")
1189            .and_then(|v| v.to_str().ok())
1190        {
1191            self.request_id = id.to_string();
1192        }
1193        self
1194    }
1195
1196    /// Get a header value by name (case-insensitive)
1197    pub fn header(&self, name: &str) -> Option<&str> {
1198        self.headers.get(name).and_then(|v| v.to_str().ok())
1199    }
1200}
1201
1202/// Parts for constructing an HTTP response (used internally by macro)
1203#[derive(Debug)]
1204pub struct ResponseParts {
1205    /// Response body bytes
1206    pub body: Vec<u8>,
1207    /// HTTP status code
1208    pub status: StatusCode,
1209    /// Response headers
1210    pub headers: HeaderMap,
1211    /// Content-Type header value
1212    pub content_type: Cow<'static, str>,
1213}
1214
1215impl ResponseParts {
1216    /// Create new ResponseParts with 200 OK status
1217    pub fn ok(body: Vec<u8>, content_type: impl Into<Cow<'static, str>>) -> Self {
1218        Self {
1219            body,
1220            status: StatusCode::OK,
1221            headers: HeaderMap::new(),
1222            content_type: content_type.into(),
1223        }
1224    }
1225
1226    /// Create new ResponseParts with custom status
1227    pub fn with_status(
1228        body: Vec<u8>,
1229        status: StatusCode,
1230        content_type: impl Into<Cow<'static, str>>,
1231    ) -> Self {
1232        Self {
1233            body,
1234            status,
1235            headers: HeaderMap::new(),
1236            content_type: content_type.into(),
1237        }
1238    }
1239
1240    /// Add X-Request-ID header
1241    pub fn with_request_id(mut self, request_id: &str) -> Self {
1242        if let Ok(val) = HeaderValue::try_from(request_id) {
1243            self.headers.insert("x-request-id", val);
1244        }
1245        self
1246    }
1247}
1248
1249// ============================================================================
1250// Legacy Error Type
1251// ============================================================================
1252
1253/// Error type for flatbed operations
1254#[derive(Debug, Clone)]
1255pub enum Error {
1256    /// Failed to deserialize FlatBuffer
1257    DeserializationError(String),
1258    /// Failed to serialize FlatBuffer
1259    SerializationError(String),
1260    /// Handler returned an error
1261    HandlerError(String),
1262    /// Custom error with message
1263    Custom(String),
1264}
1265
1266impl fmt::Display for Error {
1267    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1268        match self {
1269            Error::DeserializationError(msg) => write!(f, "Deserialization error: {}", msg),
1270            Error::SerializationError(msg) => write!(f, "Serialization error: {}", msg),
1271            Error::HandlerError(msg) => write!(f, "Handler error: {}", msg),
1272            Error::Custom(msg) => write!(f, "{}", msg),
1273        }
1274    }
1275}
1276
1277impl std::error::Error for Error {}
1278
1279impl From<flatbuffers::InvalidFlatbuffer> for Error {
1280    fn from(err: flatbuffers::InvalidFlatbuffer) -> Self {
1281        Error::DeserializationError(err.to_string())
1282    }
1283}
1284
1285impl From<String> for Error {
1286    fn from(msg: String) -> Self {
1287        Error::Custom(msg)
1288    }
1289}
1290
1291impl From<&str> for Error {
1292    fn from(msg: &str) -> Self {
1293        Error::Custom(msg.to_string())
1294    }
1295}
1296
1297/// Legacy response wrapper that contains a FlatBufferBuilder with a finalized response of type T
1298///
1299/// **Deprecated**: Use the new `Response<T>` struct instead, which works with plain structs.
1300///
1301/// This allows handlers to have type-safe signatures and include HTTP metadata:
1302/// ```ignore
1303/// fn handle(req: PingRequest) -> Result<LegacyResponse<PingResponse>, Error> {
1304///     // Build response...
1305///     LegacyResponse::new(builder)
1306///         .with_status(200)
1307///         .with_header("Content-Type", "application/octet-stream")
1308/// }
1309/// ```
1310///
1311/// The generic parameter T documents the FlatBuffer response type.
1312#[deprecated(since = "0.2.0", note = "Use the new Response<T> struct instead")]
1313pub struct LegacyResponse<T> {
1314    builder: FlatBufferBuilder<'static>,
1315    status_code: u16,
1316    headers: HashMap<String, String>,
1317    _phantom: std::marker::PhantomData<T>,
1318}
1319
1320#[allow(deprecated)]
1321impl<T> LegacyResponse<T> {
1322    /// Create a new Response from a FlatBufferBuilder that has been finalized
1323    ///
1324    /// The builder should have already called `finish()` on the response of type T.
1325    /// Default status code is 200 OK.
1326    ///
1327    /// # Example
1328    /// ```ignore
1329    /// let mut builder = FlatBufferBuilder::new();
1330    /// let response = PingResponse::create(&mut builder, &args);
1331    /// builder.finish(response, None);
1332    /// Response::<PingResponse>::new(builder)
1333    /// ```
1334    pub fn new(builder: FlatBufferBuilder<'static>) -> Self {
1335        Self {
1336            builder,
1337            status_code: 200,
1338            headers: HashMap::new(),
1339            _phantom: std::marker::PhantomData,
1340        }
1341    }
1342
1343    /// Create a Response by providing a closure that builds the response
1344    ///
1345    /// This encapsulates all the builder boilerplate. The closure receives a mutable
1346    /// reference to a FlatBufferBuilder and should create and finish the response.
1347    ///
1348    /// # Example
1349    /// ```ignore
1350    /// Response::with_body(|builder| {
1351    ///     let msg = builder.create_string("Hello");
1352    ///     let response = PingResponse::create(builder, &PingResponseArgs {
1353    ///         message: Some(msg),
1354    ///         timestamp: 123,
1355    ///         success: true,
1356    ///     });
1357    ///     builder.finish(response, None);
1358    /// })
1359    /// ```
1360    pub fn with_body<F>(f: F) -> Self
1361    where
1362        F: FnOnce(&mut FlatBufferBuilder<'static>),
1363    {
1364        let mut builder = FlatBufferBuilder::new();
1365        f(&mut builder);
1366        Self::new(builder)
1367    }
1368
1369    /// Set the HTTP status code for this response
1370    ///
1371    /// # Example
1372    /// ```ignore
1373    /// Response::new(builder).with_status(201)  // Created
1374    /// Response::new(builder).with_status(404)  // Not Found
1375    /// ```
1376    pub fn with_status(mut self, status: u16) -> Self {
1377        self.status_code = status;
1378        self
1379    }
1380
1381    /// Add an HTTP header to this response
1382    ///
1383    /// # Example
1384    /// ```ignore
1385    /// Response::new(builder)
1386    ///     .with_header("Content-Type", "application/octet-stream")
1387    ///     .with_header("X-Request-ID", "123")
1388    /// ```
1389    pub fn with_header(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
1390        self.headers.insert(key.into(), value.into());
1391        self
1392    }
1393
1394    /// Get the HTTP status code
1395    pub fn status_code(&self) -> u16 {
1396        self.status_code
1397    }
1398
1399    /// Get a reference to the headers
1400    pub fn headers(&self) -> &HashMap<String, String> {
1401        &self.headers
1402    }
1403
1404    /// Get the serialized bytes from the response
1405    pub fn into_bytes(self) -> Vec<u8> {
1406        self.builder.finished_data().to_vec()
1407    }
1408
1409    /// Get a reference to the serialized bytes
1410    pub fn as_bytes(&self) -> &[u8] {
1411        self.builder.finished_data()
1412    }
1413
1414    /// Consume self and return (bytes, status_code, headers)
1415    pub fn into_parts(self) -> (Vec<u8>, u16, HashMap<String, String>) {
1416        let bytes = self.builder.finished_data().to_vec();
1417        (bytes, self.status_code, self.headers)
1418    }
1419}
1420
1421/// Creates a new FlatBufferBuilder with the default capacity
1422pub fn new_builder() -> FlatBufferBuilder<'static> {
1423    FlatBufferBuilder::new()
1424}
1425
1426/// Creates a new FlatBufferBuilder with a specific capacity
1427pub fn new_builder_with_capacity(capacity: usize) -> FlatBufferBuilder<'static> {
1428    FlatBufferBuilder::with_capacity(capacity)
1429}
1430
1431/// Verifies a flatbuffer message
1432///
1433/// Returns true if the buffer is valid, false otherwise
1434pub fn verify_buffer<'a, T>(buf: &'a [u8]) -> bool
1435where
1436    T: flatbuffers::Follow<'a> + flatbuffers::Verifiable + 'a,
1437{
1438    flatbuffers::root::<T>(buf).is_ok()
1439}
1440
1441/// Safely reads a root table from a buffer
1442///
1443/// This performs verification before returning the root
1444pub fn get_root<'a, T>(
1445    buf: &'a [u8],
1446) -> Result<<T as flatbuffers::Follow<'a>>::Inner, flatbuffers::InvalidFlatbuffer>
1447where
1448    T: flatbuffers::Follow<'a> + flatbuffers::Verifiable + 'a,
1449{
1450    flatbuffers::root::<T>(buf)
1451}
1452
1453/// Handler function type for route registration
1454///
1455/// This represents the signature of handler functions that can be used with
1456/// the routing system. Handlers take raw bytes and content-type string,
1457/// and return a Result with response bytes and response content-type.
1458///
1459/// Supports both `application/json` and `application/x-flatbuffers` content types.
1460/// The response content-type matches the request content-type.
1461///
1462/// Note: The actual user-defined handlers return Result<Response, E>
1463/// where E can be converted to flatbed::Error. The macro wrapper handles
1464/// content-type negotiation and serialization/deserialization.
1465pub type HandlerFn = fn(Vec<u8>, &str) -> Result<(Vec<u8>, &'static str), Error>;
1466
1467/// A wire encoding a route speaks.
1468#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1469pub enum Codec {
1470    /// `application/json`.
1471    Json,
1472    /// `application/x-flatbuffers`.
1473    FlatBuffer,
1474}
1475
1476/// The codec a request's `accept` header asks the response to use, when it
1477/// names one flatbed serves. Both named, the earlier wins; neither named,
1478/// or no header, is `None` and the response follows the request's
1479/// `content-type`. A body is still decoded by `content-type` alone.
1480#[must_use]
1481pub fn accepted_codec(headers: &HeaderMap) -> Option<Codec> {
1482    let accept = headers.get("accept")?.to_str().ok()?;
1483    let json = accept.find("application/json");
1484    let flatbuffer = accept
1485        .find("application/x-flatbuffers")
1486        .or_else(|| accept.find("application/x-flat-buffers"));
1487    match (json, flatbuffer) {
1488        (Some(j), Some(f)) if f < j => Some(Codec::FlatBuffer),
1489        (Some(_), _) => Some(Codec::Json),
1490        (None, Some(_)) => Some(Codec::FlatBuffer),
1491        (None, None) => None,
1492    }
1493}
1494
1495/// Async handler function type for hyper integration
1496///
1497/// Takes request parts, body bytes, content-type, application context, and
1498/// returns response parts. The context is passed as `Arc<dyn Any>` and
1499/// downcasted to the concrete type by the `#[route]` macro wrapper.
1500pub type AsyncHandlerFn = fn(
1501    RequestParts,
1502    Vec<u8>,
1503    &str,
1504    std::sync::Arc<dyn std::any::Any + Send + Sync>,
1505) -> std::pin::Pin<
1506    Box<dyn std::future::Future<Output = Result<ResponseParts, Error>> + Send>,
1507>;
1508
1509/// HTTP method for route registration
1510#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
1511pub enum HttpMethod {
1512    Get,
1513    Post,
1514    Put,
1515    Delete,
1516    Patch,
1517    Head,
1518    Options,
1519}
1520
1521impl HttpMethod {
1522    /// Returns the method as an uppercase string
1523    pub fn as_str(&self) -> &'static str {
1524        match self {
1525            HttpMethod::Get => "GET",
1526            HttpMethod::Post => "POST",
1527            HttpMethod::Put => "PUT",
1528            HttpMethod::Delete => "DELETE",
1529            HttpMethod::Patch => "PATCH",
1530            HttpMethod::Head => "HEAD",
1531            HttpMethod::Options => "OPTIONS",
1532        }
1533    }
1534}
1535
1536impl std::fmt::Display for HttpMethod {
1537    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1538        f.write_str(self.as_str())
1539    }
1540}
1541
1542/// Route information collected from #[route] decorators
1543#[derive(Clone, Copy)]
1544pub struct RouteInfo {
1545    pub path: &'static str,
1546    pub method: HttpMethod,
1547    pub request_type: &'static str,
1548    pub response_type: &'static str,
1549    pub handler: HandlerFn,
1550    pub async_handler: AsyncHandlerFn,
1551    pub openapi: OpenApiRouteInfo,
1552}
1553
1554// Manual Debug implementation to avoid printing function pointers
1555impl std::fmt::Debug for RouteInfo {
1556    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1557        f.debug_struct("RouteInfo")
1558            .field("path", &self.path)
1559            .field("method", &self.method)
1560            .field("request_type", &self.request_type)
1561            .field("response_type", &self.response_type)
1562            .field("handler", &"<function>")
1563            .field("async_handler", &"<async_function>")
1564            .field("openapi", &self.openapi)
1565            .finish()
1566    }
1567}
1568
1569inventory::collect!(RouteInfo);
1570
1571/// Where a static mount reads its files from.
1572#[derive(Clone, Copy, Debug)]
1573pub enum StaticSource {
1574    /// A directory on the container filesystem, read at request time
1575    /// (e.g. `/app/dist`).
1576    Dir(&'static str),
1577    /// A directory compiled into the binary, served from memory.
1578    Embedded(&'static include_dir::Dir<'static>),
1579}
1580
1581/// A static-file mount registered by the `static_route!` macro.
1582///
1583/// Serves files from `source` under the `mount` URL prefix. Declared
1584/// `#[route]` routes always take precedence, so an API mounted under `/api`
1585/// keeps working alongside a static mount at `/`.
1586#[derive(Clone, Copy, Debug)]
1587pub struct StaticRouteInfo {
1588    /// URL prefix the mount answers under (e.g. `/` or `/assets`).
1589    pub mount: &'static str,
1590    /// The directory the files come from, on disk or in the binary.
1591    pub source: StaticSource,
1592    /// File served for unmatched sub-paths, enabling SPA history fallback
1593    /// (e.g. `index.html`). `None` returns 404 for a miss.
1594    pub fallback: Option<&'static str>,
1595    /// Mount-relative path prefixes under which a miss is a 404 and never
1596    /// the fallback, so a mistyped API path is not answered with the page
1597    /// shell (e.g. `&["/api/"]`, which under `mount = "/app"` covers
1598    /// `/app/api/...`).
1599    pub no_fallback: &'static [&'static str],
1600}
1601
1602inventory::collect!(StaticRouteInfo);
1603
1604/// Returns the static-file mounts registered via `static_route!`.
1605#[must_use]
1606pub fn get_static_routes() -> Vec<StaticRouteInfo> {
1607    inventory::iter::<StaticRouteInfo>
1608        .into_iter()
1609        .copied()
1610        .collect()
1611}
1612
1613// ============================================================================
1614// Worker Registration Types
1615// ============================================================================
1616
1617/// Worker function signature for background tasks
1618///
1619/// Workers receive the application context wrapped as `Arc<dyn Any + Send + Sync>`.
1620/// The `#[worker]` macro generates a wrapper that downcasts to the correct concrete type.
1621pub type WorkerFn = fn(
1622    std::sync::Arc<dyn std::any::Any + Send + Sync>,
1623) -> std::pin::Pin<
1624    Box<dyn std::future::Future<Output = Result<(), FlatbedWorkerError>> + Send>,
1625>;
1626
1627/// Drain function signature for graceful shutdown of workers.
1628///
1629/// Called during shutdown to allow workers to finish in-progress work.
1630/// Receives the application context and returns a future that completes
1631/// when draining is done.
1632pub type DrainFn = fn(
1633    std::sync::Arc<dyn std::any::Any + Send + Sync>,
1634) -> std::pin::Pin<
1635    Box<dyn std::future::Future<Output = Result<(), FlatbedWorkerError>> + Send>,
1636>;
1637
1638/// Worker metadata registered via the `register_*!` macros
1639///
1640/// This struct is collected via the `inventory` crate at runtime.
1641#[derive(Clone, Copy)]
1642pub struct WorkerInfo {
1643    /// Worker name (from attribute or function name)
1644    pub name: &'static str,
1645    /// Optional description of what the worker does
1646    pub description: Option<&'static str>,
1647    /// The worker function that receives the context
1648    pub worker: WorkerFn,
1649    /// Opt-in in-process restart policy. `None` makes the worker's first
1650    /// exit terminal.
1651    pub restart: Option<RestartPolicy>,
1652}
1653
1654// Manual Debug implementation to avoid printing function pointers
1655impl std::fmt::Debug for WorkerInfo {
1656    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1657        f.debug_struct("WorkerInfo")
1658            .field("name", &self.name)
1659            .field("description", &self.description)
1660            .field("worker", &"<worker_fn>")
1661            .field("restart", &self.restart)
1662            .finish()
1663    }
1664}
1665
1666/// Drain function metadata submitted by the `register_*!` macros alongside
1667/// the worker itself.
1668///
1669/// The shutdown path calls every registered drain once the server has stopped
1670/// accepting connections, so a worker trait's `drain` implementation is the
1671/// hook for finishing in-progress work. Workers that don't override it drain
1672/// as a no-op.
1673#[derive(Clone, Copy)]
1674pub struct WorkerDrainInfo {
1675    /// Worker name (must match the corresponding WorkerInfo name)
1676    pub name: &'static str,
1677    /// The drain function called during graceful shutdown
1678    pub drain: DrainFn,
1679}
1680
1681inventory::collect!(WorkerDrainInfo);
1682
1683/// Get all registered drain functions.
1684#[must_use]
1685pub fn get_worker_drains() -> Vec<&'static WorkerDrainInfo> {
1686    inventory::iter::<WorkerDrainInfo>.into_iter().collect()
1687}
1688
1689inventory::collect!(WorkerInfo);
1690
1691/// Get all registered workers
1692///
1693/// Returns a vector of all workers registered via `#[worker]` macros.
1694///
1695/// # Example
1696///
1697/// ```rust,ignore
1698/// for worker in get_workers() {
1699///     println!("Worker: {} - {:?}", worker.name, worker.description);
1700/// }
1701/// ```
1702#[must_use]
1703pub fn get_workers() -> Vec<&'static WorkerInfo> {
1704    inventory::iter::<WorkerInfo>.into_iter().collect()
1705}
1706
1707// ============================================================================
1708// Worker Trait
1709// ============================================================================
1710
1711/// Trait-based background worker.
1712///
1713/// Implement this trait to define a simple background worker without proc macros.
1714/// The runtime executor [`run_basic_worker`] handles context downcasting —
1715/// matching the behavior of `#[worker]`.
1716///
1717/// # Example
1718///
1719/// ```rust,ignore
1720/// use flatbed::{Worker, FlatbedWorkerError};
1721/// use std::sync::Arc;
1722///
1723/// struct HealthChecker;
1724///
1725/// impl Worker for HealthChecker {
1726///     type Context = AppContext;
1727///     const NAME: &'static str = "health-checker";
1728///     const DESCRIPTION: Option<&'static str> = Some("Periodically checks health");
1729///
1730///     fn run(
1731///         &self,
1732///         ctx: Arc<Self::Context>,
1733///     ) -> flatbed::BoxFuture<Result<(), FlatbedWorkerError>> {
1734///         Box::pin(async move {
1735///             // worker logic
1736///             Ok(())
1737///         })
1738///     }
1739/// }
1740///
1741/// flatbed::register_worker!(HealthChecker, AppContext);
1742/// ```
1743pub trait Worker: Send + Sync + 'static {
1744    /// The application context type.
1745    type Context: Send + Sync + 'static;
1746
1747    /// Worker name used for logging and identification.
1748    const NAME: &'static str;
1749
1750    /// Optional description of what this worker does.
1751    const DESCRIPTION: Option<&'static str> = None;
1752
1753    /// Opt-in in-process restart policy. `None` — the default — makes the
1754    /// worker's first exit terminal, leaving Kubernetes to restart the pod.
1755    /// Registration puts this in a `static` initializer, so the expression
1756    /// has to be const-evaluatable; the [`RestartPolicy`] builders are
1757    /// `const fn`.
1758    ///
1759    /// ```rust,ignore
1760    /// const RESTART: Option<flatbed::RestartPolicy> = Some(
1761    ///     flatbed::RestartPolicy::backoff(
1762    ///         std::time::Duration::from_secs(1),
1763    ///         std::time::Duration::from_secs(60),
1764    ///     ),
1765    /// );
1766    /// ```
1767    const RESTART: ::core::option::Option<RestartPolicy> = ::core::option::Option::None;
1768
1769    /// Run the worker. Called once after the server is ready, and expected to
1770    /// keep running for the life of the process — returning `Ok(())` marks the
1771    /// process unhealthy, and returning `Err` or panicking additionally starts
1772    /// graceful shutdown. One-shot initialisation belongs in the boot function
1773    /// passed to [`Flatbed::run`], not in a worker.
1774    fn run(&self, ctx: std::sync::Arc<Self::Context>) -> BoxFuture<Result<(), FlatbedWorkerError>>;
1775
1776    /// Finish in-progress work during graceful shutdown, within the
1777    /// configured shutdown budget. Defaults to a no-op.
1778    fn drain(
1779        &self,
1780        _ctx: std::sync::Arc<Self::Context>,
1781    ) -> BoxFuture<Result<(), FlatbedWorkerError>> {
1782        Box::pin(async { Ok(()) })
1783    }
1784}
1785
1786/// Run a [`Worker`] by downcasting the context and delegating to `Worker::run`.
1787pub fn run_basic_worker<W, C>(
1788    ctx: std::sync::Arc<dyn std::any::Any + Send + Sync>,
1789) -> BoxFuture<Result<(), FlatbedWorkerError>>
1790where
1791    W: Worker<Context = C> + Default,
1792    C: Send + Sync + 'static,
1793{
1794    Box::pin(async move {
1795        let ctx: std::sync::Arc<C> = ctx
1796            .downcast::<C>()
1797            .unwrap_or_else(|_| panic!("worker '{}' context type mismatch", W::NAME));
1798        let worker = W::default();
1799        worker.run(ctx).await
1800    })
1801}
1802
1803/// Drain a [`Worker`] by downcasting the context and delegating to
1804/// `Worker::drain`.
1805pub fn run_basic_worker_drain<W, C>(
1806    ctx: std::sync::Arc<dyn std::any::Any + Send + Sync>,
1807) -> BoxFuture<Result<(), FlatbedWorkerError>>
1808where
1809    W: Worker<Context = C> + Default,
1810    C: Send + Sync + 'static,
1811{
1812    Box::pin(async move {
1813        let ctx: std::sync::Arc<C> = ctx
1814            .downcast::<C>()
1815            .unwrap_or_else(|_| panic!("worker '{}' context type mismatch", W::NAME));
1816        W::default().drain(ctx).await
1817    })
1818}
1819
1820// ============================================================================
1821// Registration Macros
1822// ============================================================================
1823
1824/// Register a [`KubeReconciler`] implementor with the Flatbed worker system.
1825///
1826/// Submits a [`WorkerInfo`] whose worker function delegates to
1827/// [`run_kube_reconciler`], plus a [`WorkerDrainInfo`] wired to the
1828/// reconciler's `drain`.
1829///
1830/// # Usage
1831///
1832/// ```rust,ignore
1833/// struct MyReconciler;
1834/// impl flatbed::KubeReconciler for MyReconciler { /* ... */ }
1835///
1836/// flatbed::register_kube_reconciler!(MyReconciler, AppContext);
1837/// ```
1838///
1839/// The worker's `RESTART` associated const decides whether it is restarted
1840/// in place; the macro takes no configuration of its own.
1841#[cfg(all(feature = "nats", feature = "k8s"))]
1842#[macro_export]
1843macro_rules! register_kube_reconciler {
1844    ($reconciler:ty, $context:ty) => {
1845        $crate::inventory::submit! {
1846            $crate::WorkerInfo {
1847                name: <$reconciler as $crate::k8s::KubeReconciler>::NAME,
1848                description: <$reconciler as $crate::k8s::KubeReconciler>::DESCRIPTION,
1849                restart: <$reconciler as $crate::k8s::KubeReconciler>::RESTART,
1850                worker: {
1851                    fn __worker(
1852                        ctx: ::std::sync::Arc<dyn ::std::any::Any + Send + Sync>,
1853                    ) -> ::std::pin::Pin<
1854                        Box<dyn ::std::future::Future<Output = Result<(), $crate::FlatbedWorkerError>> + Send>,
1855                    > {
1856                        Box::pin($crate::k8s::run_kube_reconciler::<$reconciler, $context>(ctx))
1857                    }
1858                    __worker
1859                },
1860            }
1861        }
1862        $crate::inventory::submit! {
1863            $crate::WorkerDrainInfo {
1864                name: <$reconciler as $crate::k8s::KubeReconciler>::NAME,
1865                drain: {
1866                    fn __drain(
1867                        ctx: ::std::sync::Arc<dyn ::std::any::Any + Send + Sync>,
1868                    ) -> ::std::pin::Pin<
1869                        Box<dyn ::std::future::Future<Output = Result<(), $crate::FlatbedWorkerError>> + Send>,
1870                    > {
1871                        $crate::k8s::run_kube_reconciler_drain::<$reconciler, $context>(ctx)
1872                    }
1873                    __drain
1874                },
1875            }
1876        }
1877    };
1878}
1879
1880/// Register a [`KubeNativeReconciler`] implementor with the Flatbed
1881/// worker system.
1882///
1883/// Submits a [`WorkerInfo`] whose worker function delegates to
1884/// [`run_kube_native_reconciler`], plus a [`WorkerDrainInfo`] wired to the
1885/// reconciler's `drain`. Use this for reconcilers whose contexts implement
1886/// `HasKubeClient + HasLeaderElection` only — no JetStream context bound
1887/// required.
1888///
1889/// # Usage
1890///
1891/// ```rust,ignore
1892/// #[derive(Default)]
1893/// struct MyReconciler;
1894/// impl flatbed::KubeNativeReconciler for MyReconciler { /* ... */ }
1895///
1896/// flatbed::register_kube_native_reconciler!(MyReconciler, AppContext);
1897/// ```
1898///
1899/// The worker's `RESTART` associated const decides whether it is restarted
1900/// in place; the macro takes no configuration of its own.
1901///
1902/// [`KubeNativeReconciler`]: crate::k8s::KubeNativeReconciler
1903/// [`run_kube_native_reconciler`]: crate::k8s::run_kube_native_reconciler
1904#[cfg(feature = "k8s")]
1905#[macro_export]
1906macro_rules! register_kube_native_reconciler {
1907    ($reconciler:ty, $context:ty) => {
1908        $crate::inventory::submit! {
1909            $crate::WorkerInfo {
1910                name: <$reconciler as $crate::k8s::KubeNativeReconciler>::NAME,
1911                description: <$reconciler as $crate::k8s::KubeNativeReconciler>::DESCRIPTION,
1912                restart: <$reconciler as $crate::k8s::KubeNativeReconciler>::RESTART,
1913                worker: {
1914                    fn __worker(
1915                        ctx: ::std::sync::Arc<dyn ::std::any::Any + Send + Sync>,
1916                    ) -> ::std::pin::Pin<
1917                        Box<dyn ::std::future::Future<Output = Result<(), $crate::FlatbedWorkerError>> + Send>,
1918                    > {
1919                        Box::pin($crate::k8s::run_kube_native_reconciler::<$reconciler, $context>(ctx))
1920                    }
1921                    __worker
1922                },
1923            }
1924        }
1925        $crate::inventory::submit! {
1926            $crate::WorkerDrainInfo {
1927                name: <$reconciler as $crate::k8s::KubeNativeReconciler>::NAME,
1928                drain: {
1929                    fn __drain(
1930                        ctx: ::std::sync::Arc<dyn ::std::any::Any + Send + Sync>,
1931                    ) -> ::std::pin::Pin<
1932                        Box<dyn ::std::future::Future<Output = Result<(), $crate::FlatbedWorkerError>> + Send>,
1933                    > {
1934                        $crate::k8s::run_kube_native_reconciler_drain::<$reconciler, $context>(ctx)
1935                    }
1936                    __drain
1937                },
1938            }
1939        }
1940    };
1941}
1942
1943/// Register a [`KubeWatcher`] implementor with the Flatbed worker system.
1944///
1945/// Mirrors `register_kube_reconciler!` but uses the watcher-based executor
1946/// — the right primitive when the watched resource is owned by
1947/// kube-controller-manager (no finalizer possible) and Delete events
1948/// matter for correctness.
1949///
1950/// # Requirements
1951///
1952/// The implementor must also implement `Default` — the executor calls
1953/// `R::default()` once per executor lifetime. State that needs to
1954/// reset per-watch-burst should reset inside `on_init` (which fires
1955/// on every list+watch, including reopens after a transient error).
1956///
1957/// # Usage
1958///
1959/// ```rust,ignore
1960/// #[derive(Default)]
1961/// struct EndpointWatcher { /* per-leader state behind Arc<Mutex<..>> */ }
1962/// impl flatbed::KubeWatcher for EndpointWatcher { /* ... */ }
1963///
1964/// flatbed::register_kube_watcher!(EndpointWatcher, AppContext);
1965/// ```
1966///
1967/// The worker's `RESTART` associated const decides whether it is restarted
1968/// in place; the macro takes no configuration of its own.
1969///
1970/// [`KubeWatcher`]: crate::k8s::KubeWatcher
1971#[cfg(feature = "k8s")]
1972#[macro_export]
1973macro_rules! register_kube_watcher {
1974    ($watcher:ty, $context:ty) => {
1975        $crate::inventory::submit! {
1976            $crate::WorkerInfo {
1977                name: <$watcher as $crate::k8s::KubeWatcher>::NAME,
1978                description: <$watcher as $crate::k8s::KubeWatcher>::DESCRIPTION,
1979                restart: <$watcher as $crate::k8s::KubeWatcher>::RESTART,
1980                worker: {
1981                    fn __worker(
1982                        ctx: ::std::sync::Arc<dyn ::std::any::Any + Send + Sync>,
1983                    ) -> ::std::pin::Pin<
1984                        Box<dyn ::std::future::Future<Output = Result<(), $crate::FlatbedWorkerError>> + Send>,
1985                    > {
1986                        Box::pin($crate::k8s::run_kube_watcher::<$watcher, $context>(ctx))
1987                    }
1988                    __worker
1989                },
1990            }
1991        }
1992        $crate::inventory::submit! {
1993            $crate::WorkerDrainInfo {
1994                name: <$watcher as $crate::k8s::KubeWatcher>::NAME,
1995                drain: {
1996                    fn __drain(
1997                        ctx: ::std::sync::Arc<dyn ::std::any::Any + Send + Sync>,
1998                    ) -> ::std::pin::Pin<
1999                        Box<dyn ::std::future::Future<Output = Result<(), $crate::FlatbedWorkerError>> + Send>,
2000                    > {
2001                        $crate::k8s::run_kube_watcher_drain::<$watcher, $context>(ctx)
2002                    }
2003                    __drain
2004                },
2005            }
2006        }
2007    };
2008}
2009
2010/// Register a [`StreamWorker`] implementor with the Flatbed worker system.
2011///
2012/// Submits a [`WorkerInfo`] whose worker function delegates to
2013/// [`run_stream_worker`], plus a [`WorkerDrainInfo`] wired to the worker's
2014/// `drain`.
2015///
2016/// # Usage
2017///
2018/// ```rust,ignore
2019/// struct MyWorker;
2020/// impl flatbed::StreamWorker for MyWorker { /* ... */ }
2021///
2022/// flatbed::register_stream_worker!(MyWorker, AppContext);
2023/// ```
2024///
2025/// The worker's `RESTART` associated const decides whether it is restarted
2026/// in place; the macro takes no configuration of its own.
2027///
2028/// [`StreamWorker`]: crate::nats::StreamWorker
2029/// [`run_stream_worker`]: crate::nats::run_stream_worker
2030#[cfg(feature = "nats")]
2031#[macro_export]
2032macro_rules! register_stream_worker {
2033    ($stream_worker:ty, $context:ty) => {
2034        $crate::inventory::submit! {
2035            $crate::WorkerInfo {
2036                name: <$stream_worker as $crate::nats::StreamWorker>::NAME,
2037                description: <$stream_worker as $crate::nats::StreamWorker>::DESCRIPTION,
2038                restart: <$stream_worker as $crate::nats::StreamWorker>::RESTART,
2039                worker: {
2040                    fn __worker(
2041                        ctx: ::std::sync::Arc<dyn ::std::any::Any + Send + Sync>,
2042                    ) -> ::std::pin::Pin<
2043                        Box<dyn ::std::future::Future<Output = Result<(), $crate::FlatbedWorkerError>> + Send>,
2044                    > {
2045                        Box::pin($crate::nats::run_stream_worker::<$stream_worker, $context>(ctx))
2046                    }
2047                    __worker
2048                },
2049            }
2050        }
2051        $crate::inventory::submit! {
2052            $crate::WorkerDrainInfo {
2053                name: <$stream_worker as $crate::nats::StreamWorker>::NAME,
2054                drain: {
2055                    fn __drain(
2056                        ctx: ::std::sync::Arc<dyn ::std::any::Any + Send + Sync>,
2057                    ) -> ::std::pin::Pin<
2058                        Box<dyn ::std::future::Future<Output = Result<(), $crate::FlatbedWorkerError>> + Send>,
2059                    > {
2060                        $crate::nats::run_stream_worker_drain::<$stream_worker, $context>(ctx)
2061                    }
2062                    __drain
2063                },
2064            }
2065        }
2066    };
2067}
2068
2069/// Register a [`KvWorker`] implementor with the Flatbed worker system.
2070///
2071/// Submits a [`WorkerInfo`] whose worker function delegates to
2072/// [`run_kv_worker`], plus a [`WorkerDrainInfo`] wired to the worker's
2073/// `drain`.
2074///
2075/// # Requirements
2076///
2077/// The implementor must also implement `Default` — the executor calls
2078/// `W::default()` once at spawn. `#[derive(Default)]` on the impl
2079/// struct is usually enough.
2080///
2081/// # Usage
2082///
2083/// ```rust,ignore
2084/// #[derive(Default)]
2085/// struct CacheSubscriber;
2086/// impl flatbed::KvWorker for CacheSubscriber { /* ... */ }
2087///
2088/// flatbed::register_kv_worker!(CacheSubscriber, AppContext);
2089/// ```
2090///
2091/// The worker's `RESTART` associated const decides whether it is restarted
2092/// in place; the macro takes no configuration of its own.
2093///
2094/// [`KvWorker`]: crate::kv::KvWorker
2095/// [`run_kv_worker`]: crate::kv::run_kv_worker
2096#[cfg(feature = "nats")]
2097#[macro_export]
2098macro_rules! register_kv_worker {
2099    ($kv_worker:ty, $context:ty) => {
2100        $crate::inventory::submit! {
2101            $crate::WorkerInfo {
2102                name: <$kv_worker as $crate::kv::KvWorker>::NAME,
2103                description: <$kv_worker as $crate::kv::KvWorker>::DESCRIPTION,
2104                restart: <$kv_worker as $crate::kv::KvWorker>::RESTART,
2105                worker: {
2106                    fn __worker(
2107                        ctx: ::std::sync::Arc<dyn ::std::any::Any + Send + Sync>,
2108                    ) -> ::std::pin::Pin<
2109                        Box<dyn ::std::future::Future<Output = Result<(), $crate::FlatbedWorkerError>> + Send>,
2110                    > {
2111                        Box::pin($crate::kv::run_kv_worker::<$kv_worker, $context>(ctx))
2112                    }
2113                    __worker
2114                },
2115            }
2116        }
2117        $crate::inventory::submit! {
2118            $crate::WorkerDrainInfo {
2119                name: <$kv_worker as $crate::kv::KvWorker>::NAME,
2120                drain: {
2121                    fn __drain(
2122                        ctx: ::std::sync::Arc<dyn ::std::any::Any + Send + Sync>,
2123                    ) -> ::std::pin::Pin<
2124                        Box<dyn ::std::future::Future<Output = Result<(), $crate::FlatbedWorkerError>> + Send>,
2125                    > {
2126                        $crate::kv::run_kv_worker_drain::<$kv_worker, $context>(ctx)
2127                    }
2128                    __drain
2129                },
2130            }
2131        }
2132    };
2133}
2134
2135/// Register a [`Worker`] implementor with the Flatbed worker system.
2136///
2137/// Submits a [`WorkerInfo`] whose worker function delegates to
2138/// [`run_basic_worker`], plus a [`WorkerDrainInfo`] wired to the worker's
2139/// `drain`.
2140///
2141/// # Usage
2142///
2143/// ```rust,ignore
2144/// struct HealthChecker;
2145/// impl flatbed::Worker for HealthChecker { /* ... */ }
2146///
2147/// flatbed::register_worker!(HealthChecker, AppContext);
2148/// ```
2149///
2150/// Set the worker's [`RESTART`](Worker::RESTART) associated const to trade a
2151/// pod restart for an in-process retry with capped, jittered backoff; the
2152/// macro takes no configuration of its own.
2153#[macro_export]
2154macro_rules! register_worker {
2155    ($worker_type:ty, $context:ty) => {
2156        $crate::inventory::submit! {
2157            $crate::WorkerInfo {
2158                name: <$worker_type as $crate::Worker>::NAME,
2159                description: <$worker_type as $crate::Worker>::DESCRIPTION,
2160                restart: <$worker_type as $crate::Worker>::RESTART,
2161                worker: {
2162                    fn __worker(
2163                        ctx: ::std::sync::Arc<dyn ::std::any::Any + Send + Sync>,
2164                    ) -> ::std::pin::Pin<
2165                        Box<dyn ::std::future::Future<Output = Result<(), $crate::FlatbedWorkerError>> + Send>,
2166                    > {
2167                        $crate::run_basic_worker::<$worker_type, $context>(ctx)
2168                    }
2169                    __worker
2170                },
2171            }
2172        }
2173        $crate::inventory::submit! {
2174            $crate::WorkerDrainInfo {
2175                name: <$worker_type as $crate::Worker>::NAME,
2176                drain: {
2177                    fn __drain(
2178                        ctx: ::std::sync::Arc<dyn ::std::any::Any + Send + Sync>,
2179                    ) -> ::std::pin::Pin<
2180                        Box<dyn ::std::future::Future<Output = Result<(), $crate::FlatbedWorkerError>> + Send>,
2181                    > {
2182                        $crate::run_basic_worker_drain::<$worker_type, $context>(ctx)
2183                    }
2184                    __drain
2185                },
2186            }
2187        }
2188    };
2189}
2190
2191/// Route map type: (path, method) -> RouteInfo with handler
2192pub type RouteMap = HashMap<(String, HttpMethod), RouteInfo>;
2193
2194/// Validation error for route conflicts
2195#[derive(Debug, Clone, PartialEq, Eq)]
2196pub struct RouteConflict {
2197    pub path: String,
2198    pub method: HttpMethod,
2199    pub first_request: String,
2200    pub first_response: String,
2201    pub second_request: String,
2202    pub second_response: String,
2203}
2204
2205impl std::fmt::Display for RouteConflict {
2206    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
2207        write!(
2208            f,
2209            "Route conflict at {} '{}': ({}, {}) vs ({}, {})",
2210            self.method,
2211            self.path,
2212            self.first_request,
2213            self.first_response,
2214            self.second_request,
2215            self.second_response
2216        )
2217    }
2218}
2219
2220impl std::error::Error for RouteConflict {}
2221
2222/// Get the route map with all registered routes
2223///
2224/// Returns a HashMap mapping path strings to RouteInfo containing
2225/// handler function pointers and type information.
2226/// This is populated at runtime from all #[route] decorated functions.
2227///
2228/// # Example
2229/// ```rust,ignore
2230/// use flatbed::{route, get_routes};
2231///
2232/// #[route("/ping")]
2233/// fn handle_ping(req: Vec<u8>) -> Vec<u8> { ... }
2234///
2235/// let routes = get_routes();
2236/// let route_info = routes.get(&("/ping".to_string(), flatbed::HttpMethod::Post)).unwrap();
2237/// // Use route_info.handler to call the function
2238/// // Use route_info.request_type / response_type for metadata
2239/// ```
2240#[must_use]
2241pub fn get_routes() -> &'static RouteMap {
2242    static ROUTES: LazyLock<RouteMap> = LazyLock::new(|| {
2243        let mut map = HashMap::new();
2244        for route in inventory::iter::<RouteInfo> {
2245            map.insert((route.path.to_string(), route.method), *route);
2246        }
2247        map
2248    });
2249    &ROUTES
2250}
2251
2252/// Look up a registered [`TypeSchema`] by its bare type name.
2253///
2254/// Names are keyed bare (without the namespace), matching how routes reference
2255/// their request/response types. A bare name is therefore assumed unique across
2256/// namespaces; [`validate_type_registry`] enforces that at startup.
2257#[must_use]
2258pub fn get_type_schema(name: &str) -> Option<&'static TypeSchema> {
2259    static MAP: LazyLock<HashMap<&'static str, &'static TypeSchema>> = LazyLock::new(|| {
2260        let mut map = HashMap::new();
2261        for ty in inventory::iter::<TypeSchema> {
2262            map.insert(ty.name, ty);
2263        }
2264        map
2265    });
2266    MAP.get(name).copied()
2267}
2268
2269/// Look up a registered [`EnumSchema`] by its bare name.
2270///
2271/// Names are keyed bare (without the namespace); a bare name is assumed unique
2272/// across namespaces, which [`validate_type_registry`] enforces at startup.
2273#[must_use]
2274pub fn get_enum_schema(name: &str) -> Option<&'static EnumSchema> {
2275    static MAP: LazyLock<HashMap<&'static str, &'static EnumSchema>> = LazyLock::new(|| {
2276        let mut map = HashMap::new();
2277        for en in inventory::iter::<EnumSchema> {
2278            map.insert(en.name, en);
2279        }
2280        map
2281    });
2282    MAP.get(name).copied()
2283}
2284
2285/// A bare name registered under two different namespaces in the type registry.
2286#[derive(Debug, Clone, PartialEq, Eq)]
2287pub struct TypeConflict {
2288    pub name: String,
2289    pub first_namespace: String,
2290    pub second_namespace: String,
2291    /// `"type"` or `"enum"` — which registry the collision is in.
2292    pub kind: &'static str,
2293}
2294
2295impl std::fmt::Display for TypeConflict {
2296    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
2297        write!(
2298            f,
2299            "{} name conflict: '{}' is registered in both namespace '{}' and '{}'",
2300            self.kind, self.name, self.first_namespace, self.second_namespace
2301        )
2302    }
2303}
2304
2305impl std::error::Error for TypeConflict {}
2306
2307/// Validate that every registered type and enum has a unique bare name.
2308///
2309/// [`get_type_schema`] and [`get_enum_schema`] key on the bare name, so two
2310/// same-named types in different namespaces shadow each other in the lookup —
2311/// and anything built from the registry would silently use whichever won the
2312/// insert race. This surfaces the collision as an error instead, at startup
2313/// before the server binds.
2314pub fn validate_type_registry() -> Result<(), TypeConflict> {
2315    let types = inventory::iter::<TypeSchema>
2316        .into_iter()
2317        .map(|t| (t.name, t.namespace));
2318    find_name_collision("type", types)?;
2319    let enums = inventory::iter::<EnumSchema>
2320        .into_iter()
2321        .map(|e| (e.name, e.namespace));
2322    find_name_collision("enum", enums)?;
2323    Ok(())
2324}
2325
2326/// Return a [`TypeConflict`] if the same bare name appears under two different
2327/// namespaces in `entries` (`(name, namespace)` pairs). Repeats of an
2328/// identical `(name, namespace)` are not a conflict — only a name shared
2329/// across namespaces is, since that's what makes a bare-name lookup ambiguous.
2330fn find_name_collision<'a>(
2331    kind: &'static str,
2332    entries: impl IntoIterator<Item = (&'a str, &'a str)>,
2333) -> Result<(), TypeConflict> {
2334    let mut seen: HashMap<&str, &str> = HashMap::new();
2335    for (name, namespace) in entries {
2336        if let Some(prev) = seen.insert(name, namespace) {
2337            if prev != namespace {
2338                return Err(TypeConflict {
2339                    name: name.to_string(),
2340                    first_namespace: prev.to_string(),
2341                    second_namespace: namespace.to_string(),
2342                    kind,
2343                });
2344            }
2345        }
2346    }
2347    Ok(())
2348}
2349
2350/// Validate that all routes are unique (no duplicate paths)
2351///
2352/// Returns Ok(()) if all routes are valid, or Err with the first conflict found.
2353///
2354/// # Example
2355/// ```rust,ignore
2356/// use flatbed::validate_routes;
2357///
2358/// match validate_routes() {
2359///     Ok(()) => println!("All routes valid!"),
2360///     Err(conflict) => eprintln!("Route conflict: {}", conflict),
2361/// }
2362/// ```
2363#[allow(clippy::result_large_err)]
2364pub fn validate_routes() -> Result<(), RouteConflict> {
2365    let mut seen: HashMap<(String, HttpMethod), RouteInfo> = HashMap::new();
2366
2367    for route in inventory::iter::<RouteInfo> {
2368        let key = (route.path.to_string(), route.method);
2369
2370        if let Some(first_route) = seen.get(&key) {
2371            // Found a duplicate (path, method) with different types
2372            if first_route.request_type != route.request_type
2373                || first_route.response_type != route.response_type
2374            {
2375                return Err(RouteConflict {
2376                    path: route.path.to_string(),
2377                    method: route.method,
2378                    first_request: first_route.request_type.to_string(),
2379                    first_response: first_route.response_type.to_string(),
2380                    second_request: route.request_type.to_string(),
2381                    second_response: route.response_type.to_string(),
2382                });
2383            }
2384        }
2385
2386        seen.insert(key, *route);
2387    }
2388
2389    Ok(())
2390}
2391
2392// OpenAPI generation (enabled with "openapi" feature)
2393#[cfg(feature = "openapi")]
2394mod openapi_generation {
2395    use super::{
2396        get_routes, EnumSchema, FlatbedConfig, HttpMethod, RouteInfo, SchemaInfo, TypeFieldInfo,
2397        TypeSchema,
2398    };
2399    use std::collections::BTreeSet;
2400    use utoipa::openapi::{
2401        content::{Content, ContentBuilder},
2402        extensions::{Extensions, ExtensionsBuilder},
2403        path::{OperationBuilder, PathItemBuilder},
2404        request_body::RequestBodyBuilder,
2405        response::ResponseBuilder,
2406        schema::{
2407            AllOfBuilder, ArrayBuilder, KnownFormat, ObjectBuilder, Schema, SchemaFormat,
2408            SchemaType, Type,
2409        },
2410        server::ServerBuilder,
2411        ComponentsBuilder, Deprecated, InfoBuilder, OpenApi, OpenApiBuilder, PathsBuilder, Ref,
2412        RefOr, Required,
2413    };
2414
2415    /// Get all unique versions from registered routes
2416    pub fn get_route_versions() -> BTreeSet<String> {
2417        let mut versions = BTreeSet::new();
2418        for route in get_routes().values() {
2419            let version = route.openapi.version.unwrap_or("v1");
2420            versions.insert(version.to_string());
2421        }
2422        if versions.is_empty() {
2423            versions.insert("v1".to_string());
2424        }
2425        versions
2426    }
2427
2428    /// Get the latest version (lexicographically highest)
2429    pub fn get_latest_version() -> String {
2430        get_route_versions()
2431            .into_iter()
2432            .next_back()
2433            .unwrap_or_else(|| "v1".to_string())
2434    }
2435
2436    /// Generate FBS schema text from field info
2437    fn generate_fbs_schema_text(schema: &SchemaInfo) -> String {
2438        let mut fbs = format!("table {} {{\n", schema.name.trim_end_matches("Json"));
2439        for field in schema.fields {
2440            fbs.push_str(&format!("  {}: {};\n", field.name, field.fbs_type));
2441        }
2442        fbs.push('}');
2443        fbs
2444    }
2445
2446    /// Bare names of every registered table and enum (types, then enums).
2447    fn registered_names() -> (BTreeSet<&'static str>, BTreeSet<&'static str>) {
2448        let types = crate::inventory::iter::<TypeSchema>
2449            .into_iter()
2450            .map(|t| t.name)
2451            .collect();
2452        let enums = crate::inventory::iter::<EnumSchema>
2453            .into_iter()
2454            .map(|e| e.name)
2455            .collect();
2456        (types, enums)
2457    }
2458
2459    /// Map a scalar `fbs_type` to an OpenAPI schema with the matching type and
2460    /// `format`. 64-bit integers get the `int64` format so a client generator
2461    /// can pick a wide-enough representation.
2462    fn scalar_schema(fbs_type: &str, extensions: Option<Extensions>) -> Schema {
2463        // Both the canonical widths from `flatc` reflection (`int32`, `uint64`,
2464        // …) and the FlatBuffer source aliases (`int`, `long`, `byte`, …) are
2465        // matched, since a hand-written `ToFlatBuffer` may use either.
2466        let (schema_type, format) = match fbs_type {
2467            "bool" => (Type::Boolean, None),
2468            "string" => (Type::String, None),
2469            // OpenAPI has no standard unsigned integer format; unsigned widths
2470            // map to the same int32/int64 format as their signed counterparts.
2471            "int8" | "byte" | "uint8" | "ubyte" | "int16" | "short" | "uint16" | "ushort"
2472            | "int32" | "int" | "uint32" | "uint" => (Type::Integer, Some(KnownFormat::Int32)),
2473            "int64" | "long" | "uint64" | "ulong" => (Type::Integer, Some(KnownFormat::Int64)),
2474            "float32" | "float" => (Type::Number, Some(KnownFormat::Float)),
2475            "float64" | "double" => (Type::Number, Some(KnownFormat::Double)),
2476            // Unknown scalar — fall back to string so the spec still validates.
2477            _ => (Type::String, None),
2478        };
2479        let mut builder = ObjectBuilder::new().schema_type(SchemaType::new(schema_type));
2480        if let Some(format) = format {
2481            builder = builder.format(Some(SchemaFormat::KnownFormat(format)));
2482        }
2483        Schema::Object(builder.extensions(extensions).build())
2484    }
2485
2486    /// Schema for a bare `fbs_type`, carrying no field-level extensions (unlike
2487    /// a field's own schema): an array for `[T]`, a `$ref` for a registered
2488    /// table or enum, a scalar object otherwise.
2489    fn schema_for_fbs_type(
2490        fbs_type: &str,
2491        type_names: &BTreeSet<&str>,
2492        enum_names: &BTreeSet<&str>,
2493    ) -> RefOr<Schema> {
2494        if let Some(inner) = fbs_type.strip_prefix('[').and_then(|s| s.strip_suffix(']')) {
2495            let items = schema_for_fbs_type(inner, type_names, enum_names);
2496            return RefOr::T(Schema::Array(ArrayBuilder::new().items(items).build()));
2497        }
2498        if type_names.contains(fbs_type) || enum_names.contains(fbs_type) {
2499            RefOr::Ref(Ref::from_schema_name(fbs_type))
2500        } else {
2501            RefOr::T(scalar_schema(fbs_type, None))
2502        }
2503    }
2504
2505    /// Build the property schema for one field, carrying `x-fbs-type` /
2506    /// `x-fbs-id` vendor extensions. A table/enum reference is wrapped in
2507    /// `allOf` so the `$ref` keeps its meaning while the extensions attach to
2508    /// the wrapper — a bare `$ref` can't carry sibling keywords.
2509    fn property_schema(
2510        field: &TypeFieldInfo,
2511        type_names: &BTreeSet<&str>,
2512        enum_names: &BTreeSet<&str>,
2513    ) -> RefOr<Schema> {
2514        let extensions = || {
2515            ExtensionsBuilder::new()
2516                .add("x-fbs-type", field.fbs_type)
2517                .add("x-fbs-id", field.field_id)
2518                .build()
2519        };
2520
2521        if let Some(inner) = field
2522            .fbs_type
2523            .strip_prefix('[')
2524            .and_then(|s| s.strip_suffix(']'))
2525        {
2526            let items = schema_for_fbs_type(inner, type_names, enum_names);
2527            return RefOr::T(Schema::Array(
2528                ArrayBuilder::new()
2529                    .items(items)
2530                    .extensions(Some(extensions()))
2531                    .build(),
2532            ));
2533        }
2534
2535        if type_names.contains(field.fbs_type) || enum_names.contains(field.fbs_type) {
2536            return RefOr::T(Schema::AllOf(
2537                AllOfBuilder::new()
2538                    .item(Ref::from_schema_name(field.fbs_type))
2539                    .extensions(Some(extensions()))
2540                    .build(),
2541            ));
2542        }
2543
2544        RefOr::T(scalar_schema(field.fbs_type, Some(extensions())))
2545    }
2546
2547    /// Register a component schema for every generated table and enum. Nested
2548    /// tables and enums are referenced by `$ref`, so the components form a
2549    /// closed, resolvable graph — including tables reachable only by nesting.
2550    fn add_type_components(
2551        components_builder: &mut ComponentsBuilder,
2552        type_names: &BTreeSet<&str>,
2553        enum_names: &BTreeSet<&str>,
2554    ) {
2555        for ty in crate::inventory::iter::<TypeSchema> {
2556            let mut obj = ObjectBuilder::new().schema_type(SchemaType::new(Type::Object));
2557            for field in ty.fields {
2558                obj = obj.property(field.name, property_schema(field, type_names, enum_names));
2559                if field.required {
2560                    obj = obj.required(field.name);
2561                }
2562            }
2563            *components_builder =
2564                std::mem::take(components_builder).schema(ty.name, Schema::Object(obj.build()));
2565        }
2566
2567        for en in crate::inventory::iter::<EnumSchema> {
2568            // Enums cross the JSON wire as their variant-name string.
2569            let obj = ObjectBuilder::new()
2570                .schema_type(SchemaType::new(Type::String))
2571                .enum_values(Some(en.variants.iter().copied()))
2572                .build();
2573            *components_builder =
2574                std::mem::take(components_builder).schema(en.name, Schema::Object(obj));
2575        }
2576    }
2577
2578    /// Convert our HttpMethod to utoipa's HttpMethod
2579    fn to_utoipa_method(method: &HttpMethod) -> utoipa::openapi::path::HttpMethod {
2580        match method {
2581            HttpMethod::Get => utoipa::openapi::path::HttpMethod::Get,
2582            HttpMethod::Post => utoipa::openapi::path::HttpMethod::Post,
2583            HttpMethod::Put => utoipa::openapi::path::HttpMethod::Put,
2584            HttpMethod::Delete => utoipa::openapi::path::HttpMethod::Delete,
2585            HttpMethod::Patch => utoipa::openapi::path::HttpMethod::Patch,
2586            HttpMethod::Head => utoipa::openapi::path::HttpMethod::Head,
2587            HttpMethod::Options => utoipa::openapi::path::HttpMethod::Options,
2588        }
2589    }
2590
2591    /// The `application/x-flatbuffers` content entry. The `.fbs` text rides in
2592    /// the description because the binary body is otherwise opaque to readers.
2593    fn flatbuffers_content(schema: Option<&SchemaInfo>) -> Content {
2594        ContentBuilder::new()
2595            .schema(Some(Schema::Object(
2596                ObjectBuilder::new()
2597                    .schema_type(SchemaType::new(Type::String))
2598                    .format(Some(SchemaFormat::Custom("binary".to_string())))
2599                    .description(Some(format!(
2600                        "FlatBuffers binary data.\n\nSchema:\n```fbs\n{}\n```",
2601                        schema.map(generate_fbs_schema_text).unwrap_or_default()
2602                    )))
2603                    .build(),
2604            )))
2605            .build()
2606    }
2607
2608    /// JSON content for a route body. Generated types reference their shared
2609    /// component by `$ref`; a hand-written `ToFlatBuffer` that never registered
2610    /// a component is inlined instead, so the `$ref` never dangles.
2611    fn json_ref_content(
2612        schema: Option<&SchemaInfo>,
2613        type_names: &BTreeSet<&str>,
2614        enum_names: &BTreeSet<&str>,
2615    ) -> Content {
2616        let mut content = ContentBuilder::new();
2617        if let Some(schema) = schema {
2618            let body: RefOr<Schema> = if type_names.contains(schema.name) {
2619                RefOr::Ref(Ref::from_schema_name(schema.name))
2620            } else {
2621                RefOr::T(inline_schema(schema, type_names, enum_names))
2622            };
2623            content = content.schema(Some(body));
2624        }
2625        content.build()
2626    }
2627
2628    /// Inline object schema for a body type with no registered component;
2629    /// nested tables and enums still resolve by `$ref` where registered.
2630    fn inline_schema(
2631        schema: &SchemaInfo,
2632        type_names: &BTreeSet<&str>,
2633        enum_names: &BTreeSet<&str>,
2634    ) -> Schema {
2635        let mut obj = ObjectBuilder::new().schema_type(SchemaType::new(Type::Object));
2636        for field in schema.fields {
2637            obj = obj.property(
2638                field.name,
2639                schema_for_fbs_type(field.fbs_type, type_names, enum_names),
2640            );
2641            if field.required {
2642                obj = obj.required(field.name);
2643            }
2644        }
2645        Schema::Object(obj.build())
2646    }
2647
2648    /// Build an OpenAPI operation from a route. Generated bodies reference the
2649    /// shared component schemas by `$ref`; those components are registered once,
2650    /// up front, when the spec is assembled.
2651    fn build_operation(
2652        route: &RouteInfo,
2653        type_names: &BTreeSet<&str>,
2654        enum_names: &BTreeSet<&str>,
2655    ) -> utoipa::openapi::path::Operation {
2656        let request_body = RequestBodyBuilder::new()
2657            .content(
2658                "application/json",
2659                json_ref_content(
2660                    route.openapi.request_schema.as_ref(),
2661                    type_names,
2662                    enum_names,
2663                ),
2664            )
2665            .content(
2666                "application/x-flatbuffers",
2667                flatbuffers_content(route.openapi.request_schema.as_ref()),
2668            )
2669            .required(Some(Required::True))
2670            .build();
2671
2672        let response = ResponseBuilder::new()
2673            .description("Successful response")
2674            .content(
2675                "application/json",
2676                json_ref_content(
2677                    route.openapi.response_schema.as_ref(),
2678                    type_names,
2679                    enum_names,
2680                ),
2681            )
2682            .content(
2683                "application/x-flatbuffers",
2684                flatbuffers_content(route.openapi.response_schema.as_ref()),
2685            )
2686            .build();
2687
2688        // Build operation
2689        let mut operation_builder = OperationBuilder::new()
2690            .request_body(Some(request_body))
2691            .response("200", response);
2692
2693        if let Some(tag) = route.openapi.tag {
2694            operation_builder = operation_builder.tag(tag);
2695        }
2696        if let Some(summary) = route.openapi.summary {
2697            operation_builder = operation_builder.summary(Some(summary.to_string()));
2698        }
2699        if let Some(operation_id) = route.openapi.operation_id {
2700            operation_builder = operation_builder.operation_id(Some(operation_id.to_string()));
2701        }
2702        if route.openapi.deprecated {
2703            operation_builder = operation_builder.deprecated(Some(Deprecated::True));
2704        }
2705
2706        operation_builder.build()
2707    }
2708
2709    /// Generate OpenAPI spec for a specific version
2710    pub fn generate_openapi_spec(config: &FlatbedConfig, version: &str) -> OpenApi {
2711        let (type_names, enum_names) = registered_names();
2712        let mut paths_builder = PathsBuilder::new();
2713        let mut components_builder = ComponentsBuilder::new();
2714        add_type_components(&mut components_builder, &type_names, &enum_names);
2715
2716        // Filter routes by version and group by path
2717        let routes: Vec<&RouteInfo> = get_routes()
2718            .values()
2719            .filter(|r| r.openapi.version.unwrap_or("v1") == version)
2720            .collect();
2721
2722        // Group routes by path so multiple methods on the same path share a PathItem
2723        let mut path_groups: std::collections::BTreeMap<&str, Vec<&RouteInfo>> =
2724            std::collections::BTreeMap::new();
2725        for route in &routes {
2726            path_groups.entry(route.path).or_default().push(route);
2727        }
2728
2729        for (path, path_routes) in &path_groups {
2730            let mut path_item_builder = PathItemBuilder::new();
2731            for route in path_routes {
2732                let operation = build_operation(route, &type_names, &enum_names);
2733                let http_method = to_utoipa_method(&route.method);
2734                path_item_builder = path_item_builder.operation(http_method, operation);
2735            }
2736            paths_builder = paths_builder.path(*path, path_item_builder.build());
2737        }
2738
2739        // Build info
2740        let mut info_builder = InfoBuilder::new().title(config.title).version(version);
2741
2742        if let Some(desc) = config.description {
2743            info_builder = info_builder.description(Some(desc.to_string()));
2744        }
2745
2746        let mut builder = OpenApiBuilder::new()
2747            .info(info_builder.build())
2748            .paths(paths_builder.build())
2749            .components(Some(components_builder.build()));
2750
2751        // Add external URL if provided
2752        if let Some(ref url) = config.external_url {
2753            builder = builder.servers(Some(vec![ServerBuilder::new().url(url.clone()).build()]));
2754        }
2755
2756        builder.build()
2757    }
2758
2759    /// Generate OpenAPI spec as JSON string
2760    pub fn get_openapi_json_for_version(config: &FlatbedConfig, version: &str) -> String {
2761        let spec = generate_openapi_spec(config, version);
2762        serde_json::to_string_pretty(&spec).unwrap_or_else(|_| "{}".to_string())
2763    }
2764}
2765
2766#[cfg(feature = "openapi")]
2767pub use openapi_generation::{
2768    generate_openapi_spec, get_latest_version, get_openapi_json_for_version, get_route_versions,
2769};
2770
2771#[cfg(test)]
2772mod tests {
2773    use super::*;
2774
2775    #[test]
2776    fn test_new_builder() {
2777        let _builder = new_builder();
2778        // Builder created successfully
2779    }
2780
2781    #[test]
2782    fn test_new_builder_with_capacity() {
2783        let _builder = new_builder_with_capacity(1024);
2784        // Builder with capacity created successfully
2785    }
2786
2787    // ========================================================================
2788    // Request<T, C> tests
2789    // ========================================================================
2790
2791    #[test]
2792    fn test_request_without_context() {
2793        let req: Request<String> = Request {
2794            body: "test body".to_string(),
2795            ctx: (),
2796            headers: HeaderMap::new(),
2797            method: Method::POST,
2798            path: "/test".to_string(),
2799            path_params: HashMap::new(),
2800            query_params: HashMap::new(),
2801            request_id: "req-123".to_string(),
2802        };
2803
2804        assert_eq!(*req, "test body");
2805        assert_eq!(req.body, "test body");
2806        assert_eq!(req.path, "/test");
2807        assert_eq!(req.request_id(), "req-123");
2808    }
2809
2810    #[test]
2811    fn test_request_with_context() {
2812        #[derive(Debug)]
2813        struct AppCtx {
2814            app_name: String,
2815        }
2816
2817        let ctx = AppCtx {
2818            app_name: "test-app".to_string(),
2819        };
2820
2821        let req: Request<String, AppCtx> = Request {
2822            body: "test body".to_string(),
2823            ctx,
2824            headers: HeaderMap::new(),
2825            method: Method::POST,
2826            path: "/test".to_string(),
2827            path_params: HashMap::new(),
2828            query_params: HashMap::new(),
2829            request_id: "req-456".to_string(),
2830        };
2831
2832        assert_eq!(req.ctx.app_name, "test-app");
2833        assert_eq!(*req, "test body");
2834    }
2835
2836    #[test]
2837    fn test_request_params_and_queries() {
2838        let mut path_params = HashMap::new();
2839        path_params.insert("id".to_string(), "123".to_string());
2840
2841        let mut query_params = HashMap::new();
2842        query_params.insert("filter".to_string(), "active".to_string());
2843
2844        let req: Request<String> = Request {
2845            body: "test".to_string(),
2846            ctx: (),
2847            headers: HeaderMap::new(),
2848            method: Method::POST,
2849            path: "/users/123".to_string(),
2850            path_params,
2851            query_params,
2852            request_id: "req-789".to_string(),
2853        };
2854
2855        assert_eq!(req.param("id"), Some("123"));
2856        assert_eq!(req.param("nonexistent"), None);
2857        assert_eq!(req.query("filter"), Some("active"));
2858        assert_eq!(req.query("nonexistent"), None);
2859    }
2860
2861    #[test]
2862    fn test_request_headers() {
2863        let mut headers = HeaderMap::new();
2864        headers.insert("x-custom", HeaderValue::from_static("custom-value"));
2865
2866        let req: Request<String> = Request {
2867            body: "test".to_string(),
2868            ctx: (),
2869            headers,
2870            method: Method::POST,
2871            path: "/test".to_string(),
2872            path_params: HashMap::new(),
2873            query_params: HashMap::new(),
2874            request_id: "req-abc".to_string(),
2875        };
2876
2877        assert_eq!(req.header("x-custom"), Some("custom-value"));
2878        assert_eq!(req.header("nonexistent"), None);
2879    }
2880
2881    // ========================================================================
2882    // Response<T> tests
2883    // ========================================================================
2884
2885    #[test]
2886    fn test_response_ok() {
2887        let resp = Response::ok("success".to_string());
2888        assert_eq!(resp.body, "success");
2889        assert_eq!(resp.status, StatusCode::OK);
2890        assert!(resp.headers.is_empty());
2891    }
2892
2893    #[test]
2894    fn test_response_created() {
2895        let resp = Response::created("new resource".to_string());
2896        assert_eq!(resp.status, StatusCode::CREATED);
2897    }
2898
2899    #[test]
2900    fn test_response_with_status() {
2901        let resp = Response::with_status("accepted".to_string(), StatusCode::ACCEPTED);
2902        assert_eq!(resp.status, StatusCode::ACCEPTED);
2903    }
2904
2905    #[test]
2906    fn test_response_builder_pattern() {
2907        let resp = Response::ok("test".to_string())
2908            .status(StatusCode::CREATED)
2909            .header("x-custom", "value");
2910
2911        assert_eq!(resp.status, StatusCode::CREATED);
2912        assert_eq!(
2913            resp.headers.get("x-custom").and_then(|v| v.to_str().ok()),
2914            Some("value")
2915        );
2916    }
2917
2918    // ========================================================================
2919    // FlatbedWorkerError tests
2920    // ========================================================================
2921
2922    #[test]
2923    fn test_flatbed_worker_error_new() {
2924        let err = FlatbedWorkerError::new("TASK_ERROR", "task failed");
2925        assert_eq!(err.code, "TASK_ERROR");
2926        assert_eq!(err.message, "task failed");
2927    }
2928
2929    #[test]
2930    fn test_flatbed_worker_error_display() {
2931        let err = FlatbedWorkerError::new("CACHE_ERROR", "cache warmup failed");
2932        assert_eq!(format!("{}", err), "[CACHE_ERROR] cache warmup failed");
2933    }
2934
2935    // ========================================================================
2936    // FlatbedRouteError<D> tests
2937    // ========================================================================
2938
2939    #[test]
2940    fn test_flatbed_route_error_bad_request() {
2941        let err = FlatbedRouteError::bad_request("invalid input");
2942        assert_eq!(err.status, StatusCode::BAD_REQUEST);
2943        assert_eq!(err.message, "invalid input");
2944        assert!(err.code.is_empty());
2945        assert!(err.details.is_none());
2946    }
2947
2948    #[test]
2949    fn test_flatbed_route_error_not_found() {
2950        let err = FlatbedRouteError::not_found("resource not found");
2951        assert_eq!(err.status, StatusCode::NOT_FOUND);
2952    }
2953
2954    #[test]
2955    fn test_flatbed_route_error_internal() {
2956        let err = FlatbedRouteError::internal("something went wrong");
2957        assert_eq!(err.status, StatusCode::INTERNAL_SERVER_ERROR);
2958    }
2959
2960    #[test]
2961    fn test_flatbed_route_error_with_code() {
2962        let err = FlatbedRouteError::bad_request("validation failed").code("VALIDATION_ERROR");
2963        assert_eq!(err.code, "VALIDATION_ERROR");
2964    }
2965
2966    #[test]
2967    fn test_flatbed_route_error_with_details() {
2968        #[derive(Debug, Clone)]
2969        struct ErrorDetails {
2970            field: String,
2971        }
2972
2973        let details = ErrorDetails {
2974            field: "email".to_string(),
2975        };
2976
2977        let err = FlatbedRouteError::bad_request("validation failed").with_details(details);
2978        assert!(err.details.is_some());
2979        assert_eq!(err.details.as_ref().unwrap().field, "email");
2980    }
2981
2982    #[test]
2983    fn test_flatbed_route_error_display() {
2984        let err1 = FlatbedRouteError::bad_request("simple error");
2985        assert_eq!(format!("{}", err1), "simple error");
2986
2987        let err2 = FlatbedRouteError::bad_request("coded error").code("ERR_001");
2988        assert_eq!(format!("{}", err2), "[ERR_001] coded error");
2989    }
2990
2991    #[test]
2992    fn test_flatbed_route_error_with_header() {
2993        let err =
2994            FlatbedRouteError::unauthorized("auth required").header("www-authenticate", "Bearer");
2995        assert!(err.headers.contains_key("www-authenticate"));
2996    }
2997
2998    #[test]
2999    #[allow(deprecated)]
3000    fn test_flatbed_error_deprecated_alias() {
3001        // Test that the deprecated alias still works
3002        let err: FlatbedError = FlatbedError::bad_request("test");
3003        assert_eq!(err.status, StatusCode::BAD_REQUEST);
3004    }
3005
3006    // ========================================================================
3007    // RequestParts tests
3008    // ========================================================================
3009
3010    #[test]
3011    fn test_request_parts_new() {
3012        let parts = RequestParts::new(Method::POST, "/api/test".to_string());
3013        assert_eq!(parts.method, Method::POST);
3014        assert_eq!(parts.path, "/api/test");
3015        assert!(!parts.request_id.is_empty()); // UUID generated
3016    }
3017
3018    #[test]
3019    fn test_request_parts_request_id_from_header() {
3020        let mut parts = RequestParts::new(Method::POST, "/test".to_string());
3021        parts
3022            .headers
3023            .insert("x-request-id", HeaderValue::from_static("custom-id-123"));
3024
3025        let parts = parts.with_request_id_from_header();
3026        assert_eq!(parts.request_id, "custom-id-123");
3027    }
3028
3029    // ========================================================================
3030    // ResponseParts tests
3031    // ========================================================================
3032
3033    #[test]
3034    fn test_response_parts_ok() {
3035        let parts = ResponseParts::ok(vec![1, 2, 3], "application/json");
3036        assert_eq!(parts.body, vec![1, 2, 3]);
3037        assert_eq!(parts.status, StatusCode::OK);
3038        assert_eq!(parts.content_type, "application/json");
3039    }
3040
3041    #[test]
3042    fn test_response_parts_owned_content_type() {
3043        let ext = "js";
3044        let parts = ResponseParts::ok(vec![], format!("application/{ext}"));
3045        assert_eq!(parts.content_type, "application/js");
3046    }
3047
3048    #[test]
3049    fn test_response_raw() {
3050        let mut resp = Response::raw(b"<h1>hi</h1>".to_vec(), "text/html; charset=utf-8");
3051        assert_eq!(resp.status, StatusCode::OK);
3052        let (bytes, content_type) = resp.take_raw().expect("raw payload set");
3053        assert_eq!(bytes, b"<h1>hi</h1>");
3054        assert_eq!(content_type, "text/html; charset=utf-8");
3055    }
3056
3057    #[test]
3058    fn test_response_ok_has_no_raw() {
3059        let mut resp = Response::ok(());
3060        assert!(resp.take_raw().is_none());
3061    }
3062
3063    #[test]
3064    fn test_response_raw_status_builder_preserves_raw() {
3065        let mut resp = Response::raw(b"data".to_vec(), "text/plain").status(StatusCode::CREATED);
3066        assert_eq!(resp.status, StatusCode::CREATED);
3067        assert!(resp.take_raw().is_some(), "status() must preserve raw");
3068    }
3069
3070    #[test]
3071    fn test_response_parts_with_request_id() {
3072        let parts = ResponseParts::ok(vec![], "application/json").with_request_id("req-123");
3073        assert_eq!(
3074            parts
3075                .headers
3076                .get("x-request-id")
3077                .and_then(|v| v.to_str().ok()),
3078            Some("req-123")
3079        );
3080    }
3081
3082    // ========================================================================
3083    // FlatbedConfig tests
3084    // ========================================================================
3085
3086    #[test]
3087    fn find_name_collision_flags_same_name_across_namespaces() {
3088        let entries = [("Foo", "v_1"), ("Bar", "v_1"), ("Foo", "v_2")];
3089        let err = find_name_collision("type", entries)
3090            .expect_err("a name shared across namespaces must be a conflict");
3091        assert_eq!(err.name, "Foo");
3092        assert_eq!(err.kind, "type");
3093        assert_eq!(err.first_namespace, "v_1");
3094        assert_eq!(err.second_namespace, "v_2");
3095    }
3096
3097    #[test]
3098    fn find_name_collision_accepts_unique_names() {
3099        let entries = [("Foo", "v_1"), ("Bar", "v_1"), ("Baz", "v_2")];
3100        assert!(find_name_collision("type", entries).is_ok());
3101    }
3102
3103    #[test]
3104    fn find_name_collision_ignores_identical_reregistration() {
3105        // The same (name, namespace) appearing twice is a harmless duplicate,
3106        // not an ambiguous lookup — only a name split across namespaces is.
3107        let entries = [("Foo", "v_1"), ("Foo", "v_1")];
3108        assert!(find_name_collision("enum", entries).is_ok());
3109    }
3110
3111    #[test]
3112    fn test_flatbed_config_new() {
3113        let config = FlatbedConfig::new("Test API");
3114        assert_eq!(config.title, "Test API");
3115        assert_eq!(config.host, "0.0.0.0");
3116        assert_eq!(config.port, 8080);
3117        assert!(config.description.is_none());
3118        assert!(config.external_url.is_none());
3119    }
3120
3121    #[test]
3122    fn test_flatbed_config_default() {
3123        let config = FlatbedConfig::default();
3124        assert_eq!(config.title, "API");
3125        assert_eq!(config.host, "0.0.0.0");
3126        assert_eq!(config.port, 8080);
3127    }
3128
3129    #[test]
3130    fn test_flatbed_config_builder_pattern() {
3131        let config = FlatbedConfig::new("My API")
3132            .host("127.0.0.1")
3133            .port(3000)
3134            .description("A test API")
3135            .external_url("https://api.example.com");
3136
3137        assert_eq!(config.title, "My API");
3138        assert_eq!(config.host, "127.0.0.1");
3139        assert_eq!(config.port, 3000);
3140        assert_eq!(config.description, Some("A test API"));
3141        assert_eq!(
3142            config.external_url,
3143            Some("https://api.example.com".to_string())
3144        );
3145    }
3146
3147    #[test]
3148    fn test_flatbed_config_bind_address() {
3149        let config = FlatbedConfig::new("API").host("0.0.0.0").port(8080);
3150        assert_eq!(config.bind_address(), "0.0.0.0:8080");
3151
3152        let config2 = FlatbedConfig::new("API").host("127.0.0.1").port(3000);
3153        assert_eq!(config2.bind_address(), "127.0.0.1:3000");
3154    }
3155
3156    #[test]
3157    fn test_flatbed_config_debug() {
3158        let config = FlatbedConfig::new("Test API").host("localhost").port(9000);
3159        let debug_str = format!("{:?}", config);
3160        assert!(debug_str.contains("Test API"));
3161        assert!(debug_str.contains("localhost"));
3162        assert!(debug_str.contains("9000"));
3163    }
3164
3165    // ========================================================================
3166    // Worker trait tests
3167    // ========================================================================
3168
3169    /// Test context for Worker trait tests.
3170    struct TestWorkerCtx {
3171        pub label: String,
3172    }
3173
3174    /// A Worker implementation that records whether it was called.
3175    struct RecordingWorker;
3176
3177    impl Default for RecordingWorker {
3178        fn default() -> Self {
3179            RecordingWorker
3180        }
3181    }
3182
3183    impl Worker for RecordingWorker {
3184        type Context = TestWorkerCtx;
3185        const NAME: &'static str = "recording-worker";
3186        const DESCRIPTION: Option<&'static str> = Some("Records calls for testing");
3187
3188        fn run(
3189            &self,
3190            ctx: std::sync::Arc<Self::Context>,
3191        ) -> std::pin::Pin<
3192            Box<dyn std::future::Future<Output = Result<(), FlatbedWorkerError>> + Send>,
3193        > {
3194            Box::pin(async move {
3195                if ctx.label == "fail" {
3196                    return Err(FlatbedWorkerError::new("TEST", "intentional failure"));
3197                }
3198                Ok(())
3199            })
3200        }
3201    }
3202
3203    #[tokio::test]
3204    async fn run_basic_worker_downcasts_and_runs() {
3205        let ctx = std::sync::Arc::new(TestWorkerCtx {
3206            label: "success".to_string(),
3207        });
3208        let dyn_ctx: std::sync::Arc<dyn std::any::Any + Send + Sync> = ctx;
3209        let result = run_basic_worker::<RecordingWorker, TestWorkerCtx>(dyn_ctx).await;
3210        assert!(result.is_ok());
3211    }
3212
3213    #[tokio::test]
3214    async fn run_basic_worker_propagates_error() {
3215        let ctx = std::sync::Arc::new(TestWorkerCtx {
3216            label: "fail".to_string(),
3217        });
3218        let dyn_ctx: std::sync::Arc<dyn std::any::Any + Send + Sync> = ctx;
3219        let result = run_basic_worker::<RecordingWorker, TestWorkerCtx>(dyn_ctx).await;
3220        assert!(result.is_err());
3221        assert_eq!(result.unwrap_err().code, "TEST");
3222    }
3223
3224    #[tokio::test]
3225    async fn run_basic_worker_panics_on_context_type_mismatch() {
3226        // Provide the wrong context type (String instead of TestWorkerCtx)
3227        let wrong_ctx: std::sync::Arc<dyn std::any::Any + Send + Sync> =
3228            std::sync::Arc::new("wrong type".to_string());
3229
3230        let result = std::panic::AssertUnwindSafe(
3231            run_basic_worker::<RecordingWorker, TestWorkerCtx>(wrong_ctx),
3232        );
3233        let panic_result = futures::FutureExt::catch_unwind(result).await;
3234        assert!(
3235            panic_result.is_err(),
3236            "run_basic_worker should panic on context type mismatch"
3237        );
3238    }
3239
3240    // ========================================================================
3241    // Worker trait defaults
3242    // ========================================================================
3243
3244    #[test]
3245    fn worker_description_default_is_some_when_set() {
3246        assert_eq!(
3247            <RecordingWorker as Worker>::DESCRIPTION,
3248            Some("Records calls for testing")
3249        );
3250    }
3251
3252    /// A worker with no description override to test the default.
3253    struct MinimalWorker;
3254
3255    impl Default for MinimalWorker {
3256        fn default() -> Self {
3257            MinimalWorker
3258        }
3259    }
3260
3261    impl Worker for MinimalWorker {
3262        type Context = TestWorkerCtx;
3263        const NAME: &'static str = "minimal-worker";
3264
3265        fn run(
3266            &self,
3267            _ctx: std::sync::Arc<Self::Context>,
3268        ) -> std::pin::Pin<
3269            Box<dyn std::future::Future<Output = Result<(), FlatbedWorkerError>> + Send>,
3270        > {
3271            Box::pin(async { Ok(()) })
3272        }
3273    }
3274
3275    #[test]
3276    fn worker_description_default_is_none() {
3277        assert!(<MinimalWorker as Worker>::DESCRIPTION.is_none());
3278    }
3279
3280    // ========================================================================
3281    // register_worker! macro tests
3282    // ========================================================================
3283
3284    // Register MinimalWorker via the macro so inventory picks it up.
3285    register_worker!(MinimalWorker, TestWorkerCtx);
3286
3287    #[test]
3288    fn register_worker_macro_makes_worker_discoverable() {
3289        let workers = get_workers();
3290        let found = workers
3291            .iter()
3292            .find(|w| w.name == <MinimalWorker as Worker>::NAME);
3293        assert!(
3294            found.is_some(),
3295            "MinimalWorker should be discoverable via inventory"
3296        );
3297    }
3298
3299    #[test]
3300    fn registered_worker_name_matches_trait_constant() {
3301        let workers = get_workers();
3302        let found = workers.iter().find(|w| w.name == "minimal-worker");
3303        assert!(found.is_some());
3304        let info = found.unwrap();
3305        assert_eq!(info.name, <MinimalWorker as Worker>::NAME);
3306        assert_eq!(info.description, <MinimalWorker as Worker>::DESCRIPTION);
3307    }
3308
3309    #[tokio::test]
3310    async fn registered_worker_can_be_invoked_via_inventory() {
3311        let workers = get_workers();
3312        let info = workers
3313            .iter()
3314            .find(|w| w.name == "minimal-worker")
3315            .expect("MinimalWorker should be registered");
3316
3317        let ctx = std::sync::Arc::new(TestWorkerCtx {
3318            label: "inventory-test".to_string(),
3319        });
3320        let dyn_ctx: std::sync::Arc<dyn std::any::Any + Send + Sync> = ctx;
3321        let result = (info.worker)(dyn_ctx).await;
3322        assert!(result.is_ok());
3323    }
3324}