···
-
use anyhow::{anyhow, Result};
-
use atrium_api::com::atproto::sync::subscribe_repos::{Commit, NSID};
-
use atrium_api::client::AtpServiceClient;
-
use atrium_xrpc_client::isahc::IsahcClient;
-
use futures::StreamExt;
-
use ipld_core::ipld::Ipld;
-
use std::sync::{Arc, Mutex};
-
use tokio_tungstenite::connect_async;
-
use tokio_tungstenite::tungstenite::Message;
-
use crate::fs::{PdsFsCollection, PdsFsEntry, PdsFsRecord};
-
use indexmap::{IndexMap, IndexSet};
-
/// Frame header types for WebSocket messages
-
#[derive(Debug, Clone, PartialEq, Eq)]
-
Message(Option<String>),
-
impl TryFrom<Ipld> for FrameHeader {
-
type Error = anyhow::Error;
-
fn try_from(value: Ipld) -> Result<Self> {
-
if let Ipld::Map(map) = value {
-
if let Some(Ipld::Integer(i)) = map.get("op") {
-
let t = if let Some(Ipld::String(s)) = map.get("t") {
-
return Ok(FrameHeader::Message(t));
-
-1 => return Ok(FrameHeader::Error),
-
Err(anyhow!("invalid frame type"))
-
/// Frame types for parsed WebSocket messages
-
#[derive(Debug, Clone, PartialEq, Eq)]
-
Message(Option<String>, MessageFrame),
-
#[derive(Debug, Clone, PartialEq, Eq)]
-
pub struct MessageFrame {
-
#[derive(Debug, Clone, PartialEq, Eq)]
-
pub struct ErrorFrame {}
-
impl TryFrom<&[u8]> for Frame {
-
type Error = anyhow::Error;
-
fn try_from(value: &[u8]) -> Result<Self> {
-
let mut cursor = Cursor::new(value);
-
let (left, right) = match serde_ipld_dagcbor::from_reader::<Ipld, _>(&mut cursor) {
-
Err(serde_ipld_dagcbor::DecodeError::TrailingData) => {
-
value.split_at(cursor.position() as usize)
-
return Err(anyhow!("invalid frame type"));
-
let header = FrameHeader::try_from(serde_ipld_dagcbor::from_slice::<Ipld>(left)?)?;
-
if let FrameHeader::Message(t) = &header {
-
Ok(Frame::Message(t.clone(), MessageFrame { body: right.to_vec() }))
-
Ok(Frame::Error(ErrorFrame {}))
-
/// Subscribe to a repo's firehose and update inodes on changes
-
pub async fn subscribe_to_repo<R>(
-
inodes: Arc<Mutex<IndexSet<PdsFsEntry>>>,
-
sizes: Arc<Mutex<IndexMap<usize, u64>>>,
-
content_cache: Arc<Mutex<IndexMap<String, String>>>,
-
notifier: fuser::Notifier,
-
R: atrium_repo::blockstore::AsyncBlockStoreRead,
-
// Strip https:// or http:// prefix from PDS URL if present
-
let pds_host = pds.trim_start_matches("https://").trim_start_matches("http://");
-
let url = format!("wss://{}/xrpc/{}", pds_host, NSID);
-
println!("Connecting to firehose: {}", url);
-
let (mut stream, _) = connect_async(url).await?;
-
println!("Connected to firehose for {}", did);
-
match stream.next().await {
-
Some(Ok(Message::Binary(data))) => {
-
if let Ok(Frame::Message(Some(t), msg)) = Frame::try_from(data.as_slice()) {
-
if t.as_str() == "#commit" {
-
if let Ok(commit) = serde_ipld_dagcbor::from_reader::<Commit, _>(msg.body.as_slice()) {
-
// Only process commits for our DID
-
if commit.repo.as_str() == did {
-
if let Err(e) = handle_commit(&commit, &inodes, &sizes, &content_cache, &did, &pds, ¬ifier).await {
-
eprintln!("Error handling commit: {:?}", e);
-
Some(Ok(_)) => {} // Ignore other message types
-
eprintln!("WebSocket error: {}", e);
-
eprintln!("WebSocket closed");
-
/// Handle a commit by updating the inode tree and notifying Finder
-
async fn handle_commit(
-
inodes: &Arc<Mutex<IndexSet<PdsFsEntry>>>,
-
sizes: &Arc<Mutex<IndexMap<usize, u64>>>,
-
content_cache: &Arc<Mutex<IndexMap<String, String>>>,
-
notifier: &fuser::Notifier,
-
let did_entry = PdsFsEntry::Did(did.to_string());
-
let inodes_lock = inodes.lock().unwrap();
-
inodes_lock.get_index_of(&did_entry)
-
let Some(did_inode) = did_inode else {
-
return Err(anyhow!("DID not found in inodes"));
-
for op in &commit.ops {
-
let Some((collection, rkey)) = op.path.split_once('/') else {
-
match op.action.as_str() {
-
// Fetch the record from PDS
-
let record_key = format!("{}/{}", collection, rkey);
-
let cache_key = format!("{}/{}", did, record_key);
-
// Fetch record content from PDS
-
match fetch_record(pds, did, collection, rkey).await {
-
let content_len = content.len() as u64;
-
// Add the record to inodes
-
let (collection_inode, record_inode) = {
-
let mut inodes_lock = inodes.lock().unwrap();
-
// Ensure collection exists
-
let collection_entry = PdsFsEntry::Collection(PdsFsCollection {
-
nsid: collection.to_string(),
-
let (collection_inode, _) = inodes_lock.insert_full(collection_entry);
-
let record_entry = PdsFsEntry::Record(PdsFsRecord {
-
parent: collection_inode,
-
rkey: rkey.to_string(),
-
let (record_inode, _) = inodes_lock.insert_full(record_entry);
-
(collection_inode, record_inode)
-
// Cache the content and size
-
content_cache.lock().unwrap().insert(cache_key, content);
-
sizes.lock().unwrap().insert(record_inode, content_len);
-
// Notify Finder about the new file (release lock first)
-
let filename = format!("{}.json", rkey);
-
if let Err(e) = notifier.inval_entry(collection_inode as u64, filename.as_ref()) {
-
eprintln!("Failed to invalidate entry for {}: {}", filename, e);
-
println!("Created: {}/{}", collection, rkey);
-
eprintln!("Failed to fetch record {}/{}: {}", collection, rkey, e);
-
// Get inodes before removing
-
let (collection_inode_opt, child_inode_opt) = {
-
let mut inodes_lock = inodes.lock().unwrap();
-
let collection_entry = PdsFsEntry::Collection(PdsFsCollection {
-
nsid: collection.to_string(),
-
let collection_inode = inodes_lock.get_index_of(&collection_entry);
-
// Find and remove the record
-
let child_inode = if let Some(coll_ino) = collection_inode {
-
let record_entry = PdsFsEntry::Record(PdsFsRecord {
-
rkey: rkey.to_string(),
-
let child_ino = inodes_lock.get_index_of(&record_entry);
-
inodes_lock.shift_remove(&record_entry);
-
(collection_inode, child_inode)
-
// Notify Finder about the deletion (release lock first)
-
if let (Some(coll_ino), Some(child_ino)) = (collection_inode_opt, child_inode_opt) {
-
sizes.lock().unwrap().shift_remove(&child_ino);
-
let cache_key = format!("{}/{}/{}", did, collection, rkey);
-
content_cache.lock().unwrap().shift_remove(&cache_key);
-
let filename = format!("{}.json", rkey);
-
if let Err(e) = notifier.delete(coll_ino as u64, child_ino as u64, filename.as_ref()) {
-
eprintln!("Failed to notify deletion for {}: {}", filename, e);
-
println!("Deleted: {}/{}", collection, rkey);
-
// For updates, invalidate the inode so content is re-fetched
-
let record_inode_opt = {
-
let inodes_lock = inodes.lock().unwrap();
-
let collection_entry = PdsFsEntry::Collection(PdsFsCollection {
-
nsid: collection.to_string(),
-
if let Some(collection_inode) = inodes_lock.get_index_of(&collection_entry) {
-
let record_entry = PdsFsEntry::Record(PdsFsRecord {
-
parent: collection_inode,
-
rkey: rkey.to_string(),
-
inodes_lock.get_index_of(&record_entry)
-
// Notify Finder to invalidate the inode (release lock first)
-
if let Some(record_ino) = record_inode_opt {
-
// Clear caches so content is recalculated
-
sizes.lock().unwrap().shift_remove(&record_ino);
-
let cache_key = format!("{}/{}/{}", did, collection, rkey);
-
content_cache.lock().unwrap().shift_remove(&cache_key);
-
// Invalidate the entire inode (metadata and all data)
-
if let Err(e) = notifier.inval_inode(record_ino as u64, 0, 0) {
-
eprintln!("Failed to invalidate inode for {}/{}: {}", collection, rkey, e);
-
println!("Updated: {}/{}", collection, rkey);
-
/// Fetch a record from the PDS
-
async fn fetch_record(pds: &str, did: &str, collection: &str, rkey: &str) -> Result<String> {
-
let client = AtpServiceClient::new(IsahcClient::new(pds));
-
let did = types::string::Did::new(did.to_string()).map_err(|e| anyhow!(e))?;
-
let collection_nsid = types::string::Nsid::new(collection.to_string()).map_err(|e| anyhow!(e))?;
-
let record_key = types::string::RecordKey::new(rkey.to_string()).map_err(|e| anyhow!(e))?;
-
.get_record(com::atproto::repo::get_record::Parameters::from(
-
com::atproto::repo::get_record::ParametersData {
-
collection: collection_nsid,
-
repo: types::string::AtIdentifier::Did(did),
-
Ok(serde_json::to_string_pretty(&response.value)?)
danabra.mov