1use std::sync::Arc;
19
20use dig_rpc_protocol::{
21 envelope::{JsonRpcRequest, JsonRpcResponse, RequestId},
22 openrpc, ErrorCode, ErrorOrigin, Method, RpcError, Tier,
23};
24use serde_json::Value;
25
26use crate::handler::RpcHandler;
27
28#[derive(Debug, Clone, Copy, PartialEq, Eq)]
31pub enum Surface {
32 Loopback,
35 PublicRead,
38 Peer,
41}
42
43impl Surface {
44 pub const fn discriminant(self) -> u8 {
46 match self {
47 Surface::Loopback => 0,
48 Surface::PublicRead => 1,
49 Surface::Peer => 2,
50 }
51 }
52
53 fn allows(self, method: Method) -> bool {
55 match self {
56 Surface::Loopback => true,
58 Surface::PublicRead => method.tier() == Tier::PublicRead,
60 Surface::Peer => method.is_peer_reachable(),
62 }
63 }
64
65 fn rejection(self, method: Method) -> RpcError {
70 if method.tier() == Tier::Control && self != Surface::Loopback {
71 RpcError::new(
72 ErrorCode::Unauthorized,
73 format!(
74 "{} is a control method; reachable only on the loopback surface",
75 method.name()
76 ),
77 ErrorOrigin::Control,
78 )
79 } else {
80 RpcError::of(
81 ErrorCode::MethodNotFound,
82 format!("method {} not available on this surface", method.name()),
83 )
84 }
85 }
86}
87
88pub async fn dispatch<H: RpcHandler + ?Sized>(
92 handler: &H,
93 surface: Surface,
94 req: JsonRpcRequest<Value>,
95) -> JsonRpcResponse<Value> {
96 let id = req.id.clone();
97
98 let Some(method) = Method::from_name(&req.method) else {
100 return JsonRpcResponse::error(
101 id,
102 RpcError::of(
103 ErrorCode::MethodNotFound,
104 format!("method {:?} not found", req.method),
105 ),
106 );
107 };
108
109 if !surface.allows(method) {
111 return JsonRpcResponse::error(id, surface.rejection(method));
112 }
113
114 if method == Method::RpcDiscover {
117 return JsonRpcResponse::success(id, openrpc::openrpc_document(&handler.version()));
118 }
119
120 let params = req.params.unwrap_or(Value::Null);
121 match handler.handle(method, params).await {
122 Ok(result) => JsonRpcResponse::success(id, result),
123 Err(err) => JsonRpcResponse::error(id, err),
124 }
125}
126
127pub fn parse_error_response(id: RequestId, message: impl Into<String>) -> JsonRpcResponse<Value> {
131 JsonRpcResponse::error(id, RpcError::of(ErrorCode::ParseError, message))
132}
133
134pub type SharedHandler = Arc<dyn RpcHandler>;
136
137#[cfg(test)]
138mod tests {
139 use super::*;
140 use async_trait::async_trait;
141 use dig_rpc_protocol::envelope::JsonRpcResponseBody;
142 use serde_json::json;
143
144 struct Echo;
147 #[async_trait]
148 impl RpcHandler for Echo {
149 async fn handle(&self, method: Method, _params: Value) -> Result<Value, RpcError> {
150 Ok(json!({ "method": method.name() }))
151 }
152 }
153
154 fn req(method: &str) -> JsonRpcRequest<Value> {
155 JsonRpcRequest::new(1, method, json!({}))
156 }
157
158 fn err_of(resp: &JsonRpcResponse<Value>) -> &RpcError {
159 match &resp.body {
160 JsonRpcResponseBody::Error { error } => error,
161 _ => panic!("expected error, got {resp:?}"),
162 }
163 }
164
165 #[tokio::test]
167 async fn unknown_method_not_found() {
168 let resp = dispatch(&Echo, Surface::Loopback, req("dig.nope")).await;
169 assert_eq!(err_of(&resp).code, ErrorCode::MethodNotFound);
170 assert_eq!(resp.id, RequestId::Num(1));
171 }
172
173 #[tokio::test]
177 async fn control_method_gated_to_loopback() {
178 let ok = dispatch(&Echo, Surface::Loopback, req("cache.clear")).await;
179 assert!(matches!(ok.body, JsonRpcResponseBody::Success { .. }));
180
181 for surface in [Surface::Peer, Surface::PublicRead] {
182 let resp = dispatch(&Echo, surface, req("cache.clear")).await;
183 assert_eq!(err_of(&resp).code, ErrorCode::Unauthorized, "{surface:?}");
184 assert_eq!(err_of(&resp).data.origin, ErrorOrigin::Control);
185 }
186 }
187
188 #[tokio::test]
192 async fn public_read_not_on_peer_unless_allowlisted() {
193 let resp = dispatch(&Echo, Surface::Peer, req("dig.getManifest")).await;
195 assert_eq!(err_of(&resp).code, ErrorCode::MethodNotFound);
196
197 let ok = dispatch(&Echo, Surface::Peer, req("dig.getContent")).await;
199 assert!(matches!(ok.body, JsonRpcResponseBody::Success { .. }));
200 }
201
202 #[tokio::test]
205 async fn anchored_reads_served_on_peer() {
206 for m in [
207 "dig.getAnchoredRoot",
208 "dig.getCollection",
209 "dig.listCollectionItems",
210 ] {
211 let resp = dispatch(&Echo, Surface::Peer, req(m)).await;
212 assert!(
213 matches!(resp.body, JsonRpcResponseBody::Success { .. }),
214 "{m}"
215 );
216 }
217 }
218
219 #[tokio::test]
222 async fn rpc_discover_served_from_generator() {
223 let resp = dispatch(&Echo, Surface::Loopback, req("rpc.discover")).await;
224 match resp.body {
225 JsonRpcResponseBody::Success { result } => {
226 assert_eq!(result["openrpc"], "1.2.6");
227 assert!(result.get("methods").is_some());
229 }
230 _ => panic!("expected discovery document"),
231 }
232 let peer = dispatch(&Echo, Surface::Peer, req("rpc.discover")).await;
234 assert_eq!(err_of(&peer).code, ErrorCode::Unauthorized);
235 }
236
237 #[tokio::test]
239 async fn handler_error_propagates() {
240 struct Failing;
241 #[async_trait]
242 impl RpcHandler for Failing {
243 async fn handle(&self, _m: Method, _p: Value) -> Result<Value, RpcError> {
244 Err(RpcError::of(ErrorCode::RootNotAnchored, "stale root"))
245 }
246 }
247 let resp = dispatch(&Failing, Surface::Loopback, req("dig.getContent")).await;
248 assert_eq!(err_of(&resp).code, ErrorCode::RootNotAnchored);
249 assert_eq!(err_of(&resp).data.code, "ROOT_NOT_ANCHORED");
250 }
251
252 #[test]
254 fn parse_error_shape() {
255 let resp = parse_error_response(RequestId::Null, "bad json");
256 assert_eq!(err_of(&resp).code, ErrorCode::ParseError);
257 }
258}