pds: more bsky posts/feeds/follow/like progress

6 files changed, 427 insertions, 52 deletions

diff --git a/adenosine-pds/Cargo.toml b/adenosine-pds/Cargo.toml
index 7075e65..1637164 100644
--- a/adenosine-pds/Cargo.toml
+++ b/adenosine-pds/Cargo.toml
@@ -44,6 +44,7 @@ bs58 = "*"
 async-trait = "*"
 dotenv = "*"
 askama = "0.11"
+time = { version = "*", features = ["formatting"] }
 
 # for vendored iroh-car
 thiserror = "1.0"
diff --git a/adenosine-pds/src/atp_db.sql b/adenosine-pds/src/atp_db.sql
index 71fbf6d..9a6e30c 100644
--- a/adenosine-pds/src/atp_db.sql
+++ b/adenosine-pds/src/atp_db.sql
@@ -33,19 +33,22 @@ CREATE TABLE bsky_post(
     tid                 TEXT NOT NULL,
     cid                 TEXT NOT NULL,
     record_json         TEXT NOT NULL,
-    reply_root_uri      TEXT,
+    reply_to_parent_uri    TEXT,
+    reply_to_root_uri      TEXT,
     created_at          TIMESTAMP WITH TIME ZONE  NOT NULL,
     indexed_at          TIMESTAMP WITH TIME ZONE  NOT NULL DEFAULT ( DATETIME('now') ),
     PRIMARY KEY(did, tid)
 );
-CREATE INDEX bsky_post_reply_root_uri_idx on bsky_post(reply_root_uri);
+CREATE INDEX bsky_post_reply_to_parent_uri_idx on bsky_post(reply_to_parent_uri);
+CREATE INDEX bsky_post_reply_to_root_uri_idx on bsky_post(reply_to_root_uri);
 
 CREATE TABLE bsky_ref(
     ref_type            TEXT NOT NULL,
     did                 TEXT NOT NULL,
     tid                 TEXT NOT NULL,
     subject_uri         TEXT NOT NULL,
-    subject_cid         TEXT NOT NULL,
+    -- TODO: NOT NULL on subject_cid
+    subject_cid         TEXT,
     created_at          TIMESTAMP WITH TIME ZONE  NOT NULL,
     indexed_at          TIMESTAMP WITH TIME ZONE  NOT NULL DEFAULT ( DATETIME('now') ),
     PRIMARY KEY(ref_type, did, tid)
@@ -56,7 +59,8 @@ CREATE TABLE bsky_follow(
     did                 TEXT NOT NULL,
     tid                 TEXT NOT NULL,
     subject_did         TEXT NOT NULL,
-    subject_cid         TEXT NOT NULL,
+    -- TODO: NOT NULL on subject_cid
+    subject_cid         TEXT,
     created_at          TIMESTAMP WITH TIME ZONE  NOT NULL,
     indexed_at          TIMESTAMP WITH TIME ZONE  NOT NULL DEFAULT ( DATETIME('now') ),
     PRIMARY KEY(did, tid)
@@ -68,7 +72,8 @@ CREATE TABLE bsky_notification(
     pk                  INTEGER PRIMARY KEY AUTOINCREMENT,
     user_did            TEXT NOT NULL,
     subject_uri         TEXT NOT NULL,
-    subject_cid         TEXT NOT NULL,
+    -- TODO: NOT NULL on subject_cid
+    subject_cid         TEXT,
     reason              TEXT NOT NULL,
     seen_at             TIMESTAMP WITH TIME ZONE,
     indexed_at          TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT ( DATETIME('now') )
diff --git a/adenosine-pds/src/bsky.rs b/adenosine-pds/src/bsky.rs
index eeab9e3..66b05eb 100644
--- a/adenosine-pds/src/bsky.rs
+++ b/adenosine-pds/src/bsky.rs
@@ -91,14 +91,13 @@ pub fn bsky_get_profile(srv: &mut AtpService, did: &Did) -> Result<Profile> {
     let mut stmt = srv
         .atp_db
         .conn
-        .prepare_cached("SELECT COUNT(*) FROM bsky_ref WHERE ref_type = 'follow' AND did = $1")?;
+        .prepare_cached("SELECT COUNT(*) FROM bsky_follow WHERE did = $1")?;
     let follows_count: u64 = stmt.query_row(params!(did.to_string()), |row| row.get(0))?;
-    let uri = format!("at://{}", did);
     let mut stmt = srv
         .atp_db
         .conn
-        .prepare_cached("SELECT COUNT(*) FROM bsky_ref WHERE ref_type = 'follow' AND uri = $1")?;
-    let followers_count: u64 = stmt.query_row(params!(uri), |row| row.get(0))?;
+        .prepare_cached("SELECT COUNT(*) FROM bsky_follow WHERE subject_did = $1")?;
+    let followers_count: u64 = stmt.query_row(params!(did.to_string()), |row| row.get(0))?;
     Ok(Profile {
         did: did.to_string(),
         handle: handle,
@@ -120,7 +119,7 @@ pub fn bsky_update_profile(srv: &mut AtpService, did: &Did, profile: ProfileReco
     let prefix = "/app.bsky.actor.profile/";
     for (mst_key, _cid) in full_map.iter() {
         if mst_key.starts_with(&prefix) {
-            profile_tid = Some(Tid::from_str(mst_key.split('/').nth(1).unwrap())?);
+            profile_tid = Some(Tid::from_str(mst_key.split('/').nth(2).unwrap())?);
         }
     }
     let profile_tid: Tid = profile_tid.unwrap_or(srv.tid_gen.next_tid());
@@ -139,6 +138,7 @@ struct FeedRow {
     pub item_handle: String,
     pub item_post_tid: Tid,
     pub item_post_cid: Cid,
+    pub indexed_at: String,
 }
 
 fn feed_row(row: &rusqlite::Row) -> Result<FeedRow> {
@@ -149,45 +149,66 @@ fn feed_row(row: &rusqlite::Row) -> Result<FeedRow> {
     let item_post_tid = Tid::from_str(&item_post_tid)?;
     let cid_string: String = row.get(3)?;
     let item_post_cid = Cid::from_str(&cid_string)?;
+    let indexed_at: String = row.get(4)?;
     Ok(FeedRow {
         item_did,
         item_handle,
         item_post_tid,
         item_post_cid,
+        indexed_at,
     })
 }
 
 fn feed_row_to_item(srv: &mut AtpService, row: FeedRow) -> Result<FeedItem> {
     let record_ipld = srv.repo.get_ipld(&row.item_post_cid)?;
+    let uri = format!(
+        "at://{}/{}/{}",
+        row.item_did, "app.bsky.feed.post", row.item_post_tid
+    );
+
+    let mut stmt = srv.atp_db.conn.prepare_cached(
+        "SELECT COUNT(*) FROM bsky_ref WHERE ref_type = 'like' AND subject_uri = $1",
+    )?;
+    let like_count: u64 = stmt.query_row(params!(uri), |row| row.get(0))?;
+
+    let mut stmt = srv.atp_db.conn.prepare_cached(
+        "SELECT COUNT(*) FROM bsky_ref WHERE ref_type = 'repost' AND subject_uri = $1",
+    )?;
+    let repost_count: u64 = stmt.query_row(params!(uri), |row| row.get(0))?;
+
+    let mut stmt = srv
+        .atp_db
+        .conn
+        .prepare_cached("SELECT COUNT(*) FROM bsky_post WHERE reply_to_parent_uri = $1")?;
+    let reply_count: u64 = stmt.query_row(params!(uri), |row| row.get(0))?;
+
     let feed_item = FeedItem {
-        uri: format!(
-            "at://{}/{}/{}",
-            row.item_did, "app.bsky.feed.post", row.item_post_tid
-        ),
-        cid: row.item_post_cid.to_string(),
+        uri: uri,
+        cid: Some(row.item_post_cid.to_string()),
         author: User {
             did: row.item_did.to_string(),
             handle: row.item_handle,
-            displayName: None, // TODO
+            displayName: None, // TODO: fetch from profile (or cache)
         },
         repostedBy: None,
         record: ipld_into_json_value(record_ipld),
         embed: None,
-        replyCount: 0,                 // TODO
-        repostCount: 0,                // TODO
-        likeCount: 0,                  // TODO
-        indexedAt: "TODO".to_string(), // TODO
+        replyCount: reply_count,
+        repostCount: repost_count,
+        likeCount: like_count,
+        indexedAt: row.indexed_at,
         myState: None,
     };
     Ok(feed_item)
 }
 
-pub fn bsky_get_author_feed(srv: &mut AtpService, did: &Did) -> Result<GenericFeed> {
+pub fn bsky_get_timeline(srv: &mut AtpService, did: &Did) -> Result<GenericFeed> {
     let mut feed: Vec<FeedItem> = vec![];
+    // TODO: also handle reposts
     let rows = {
         let mut stmt = srv.atp_db
             .conn
-            .prepare_cached("SELECT account.did, account.handle, bsky_post.tid, bsky_post.cid, FROM bsky_post LEFT JOIN account ON bsky_post.did = account.did LEFT JOIN bsky_follow ON bsky_post.did = bsky_follow.subject_did WHERE bsky_follow.did = ?1 ORDER BY bsky_post.tid DESC LIMIT 20")?;
+            .prepare_cached("SELECT account.did, account.handle, bsky_post.tid, bsky_post.cid, bsky_post.indexed_at FROM bsky_post LEFT JOIN account ON bsky_post.did = account.did LEFT JOIN bsky_follow ON bsky_post.did = bsky_follow.subject_did WHERE bsky_follow.did = ?1 AND account.did IS NOT NULL ORDER BY bsky_post.tid DESC LIMIT 20")?;
         let mut sql_rows = stmt.query(params!(did.to_string()))?;
         let mut rows: Vec<FeedRow> = vec![];
         while let Some(sql_row) = sql_rows.next()? {
@@ -202,12 +223,13 @@ pub fn bsky_get_author_feed(srv: &mut AtpService, did: &Did) -> Result<GenericFe
     Ok(GenericFeed { feed })
 }
 
-pub fn bsky_get_timeline(srv: &mut AtpService, did: &Did) -> Result<GenericFeed> {
+pub fn bsky_get_author_feed(srv: &mut AtpService, did: &Did) -> Result<GenericFeed> {
     let mut feed: Vec<FeedItem> = vec![];
+    // TODO: also handle reposts
     let rows = {
         let mut stmt = srv.atp_db
             .conn
-            .prepare_cached("SELECT account.did, account.handle, bsky_post.tid, bsky_post.cid, FROM bsky_post LEFT JOIN account ON bsky_post.did = account.did WHERE bsky_post.did = ?1 ORDER BY bsky_post.tid DESC LIMIT 20")?;
+            .prepare_cached("SELECT account.did, account.handle, bsky_post.tid, bsky_post.cid, bsky_post.indexed_at FROM bsky_post LEFT JOIN account ON bsky_post.did = account.did WHERE bsky_post.did = ?1 ORDER BY bsky_post.tid DESC LIMIT 20")?;
         let mut sql_rows = stmt.query(params!(did.to_string()))?;
         let mut rows: Vec<FeedRow> = vec![];
         while let Some(sql_row) = sql_rows.next()? {
@@ -231,3 +253,315 @@ pub fn bsky_get_thread(
     // start?
     unimplemented!()
 }
+
+#[test]
+fn test_bsky_profile() {
+    use crate::{create_account, created_at_now};
+    use libipld::ipld;
+
+    let post_nsid = Nsid::from_str("app.bsky.feed.post").unwrap();
+    let follow_nsid = Nsid::from_str("app.bsky.graph.follow").unwrap();
+
+    let mut srv = AtpService::new_ephemeral().unwrap();
+    let req = AccountRequest {
+        email: "test@bogus.com".to_string(),
+        handle: "handle.test".to_string(),
+        password: "bogus".to_string(),
+        inviteCode: None,
+        recoveryKey: None,
+    };
+    let session = create_account(&mut srv, &req, true).unwrap();
+    let did = Did::from_str(&session.did).unwrap();
+    let profile = bsky_get_profile(&mut srv, &did).unwrap();
+    assert_eq!(profile.did, session.did);
+    assert_eq!(profile.handle, req.handle);
+    assert_eq!(profile.displayName, None);
+    assert_eq!(profile.description, None);
+    assert_eq!(profile.followersCount, 0);
+    assert_eq!(profile.followsCount, 0);
+    assert_eq!(profile.postsCount, 0);
+
+    let record = ProfileRecord {
+        displayName: "Test Name".to_string(),
+        description: Some("short description".to_string()),
+    };
+    bsky_update_profile(&mut srv, &did, record.clone()).unwrap();
+    let profile = bsky_get_profile(&mut srv, &did).unwrap();
+    assert_eq!(profile.displayName, Some(record.displayName));
+    assert_eq!(profile.description, record.description);
+
+    let record = ProfileRecord {
+        displayName: "New Test Name".to_string(),
+        description: Some("longer description".to_string()),
+    };
+    bsky_update_profile(&mut srv, &did, record.clone()).unwrap();
+    let profile = bsky_get_profile(&mut srv, &did).unwrap();
+    assert_eq!(profile.displayName, Some(record.displayName));
+    assert_eq!(profile.description, record.description);
+
+    let mutations = vec![
+        Mutation::Create(
+            follow_nsid.clone(),
+            srv.tid_gen.next_tid(),
+            ipld!({"subject": {"did": session.did}, "createdAt": created_at_now()}),
+        ),
+        Mutation::Create(
+            follow_nsid.clone(),
+            srv.tid_gen.next_tid(),
+            ipld!({"subject": {"did": "did:web:external.domain"}, "createdAt": created_at_now()}),
+        ),
+        Mutation::Create(
+            post_nsid.clone(),
+            srv.tid_gen.next_tid(),
+            ipld!({"text": "first post"}),
+        ),
+        Mutation::Create(
+            post_nsid.clone(),
+            srv.tid_gen.next_tid(),
+            ipld!({"text": "second post"}),
+        ),
+        Mutation::Create(
+            post_nsid.clone(),
+            srv.tid_gen.next_tid(),
+            ipld!({"text": "third post"}),
+        ),
+    ];
+    srv.repo
+        .mutate_repo(&did, &mutations, &srv.pds_keypair)
+        .unwrap();
+    bsky_mutate_db(&mut srv.atp_db, &did, mutations).unwrap();
+
+    let profile = bsky_get_profile(&mut srv, &did).unwrap();
+    assert_eq!(profile.followersCount, 1);
+    assert_eq!(profile.followsCount, 2);
+    assert_eq!(profile.postsCount, 3);
+}
+
+#[test]
+fn test_bsky_feeds() {
+    use crate::{create_account, created_at_now};
+    use libipld::ipld;
+
+    let post_nsid = Nsid::from_str("app.bsky.feed.post").unwrap();
+    let like_nsid = Nsid::from_str("app.bsky.feed.like").unwrap();
+    let repost_nsid = Nsid::from_str("app.bsky.feed.repost").unwrap();
+    let follow_nsid = Nsid::from_str("app.bsky.graph.follow").unwrap();
+
+    let mut srv = AtpService::new_ephemeral().unwrap();
+    let alice_did = {
+        let req = AccountRequest {
+            email: "alice@bogus.com".to_string(),
+            handle: "alice.test".to_string(),
+            password: "bogus".to_string(),
+            inviteCode: None,
+            recoveryKey: None,
+        };
+        let session = create_account(&mut srv, &req, true).unwrap();
+        Did::from_str(&session.did).unwrap()
+    };
+    let bob_did = {
+        let req = AccountRequest {
+            email: "bob@bogus.com".to_string(),
+            handle: "bob.test".to_string(),
+            password: "bogus".to_string(),
+            inviteCode: None,
+            recoveryKey: None,
+        };
+        let session = create_account(&mut srv, &req, true).unwrap();
+        Did::from_str(&session.did).unwrap()
+    };
+    let carol_did = {
+        let req = AccountRequest {
+            email: "carol@bogus.com".to_string(),
+            handle: "carol.test".to_string(),
+            password: "bogus".to_string(),
+            inviteCode: None,
+            recoveryKey: None,
+        };
+        let session = create_account(&mut srv, &req, true).unwrap();
+        Did::from_str(&session.did).unwrap()
+    };
+
+    // all feeds and timelines should be empty
+    let alice_feed = bsky_get_author_feed(&mut srv, &alice_did).unwrap();
+    let alice_timeline = bsky_get_timeline(&mut srv, &alice_did).unwrap();
+    assert!(alice_feed.feed.is_empty());
+    assert!(alice_timeline.feed.is_empty());
+    let bob_feed = bsky_get_author_feed(&mut srv, &bob_did).unwrap();
+    let bob_timeline = bsky_get_timeline(&mut srv, &bob_did).unwrap();
+    assert!(bob_feed.feed.is_empty());
+    assert!(bob_timeline.feed.is_empty());
+    let carol_feed = bsky_get_author_feed(&mut srv, &carol_did).unwrap();
+    let carol_timeline = bsky_get_timeline(&mut srv, &carol_did).unwrap();
+    assert!(carol_feed.feed.is_empty());
+    assert!(carol_timeline.feed.is_empty());
+
+    // alice does some posts
+    let alice_post1_tid = srv.tid_gen.next_tid();
+    let alice_post2_tid = srv.tid_gen.next_tid();
+    let alice_post3_tid = srv.tid_gen.next_tid();
+    assert!(alice_post1_tid < alice_post2_tid && alice_post2_tid < alice_post3_tid);
+    let mutations = vec![
+        Mutation::Create(
+            post_nsid.clone(),
+            alice_post1_tid.clone(),
+            ipld!({"text": "alice first post"}),
+        ),
+        Mutation::Create(
+            post_nsid.clone(),
+            alice_post2_tid.clone(),
+            ipld!({"text": "alice second post"}),
+        ),
+        Mutation::Create(
+            post_nsid.clone(),
+            alice_post3_tid.clone(),
+            ipld!({"text": "alice third post"}),
+        ),
+    ];
+    srv.repo
+        .mutate_repo(&alice_did, &mutations, &srv.pds_keypair)
+        .unwrap();
+    bsky_mutate_db(&mut srv.atp_db, &alice_did, mutations).unwrap();
+
+    // bob follows alice, likes first post, reposts second, replies third
+    let alice_post3_uri = format!(
+        "at://{}/{}/{}",
+        alice_did.to_string(),
+        post_nsid.to_string(),
+        alice_post3_tid.to_string()
+    );
+    let mutations = vec![
+        Mutation::Create(
+            follow_nsid.clone(),
+            srv.tid_gen.next_tid(),
+            ipld!({"subject": {"did": alice_did.to_string()}, "createdAt": created_at_now()}),
+        ),
+        Mutation::Create(
+            like_nsid.clone(),
+            srv.tid_gen.next_tid(),
+            ipld!({"subject": {"uri": format!("at://{}/{}/{}", alice_did.to_string(), post_nsid.to_string(), alice_post1_tid.to_string())}, "createdAt": created_at_now()}),
+        ),
+        Mutation::Create(
+            repost_nsid.clone(),
+            srv.tid_gen.next_tid(),
+            ipld!({"subject": {"uri": format!("at://{}/{}/{}", alice_did.to_string(), post_nsid.to_string(), alice_post2_tid.to_string())}, "createdAt": created_at_now()}),
+        ),
+        Mutation::Create(
+            post_nsid.clone(),
+            srv.tid_gen.next_tid(),
+            ipld!({"text": "bob comment on alice post3", "reply": {"parent": {"uri": alice_post3_uri.clone()}, "root": {"uri": alice_post3_uri.clone()}}}),
+        ),
+    ];
+    srv.repo
+        .mutate_repo(&bob_did, &mutations, &srv.pds_keypair)
+        .unwrap();
+    bsky_mutate_db(&mut srv.atp_db, &bob_did, mutations).unwrap();
+
+    // carol follows bob
+    let mutations = vec![Mutation::Create(
+        follow_nsid.clone(),
+        srv.tid_gen.next_tid(),
+        ipld!({"subject": {"did": bob_did.to_string()}, "createdAt": created_at_now()}),
+    )];
+    srv.repo
+        .mutate_repo(&bob_did, &mutations, &srv.pds_keypair)
+        .unwrap();
+    bsky_mutate_db(&mut srv.atp_db, &carol_did, mutations).unwrap();
+
+    // test alice profile: counts should be updated
+    let alice_profile = bsky_get_profile(&mut srv, &alice_did).unwrap();
+    assert_eq!(alice_profile.followersCount, 1);
+    assert_eq!(alice_profile.followsCount, 0);
+    assert_eq!(alice_profile.postsCount, 3);
+
+    // test alice timeline: still empty (?)
+    let alice_timeline = bsky_get_timeline(&mut srv, &alice_did).unwrap();
+    println!("{:?}", alice_timeline);
+    assert!(alice_timeline.feed.is_empty());
+
+    // test alice feed: should have 3 posts, with correct counts
+    let alice_feed = bsky_get_author_feed(&mut srv, &alice_did).unwrap();
+    assert_eq!(alice_feed.feed.len(), 3);
+
+    assert_eq!(
+        alice_feed.feed[2].uri,
+        format!(
+            "at://{}/{}/{}",
+            alice_did.to_string(),
+            post_nsid.to_string(),
+            alice_post1_tid.to_string()
+        )
+    );
+    // TODO: CID
+    assert_eq!(alice_feed.feed[2].author.did, alice_did.to_string());
+    assert_eq!(alice_feed.feed[2].author.handle, "alice.test");
+    assert_eq!(alice_feed.feed[2].repostedBy, None);
+    assert_eq!(
+        alice_feed.feed[2].record["text"].as_str().unwrap(),
+        "alice first post"
+    );
+    assert_eq!(alice_feed.feed[2].embed, None);
+    assert_eq!(alice_feed.feed[2].replyCount, 0);
+    assert_eq!(alice_feed.feed[2].repostCount, 0);
+    assert_eq!(alice_feed.feed[2].likeCount, 1);
+
+    assert_eq!(alice_feed.feed[1].author.did, alice_did.to_string());
+    assert_eq!(alice_feed.feed[1].replyCount, 0);
+    assert_eq!(alice_feed.feed[1].repostCount, 1);
+    assert_eq!(alice_feed.feed[1].likeCount, 0);
+
+    assert_eq!(alice_feed.feed[0].author.did, alice_did.to_string());
+    assert_eq!(alice_feed.feed[0].replyCount, 1);
+    assert_eq!(alice_feed.feed[0].repostCount, 0);
+    assert_eq!(alice_feed.feed[0].likeCount, 0);
+
+    // test bob timeline: should include alice posts
+    let bob_timeline = bsky_get_timeline(&mut srv, &bob_did).unwrap();
+    println!("BOB TIMELINE ======");
+    for item in bob_timeline.feed.iter() {
+        println!("{:?}", item);
+    }
+    assert_eq!(bob_timeline.feed.len(), 3);
+    assert_eq!(
+        bob_timeline.feed[2].uri,
+        format!(
+            "at://{}/{}/{}",
+            alice_did.to_string(),
+            post_nsid.to_string(),
+            alice_post1_tid.to_string()
+        )
+    );
+    // TODO: CID
+    assert_eq!(bob_timeline.feed[2].author.did, alice_did.to_string());
+    assert_eq!(bob_timeline.feed[2].author.handle, "alice.test");
+
+    // test bob feed: should include repost and reply
+    let bob_feed = bsky_get_author_feed(&mut srv, &bob_did).unwrap();
+    assert_eq!(bob_feed.feed.len(), 1);
+    // TODO: handle reposts
+    /*
+    assert_eq!(bob_feed.feed.len(), 2);
+    assert_eq!(bob_feed.feed[1].uri, format!("at://{}/{}/{}", alice_did.to_string(), post_nsid.to_string(), alice_post1_tid.to_string()));
+    // TODO: CID
+    assert_eq!(bob_feed.feed[1].author.did, alice_did.to_string());
+    assert_eq!(bob_feed.feed[1].author.handle, "alice.test");
+    assert_eq!(bob_feed.feed[1].repostedBy.as_ref().unwrap().did, bob_did.to_string());
+    assert_eq!(bob_feed.feed[1].repostedBy.as_ref().unwrap().handle, "bob.test");
+    // TODO: "is a repost" (check record?)
+    */
+
+    assert_eq!(bob_feed.feed[0].author.did, bob_did.to_string());
+    assert_eq!(bob_feed.feed[0].author.handle, "bob.test");
+
+    // test carol timeline: should include bob's repost and reply
+    let carol_timeline = bsky_get_timeline(&mut srv, &carol_did).unwrap();
+    // TODO: handle re-posts (+1 here)
+    assert_eq!(carol_timeline.feed.len(), 1);
+    // TODO: details
+
+    // test carol feed: still empty
+    let carol_feed = bsky_get_author_feed(&mut srv, &carol_did).unwrap();
+    assert!(carol_feed.feed.is_empty());
+}
+
+// TODO: post threads (include in the above test?)
diff --git a/adenosine-pds/src/db.rs b/adenosine-pds/src/db.rs
index 384b498..051588d 100644
--- a/adenosine-pds/src/db.rs
+++ b/adenosine-pds/src/db.rs
@@ -1,6 +1,6 @@
 use crate::models::{FollowRecord, Post, RefRecord};
 /// ATP database (as distinct from blockstore)
-use crate::{ipld_into_json_value, AtpSession, Did, KeyPair, Tid};
+use crate::{created_at_now, ipld_into_json_value, AtpSession, Did, KeyPair, Tid};
 use anyhow::{anyhow, Result};
 use lazy_static::lazy_static;
 use libipld::cbor::DagCborCodec;
@@ -167,15 +167,21 @@ impl AtpDatabase {
             let block = Block::<DefaultParams>::encode(DagCborCodec, Code::Sha2_256, &val)?;
             let cid = *block.cid();
             let post: Post = serde_json::from_value(ipld_into_json_value(val))?;
+            let (reply_to_parent_uri, reply_to_root_uri) = match post.reply {
+                Some(ref reply) => (Some(reply.parent.uri.clone()), Some(reply.root.uri.clone())),
+                None => (None, None),
+            };
             let mut stmt = self
                 .conn
-                .prepare_cached("INSERT INTO bsky_post (did, tid, cid, record_json, created_at) VALUES (?1, ?2, ?3, ?4, ?5)")?;
+                .prepare_cached("INSERT INTO bsky_post (did, tid, cid, reply_to_parent_uri, reply_to_root_uri, record_json, created_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)")?;
             stmt.execute(params!(
                 did.to_string(),
                 tid.to_string(),
                 cid.to_string(),
+                reply_to_parent_uri,
+                reply_to_root_uri,
                 serde_json::to_string(&post)?,
-                post.createdAt
+                post.createdAt.unwrap_or_else(|| created_at_now())
             ))?;
         } else {
             let mut stmt = self
diff --git a/adenosine-pds/src/lib.rs b/adenosine-pds/src/lib.rs
index c7bb336..738a061 100644
--- a/adenosine-pds/src/lib.rs
+++ b/adenosine-pds/src/lib.rs
@@ -138,6 +138,16 @@ impl AtpService {
         })
     }
 
+    pub fn new_ephemeral() -> Result<Self> {
+        Ok(AtpService {
+            repo: RepoStore::open_ephemeral()?,
+            atp_db: AtpDatabase::open_ephemeral()?,
+            pds_keypair: KeyPair::new_random(),
+            tid_gen: TidLord::new(),
+            config: AtpServiceConfig::default(),
+        })
+    }
+
     pub fn run_server(self) -> Result<()> {
         let config = self.config.clone();
         let srv = Mutex::new(self);
@@ -389,6 +399,29 @@ fn xrpc_get_handler(
             };
             Ok(json!(desc))
         }
+        // =========== app.bsky methods
+        "app.bsky.actor.getProfile" => {
+            // TODO did or handle
+            let did = Did::from_str(&xrpc_required_param(request, "user")?)?;
+            let mut srv = srv.lock().unwrap();
+            Ok(json!(bsky_get_profile(&mut srv, &did)?))
+        }
+        "app.bsky.feed.getAuthorFeed" => {
+            // TODO did or handle
+            let did = Did::from_str(&xrpc_required_param(request, "author")?)?;
+            let mut srv = srv.lock().unwrap();
+            Ok(json!(bsky_get_author_feed(&mut srv, &did)?))
+        }
+        "app.bsky.feed.getTimeline" => {
+            let mut srv = srv.lock().unwrap();
+            let auth_did = &xrpc_check_auth_header(&mut srv, request, None)?;
+            Ok(json!(bsky_get_timeline(&mut srv, &auth_did)?))
+        }
+        "app.bsky.feed.getPostThread" => {
+            let uri = AtUri::from_str(&xrpc_required_param(request, "uri")?)?;
+            let mut srv = srv.lock().unwrap();
+            Ok(json!(bsky_get_thread(&mut srv, &uri, None)?))
+        }
         _ => Err(anyhow!(XrpcError::NotFound(format!(
             "XRPC endpoint handler not found: {}",
             method
@@ -609,12 +642,6 @@ fn xrpc_post_handler(
             Ok(json!({}))
         }
         // =========== app.bsky methods
-        "app.bsky.actor.getProfile" => {
-            // TODO did or handle
-            let did = Did::from_str(&xrpc_required_param(request, "user")?)?;
-            let mut srv = srv.lock().unwrap();
-            Ok(json!(bsky_get_profile(&mut srv, &did)?))
-        }
         "app.bsky.actor.updateProfile" => {
             let profile: ProfileRecord = rouille::input::json_input(request)?;
             let mut srv = srv.lock().unwrap();
@@ -622,22 +649,6 @@ fn xrpc_post_handler(
             bsky_update_profile(&mut srv, &auth_did, profile)?;
             Ok(json!({}))
         }
-        "app.bsky.feed.getAuthorFeed" => {
-            // TODO did or handle
-            let did = Did::from_str(&xrpc_required_param(request, "author")?)?;
-            let mut srv = srv.lock().unwrap();
-            Ok(json!(bsky_get_author_feed(&mut srv, &did)?))
-        }
-        "app.bsky.feed.getTimeline" => {
-            let mut srv = srv.lock().unwrap();
-            let auth_did = &xrpc_check_auth_header(&mut srv, request, None)?;
-            Ok(json!(bsky_get_timeline(&mut srv, &auth_did)?))
-        }
-        "app.bsky.feed.getPostThread" => {
-            let uri = AtUri::from_str(&xrpc_required_param(request, "uri")?)?;
-            let mut srv = srv.lock().unwrap();
-            Ok(json!(bsky_get_thread(&mut srv, &uri, None)?))
-        }
         _ => Err(anyhow!(XrpcError::NotFound(format!(
             "XRPC endpoint handler not found: {}",
             method
@@ -810,3 +821,10 @@ fn record_handler(
     }
     .render()?)
 }
+
+/// Helper to generate the current timestamp as right now, UTC, formatted as a string
+pub fn created_at_now() -> String {
+    let now = time::OffsetDateTime::now_utc();
+    now.format(&time::format_description::well_known::Rfc3339)
+        .unwrap()
+}
diff --git a/adenosine-pds/src/models.rs b/adenosine-pds/src/models.rs
index 116ac53..0f47c2d 100644
--- a/adenosine-pds/src/models.rs
+++ b/adenosine-pds/src/models.rs
@@ -59,7 +59,8 @@ pub struct RepoBatchWrite {
 #[derive(Debug, serde::Serialize, serde::Deserialize, Clone, PartialEq, Eq)]
 pub struct Subject {
     pub uri: String,
-    pub cid: String,
+    // TODO: CID is required
+    pub cid: Option<String>,
 }
 
 /// Generic over Re-post and Like
@@ -122,7 +123,8 @@ pub struct User {
 #[derive(Debug, serde::Serialize, serde::Deserialize, Clone, PartialEq, Eq)]
 pub struct FeedItem {
     pub uri: String,
-    pub cid: String,
+    // TODO: cid is required
+    pub cid: Option<String>,
     pub author: User,
     pub repostedBy: Option<User>,
     pub record: Value,
@@ -139,11 +141,19 @@ pub struct FeedItem {
 #[derive(Debug, serde::Serialize, serde::Deserialize, Clone, PartialEq, Eq)]
 pub struct Post {
     pub text: String,
+    pub reply: Option<PostReply>,
     pub createdAt: Option<String>,
 }
 
 #[allow(non_snake_case)]
 #[derive(Debug, serde::Serialize, serde::Deserialize, Clone, PartialEq, Eq)]
+pub struct PostReply {
+    pub parent: Subject,
+    pub root: Subject,
+}
+
+#[allow(non_snake_case)]
+#[derive(Debug, serde::Serialize, serde::Deserialize, Clone, PartialEq, Eq)]
 pub struct PostThread {
     pub thread: ThreadItem,
 }
@@ -152,7 +162,8 @@ pub struct PostThread {
 #[derive(Debug, serde::Serialize, serde::Deserialize, Clone, PartialEq, Eq)]
 pub struct ThreadItem {
     pub uri: String,
-    pub cid: String,
+    // TODO: CID is required
+    pub cid: Option<String>,
     pub author: User,
     pub record: Value,
     //pub embed?: RecordEmbed | ExternalEmbed | UnknownEmbed,