+143

src/api.rs

··· 1 1 + use crate::types::{DidDocument, JsonCid, PlcEntry, PlcOperationType}; 2 2 + use crate::{ApiContext, db}; 3 3 + use dropshot::{ClientErrorStatusCode, HttpError, HttpResponseOk, Path, RequestContext, endpoint}; 4 4 + use ipld_core::cid::Cid; 5 5 + use schemars::JsonSchema; 6 6 + use serde::Deserialize; 7 7 + use std::str::FromStr; 8 8 + 9 9 + fn err_did_not_found(did: &str) -> HttpError { 10 10 + HttpError::for_not_found(None, format!("DID not registered: {did}")) 11 11 + } 12 12 + 13 13 + fn err_did_tombstoned(did: &str) -> HttpError { 14 14 + HttpError::for_client_error( 15 15 + None, 16 16 + ClientErrorStatusCode::GONE, 17 17 + format!("DID not available: {did}"), 18 18 + ) 19 19 + } 20 20 + 21 21 + #[derive(Debug, JsonSchema, Deserialize)] 22 22 + pub struct DidPathParams { 23 23 + pub did: String, 24 24 + } 25 25 + 26 26 + #[endpoint { 27 27 + method = GET, 28 28 + path = "/{did}", 29 29 + }] 30 30 + pub async fn resolve_did( 31 31 + rqctx: RequestContext<ApiContext>, 32 32 + path: Path<DidPathParams>, 33 33 + ) -> Result<HttpResponseOk<DidDocument>, HttpError> { 34 34 + let conn = rqctx.context().get_conn().await?; 35 35 + let did = path.into_inner().did; 36 36 + 37 37 + let op = db::get_latest_operation(&conn, &did) 38 38 + .await 39 39 + .map_err(|v| HttpError::for_internal_error(v.to_string()))? 40 40 + .ok_or(err_did_not_found(&did))?; 41 41 + 42 42 + let decoded: PlcOperationType = serde_json::from_value(op.operation) 43 43 + .map_err(|v| HttpError::for_internal_error(v.to_string()))?; 44 44 + 45 45 + let doc = match decoded { 46 46 + PlcOperationType::Tombstone(_) => return Err(err_did_tombstoned(&did)), 47 47 + PlcOperationType::Create(op) => DidDocument::from_create(&did, op), 48 48 + PlcOperationType::Operation(op) => DidDocument::from_plc_op(&did, op), 49 49 + }; 50 50 + 51 51 + Ok(HttpResponseOk(doc)) 52 52 + } 53 53 + 54 54 + #[endpoint { 55 55 + method = GET, 56 56 + path = "/{did}/log", 57 57 + }] 58 58 + pub async fn get_plc_op_log( 59 59 + rqctx: RequestContext<ApiContext>, 60 60 + path: Path<DidPathParams>, 61 61 + ) -> Result<HttpResponseOk<Vec<PlcOperationType>>, HttpError> { 62 62 + let conn = rqctx.context().get_conn().await?; 63 63 + let did = path.into_inner().did; 64 64 + 65 65 + let ops = db::get_operations(&conn, &did) 66 66 + .await 67 67 + .map_err(|v| HttpError::for_internal_error(v.to_string()))?; 68 68 + 69 69 + if ops.is_empty() { 70 70 + return Err(err_did_not_found(&did)); 71 71 + } 72 72 + 73 73 + let ops = ops 74 74 + .into_iter() 75 75 + .map(|op| serde_json::from_value(op.operation)) 76 76 + .collect::<Result<Vec<_>, _>>() 77 77 + .map_err(|v| HttpError::for_internal_error(v.to_string()))?; 78 78 + 79 79 + Ok(HttpResponseOk(ops)) 80 80 + } 81 81 + 82 82 + #[endpoint { 83 83 + method = GET, 84 84 + path = "/{did}/log/audit", 85 85 + }] 86 86 + pub async fn get_plc_audit_log( 87 87 + rqctx: RequestContext<ApiContext>, 88 88 + path: Path<DidPathParams>, 89 89 + ) -> Result<HttpResponseOk<Vec<PlcEntry>>, HttpError> { 90 90 + let conn = rqctx.context().get_conn().await?; 91 91 + let did = path.into_inner().did; 92 92 + 93 93 + let ops = db::get_operations(&conn, &did) 94 94 + .await 95 95 + .map_err(|v| HttpError::for_internal_error(v.to_string()))?; 96 96 + 97 97 + if ops.is_empty() { 98 98 + return Err(err_did_not_found(&did)); 99 99 + } 100 100 + 101 101 + let entries = ops 102 102 + .into_iter() 103 103 + .map(|op| { 104 104 + let operation = serde_json::from_value(op.operation)?; 105 105 + let cid = Cid::from_str(&op.hash).unwrap(); 106 106 + 107 107 + let entry = PlcEntry { 108 108 + did: did.clone(), 109 109 + operation, 110 110 + cid: JsonCid(cid), 111 111 + nullified: op.nullified, 112 112 + created_at: op.created_at, 113 113 + }; 114 114 + 115 115 + Ok(entry) 116 116 + }) 117 117 + .collect::<Result<Vec<_>, _>>() 118 118 + .map_err(|v: eyre::Report| HttpError::for_internal_error(v.to_string()))?; 119 119 + 120 120 + Ok(HttpResponseOk(entries)) 121 121 + } 122 122 + 123 123 + #[endpoint { 124 124 + method = GET, 125 125 + path = "/{did}/log/last", 126 126 + }] 127 127 + pub async fn get_last_op( 128 128 + rqctx: RequestContext<ApiContext>, 129 129 + path: Path<DidPathParams>, 130 130 + ) -> Result<HttpResponseOk<PlcOperationType>, HttpError> { 131 131 + let conn = rqctx.context().get_conn().await?; 132 132 + let did = path.into_inner().did; 133 133 + 134 134 + let op = db::get_latest_operation(&conn, &did) 135 135 + .await 136 136 + .map_err(|v| HttpError::for_internal_error(v.to_string()))? 137 137 + .ok_or(err_did_not_found(&did))?; 138 138 + 139 139 + let decoded = serde_json::from_value(op.operation) 140 140 + .map_err(|v| HttpError::for_internal_error(v.to_string()))?; 141 141 + 142 142 + Ok(HttpResponseOk(decoded)) 143 143 + }

+56

src/db.rs

··· 1 1 + use chrono::{DateTime, Utc}; 2 2 + use deadpool_postgres::Object; 3 3 + use serde_json::Value; 4 4 + use tokio_postgres::Row; 5 5 + 6 6 + pub struct Operation { 7 7 + pub did: String, 8 8 + pub hash: String, 9 9 + pub prev: Option<String>, 10 10 + pub sig: String, 11 11 + pub nullified: bool, 12 12 + pub operation: Value, 13 13 + pub created_at: DateTime<Utc>, 14 14 + } 15 15 + 16 16 + impl From<Row> for Operation { 17 17 + fn from(row: Row) -> Self { 18 18 + Operation { 19 19 + did: row.get(0), 20 20 + hash: row.get(1), 21 21 + prev: row.get(2), 22 22 + sig: row.get(3), 23 23 + nullified: row.get(4), 24 24 + operation: row.get(5), 25 25 + created_at: row.get(6), 26 26 + } 27 27 + } 28 28 + } 29 29 + 30 30 + pub async fn get_latest_operation( 31 31 + conn: &Object, 32 32 + did: &str, 33 33 + ) -> Result<Option<Operation>, tokio_postgres::Error> { 34 34 + let maybe_op = conn 35 35 + .query_opt( 36 36 + "SELECT * FROM operations WHERE did=$1 ORDER BY created_at DESC LIMIT 1", 37 37 + &[&did], 38 38 + ) 39 39 + .await?; 40 40 + 41 41 + Ok(maybe_op.map(Operation::from)) 42 42 + } 43 43 + 44 44 + pub async fn get_operations( 45 45 + conn: &Object, 46 46 + did: &str, 47 47 + ) -> Result<Vec<Operation>, tokio_postgres::Error> { 48 48 + let ops = conn 49 49 + .query( 50 50 + "SELECT * FROM operations WHERE did=$1 ORDER BY created_at", 51 51 + &[&did], 52 52 + ) 53 53 + .await?; 54 54 + 55 55 + Ok(ops.into_iter().map(Operation::from).collect()) 56 56 + }

+15 -5

src/lib.rs

··· 1 1 - use deadpool_postgres::{Manager, Pool}; 2 2 - use dropshot::{ApiDescription, ConfigLogging, ConfigLoggingLevel}; 1 1 + use deadpool_postgres::{Manager, Object, Pool}; 2 2 + use dropshot::{ApiDescription, ConfigLogging, ConfigLoggingLevel, HttpError}; 3 3 use eyre::Context; 4 4 use slog::Logger; 5 5 use std::env::var; 6 6 use std::str::FromStr; 7 7 use tokio_postgres::{Config, NoTls}; 8 8 9 9 + mod api; 9 10 mod db; 10 11 pub mod import; 11 12 mod types; ··· 16 17 pub pool: Pool, 17 18 } 18 19 20 20 + impl ApiContext { 21 21 + pub async fn get_conn(&self) -> Result<Object, HttpError> { 22 22 + self.pool.get().await.map_err(|err| HttpError::for_internal_error(err.to_string())) 23 23 + } 24 24 + } 25 25 + 19 26 pub fn create_logger() -> eyre::Result<Logger> { 20 27 let log = ConfigLogging::StderrTerminal { level: ConfigLoggingLevel::Info } 21 28 .to_logger("plc-mirror")?; ··· 24 31 } 25 32 26 33 pub fn create_api() -> eyre::Result<ApiDescription<ApiContext>> { 27 27 - let mut api = ApiDescription::new(); 34 34 + let mut api_desc = ApiDescription::new(); 28 35 29 29 - // TODO: api endpoints 36 36 + api_desc.register(api::get_plc_op_log)?; 37 37 + api_desc.register(api::get_plc_audit_log)?; 38 38 + api_desc.register(api::get_last_op)?; 39 39 + api_desc.register(api::resolve_did)?; 30 40 31 31 - Ok(api) 41 41 + Ok(api_desc) 32 42 } 33 43 34 44 pub async fn connect_db() -> eyre::Result<Pool> {