pds: start implementing bsky database ops and XRPC endpoints

5 files changed, 502 insertions, 24 deletions

diff --git a/adenosine-pds/src/atp_db.sql b/adenosine-pds/src/atp_db.sql
index b356abc..71fbf6d 100644
--- a/adenosine-pds/src/atp_db.sql
+++ b/adenosine-pds/src/atp_db.sql
@@ -40,33 +40,26 @@ CREATE TABLE bsky_post(
 );
 CREATE INDEX bsky_post_reply_root_uri_idx on bsky_post(reply_root_uri);
 
-CREATE TABLE bsky_repost(
-    did                 TEXT NOT NULL,
-    subject_uri         TEXT NOT NULL,
-    cid                 TEXT NOT NULL,
-    created_at          TIMESTAMP WITH TIME ZONE  NOT NULL,
-    indexed_at          TIMESTAMP WITH TIME ZONE  NOT NULL DEFAULT ( DATETIME('now') ),
-    PRIMARY KEY(did, subject_uri)
-);
-CREATE INDEX bsky_repost_subject_uri_idx on bsky_repost(subject_uri);
-
-CREATE TABLE bsky_like(
+CREATE TABLE bsky_ref(
+    ref_type            TEXT NOT NULL,
     did                 TEXT NOT NULL,
+    tid                 TEXT NOT NULL,
     subject_uri         TEXT NOT NULL,
-    cid                 TEXT NOT NULL,
+    subject_cid         TEXT NOT NULL,
     created_at          TIMESTAMP WITH TIME ZONE  NOT NULL,
     indexed_at          TIMESTAMP WITH TIME ZONE  NOT NULL DEFAULT ( DATETIME('now') ),
-    PRIMARY KEY(did, subject_uri)
+    PRIMARY KEY(ref_type, did, tid)
 );
-CREATE INDEX bsky_like_subject_uri_idx on bsky_like(subject_uri);
+CREATE INDEX bsky_ref_subject_uri_idx on bsky_ref(subject_uri);
 
 CREATE TABLE bsky_follow(
     did                 TEXT NOT NULL,
+    tid                 TEXT NOT NULL,
     subject_did         TEXT NOT NULL,
-    cid                 TEXT NOT NULL,
+    subject_cid         TEXT NOT NULL,
     created_at          TIMESTAMP WITH TIME ZONE  NOT NULL,
     indexed_at          TIMESTAMP WITH TIME ZONE  NOT NULL DEFAULT ( DATETIME('now') ),
-    PRIMARY KEY(did, subject_did)
+    PRIMARY KEY(did, tid)
 );
 CREATE INDEX bsky_follow_subject_did_idx on bsky_follow(subject_did);
 
diff --git a/adenosine-pds/src/bsky.rs b/adenosine-pds/src/bsky.rs
new file mode 100644
index 0000000..eeab9e3
--- /dev/null
+++ b/adenosine-pds/src/bsky.rs
@@ -0,0 +1,233 @@
+use crate::models::*;
+use crate::repo::Mutation;
+/// Helper functions for doing database and repo operations relating to bluesky endpoints and
+/// records
+use crate::{
+    ipld_into_json_value, json_value_into_ipld, AtpDatabase, AtpService, Did, Result, Tid,
+};
+use adenosine_cli::identifiers::{AtUri, Nsid};
+use libipld::Cid;
+use rusqlite::params;
+use serde_json::json;
+use std::str::FromStr;
+
+/// Handles updating the database with creation, update, deletion of arbitrary records
+pub fn bsky_mutate_db(db: &mut AtpDatabase, did: &Did, mutations: Vec<Mutation>) -> Result<()> {
+    // TODO: this function could probably be refactored
+    let bsky_post: Nsid = Nsid::from_str("app.bsky.feed.post").unwrap();
+    let bsky_repost: Nsid = Nsid::from_str("app.bsky.feed.repost").unwrap();
+    let bsky_like: Nsid = Nsid::from_str("app.bsky.feed.like").unwrap();
+    let bsky_follow: Nsid = Nsid::from_str("app.bsky.graph.follow").unwrap();
+    for m in mutations.into_iter() {
+        match m {
+            Mutation::Create(ref_type, tid, val) | Mutation::Update(ref_type, tid, val)
+                if ref_type == bsky_post =>
+            {
+                db.bsky_upsert_post(did, &tid, Some(val))?
+            }
+            Mutation::Delete(ref_type, tid) if ref_type == bsky_post => {
+                db.bsky_upsert_post(did, &tid, None)?
+            }
+            Mutation::Create(ref_type, tid, val) | Mutation::Update(ref_type, tid, val)
+                if ref_type == bsky_repost =>
+            {
+                db.bsky_upsert_ref("repost", did, &tid, Some(val))?
+            }
+            Mutation::Delete(ref_type, tid) if ref_type == bsky_repost => {
+                db.bsky_upsert_ref("repost", did, &tid, None)?
+            }
+            Mutation::Create(ref_type, tid, val) | Mutation::Update(ref_type, tid, val)
+                if ref_type == bsky_like =>
+            {
+                db.bsky_upsert_ref("like", did, &tid, Some(val))?
+            }
+            Mutation::Delete(ref_type, tid) if ref_type == bsky_like => {
+                db.bsky_upsert_ref("like", did, &tid, None)?
+            }
+            Mutation::Create(ref_type, tid, val) | Mutation::Update(ref_type, tid, val)
+                if ref_type == bsky_follow =>
+            {
+                db.bsky_upsert_follow(did, &tid, Some(val))?
+            }
+            Mutation::Delete(ref_type, tid) if ref_type == bsky_follow => {
+                db.bsky_upsert_follow(did, &tid, None)?
+            }
+            _ => (),
+        }
+    }
+    Ok(())
+}
+
+pub fn bsky_get_profile(srv: &mut AtpService, did: &Did) -> Result<Profile> {
+    // first get the profile record
+    let mut profile_cid: Option<Cid> = None;
+    let commit_cid = &srv.repo.lookup_commit(&did)?.unwrap();
+    let last_commit = srv.repo.get_commit(commit_cid)?;
+    let full_map = srv.repo.mst_to_map(&last_commit.mst_cid)?;
+    let prefix = "/app.bsky.actor.profile/";
+    for (mst_key, cid) in full_map.iter() {
+        if mst_key.starts_with(&prefix) {
+            profile_cid = Some(*cid);
+        }
+    }
+    let (display_name, description): (Option<String>, Option<String>) =
+        if let Some(cid) = profile_cid {
+            let record: ProfileRecord =
+                serde_json::from_value(ipld_into_json_value(srv.repo.get_ipld(&cid)?))?;
+            (Some(record.displayName), record.description)
+        } else {
+            (None, None)
+        };
+    let mut stmt = srv
+        .atp_db
+        .conn
+        .prepare_cached("SELECT handle FROM account WHERE did = $1")?;
+    let handle: String = stmt.query_row(params!(did.to_string()), |row| row.get(0))?;
+    let mut stmt = srv
+        .atp_db
+        .conn
+        .prepare_cached("SELECT COUNT(*) FROM bsky_post WHERE did = $1")?;
+    let post_count: u64 = stmt.query_row(params!(did.to_string()), |row| row.get(0))?;
+    let mut stmt = srv
+        .atp_db
+        .conn
+        .prepare_cached("SELECT COUNT(*) FROM bsky_ref WHERE ref_type = 'follow' AND 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))?;
+    Ok(Profile {
+        did: did.to_string(),
+        handle: handle,
+        displayName: display_name,
+        description: description,
+        followersCount: followers_count,
+        followsCount: follows_count,
+        postsCount: post_count,
+        myState: json!({}),
+    })
+}
+
+pub fn bsky_update_profile(srv: &mut AtpService, did: &Did, profile: ProfileRecord) -> Result<()> {
+    // get the profile record
+    let mut profile_tid: Option<Tid> = None;
+    let commit_cid = &srv.repo.lookup_commit(&did)?.unwrap();
+    let last_commit = srv.repo.get_commit(commit_cid)?;
+    let full_map = srv.repo.mst_to_map(&last_commit.mst_cid)?;
+    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())?);
+        }
+    }
+    let profile_tid: Tid = profile_tid.unwrap_or(srv.tid_gen.next_tid());
+    let mutations: Vec<Mutation> = vec![Mutation::Update(
+        Nsid::from_str("app.bsky.actor.profile")?,
+        profile_tid,
+        json_value_into_ipld(serde_json::to_value(profile)?),
+    )];
+    let keypair = srv.pds_keypair.clone();
+    srv.repo.mutate_repo(&did, &mutations, &keypair)?;
+    Ok(())
+}
+
+struct FeedRow {
+    pub item_did: Did,
+    pub item_handle: String,
+    pub item_post_tid: Tid,
+    pub item_post_cid: Cid,
+}
+
+fn feed_row(row: &rusqlite::Row) -> Result<FeedRow> {
+    let item_did: String = row.get(0)?;
+    let item_did = Did::from_str(&item_did)?;
+    let item_handle = row.get(1)?;
+    let item_post_tid: String = row.get(2)?;
+    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)?;
+    Ok(FeedRow {
+        item_did,
+        item_handle,
+        item_post_tid,
+        item_post_cid,
+    })
+}
+
+fn feed_row_to_item(srv: &mut AtpService, row: FeedRow) -> Result<FeedItem> {
+    let record_ipld = srv.repo.get_ipld(&row.item_post_cid)?;
+    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(),
+        author: User {
+            did: row.item_did.to_string(),
+            handle: row.item_handle,
+            displayName: None, // TODO
+        },
+        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
+        myState: None,
+    };
+    Ok(feed_item)
+}
+
+pub fn bsky_get_author_feed(srv: &mut AtpService, did: &Did) -> Result<GenericFeed> {
+    let mut feed: Vec<FeedItem> = vec![];
+    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")?;
+        let mut sql_rows = stmt.query(params!(did.to_string()))?;
+        let mut rows: Vec<FeedRow> = vec![];
+        while let Some(sql_row) = sql_rows.next()? {
+            let row = feed_row(sql_row)?;
+            rows.push(row);
+        }
+        rows
+    };
+    for row in rows {
+        feed.push(feed_row_to_item(srv, row)?);
+    }
+    Ok(GenericFeed { feed })
+}
+
+pub fn bsky_get_timeline(srv: &mut AtpService, did: &Did) -> Result<GenericFeed> {
+    let mut feed: Vec<FeedItem> = vec![];
+    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")?;
+        let mut sql_rows = stmt.query(params!(did.to_string()))?;
+        let mut rows: Vec<FeedRow> = vec![];
+        while let Some(sql_row) = sql_rows.next()? {
+            let row = feed_row(sql_row)?;
+            rows.push(row);
+        }
+        rows
+    };
+    for row in rows {
+        feed.push(feed_row_to_item(srv, row)?);
+    }
+    Ok(GenericFeed { feed })
+}
+
+pub fn bsky_get_thread(
+    _srv: &mut AtpService,
+    _uri: &AtUri,
+    _depth: Option<u64>,
+) -> Result<GenericFeed> {
+    // TODO: what is the best way to implement this? recurisvely? just first-level children to
+    // start?
+    unimplemented!()
+}
diff --git a/adenosine-pds/src/db.rs b/adenosine-pds/src/db.rs
index 2249637..384b498 100644
--- a/adenosine-pds/src/db.rs
+++ b/adenosine-pds/src/db.rs
@@ -1,7 +1,11 @@
+use crate::models::{FollowRecord, Post, RefRecord};
 /// ATP database (as distinct from blockstore)
-use crate::{AtpSession, Did, KeyPair};
+use crate::{ipld_into_json_value, AtpSession, Did, KeyPair, Tid};
 use anyhow::{anyhow, Result};
 use lazy_static::lazy_static;
+use libipld::cbor::DagCborCodec;
+use libipld::multihash::Code;
+use libipld::{Block, DefaultParams, Ipld};
 use log::debug;
 use rusqlite::{params, Connection, OptionalExtension};
 use rusqlite_migration::{Migrations, M};
@@ -29,7 +33,7 @@ lazy_static! {
 
 #[derive(Debug)]
 pub struct AtpDatabase {
-    conn: Connection,
+    pub conn: Connection,
 }
 
 impl AtpDatabase {
@@ -156,4 +160,83 @@ impl AtpDatabase {
         let doc_json: String = stmt.query_row(params!(did.to_string()), |row| row.get(0))?;
         Ok(Value::from_str(&doc_json)?)
     }
+
+    pub fn bsky_upsert_post(&mut self, did: &Did, tid: &Tid, val: Option<Ipld>) -> Result<()> {
+        if let Some(val) = val {
+            // need to re-compute the CID from DagCbor re-encoding, I guess. bleh.
+            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 mut stmt = self
+                .conn
+                .prepare_cached("INSERT INTO bsky_post (did, tid, cid, record_json, created_at) VALUES (?1, ?2, ?3, ?4, ?5)")?;
+            stmt.execute(params!(
+                did.to_string(),
+                tid.to_string(),
+                cid.to_string(),
+                serde_json::to_string(&post)?,
+                post.createdAt
+            ))?;
+        } else {
+            let mut stmt = self
+                .conn
+                .prepare_cached("DELETE FROM bsky_post WHERE did = ?1 AND tid = ?2")?;
+            stmt.execute(params!(did.to_string(), tid.to_string()))?;
+        }
+        Ok(())
+    }
+
+    pub fn bsky_upsert_ref(
+        &mut self,
+        ref_type: &str,
+        did: &Did,
+        tid: &Tid,
+        val: Option<Ipld>,
+    ) -> Result<()> {
+        if let Some(val) = val {
+            let ref_obj: RefRecord = serde_json::from_value(ipld_into_json_value(val))?;
+            let mut stmt = self
+                .conn
+                .prepare_cached("INSERT INTO bsky_ref (ref_type, did, tid, subject_uri, subject_cid, created_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6)")?;
+            stmt.execute(params!(
+                ref_type.to_string(),
+                did.to_string(),
+                tid.to_string(),
+                ref_obj.subject.uri,
+                ref_obj.subject.cid,
+                ref_obj.createdAt
+            ))?;
+        } else {
+            let mut stmt = self.conn.prepare_cached(
+                "DELETE FROM bsky_ref WHERE ref_type = ?1 AND did = ?2 AND tid = ?3",
+            )?;
+            stmt.execute(params!(
+                ref_type.to_string(),
+                did.to_string(),
+                tid.to_string()
+            ))?;
+        }
+        Ok(())
+    }
+
+    pub fn bsky_upsert_follow(&mut self, did: &Did, tid: &Tid, val: Option<Ipld>) -> Result<()> {
+        if let Some(val) = val {
+            let follow: FollowRecord = serde_json::from_value(ipld_into_json_value(val))?;
+            let mut stmt = self
+                .conn
+                .prepare_cached("INSERT INTO bsky_follow (did, tid, subject_did, created_at) VALUES (?1, ?2, ?3, ?4)")?;
+            stmt.execute(params!(
+                did.to_string(),
+                tid.to_string(),
+                follow.subject.did,
+                follow.createdAt
+            ))?;
+        } else {
+            let mut stmt = self
+                .conn
+                .prepare_cached("DELETE FROM bsky_follow WHERE did = ?2 AND tid = ?3")?;
+            stmt.execute(params!(did.to_string(), tid.to_string()))?;
+        }
+        Ok(())
+    }
 }
diff --git a/adenosine-pds/src/lib.rs b/adenosine-pds/src/lib.rs
index 468cc6e..c7bb336 100644
--- a/adenosine-pds/src/lib.rs
+++ b/adenosine-pds/src/lib.rs
@@ -1,4 +1,4 @@
-use adenosine_cli::identifiers::{Did, Nsid, Tid, TidLord};
+use adenosine_cli::identifiers::{AtUri, Did, Nsid, Tid, TidLord};
 use anyhow::{anyhow, Result};
 use askama::Template;
 use libipld::Cid;
@@ -13,6 +13,7 @@ use std::path::PathBuf;
 use std::str::FromStr;
 use std::sync::Mutex;
 
+mod bsky;
 mod car;
 mod crypto;
 mod db;
@@ -24,6 +25,7 @@ mod ucan_p256;
 mod vendored;
 mod web;
 
+use bsky::*;
 pub use crypto::{KeyPair, PubKey};
 pub use db::AtpDatabase;
 pub use did::DidDocMeta;
@@ -359,7 +361,7 @@ fn xrpc_get_handler(
             let full_map = srv.repo.mst_to_map(&last_commit.mst_cid)?;
             let prefix = format!("/{}/", collection);
             for (mst_key, cid) in full_map.iter() {
-                debug!("{}", mst_key);
+                //debug!("{}", mst_key);
                 if mst_key.starts_with(&prefix) {
                     let record = srv.repo.get_ipld(cid)?;
                     record_list.push(json!({
@@ -541,7 +543,7 @@ fn xrpc_post_handler(
             }
             let keypair = srv.pds_keypair.clone();
             srv.repo.mutate_repo(&did, &mutations, &keypair)?;
-            // TODO: next handle updates to database
+            bsky_mutate_db(&mut srv.atp_db, &did, mutations)?;
             Ok(json!({}))
         }
         "com.atproto.repo.createRecord" => {
@@ -558,7 +560,7 @@ fn xrpc_post_handler(
             )];
             let keypair = srv.pds_keypair.clone();
             srv.repo.mutate_repo(&did, &mutations, &keypair)?;
-            // TODO: next handle updates to database
+            bsky_mutate_db(&mut srv.atp_db, &did, mutations)?;
             Ok(json!({}))
         }
         "com.atproto.repo.putRecord" => {
@@ -577,7 +579,7 @@ fn xrpc_post_handler(
             )];
             let keypair = srv.pds_keypair.clone();
             srv.repo.mutate_repo(&did, &mutations, &keypair)?;
-            // TODO: next handle updates to database
+            bsky_mutate_db(&mut srv.atp_db, &did, mutations)?;
             Ok(json!({}))
         }
         "com.atproto.repo.deleteRecord" => {
@@ -590,7 +592,7 @@ fn xrpc_post_handler(
             let mutations: Vec<Mutation> = vec![Mutation::Delete(collection, tid)];
             let keypair = srv.pds_keypair.clone();
             srv.repo.mutate_repo(&did, &mutations, &keypair)?;
-            // TODO: next handle updates to database
+            bsky_mutate_db(&mut srv.atp_db, &did, mutations)?;
             Ok(json!({}))
         }
         "com.atproto.sync.updateRepo" => {
@@ -603,8 +605,39 @@ fn xrpc_post_handler(
             let _auth_did = &xrpc_check_auth_header(&mut srv, request, Some(&did))?;
             srv.repo
                 .import_car_bytes(&car_bytes, Some(did.to_string()))?;
+            // TODO: need to update atp_db
             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();
+            let auth_did = &xrpc_check_auth_header(&mut srv, request, None)?;
+            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
diff --git a/adenosine-pds/src/models.rs b/adenosine-pds/src/models.rs
index f172819..116ac53 100644
--- a/adenosine-pds/src/models.rs
+++ b/adenosine-pds/src/models.rs
@@ -1,3 +1,7 @@
+use serde_json::Value;
+
+// =========== com.atproto types (manually entered)
+
 #[allow(non_snake_case)]
 #[derive(Debug, serde::Deserialize, serde::Serialize, PartialEq, Eq)]
 pub struct AccountRequest {
@@ -49,3 +53,135 @@ pub struct RepoBatchWrite {
     pub rkey: Option<String>,
     pub value: serde_json::Value,
 }
+
+// =========== app.bsky types (manually entered)
+
+#[derive(Debug, serde::Serialize, serde::Deserialize, Clone, PartialEq, Eq)]
+pub struct Subject {
+    pub uri: String,
+    pub cid: String,
+}
+
+/// Generic over Re-post and Like
+#[allow(non_snake_case)]
+#[derive(Debug, serde::Serialize, serde::Deserialize, Clone, PartialEq, Eq)]
+pub struct RefRecord {
+    pub subject: Subject,
+    pub createdAt: String,
+}
+
+#[allow(non_snake_case)]
+#[derive(Debug, serde::Serialize, serde::Deserialize, Clone, PartialEq, Eq)]
+pub struct FollowSubject {
+    pub did: String,
+    // pub declarationCid: String,
+}
+
+#[allow(non_snake_case)]
+#[derive(Debug, serde::Serialize, serde::Deserialize, Clone, PartialEq, Eq)]
+pub struct FollowRecord {
+    pub subject: FollowSubject,
+    pub createdAt: String,
+}
+
+#[allow(non_snake_case)]
+#[derive(Debug, serde::Serialize, serde::Deserialize, Clone, PartialEq, Eq)]
+pub struct ProfileRecord {
+    pub displayName: String,
+    pub description: Option<String>,
+}
+
+#[allow(non_snake_case)]
+#[derive(Debug, serde::Serialize, serde::Deserialize, Clone, PartialEq, Eq)]
+pub struct Profile {
+    pub did: String,
+    pub handle: String,
+    pub displayName: Option<String>,
+    pub description: Option<String>,
+    pub followersCount: u64,
+    pub followsCount: u64,
+    pub postsCount: u64,
+    pub myState: serde_json::Value,
+}
+
+#[allow(non_snake_case)]
+#[derive(Debug, serde::Serialize, serde::Deserialize, Clone, PartialEq, Eq)]
+pub struct GenericFeed {
+    pub feed: Vec<FeedItem>,
+}
+
+#[allow(non_snake_case)]
+#[derive(Debug, serde::Serialize, serde::Deserialize, Clone, PartialEq, Eq)]
+pub struct User {
+    pub did: String,
+    pub handle: String,
+    pub displayName: Option<String>,
+}
+
+#[allow(non_snake_case)]
+#[derive(Debug, serde::Serialize, serde::Deserialize, Clone, PartialEq, Eq)]
+pub struct FeedItem {
+    pub uri: String,
+    pub cid: String,
+    pub author: User,
+    pub repostedBy: Option<User>,
+    pub record: Value,
+    //pub embed?: RecordEmbed | ExternalEmbed | UnknownEmbed,
+    pub embed: Option<Value>,
+    pub replyCount: u64,
+    pub repostCount: u64,
+    pub likeCount: u64,
+    pub indexedAt: String,
+    pub myState: Option<Value>,
+}
+
+#[allow(non_snake_case)]
+#[derive(Debug, serde::Serialize, serde::Deserialize, Clone, PartialEq, Eq)]
+pub struct Post {
+    pub text: String,
+    pub createdAt: Option<String>,
+}
+
+#[allow(non_snake_case)]
+#[derive(Debug, serde::Serialize, serde::Deserialize, Clone, PartialEq, Eq)]
+pub struct PostThread {
+    pub thread: ThreadItem,
+}
+
+#[allow(non_snake_case)]
+#[derive(Debug, serde::Serialize, serde::Deserialize, Clone, PartialEq, Eq)]
+pub struct ThreadItem {
+    pub uri: String,
+    pub cid: String,
+    pub author: User,
+    pub record: Value,
+    //pub embed?: RecordEmbed | ExternalEmbed | UnknownEmbed,
+    pub embed: Option<Value>,
+    pub parent: Box<ThreadItem>,
+    pub replyCount: u64,
+    pub replies: Option<Vec<ThreadItem>>,
+    pub likeCount: u64,
+    pub repostCount: u64,
+    pub indexedAt: String,
+    pub myState: Option<Value>,
+}
+
+#[allow(non_snake_case)]
+#[derive(Debug, serde::Serialize, serde::Deserialize, Clone, PartialEq, Eq)]
+pub struct FollowTarget {
+    // TODO: nested follow list?
+    pub subject: Subject,
+    pub did: String,
+    pub handle: String,
+    pub displayName: Option<String>,
+    pub createdAt: Option<String>,
+    pub indexedAt: String,
+}
+
+#[allow(non_snake_case)]
+#[derive(Debug, serde::Serialize, serde::Deserialize, Clone, PartialEq, Eq)]
+pub struct Follow {
+    // TODO: nested follow list?
+    pub subject: Subject,
+    pub follows: FollowTarget,
+}