Something went wrong. Try again.
[ARCHIVED] A PLC Mirror written in Rust
Something went wrong. Try again.
4.7 kB · 157 lines
Rust
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158use crate::types::{PlcEntry, PlcOperationType};use deadpool_postgres::{Object, Pool};use reqwest::Client;use slog::{Logger, debug, error, info};use std::env::var;use std::time::{Duration, Instant};use tokio::sync::OnceCell;use tokio_postgres::Statement;use tokio_postgres::types::Type;
// PLC rate limit is 100req/min - we aim a little lowerconst SLEEP_TIME_F: f32 = (60. / 90.) * 1000.;const SLEEP_TIME: Duration = Duration::from_millis(SLEEP_TIME_F as u64);
static INSERT_DID_STMT: OnceCell<Statement> = OnceCell::const_new();static INSERT_OP_STMT: OnceCell<Statement> = OnceCell::const_new();
pub async fn importer(log: Logger, pool: Pool, start_after: Option<String>) { let plc_main = var("PLC_UPSTREAM").unwrap_or("https://plc.directory".to_string()); let user_agent = var("PLC_USER_AGENT").unwrap_or("parakeet-plc".to_string());
let mut after = start_after;
if let Some(after) = &after { info!(log, "starting import at {after}"); }
let client = Client::builder().user_agent(user_agent).build().unwrap(); let conn = pool.get().await.unwrap();
loop { let start = Instant::now();
match upstream_export(&client, &plc_main, after.clone()).await { Ok((entries, errors)) => { debug!( log, "exported {} entries, {} failed", entries.len(), errors.len() );
if !errors.is_empty() { error!(log, "{} entries failed to decode", errors.len()); }
for entry in &entries { if let Err(e) = insert_did_and_op(&conn, entry).await { error!(log, "failed to insert operation for did {}: {e}", entry.did); } }
if let Some(new_after) = entries.last() { let new_after = new_after.created_at; after = Some(new_after.to_rfc3339_opts(chrono::SecondsFormat::Millis, true)); } } Err(err) => error!(log, "failed to export PLC directory: {err}"), }
let end = start.elapsed(); tokio::time::sleep(SLEEP_TIME.saturating_sub(end)).await; }}
async fn upstream_export( client: &Client, plc: &str, after: Option<String>,) -> eyre::Result<(Vec<PlcEntry>, Vec<serde_json::Error>)> { let mut uri = format!("{plc}/export?count=1000"); if let Some(after) = after { uri += format!("&after={}", after).as_str(); }
let export_res = client.get(uri).send().await?;
let export = export_res.text().await?; let (exports, errors) = export .lines() .map(serde_json::from_str) .fold((Vec::new(), Vec::new()), fold_exports);
Ok((exports, errors))}
fn fold_exports<A, B>( (mut oks, mut errors): (Vec<A>, Vec<B>), current: Result<A, B>,) -> (Vec<A>, Vec<B>) { match current { Ok(data) => oks.push(data), Err(err) => errors.push(err), };
(oks, errors)}
async fn insert_did_and_op(conn: &Object, entry: &PlcEntry) -> eyre::Result<()> { let insert_did_stmt = INSERT_DID_STMT .get_or_init(|| init_insert_did_stmt(conn)) .await; let insert_op_stmt = INSERT_OP_STMT .get_or_init(|| init_insert_op_stmt(conn)) .await;
let operation = serde_json::to_value(&entry.operation)?;
let (prev, sig) = match &entry.operation { PlcOperationType::Create(op) => (op.prev.map(|v| v.to_string()), &op.sig), PlcOperationType::Tombstone(op) => (op.prev.map(|v| v.to_string()), &op.sig), PlcOperationType::Operation(op) => (op.prev.map(|v| v.to_string()), &op.sig), };
conn.execute(insert_did_stmt, &[&entry.did]).await?; conn.execute( insert_op_stmt, &[ &entry.did, &entry.cid.to_string(), &prev, &sig, &entry.nullified, &operation, &entry.created_at, ], ) .await?;
Ok(())}
async fn init_insert_did_stmt(conn: &Object) -> Statement { conn.prepare_typed( "INSERT INTO dids (did) VALUES ($1) ON CONFLICT DO NOTHING", &[Type::TEXT], ) .await .unwrap()}
async fn init_insert_op_stmt(conn: &Object) -> Statement { conn.prepare_typed( include_str!("sql/insert_op.sql"), &[ Type::TEXT, Type::TEXT, Type::TEXT, Type::TEXT, Type::BOOL, Type::JSONB, Type::TIMESTAMPTZ, ], ) .await .unwrap()}