Something went wrong. Try again.
printer on atproto
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272use std::sync::Arc;use std::time::Duration;
use chrono::{DateTime, Utc};use futures_util::StreamExt;use miette::{IntoDiagnostic, Result};use serde::{Deserialize, Serialize};use tokio::sync::mpsc;use tokio_tungstenite::tungstenite::Message;
use crate::printer::{Printer, PrinterEndpoint};use crate::resolver::Resolver;
mod printer;mod resolver;
const LEX: &str = "net.klbr.spool.job";const JETSTREAM: [&str; 3] = [ "wss://jetstream2.fr.hose.cam", "wss://jetstream2.us-east.bsky.network", "wss://jetstream2.us-west.bsky.network",];
#[derive(Deserialize)]struct Event { #[serde(rename = "time_us")] cursor: u64, did: String, kind: String, commit: Option<Commit>,}
#[derive(Deserialize)]struct Commit { collection: String, operation: String, record: Option<Box<serde_json::value::RawValue>>,}
#[derive(Deserialize, Serialize, Debug)]#[serde(rename = "net.klbr.spool.job", rename_all = "camelCase", tag = "$type")]struct Job { content: JobContent,}
#[derive(Deserialize, Serialize, Debug)]#[serde(tag = "$type")]enum JobContent { #[serde(rename = "net.klbr.spool.job.content.text")] Text(TextContent),}
impl JobContent { fn kind(&self) -> &'static str { match self { JobContent::Text(_) => "text", } }}
#[derive(Deserialize, Serialize, Debug)]#[serde(rename_all = "camelCase")]struct TextContent { text: String,}
struct Task { date: DateTime<Utc>, identifier: String, job: Job,}
type TaskTx = mpsc::Sender<Task>;type TaskRx = mpsc::Receiver<Task>;
#[tokio::main]async fn main() -> Result<()> { tracing_subscriber::fmt().init();
let printer = Printer::connect( PrinterEndpoint::Usb { vendor_id: 0x0416, product_id: 0x5011, }, Default::default(), )?;
let password = std::env::var("SPOOL_PASSWORD").ok();
let allowed_dids: Vec<String> = tokio::fs::read_to_string("allowed.txt") .await .into_diagnostic()? .lines() .map(|l| l.find('#').map_or(l, |i| &l[..i]).trim().to_string()) .filter(|l| !l.is_empty()) .collect();
// prewarm identity cache let resolver = Arc::new(Resolver::new()); for did in &allowed_dids { resolver.resolve(did).await?; }
let mut jetstream_query = format!("wantedCollections={LEX}"); for did in &allowed_dids { jetstream_query.push_str(&format!("&wantedDids={did}")); }
let (task_tx, task_rx) = mpsc::channel(16); let job_handler = tokio::task::spawn_blocking(move || handle_jobs(printer, task_rx)); let job_stream = tokio::spawn(stream_jobs(jetstream_query, task_tx.clone(), resolver)); let api = tokio::spawn(serve_api(password, task_tx));
tokio::select! { r = job_stream => r.into_diagnostic().flatten(), r = job_handler => r.into_diagnostic().flatten(), r = api => r.into_diagnostic().flatten(), }}
async fn stream_jobs(query: String, task_tx: TaskTx, resolver: Arc<Resolver>) -> Result<()> { let mut jetstream_idx = 0; loop { let jetstream = JETSTREAM[jetstream_idx]; let url = format!("{jetstream}/subscribe?{query}"); match run_stream(&url, &task_tx, &resolver).await { Ok(()) => tracing::warn!("{url} disconnected, reconnecting..."), Err(e) => { tracing::warn!(err = %e, "{url} error, reconnecting..."); jetstream_idx = (jetstream_idx + 1) % JETSTREAM.len(); } } tokio::time::sleep(Duration::from_secs(5)).await; }}
async fn run_stream( url: &str, task_tx: &TaskTx, resolver: &Resolver,) -> Result<(), tokio_tungstenite::tungstenite::Error> { let (mut ws, _) = tokio_tungstenite::connect_async(url).await?; while let Some(msg) = ws.next().await { let text = match msg? { Message::Text(t) => t, _ => continue, }; let event: Event = match serde_json::from_str(&text) { Ok(e) => e, Err(e) => { tracing::warn!(err = %e, "failed to parse event"); continue; } };
// invalidate identity cache if needed if event.kind == "identity" { resolver.invalidate(&event.did).await; continue; }
// handle records let Some(commit) = event.commit else { continue }; if commit.operation != "create" || commit.collection != LEX { continue; } let Some(record_raw) = commit.record else { continue; }; let job: Job = match serde_json::from_str(record_raw.get()) { Ok(j) => j, Err(e) => { tracing::warn!(err = %e, "invalid record"); continue; } }; let identifier = match resolver.resolve(&event.did).await { Ok(h) => h, Err(e) => { tracing::error!(err = %e, "can't resolve did"); resolver .cached(&event.did) .await .unwrap_or_else(|| event.did.clone()) } }; let date = DateTime::from_timestamp_micros(event.cursor as i64).unwrap_or_else(Utc::now); let _ = task_tx .send(Task { identifier, job, date, }) .await; } Ok(())}
fn handle_jobs(mut printer: Printer, mut task_rx: TaskRx) -> Result<()> { while let Some(task) = task_rx.blocking_recv() { tracing::info!( id = task.identifier, kind = task.job.content.kind(), "handling job" ); match task.job.content { JobContent::Text(c) if !c.text.is_empty() => { printer .centered(&task.identifier, 0xCD)? .markdown(&c.text)? .newline()? .centered(&task.date.to_string(), 0xCD)? .newline()? .cut()?; } _ => continue, } }
Ok(())}
async fn serve_api(password: Option<String>, task_tx: TaskTx) -> Result<()> { use async_tiny::{Response, Server};
let mut server = Server::http("0.0.0.0:9889", true).await.into_diagnostic()?;
while let Some(request) = server.next().await { let url = request.url().to_string(); let (path, query_str) = url.split_once('?').unwrap_or((&url, ""));
let params: std::collections::HashMap<&str, &str> = query_str .split('&') .filter_map(|p| p.split_once('=')) .collect();
if let Some(ref pw) = password { if params.get("token").copied() != Some(pw.as_str()) { let _ = request.respond(Response::from_status_and_string(400, "400 Bad Request")); continue; } }
let Some(&name) = params.get("name") else { let _ = request.respond(Response::from_status_and_string(400, "400 Bad Request")); continue; }; let name = name.to_string();
let response = match path { "/print" => { let text = String::from_utf8_lossy(request.body()).into_owned(); let _ = task_tx .send(Task { date: Utc::now(), identifier: name, job: Job { content: JobContent::Text(TextContent { text }), }, }) .await; Response::from_string("ok").with_content_type("application/json") } _ => Response::from_status_and_string(404, "404 Not Found"), };
let _ = request.respond(response); }
Ok(())}