pds: more skeleton progress

1 files changed, 21 insertions, 50 deletions

diff --git a/adenosine-pds/src/lib.rs b/adenosine-pds/src/lib.rs
index 1785640..a8dc46a 100644
--- a/adenosine-pds/src/lib.rs
+++ b/adenosine-pds/src/lib.rs
@@ -1,23 +1,29 @@
 use anyhow::Result;
 use log::{error, info};
 use rouille::{router, Request, Response};
+use std::path::PathBuf;
+use std::sync::Mutex;
 
-use futures::TryStreamExt;
 use ipfs_sqlite_block_store::BlockStore;
-use iroh_car::CarReader;
-use libipld::block::Block;
-use std::path::PathBuf;
-use tokio::fs::File;
-use tokio::io::BufReader;
 
+mod car;
+mod db;
+mod models;
 mod mst;
 
+pub use car::{load_car_to_blockstore, load_car_to_sqlite};
+pub use db::AtpDatabase;
+pub use models::*;
 pub use mst::{dump_mst_keys, repro_mst};
 
-pub fn run_server(port: u16) -> Result<()> {
-    // TODO: log access requests
+pub fn run_server(port: u16, blockstore_db_path: &PathBuf, atp_db_path: &PathBuf) -> Result<()> {
     // TODO: some static files? https://github.com/tomaka/rouille/blob/master/examples/static-files.rs
 
+    // TODO: could just open connection on every request?
+    let db = Mutex::new(AtpDatabase::open(atp_db_path)?);
+    let mut _blockstore: BlockStore<libipld::DefaultParams> =
+        BlockStore::open(blockstore_db_path, Default::default())?;
+
     let log_ok = |req: &Request, _resp: &Response, elap: std::time::Duration| {
         info!("{} {} ({:?})", req.method(), req.raw_url(), elap);
     };
@@ -43,50 +49,15 @@ pub fn run_server(port: u16) -> Result<()> {
                     Response::text("didn't get other thing")
                     // TODO: parse and echo back JSON body
                 },
+
+                (GET) ["/xrpc/com.atproto.getRecord"] => {
+                    // TODO: JSON response
+                    // TODO: handle error
+                    let mut db = db.lock().unwrap().new_connection().unwrap();
+                    Response::text(db.get_record("asdf", "123", "blah").unwrap().to_string())
+                },
                 _ => rouille::Response::empty_404()
             )
         })
     });
 }
-
-pub fn load_car_to_sqlite(db_path: &PathBuf, car_path: &PathBuf) -> Result<()> {
-    let mut db: BlockStore<libipld::DefaultParams> =
-        { BlockStore::open(db_path, Default::default())? };
-
-    load_car_to_blockstore(&mut db, car_path)
-}
-
-pub fn load_car_to_blockstore(db: &mut BlockStore<libipld::DefaultParams>, car_path: &PathBuf) -> Result<()> {
-    let rt = tokio::runtime::Builder::new_current_thread()
-        .enable_all()
-        .build()?;
-    rt.block_on(inner_car_loader(db, car_path))
-}
-
-// this async function is wrapped in the sync version above
-async fn inner_car_loader(db: &mut BlockStore<libipld::DefaultParams>, car_path: &PathBuf) -> Result<()> {
-    println!("{} - {}", std::env::current_dir()?.display(), car_path.display());
-    let car_reader = {
-        let file = File::open(car_path).await?;
-        let buf_reader = BufReader::new(file);
-        CarReader::new(buf_reader).await?
-    };
-    let car_header = car_reader.header().clone();
-
-    car_reader
-        .stream()
-        .try_for_each(|(cid, raw)| {
-            // TODO: error handling here instead of unwrap (?)
-            let block = Block::new(cid, raw).unwrap();
-            db.put_block(block, None).unwrap();
-            futures::future::ready(Ok(()))
-        })
-        .await?;
-
-    // pin the header
-    if car_header.roots().len() >= 1 {
-        db.alias(b"import".to_vec(), Some(&car_header.roots()[0]))?;
-    }
-
-    Ok(())
-}