fts_server/
lib.rs

1#![warn(missing_docs)]
2// Note: this overwrites the link in the README to point to the rust docs of the fts-demo crate.
3//! [fts_core]: https://docs.rs/fts_core/latest/fts_core/index.html
4//! [fts_server]: https://docs.rs/fts_server/latest/fts_server/index.html
5//! [fts_solver]: https://docs.rs/fts_solver/latest/fts_solver/index.html
6//! [fts_demo]: https://docs.rs/fts_demo/latest/fts_demo/index.html
7#![doc = include_str!("../README.md")]
8use fts_core::{
9    models::{BidderId, Outcome, ProductId, RawAuctionInput},
10    ports::{AuctionRepository, MarketRepository},
11};
12
13use axum::Json;
14use axum::Router;
15use axum::http::header;
16use axum::response::sse::Event;
17use fts_solver::Solver;
18use fxhash::FxBuildHasher;
19use openapi::openapi_router;
20use serde::Serialize;
21use std::sync::Arc;
22use std::{convert::Infallible, net::SocketAddr};
23use time::OffsetDateTime;
24use tokio::sync::{mpsc, watch};
25use tokio::task::JoinHandle;
26use tokio::try_join;
27use tower_http::cors;
28
29mod openapi;
30mod routes;
31mod utils;
32
33pub use crate::openapi::MarketplaceApi;
34pub use utils::CustomJWTClaims;
35use utils::JWTVerifier;
36pub use utils::Now;
37pub use utils::generate_jwt;
38
39/// Type alias for a thread-safe map of senders for SSE events.
40///
41/// Used to maintain channels for server-sent events for real-time updates.
42type SenderMap<T> =
43    Arc<dashmap::DashMap<T, watch::Sender<Result<Event, Infallible>>, FxBuildHasher>>;
44
45/// Application state shared across all request handlers.
46///
47/// Contains references to repositories, communication channels, and
48/// authentication services used throughout the application.
49#[derive(Clone)]
50pub struct AppState<T: MarketRepository> {
51    /// JWT verification service
52    jwt: JWTVerifier,
53    /// Market repository for data access
54    market: T,
55    /// Channel for queueing auction solves
56    solve_queue: mpsc::Sender<RawAuctionInput<T::AuctionId>>,
57    /// Channel for receiving activity updates
58    activity_receiver: watch::Receiver<Result<Event, Infallible>>,
59    /// Channels for product-specific updates
60    product_sender: SenderMap<ProductId>,
61    /// Channels for bidder-specific updates
62    bidder_sender: SenderMap<BidderId>,
63}
64
65/// Represents an update to be sent via server-sent events.
66///
67/// Contains auction outcome data along with its time range.
68#[derive(Serialize)]
69pub struct Update<T> {
70    /// Start time of the auction period
71    #[serde(with = "time::serde::rfc3339")]
72    pub from: OffsetDateTime,
73    /// End time of the auction period
74    #[serde(with = "time::serde::rfc3339")]
75    pub thru: OffsetDateTime,
76    /// Outcome data from the auction
77    #[serde(flatten)]
78    pub outcome: T,
79}
80
81/// Response for the health check endpoint
82#[derive(Serialize)]
83struct HealthResponse {
84    status: String,
85}
86
87/// Simple health check endpoint
88async fn health_check() -> Json<HealthResponse> {
89    Json(HealthResponse {
90        status: "ok".to_string(),
91    })
92}
93
94/// Creates the application state and solver background task.
95pub fn state<T: MarketRepository>(
96    api_secret: &str,
97    market: T,
98) -> (AppState<T>, JoinHandle<Result<(), T::Error>>) {
99    // We create a FIFO queue for solving auctions
100    let (solve_sender, mut solve_receiver) = mpsc::channel::<RawAuctionInput<T::AuctionId>>(24);
101
102    // These channels are for reporting activity to SSE subscribers on /activity
103    let (activity_sender, activity_receiver) = watch::channel(Ok(Event::default().comment("")));
104
105    let product_sender: SenderMap<ProductId> = Default::default();
106    let bidder_sender: SenderMap<BidderId> = Default::default();
107
108    let solver = {
109        let market = market.clone();
110        let activity_sender = activity_sender.clone();
111        let product_sender = product_sender.clone();
112        let bidder_sender = bidder_sender.clone();
113        tokio::spawn(async move {
114            let solver = T::solver();
115            while let Some(auction) = solve_receiver.recv().await {
116                let id = auction.id.clone();
117
118                // Convert the auction into a format the solver understands
119                let submissions = auction.into_solver();
120
121                // TODO: this is where warm-starting would be used
122                let fts_solver::AuctionOutcome { bidders, products } = solver.solve(&submissions);
123
124                // TODO: update the API to scope the auth_id the bidder_id
125                let auth_outcomes = bidders
126                    .values()
127                    .flat_map(|outcome| {
128                        outcome.auths.iter().map(|(auth_id, auth_outcome)| {
129                            (
130                                auth_id.clone(),
131                                Outcome {
132                                    price: auth_outcome.price,
133                                    trade: auth_outcome.trade,
134                                    data: None,
135                                },
136                            )
137                        })
138                    })
139                    .collect::<Vec<_>>();
140
141                let product_outcomes = products
142                    .iter()
143                    .map(|(product_id, product_outcome)| {
144                        (
145                            product_id.clone(),
146                            Outcome {
147                                price: product_outcome.price,
148                                trade: product_outcome.trade,
149                                data: None,
150                            },
151                        )
152                    })
153                    .collect::<Vec<_>>();
154
155                let now = OffsetDateTime::now_utc().into();
156
157                // We are basically copy/pasting solver::*Outcome into crate::Outcome, which seems silly.
158                // But, if the server wants to augment or transform the data somehow, this a reasonable
159                // indirection to have.
160                let metadata = AuctionRepository::report(
161                    &market.clone(),
162                    id,
163                    auth_outcomes.into_iter(),
164                    product_outcomes.into_iter(),
165                    now,
166                )
167                .await?;
168
169                // Now that we've stored the outcomes, we push updates to the SSE listeners
170                if let Some(metadata) = metadata {
171                    let _ = activity_sender.send_replace(Ok(Event::default()
172                        .event("outcome")
173                        .data(serde_json::to_string(&metadata).expect("infallible!"))));
174
175                    // We also individually broadcast the results to any listeners.
176                    // TODO: think about how to cleanup the dashmap over time
177                    for (product_id, outcome) in products {
178                        if let Some(channel) = product_sender.get(&product_id) {
179                            let update = Update {
180                                from: metadata.from,
181                                thru: metadata.thru,
182                                outcome,
183                            };
184                            let _ = channel.send_replace(Ok(Event::default()
185                                .event("outcome")
186                                .data(serde_json::to_string(&update).expect("infallible!"))));
187                        };
188                    }
189
190                    for (bidder_id, outcome) in bidders {
191                        if let Some(channel) = bidder_sender.get(&bidder_id) {
192                            let update = Update {
193                                from: metadata.from,
194                                thru: metadata.thru,
195                                outcome,
196                            };
197                            let _ = channel.send_replace(Ok(Event::default()
198                                .event("outcome")
199                                .data(serde_json::to_string(&update).expect("infallible!"))));
200                        }
201                    }
202                }
203            }
204            Result::<(), T::Error>::Ok(())
205        })
206    };
207
208    let state = AppState {
209        jwt: JWTVerifier::from(api_secret),
210        market,
211        solve_queue: solve_sender,
212        activity_receiver,
213        product_sender,
214        bidder_sender,
215    };
216
217    (state, solver)
218}
219
220/// Creates the application router with all routes configured.
221pub fn router<T: MarketRepository>(state: AppState<T>) -> Router {
222    // To allow for web app access, we use a permissive CORS policy. Notably,
223    // this strips any implicit authorization, making this a pretty decent policy.
224    let policy = cors::CorsLayer::new()
225        .allow_origin(cors::Any)
226        .allow_methods(cors::Any)
227        .allow_headers([header::AUTHORIZATION, header::CONTENT_TYPE]);
228
229    // Wire it together
230    let app = Router::new()
231        .route("/health", axum::routing::get(health_check))
232        .nest("/v0/auths", routes::auths::router())
233        .nest("/v0/costs", routes::costs::router())
234        // Bidder-specific routes for their submission
235        .nest("/v0/submissions", routes::submission::router(state.clone()))
236        // View products and their results
237        .nest("/v0/products", routes::products::router())
238        // These are the SSE-routes for live-updates
239        .nest("/v0/outcomes", routes::outcomes::router())
240        .nest("/admin", routes::admin::router(state.clone()));
241
242    app.layer(policy).with_state(state)
243}
244
245/// Starts the HTTP server with the configured application.
246pub async fn start<T: MarketRepository>(api_port: u16, api_secret: String, market: T) {
247    // Setup the HTTP server
248    let listener = tokio::net::TcpListener::bind(SocketAddr::new([0, 0, 0, 0].into(), api_port))
249        .await
250        .expect("Unable to bind local port");
251    tracing::info!(
252        "Listening for requests on {}",
253        listener.local_addr().unwrap()
254    );
255
256    let (app_state, solver) = state(&api_secret, market);
257
258    // Setup the combined application state and serve it with our router
259    let server = tokio::spawn(async move {
260        axum::serve(listener, router(app_state).merge(openapi_router())).await
261    });
262
263    let _ = try_join!(solver, server).expect("shutdown");
264}
fts_server/lib.rs

fts_server/
lib.rs
