1use std::{net::SocketAddr, str::FromStr};
2use std::sync::Arc;
3
4use axum::{
5 extract::State,
6 http::StatusCode,
7 response::IntoResponse,
8 routing::post,
9 Json,
10 Router,
11};
12use serde::{Deserialize, Serialize};
13use serde_json::{json, Value};
14use solana_sdk::pubkey::Pubkey;
15
16use super::store::{StoreError, TapeStore};
17
18#[repr(i64)]
19#[derive(Copy, Clone)]
20pub enum ErrorCode {
21 ParseError = -32700,
22 InvalidRequest = -32600,
23 MethodNotFound = -32601,
24 InvalidParams = -32602,
25 InternalError = -32603,
26 ServerError = -32000,
27}
28
29impl ErrorCode {
30 pub fn code(self) -> i64 {
31 self as i64
32 }
33}
34
35#[derive(Deserialize)]
36struct RpcRequest {
37 method: String,
38 params: Value,
39 id: Option<Value>,
40}
41
42#[derive(Serialize)]
43pub struct RpcError {
44 code: i64,
45 message: String,
46}
47
48#[derive(Serialize)]
49struct RpcResponse {
50 jsonrpc: String,
51 #[serde(skip_serializing_if = "Option::is_none")]
52 result: Option<Value>,
53 #[serde(skip_serializing_if = "Option::is_none")]
54 error: Option<RpcError>,
55 id: Option<Value>,
56}
57
58fn make_response(
60 id: Option<Value>,
61 result: Result<Value, RpcError>,
62) -> (StatusCode, Json<RpcResponse>) {
63 let (res, err) = match result {
64 Ok(val) => (Some(val), None),
65 Err(e) => (None, Some(e)),
66 };
67 let resp = RpcResponse {
68 jsonrpc: "2.0".into(),
69 result: res,
70 error: err,
71 id,
72 };
73 (StatusCode::OK, Json(resp))
74}
75
76pub fn rpc_get_health(store: &TapeStore, _params: &Value) -> Result<Value, RpcError> {
85 let (last_processed_slot, drift) = store
86 .get_health()
87 .map_err(|e| RpcError {
88 code: ErrorCode::ServerError.code(),
89 message: e.to_string(),
90 })?;
91 Ok(json!({ "last_processed_slot": last_processed_slot, "drift": drift }))
92}
93
94pub fn rpc_get_tape_address(store: &TapeStore, params: &Value) -> Result<Value, RpcError> {
108 let tn = params
109 .get("tape_number")
110 .and_then(Value::as_u64)
111 .ok_or(RpcError {
112 code: ErrorCode::InvalidParams.code(),
113 message: "invalid or missing tape_number".into(),
114 })?;
115
116 store
117 .get_tape_address(tn)
118 .map(|pk| json!(pk.to_string()))
119 .map_err(|e| match e {
120 StoreError::TapeNotFound(n) => RpcError {
121 code: ErrorCode::ServerError.code(),
122 message: format!("tape {} not found", n),
123 },
124 other => RpcError {
125 code: ErrorCode::ServerError.code(),
126 message: other.to_string(),
127 },
128 })
129}
130
131pub fn rpc_get_tape_number(store: &TapeStore, params: &Value) -> Result<Value, RpcError> {
145 let addr = params
146 .get("tape_address")
147 .and_then(Value::as_str)
148 .ok_or(RpcError {
149 code: ErrorCode::InvalidParams.code(),
150 message: "invalid or missing tape_address".into(),
151 })?;
152
153 let pk = Pubkey::from_str(addr).map_err(|e| RpcError {
154 code: ErrorCode::InvalidParams.code(),
155 message: format!("invalid pubkey: {}", e),
156 })?;
157
158 store
159 .get_tape_number(&pk)
160 .map(|num| json!(num))
161 .map_err(|e| match e {
162 StoreError::TapeNotFoundForAddress(_) => RpcError {
163 code: ErrorCode::ServerError.code(),
164 message: "tape not found for address".into(),
165 },
166 other => RpcError {
167 code: ErrorCode::ServerError.code(),
168 message: other.to_string(),
169 },
170 })
171}
172
173pub fn rpc_get_segment(store: &TapeStore, params: &Value) -> Result<Value, RpcError> {
188 let addr = params
189 .get("tape_address")
190 .and_then(Value::as_str)
191 .ok_or(RpcError {
192 code: ErrorCode::InvalidParams.code(),
193 message: "invalid or missing tape_address".into(),
194 })?;
195
196 let sn = params
197 .get("segment_number")
198 .and_then(Value::as_u64)
199 .ok_or(RpcError {
200 code: ErrorCode::InvalidParams.code(),
201 message: "invalid or missing segment_number".into(),
202 })?;
203
204 let pk = Pubkey::from_str(addr).map_err(|e| RpcError {
205 code: ErrorCode::InvalidParams.code(),
206 message: format!("invalid pubkey: {}", e),
207 })?;
208
209 store
210 .get_segment(&pk, sn)
211 .map(|data| json!(base64::encode(data)))
212 .map_err(|e| match e {
213 StoreError::SegmentNotFound(_, num) => RpcError {
214 code: ErrorCode::ServerError.code(),
215 message: format!("segment {} not found", num),
216 },
217 other => RpcError {
218 code: ErrorCode::ServerError.code(),
219 message: other.to_string(),
220 },
221 })
222}
223
224pub fn rpc_get_tape(store: &TapeStore, params: &Value) -> Result<Value, RpcError> {
239 let addr = params
240 .get("tape_address")
241 .and_then(Value::as_str)
242 .ok_or(RpcError {
243 code: ErrorCode::InvalidParams.code(),
244 message: "invalid or missing tape_address".into(),
245 })?;
246
247 let pk = Pubkey::from_str(addr).map_err(|e| RpcError {
248 code: ErrorCode::InvalidParams.code(),
249 message: format!("invalid pubkey: {}", e),
250 })?;
251
252 let segments = store.get_tape_segments(&pk).map_err(|e| RpcError {
253 code: ErrorCode::ServerError.code(),
254 message: e.to_string(),
255 })?;
256
257 let arr: Vec<Value> = segments
258 .into_iter()
259 .map(|(num, data)| {
260 json!({
261 "segment_number": num,
262 "data": base64::encode(data),
263 })
264 })
265 .collect();
266
267 Ok(json!(arr))
268}
269
270async fn rpc_handler(
271 State(store): State<Arc<TapeStore>>,
272 Json(req): Json<RpcRequest>,
273) -> impl IntoResponse {
274
275 let id = req.id.clone();
276 let outcome = match req.method.as_str() {
277 "getHealth" => rpc_get_health(&store, &req.params),
278 "getTapeAddress" => rpc_get_tape_address(&store, &req.params),
279 "getTapeNumber" => rpc_get_tape_number(&store, &req.params),
280 "getSegment" => rpc_get_segment(&store, &req.params),
281 "getTape" => rpc_get_tape(&store, &req.params),
282 _ => Err(RpcError {
283 code: ErrorCode::MethodNotFound.code(),
284 message: "method not found".into(),
285 }),
286 };
287
288 make_response(id, outcome)
289}
290
291pub async fn web_loop(
292 store: TapeStore,
293 port: u16,
294) -> anyhow::Result<()> {
295 let store = Arc::new(store);
296
297 {
299 let store = Arc::clone(&store);
300 tokio::spawn(async move {
301 let interval = std::time::Duration::from_secs(15);
302 loop {
303 store.catch_up_with_primary().unwrap();
304 tokio::time::sleep(interval).await;
305 }
306 });
307 }
308
309 let app = Router::new()
310 .route("/api", post(rpc_handler))
311 .with_state(store);
312
313 let addr = SocketAddr::from(([127, 0, 0, 1], port));
314 let listener = tokio::net::TcpListener::bind(&addr).await?;
315
316 axum::serve(listener, app).await?;
317
318 Ok(())
319}