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.
3.4 kB · 110 lines
Rust
at test
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111mod external_data;mod jetstream;mod records;mod server;mod store;mod xrpc;
use jacquard::api::app_bsky::graph::follow::Follow;use jacquard::common::CowStr;use jacquard::deps::fluent_uri::Uri;use jacquard::{oauth::resolver::OAuthResolver, types::ident::AtIdentifier, xrpc::XrpcExt};use jacquard_api::com_atproto::repo::list_records::ListRecords;use jacquard_identity::JacquardResolver;use jacquard_identity::resolver::ResolverOptions;use serde::Deserialize;use sqlx::migrate::MigrateDatabase;
#[tokio::main]async fn main() -> anyhow::Result<()> { dotenv::dotenv().ok();
let db_url = std::env::var("DATABASE_URL")?;
if !sqlx::Sqlite::database_exists(&db_url).await? { sqlx::Sqlite::create_database(&db_url).await?; }
let pool = sqlx::SqlitePool::connect(&db_url).await?;
// run migrations sqlx::migrate!("./migrations").run(&pool).await?;
let users_did = std::env::var("USERS_DID").unwrap(); let jetstream_host = std::env::var("JETSTREAM_HOST").unwrap();
let resolver = JacquardResolver::new(reqwest::Client::new(), ResolverOptions::default());
match get_pds_endpoint(&resolver, &users_did).await { Some(endpoint) => seed_follows(&pool, &users_did, endpoint).await?, None => Err("no endpoint found for user").unwrap(), }
tokio::join!( server::run_server(resolver.clone(), &pool), jetstream::run_posts_jetstream(&users_did, &jetstream_host, resolver.clone(), &pool), ); Ok(())}
async fn get_pds_endpoint(resolver: &JacquardResolver, users_did: &String) -> Option<String> { let (_, identity) = resolver.resolve_from_identity(users_did).await.unwrap();
match identity.pds_endpoint().iter().next() { Some(x) => Some(x.to_string()), None => None, }}
async fn seed_follows( pool: &sqlx::SqlitePool, did: &String, endpoint: String,) -> anyhow::Result<()> { let did = AtIdentifier::new(did.as_str()).unwrap();
let http = reqwest::Client::new(); let base = Uri::parse(endpoint).to_owned().unwrap(); let mut done = false; let mut cursor: String = "".to_string();
store::purge_follows(&pool).await;
while !done { let req = ListRecords::new() .repo(did.clone()) .collection("app.bsky.graph.follow".to_string()) .limit(100) .cursor(CowStr::copy_from_str(cursor.as_str())) .build();
let response = http.xrpc(base.to_owned()).send(&req).await?; let a = response.into_output().unwrap();
match a.cursor { Some(c) => cursor = c.to_string(), None => done = true, }
for record in a.records.iter() { match Follow::deserialize(&record.value) { Ok(follow) => { let rec = store::FollowRecord { subject: follow.subject.to_string(), };
match record.uri.rkey() { Some(rkey) => { store::insert_follow_record(&pool, &rec, rkey.0.to_string()).await } None => println!("no rkey"), } } Err(e) => { eprintln!("skipping {}: couldn't decode as a follow: {e}", record.uri) } } } } Ok(())}