diff --git a/pocket/.gitignore b/pocket/.gitignore new file mode 100644 index 0000000..540ddfe --- /dev/null +++ b/pocket/.gitignore @@ -0,0 +1 @@ +prefs.sqlite3* diff --git a/pocket/src/lib.rs b/pocket/src/lib.rs index c6a7fd8..67b3a79 100644 --- a/pocket/src/lib.rs +++ b/pocket/src/lib.rs @@ -1,5 +1,7 @@ mod server; +mod storage; mod token; pub use server::serve; +pub use storage::Storage; pub use token::TokenVerifier; diff --git a/pocket/src/main.rs b/pocket/src/main.rs index 46a7ad9..986bef6 100644 --- a/pocket/src/main.rs +++ b/pocket/src/main.rs @@ -1,8 +1,34 @@ -use pocket::serve; +use clap::Parser; +use pocket::{Storage, serve}; +use std::path::PathBuf; + +/// Slingshot record edge cache +#[derive(Parser, Debug, Clone)] +#[command(version, about, long_about = None)] +struct Args { + /// path to the sqlite db file + #[arg(long)] + db: Option, + /// just initialize the db and exit + #[arg(long, action)] + init_db: bool, + /// the domain for serving a did doc (unused if running behind reflector) + #[arg(long)] + domain: Option, +} #[tokio::main] async fn main() { tracing_subscriber::fmt::init(); - println!("Hello, world!"); - serve("mac.cinnebar-tet.ts.net").await + log::info!("👖 hi"); + let args = Args::parse(); + let domain = args.domain.unwrap_or("bad-example.com".into()); + let db_path = args.db.unwrap_or("prefs.sqlite3".into()); + if args.init_db { + Storage::init(&db_path).unwrap(); + log::info!("👖 initialized db at {db_path:?}. bye") + } else { + let storage = Storage::connect(db_path).unwrap(); + serve(&domain, storage).await + } } diff --git a/pocket/src/server.rs b/pocket/src/server.rs index 309c107..f507db1 100644 --- a/pocket/src/server.rs +++ b/pocket/src/server.rs @@ -1,10 +1,10 @@ -use crate::TokenVerifier; +use crate::{Storage, TokenVerifier}; use poem::{ Endpoint, EndpointExt, Route, Server, endpoint::{StaticFileEndpoint, make_sync}, http::Method, listener::TcpListener, - middleware::{CatchPanic, Cors, SizeLimit, Tracing}, + middleware::{CatchPanic, Cors, Tracing}, }; use poem_openapi::{ ApiResponse, ContactObject, ExternalDocumentObject, Object, OpenApi, OpenApiService, @@ -15,6 +15,7 @@ use poem_openapi::{ }; use serde::Serialize; use serde_json::{Value, json}; +use std::sync::{Arc, Mutex}; #[derive(Debug, SecurityScheme)] #[oai(ty = "bearer")] @@ -51,13 +52,13 @@ fn xrpc_error(error: impl AsRef, message: impl AsRef) -> XrpcError { }) } -#[derive(Object)] +#[derive(Debug, Object)] #[oai(example = true)] -struct GetBskyPrefsResponseObject { +struct BskyPrefsObject { /// at-uri for this record preferences: Value, } -impl Example for GetBskyPrefsResponseObject { +impl Example for BskyPrefsObject { fn example() -> Self { Self { preferences: json!({ @@ -71,7 +72,7 @@ impl Example for GetBskyPrefsResponseObject { enum GetBskyPrefsResponse { /// Record found #[oai(status = 200)] - Ok(Json), + Ok(Json), /// Bad request or no preferences to return #[oai(status = 400)] BadRequest(XrpcError), @@ -92,6 +93,7 @@ enum PutBskyPrefsResponse { struct Xrpc { verifier: TokenVerifier, + storage: Arc>, } #[OpenApi] @@ -114,8 +116,40 @@ impl Xrpc { Err(e) => return GetBskyPrefsResponse::BadRequest(xrpc_error("boooo", e.to_string())), }; log::info!("verified did: {did}/{aud}"); - // TODO: fetch from storage - GetBskyPrefsResponse::Ok(Json(GetBskyPrefsResponseObject::example())) + + let storage = self.storage.clone(); + + let Ok(Ok(res)) = tokio::task::spawn_blocking(move || { + storage + .lock() + .unwrap() + .get(&did, &aud) + .inspect_err(|e| log::error!("failed to get prefs: {e}")) + }) + .await + else { + return GetBskyPrefsResponse::BadRequest(xrpc_error("boooo", "failed to get from db")); + }; + + let Some(serialized) = res else { + return GetBskyPrefsResponse::BadRequest(xrpc_error( + "NotFound", + "could not find prefs for u", + )); + }; + + let preferences = match serde_json::from_str(&serialized) { + Ok(v) => v, + Err(e) => { + log::error!("failed to deserialize prefs: {e}"); + return GetBskyPrefsResponse::BadRequest(xrpc_error( + "boooo", + "failed to deserialize prefs", + )); + } + }; + + GetBskyPrefsResponse::Ok(Json(BskyPrefsObject { preferences })) } /// com.bad-example.pocket.putPreferences @@ -129,7 +163,7 @@ impl Xrpc { async fn pocket_put_prefs( &self, XrpcAuth(auth): XrpcAuth, - Json(prefs): Json, + Json(prefs): Json, ) -> PutBskyPrefsResponse { let (did, aud) = match self .verifier @@ -141,8 +175,23 @@ impl Xrpc { }; log::info!("verified did: {did}/{aud}"); log::warn!("received prefs: {prefs:?}"); - // TODO: put prefs into storage - PutBskyPrefsResponse::Ok(PlainText("hiiiiii".to_string())) + + let storage = self.storage.clone(); + let serialized = prefs.preferences.to_string(); + + let Ok(Ok(())) = tokio::task::spawn_blocking(move || { + storage + .lock() + .unwrap() + .put(&did, &aud, &serialized) + .inspect_err(|e| log::error!("failed to insert prefs: {e}")) + }) + .await + else { + return PutBskyPrefsResponse::BadRequest(xrpc_error("boooo", "failed to put to db")); + }; + + PutBskyPrefsResponse::Ok(PlainText("saved.".to_string())) } } @@ -178,25 +227,31 @@ fn get_did_doc(domain: &str) -> impl Endpoint + use<> { make_sync(move |_| doc.clone()) } -pub async fn serve(domain: &str) -> () { +pub async fn serve(domain: &str, storage: Storage) -> () { let verifier = TokenVerifier::default(); - let api_service = OpenApiService::new(Xrpc { verifier }, "Pocket", env!("CARGO_PKG_VERSION")) - .server(domain) - .url_prefix("/xrpc") - .contact( - ContactObject::new() - .name("@microcosm.blue") - .url("https://bsky.app/profile/microcosm.blue"), - ) - .description(include_str!("../api-description.md")) - .external_document(ExternalDocumentObject::new("https://microcosm.blue/pocket")); + let api_service = OpenApiService::new( + Xrpc { + verifier, + storage: Arc::new(Mutex::new(storage)), + }, + "Pocket", + env!("CARGO_PKG_VERSION"), + ) + .server(domain) + .url_prefix("/xrpc") + .contact( + ContactObject::new() + .name("@microcosm.blue") + .url("https://bsky.app/profile/microcosm.blue"), + ) + .description(include_str!("../api-description.md")) + .external_document(ExternalDocumentObject::new("https://microcosm.blue/pocket")); let app = Route::new() .nest("/openapi", api_service.spec_endpoint()) .nest("/xrpc/", api_service) .at("/.well-known/did.json", get_did_doc(domain)) .at("/", StaticFileEndpoint::new("./static/index.html")) - .with(SizeLimit::new(100 * 2_usize.pow(10))) .with( Cors::new() .allow_method(Method::GET) diff --git a/pocket/src/storage.rs b/pocket/src/storage.rs new file mode 100644 index 0000000..1b6ff96 --- /dev/null +++ b/pocket/src/storage.rs @@ -0,0 +1,50 @@ +use rusqlite::{Connection, OptionalExtension, Result}; +use std::path::Path; + +pub struct Storage { + con: Connection, +} + +impl Storage { + pub fn connect(path: impl AsRef) -> Result { + let con = Connection::open(path)?; + con.pragma_update(None, "journal_mode", "WAL")?; + con.pragma_update(None, "synchronous", "NORMAL")?; + con.pragma_update(None, "busy_timeout", "100")?; + con.pragma_update(None, "foreign_keys", "ON")?; + Ok(Self { con }) + } + pub fn init(path: impl AsRef) -> Result { + let me = Self::connect(path)?; + me.con.execute( + r#" + create table prefs ( + actor text not null, + aud text not null, + pref text not null, + primary key (actor, aud) + ) strict"#, + (), + )?; + Ok(me) + } + pub fn put(&self, actor: &str, aud: &str, pref: &str) -> Result<()> { + self.con.execute( + r#"insert into prefs (actor, aud, pref) + values (?1, ?2, ?3) + on conflict do update set pref = excluded.pref"#, + [actor, aud, pref], + )?; + Ok(()) + } + pub fn get(&self, actor: &str, aud: &str) -> Result> { + self.con + .query_one( + r#"select pref from prefs + where actor = ?1 and aud = ?2"#, + [actor, aud], + |row| row.get(0), + ) + .optional() + } +}