Something went wrong. Try again.
A personal app view to see Bsky posts of your followers (for when their app view goes down)
Something went wrong. Try again.
7.8 kB · 269 lines
Rust
at test
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270use futures::TryStreamExt;use jacquard::CowStr;use serde::Deserialize;use sqlx::{FromRow, Row, types::chrono};
#[derive(Debug, FromRow, Clone, Deserialize)]pub struct PostRecord { pub created: i64, pub indexed: chrono::DateTime<chrono::Utc>, pub author: String, pub rkey: String, pub cid: String, pub subject: String, pub raw_blob: Vec<u8>, pub subject_raw_blob: Vec<u8>,}
#[derive(Deserialize, FromRow)]pub struct FollowRecord { pub subject: String,}
#[derive(Debug, FromRow, Deserialize)]pub struct DidDocRecord { pub did: String, pub pds_endpoint: String, pub handle: String,}
pub async fn purge_follows(pool: &sqlx::SqlitePool) { let result = sqlx::query("DELETE FROM follows").execute(pool).await; if result.is_err() { println!("Error deleting all follows from the database: {result:?}"); }}
pub async fn insert_follow_record(pool: &sqlx::SqlitePool, rec: &FollowRecord, rkey: String) { let result = sqlx::query("INSERT INTO follows (subject, rkey) VALUES ($1, $2)") .bind(&rec.subject) .bind(rkey) .execute(pool) .await; if result.is_err() { println!("Error inserting follow into the database: {result:?}"); }}
pub async fn get_follows(pool: &sqlx::SqlitePool) -> Vec<String> { let follows = sqlx::query_as::<_, FollowRecord>("SELECT subject FROM follows") .fetch_all(pool) .await .unwrap();
let mut dids = Vec::new(); for follow in follows { dids.push(follow.subject); }
dids}
pub async fn delete_follow_record(pool: &sqlx::SqlitePool, rkey: String) { let result = sqlx::query("DELETE FROM follows WHERE rkey = $1") .bind(&rkey) .execute(pool) .await; if result.is_err() { println!("Error deleting follow from the database: {result:?}"); }}
pub async fn insert_post(pool: &sqlx::SqlitePool, post: PostRecord) { let result = sqlx::query( "INSERT INTO posts (created, indexed, author, rkey, cid, raw_blob) VALUES ($1, $2, $3, $4, $5, $6) ON CONFLICT(author, rkey) DO NOTHING", ) .bind(post.created) .bind(post.indexed) .bind(post.author) .bind(post.rkey) .bind(post.cid) .bind(post.raw_blob) .execute(pool) .await;
if result.is_err() { println!("Error inserting post into the database: {result:?}"); }}
pub async fn insert_repost(pool: &sqlx::SqlitePool, post: PostRecord) { let result = sqlx::query( "INSERT INTO posts (created, indexed, author, rkey, cid, raw_blob, subject, subject_raw_blob) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) ON CONFLICT(author, rkey) DO NOTHING", ) .bind(post.created) .bind(post.indexed) .bind(post.author) .bind(post.rkey) .bind(post.cid) .bind(post.raw_blob) .bind(post.subject) .bind(post.subject_raw_blob) .execute(pool) .await;
if result.is_err() { println!("Error inserting post into the database: {result:?}"); }}
pub async fn delete_post(pool: &sqlx::SqlitePool, rkey: String) { let result = sqlx::query("DELETE FROM posts WHERE rkey = $1") .bind(rkey) .execute(pool) .await;
if result.is_err() { println!("Error deleting post from the database: {result:?}"); }}
pub async fn delete_repost(pool: &sqlx::SqlitePool, did: String, rkey: String) { let result = sqlx::query("DELETE FROM posts WHERE author = $1 and rkey = $2") .bind(did) .bind(rkey) .execute(pool) .await;
if result.is_err() { println!("Error deleting repost from the database: {result:?}"); }}
pub async fn get_follow_by_rkey(pool: &sqlx::SqlitePool, rkey: &String) -> String { let result = sqlx::query("SELECT subject FROM follows where rkey = ? LIMIT 1") .bind(rkey) .fetch_one(pool) .await;
match result { Ok(row) => { if row.len() == 0 { println!("did not find subject in follows with rkey: {rkey}"); return "".to_string(); } let subject = row.get::<String, _>(0);
return subject; } Err(e) => { println!("error getting follow {e}"); return "".to_string(); } }}
pub async fn get_posts<'a>( pool: &sqlx::SqlitePool, cursor: Option<CowStr<'a>>, req_limit: Option<i64>,) -> anyhow::Result<Vec<PostRecord>> { let query_str = "SELECT created, indexed, author, rkey, cid, subject, raw_blob, subject_raw_blob FROM posts";
let mut cursor_str: String = "".to_string(); if let Some(cur) = cursor { match cur.parse::<i64>() { Ok(v) => cursor_str = v.to_string(), Err(e) => println!("invalid cursor: {e}"), } }
let mut where_clause: String = "".to_string(); if cursor_str != "" { where_clause = "WHERE created < $1".to_string(); }
let mut limit = 20; if let Some(li) = req_limit { limit = li }
let order_clause = format!("ORDER BY created desc LIMIT {}", limit); let full_query = format!("{} {} {}", query_str, where_clause, order_clause);
let mut rows = sqlx::query(&full_query).bind(cursor_str).fetch(pool);
let mut results = Vec::new();
while let Some(row) = rows.try_next().await? { let post = PostRecord { created: row.get::<i64, _>(0), indexed: row.get::<chrono::DateTime<chrono::Utc>, _>(1), author: row.get::<String, _>(2), rkey: row.get::<String, _>(3), cid: row.get::<String, _>(4), subject: row.get::<String, _>(5), raw_blob: row.get::<Vec<u8>, _>(6), subject_raw_blob: row.get::<Vec<u8>, _>(7), };
results.push(post); }
Ok(results)}
pub async fn insert_did_doc_record(pool: &sqlx::SqlitePool, doc: &DidDocRecord) { let result = sqlx::query("INSERT INTO did_docs (did, pds_endpoint, handle) VALUES ($1, $2, $3) ON CONFLICT(did) DO NOTHING") .bind(&doc.did) .bind(&doc.pds_endpoint) .bind(&doc.handle) .execute(pool) .await; if result.is_err() { println!("Error inserting did doc into the database: {result:?}"); }}
pub async fn get_did_doc_by_did(pool: &sqlx::SqlitePool, did: &String) -> Option<DidDocRecord> { let result = sqlx::query("SELECT pds_endpoint, handle FROM did_docs where did = ?") .bind(did) .fetch_one(pool) .await;
match result { Err(_) => None, Ok(row) => { if row.is_empty() { return None; } else { Some(DidDocRecord { did: did.to_string(), pds_endpoint: row.get(0), handle: row.get(1), }) } } }}
pub async fn insert_profile(pool: &sqlx::SqlitePool, did: String, blob: Vec<u8>) { let result = sqlx::query("INSERT INTO profiles (did, blob) VALUES ($1, $2) ON CONFLICT(did) DO NOTHING") .bind(&did) .bind(&blob) .execute(pool) .await; if result.is_err() { println!("Error inserting profile into the database: {result:?}"); }}
pub async fn get_profile_blob_did(pool: &sqlx::SqlitePool, did: &String) -> Option<Vec<u8>> { let result = sqlx::query("SELECT blob FROM profiles where did = ?") .bind(did) .fetch_one(pool) .await;
match result { Err(_) => None, Ok(row) => { if row.is_empty() { return None; } else { Some(row.get(0)) } } }}