1use 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#[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#[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#[cfg(any(feature = "nats", feature = "k8s"))]
70#[doc(hidden)]
71pub use futures;
72
73#[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#[cfg(any(feature = "nats", feature = "k8s"))]
90mod context;
91#[cfg(any(feature = "nats", feature = "k8s"))]
92pub use context::FlatbedContext;
93
94#[cfg(feature = "k8s")]
96#[doc(hidden)]
97pub use kube;
98
99pub mod hyper;
101
102pub mod readiness;
103pub use readiness::{Readiness, ReadinessGate};
104
105mod supervisor;
107pub use supervisor::{worker_states, RestartPolicy, WorkerState};
108
109#[cfg(feature = "telemetry")]
111pub mod telemetry;
112#[cfg(feature = "telemetry")]
113pub use telemetry::{TelemetryConfig, TelemetryError, TelemetryService};
114
115pub use flatbuffers;
118pub use serde;
119pub use serde_json;
120#[cfg(feature = "openapi")]
121pub use utoipa;
122
123#[doc(hidden)]
126pub use include_dir;
127
128#[doc(hidden)]
130pub use tokio;
131
132pub type BoxFuture<T> = std::pin::Pin<Box<dyn std::future::Future<Output = T> + Send>>;
137
138pub use http::{HeaderMap, HeaderName, HeaderValue, Method, StatusCode};
140
141#[doc(hidden)]
143pub use uuid;
144
145pub type BeforeRequestHook =
151 std::sync::Arc<dyn Fn(&RequestParts) -> Result<(), FlatbedRouteError> + Send + Sync>;
152
153#[derive(Clone)]
177pub struct FlatbedConfig {
178 pub title: &'static str,
180 pub description: Option<&'static str>,
182 pub external_url: Option<String>,
184 pub host: String,
186 pub port: u16,
188 pub splash: Option<String>,
190 pub before_request: Option<BeforeRequestHook>,
193 pub readiness: Readiness,
197 #[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 pub fn new(title: &'static str) -> Self {
246 Self {
247 title,
248 ..Self::default()
249 }
250 }
251
252 pub fn host(mut self, host: &str) -> Self {
254 self.host = host.to_string();
255 self
256 }
257
258 pub fn port(mut self, port: u16) -> Self {
260 self.port = port;
261 self
262 }
263
264 pub fn description(mut self, desc: &'static str) -> Self {
266 self.description = Some(desc);
267 self
268 }
269
270 pub fn splash(mut self, splash: impl Into<String>) -> Self {
284 self.splash = Some(splash.into());
285 self
286 }
287
288 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 pub fn readiness(mut self, readiness: Readiness) -> Self {
327 self.readiness = readiness;
328 self
329 }
330
331 pub fn external_url(mut self, url: impl Into<String>) -> Self {
333 self.external_url = Some(url.into());
334 self
335 }
336
337 #[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 pub fn bind_address(&self) -> String {
351 format!("{}:{}", self.host, self.port)
352 }
353}
354
355pub struct Flatbed;
422
423impl Flatbed {
424 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 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 let router = Arc::new(hyper::build_router());
473
474 let addr: std::net::SocketAddr = config
476 .bind_address()
477 .parse()
478 .map_err(|e| Error::Custom(format!("Invalid bind address: {}", e)))?;
479
480 let (healthz_tx, healthz_rx) = watch::channel(false);
482 let (booted_tx, booted_rx) = watch::channel(false);
483
484 let context: Arc<RwLock<Option<Arc<C>>>> = Arc::new(RwLock::new(None));
486
487 let boot_config = config.clone();
489
490 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 let server = hyper::AutoServer::new(addr, service_ctx, healthz_tx.clone());
502 let server_handle = tokio::spawn(async move { server.serve().await });
503
504 let _ = healthz_tx.send(true);
507
508 let ctx = boot(boot_config)
510 .await
511 .map_err(|e| Error::Custom(format!("Boot error: {}", e)))?;
512
513 {
516 let mut guard = context.write().await;
517 *guard = Some(Arc::new(ctx));
518 }
519 let _ = booted_tx.send(true);
520
521 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#[derive(Clone, Copy, Debug)]
531pub struct FieldInfo {
532 pub name: &'static str,
534 pub fbs_type: &'static str,
536 pub required: bool,
539}
540
541#[derive(Clone, Copy, Debug)]
543pub struct SchemaInfo {
544 pub name: &'static str,
546 pub fields: &'static [FieldInfo],
548}
549
550#[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#[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#[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#[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#[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
615pub trait ToFlatBuffer: serde::de::DeserializeOwned + serde::Serialize {
634 const SCHEMA_FIELDS: &'static [FieldInfo];
636
637 const SCHEMA_NAME: &'static str;
639
640 fn to_flatbuffer(&self) -> Vec<u8>;
642
643 fn schema_fields() -> &'static [FieldInfo] {
645 Self::SCHEMA_FIELDS
646 }
647
648 fn schema_name() -> &'static str {
650 Self::SCHEMA_NAME
651 }
652}
653
654impl 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
667pub trait FromFlatBuffer: Sized {
681 fn from_flatbuffer(bytes: &[u8]) -> Result<Self, flatbuffers::InvalidFlatbuffer>;
688}
689
690impl FromFlatBuffer for () {
693 fn from_flatbuffer(_bytes: &[u8]) -> Result<Self, flatbuffers::InvalidFlatbuffer> {
694 Ok(())
695 }
696}
697
698pub trait HasJsonCompanion {
714 type Json: ToFlatBuffer;
716}
717
718pub trait HasPlainStruct {
731 type Plain: ToFlatBuffer;
733 type FlatBuffer<'a>;
735}
736
737#[derive(Clone, Copy, Debug, Default)]
739pub struct OpenApiRouteInfo {
740 pub version: Option<&'static str>, 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
749pub use flatbed_macros::route;
751
752pub use flatbed_macros::static_route;
753
754pub use flatbed_macros::main;
756
757#[doc(hidden)]
759pub use inventory;
760
761#[derive(Debug)]
787pub struct Request<T, C = ()> {
788 pub body: T,
790 pub ctx: C,
792 pub headers: HeaderMap,
794 pub method: Method,
796 pub path: String,
798 pub path_params: HashMap<String, String>,
800 pub query_params: HashMap<String, String>,
802 pub request_id: String,
804}
805
806impl<T, C> Request<T, C> {
807 pub fn header(&self, name: &str) -> Option<&str> {
809 self.headers.get(name).and_then(|v| v.to_str().ok())
810 }
811
812 pub fn param(&self, name: &str) -> Option<&str> {
814 self.path_params.get(name).map(|s| s.as_str())
815 }
816
817 pub fn query(&self, name: &str) -> Option<&str> {
819 self.query_params.get(name).map(|s| s.as_str())
820 }
821
822 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#[derive(Debug)]
851pub struct Response<T> {
852 pub body: T,
854 pub status: StatusCode,
856 pub headers: HeaderMap,
858 raw: Option<(Vec<u8>, Cow<'static, str>)>,
859}
860
861impl<T> Response<T> {
862 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 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 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 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 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 pub fn status(mut self, status: StatusCode) -> Self {
914 self.status = status;
915 self
916 }
917
918 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 #[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 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#[derive(Debug, Clone)]
986pub struct FlatbedWorkerError {
987 pub code: String,
989 pub message: String,
991}
992
993impl FlatbedWorkerError {
994 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#[derive(Debug, Clone)]
1040pub struct FlatbedRouteError<D = ()> {
1041 pub status: StatusCode,
1043 pub code: String,
1045 pub message: String,
1047 pub headers: HeaderMap,
1049 pub details: Option<D>,
1051}
1052
1053#[deprecated(since = "0.2.0", note = "Use FlatbedRouteError instead")]
1057pub type FlatbedError<D = ()> = FlatbedRouteError<D>;
1058
1059impl FlatbedRouteError<()> {
1060 pub fn bad_request(message: impl Into<String>) -> Self {
1062 Self::with_status(StatusCode::BAD_REQUEST, message)
1063 }
1064
1065 pub fn unauthorized(message: impl Into<String>) -> Self {
1067 Self::with_status(StatusCode::UNAUTHORIZED, message)
1068 }
1069
1070 pub fn forbidden(message: impl Into<String>) -> Self {
1072 Self::with_status(StatusCode::FORBIDDEN, message)
1073 }
1074
1075 pub fn not_found(message: impl Into<String>) -> Self {
1077 Self::with_status(StatusCode::NOT_FOUND, message)
1078 }
1079
1080 pub fn conflict(message: impl Into<String>) -> Self {
1082 Self::with_status(StatusCode::CONFLICT, message)
1083 }
1084
1085 pub fn unprocessable(message: impl Into<String>) -> Self {
1087 Self::with_status(StatusCode::UNPROCESSABLE_ENTITY, message)
1088 }
1089
1090 pub fn internal(message: impl Into<String>) -> Self {
1092 Self::with_status(StatusCode::INTERNAL_SERVER_ERROR, message)
1093 }
1094
1095 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 pub fn code(mut self, code: impl Into<String>) -> Self {
1110 self.code = code.into();
1111 self
1112 }
1113
1114 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 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 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#[derive(Debug, Clone)]
1156pub struct RequestParts {
1157 pub headers: HeaderMap,
1159 pub method: Method,
1161 pub path: String,
1163 pub path_params: HashMap<String, String>,
1165 pub query_params: HashMap<String, String>,
1167 pub request_id: String,
1169}
1170
1171impl RequestParts {
1172 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 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 pub fn header(&self, name: &str) -> Option<&str> {
1198 self.headers.get(name).and_then(|v| v.to_str().ok())
1199 }
1200}
1201
1202#[derive(Debug)]
1204pub struct ResponseParts {
1205 pub body: Vec<u8>,
1207 pub status: StatusCode,
1209 pub headers: HeaderMap,
1211 pub content_type: Cow<'static, str>,
1213}
1214
1215impl ResponseParts {
1216 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 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 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#[derive(Debug, Clone)]
1255pub enum Error {
1256 DeserializationError(String),
1258 SerializationError(String),
1260 HandlerError(String),
1262 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#[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 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 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 pub fn with_status(mut self, status: u16) -> Self {
1377 self.status_code = status;
1378 self
1379 }
1380
1381 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 pub fn status_code(&self) -> u16 {
1396 self.status_code
1397 }
1398
1399 pub fn headers(&self) -> &HashMap<String, String> {
1401 &self.headers
1402 }
1403
1404 pub fn into_bytes(self) -> Vec<u8> {
1406 self.builder.finished_data().to_vec()
1407 }
1408
1409 pub fn as_bytes(&self) -> &[u8] {
1411 self.builder.finished_data()
1412 }
1413
1414 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
1421pub fn new_builder() -> FlatBufferBuilder<'static> {
1423 FlatBufferBuilder::new()
1424}
1425
1426pub fn new_builder_with_capacity(capacity: usize) -> FlatBufferBuilder<'static> {
1428 FlatBufferBuilder::with_capacity(capacity)
1429}
1430
1431pub 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
1441pub 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
1453pub type HandlerFn = fn(Vec<u8>, &str) -> Result<(Vec<u8>, &'static str), Error>;
1466
1467#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1469pub enum Codec {
1470 Json,
1472 FlatBuffer,
1474}
1475
1476#[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
1495pub 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#[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 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#[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
1554impl 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#[derive(Clone, Copy, Debug)]
1573pub enum StaticSource {
1574 Dir(&'static str),
1577 Embedded(&'static include_dir::Dir<'static>),
1579}
1580
1581#[derive(Clone, Copy, Debug)]
1587pub struct StaticRouteInfo {
1588 pub mount: &'static str,
1590 pub source: StaticSource,
1592 pub fallback: Option<&'static str>,
1595 pub no_fallback: &'static [&'static str],
1600}
1601
1602inventory::collect!(StaticRouteInfo);
1603
1604#[must_use]
1606pub fn get_static_routes() -> Vec<StaticRouteInfo> {
1607 inventory::iter::<StaticRouteInfo>
1608 .into_iter()
1609 .copied()
1610 .collect()
1611}
1612
1613pub 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
1627pub 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#[derive(Clone, Copy)]
1642pub struct WorkerInfo {
1643 pub name: &'static str,
1645 pub description: Option<&'static str>,
1647 pub worker: WorkerFn,
1649 pub restart: Option<RestartPolicy>,
1652}
1653
1654impl 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#[derive(Clone, Copy)]
1674pub struct WorkerDrainInfo {
1675 pub name: &'static str,
1677 pub drain: DrainFn,
1679}
1680
1681inventory::collect!(WorkerDrainInfo);
1682
1683#[must_use]
1685pub fn get_worker_drains() -> Vec<&'static WorkerDrainInfo> {
1686 inventory::iter::<WorkerDrainInfo>.into_iter().collect()
1687}
1688
1689inventory::collect!(WorkerInfo);
1690
1691#[must_use]
1703pub fn get_workers() -> Vec<&'static WorkerInfo> {
1704 inventory::iter::<WorkerInfo>.into_iter().collect()
1705}
1706
1707pub trait Worker: Send + Sync + 'static {
1744 type Context: Send + Sync + 'static;
1746
1747 const NAME: &'static str;
1749
1750 const DESCRIPTION: Option<&'static str> = None;
1752
1753 const RESTART: ::core::option::Option<RestartPolicy> = ::core::option::Option::None;
1768
1769 fn run(&self, ctx: std::sync::Arc<Self::Context>) -> BoxFuture<Result<(), FlatbedWorkerError>>;
1775
1776 fn drain(
1779 &self,
1780 _ctx: std::sync::Arc<Self::Context>,
1781 ) -> BoxFuture<Result<(), FlatbedWorkerError>> {
1782 Box::pin(async { Ok(()) })
1783 }
1784}
1785
1786pub 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
1803pub 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#[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#[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#[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#[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#[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#[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
2191pub type RouteMap = HashMap<(String, HttpMethod), RouteInfo>;
2193
2194#[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#[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#[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#[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#[derive(Debug, Clone, PartialEq, Eq)]
2287pub struct TypeConflict {
2288 pub name: String,
2289 pub first_namespace: String,
2290 pub second_namespace: String,
2291 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
2307pub 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
2326fn 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#[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 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#[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 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 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 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 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 fn scalar_schema(fbs_type: &str, extensions: Option<Extensions>) -> Schema {
2463 let (schema_type, format) = match fbs_type {
2467 "bool" => (Type::Boolean, None),
2468 "string" => (Type::String, None),
2469 "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 _ => (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 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 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 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 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 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 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 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 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 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 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 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 let routes: Vec<&RouteInfo> = get_routes()
2718 .values()
2719 .filter(|r| r.openapi.version.unwrap_or("v1") == version)
2720 .collect();
2721
2722 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 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 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 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 }
2780
2781 #[test]
2782 fn test_new_builder_with_capacity() {
2783 let _builder = new_builder_with_capacity(1024);
2784 }
2786
2787 #[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 #[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 #[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 #[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 let err: FlatbedError = FlatbedError::bad_request("test");
3003 assert_eq!(err.status, StatusCode::BAD_REQUEST);
3004 }
3005
3006 #[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()); }
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 #[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 #[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 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 struct TestWorkerCtx {
3171 pub label: String,
3172 }
3173
3174 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 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 #[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 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 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}