···
1
-
use anyhow::{anyhow, Result};
2
-
use atrium_api::com::atproto::sync::subscribe_repos::{Commit, NSID};
3
-
use atrium_api::client::AtpServiceClient;
5
-
use atrium_api::types;
6
-
use atrium_xrpc_client::isahc::IsahcClient;
7
-
use futures::StreamExt;
8
-
use ipld_core::ipld::Ipld;
10
-
use std::sync::{Arc, Mutex};
11
-
use tokio_tungstenite::connect_async;
12
-
use tokio_tungstenite::tungstenite::Message;
14
-
use crate::fs::{PdsFsCollection, PdsFsEntry, PdsFsRecord};
15
-
use indexmap::{IndexMap, IndexSet};
17
-
/// Frame header types for WebSocket messages
18
-
#[derive(Debug, Clone, PartialEq, Eq)]
20
-
Message(Option<String>),
24
-
impl TryFrom<Ipld> for FrameHeader {
25
-
type Error = anyhow::Error;
27
-
fn try_from(value: Ipld) -> Result<Self> {
28
-
if let Ipld::Map(map) = value {
29
-
if let Some(Ipld::Integer(i)) = map.get("op") {
32
-
let t = if let Some(Ipld::String(s)) = map.get("t") {
37
-
return Ok(FrameHeader::Message(t));
39
-
-1 => return Ok(FrameHeader::Error),
44
-
Err(anyhow!("invalid frame type"))
48
-
/// Frame types for parsed WebSocket messages
49
-
#[derive(Debug, Clone, PartialEq, Eq)]
51
-
Message(Option<String>, MessageFrame),
55
-
#[derive(Debug, Clone, PartialEq, Eq)]
56
-
pub struct MessageFrame {
60
-
#[derive(Debug, Clone, PartialEq, Eq)]
61
-
pub struct ErrorFrame {}
63
-
impl TryFrom<&[u8]> for Frame {
64
-
type Error = anyhow::Error;
66
-
fn try_from(value: &[u8]) -> Result<Self> {
67
-
let mut cursor = Cursor::new(value);
68
-
let (left, right) = match serde_ipld_dagcbor::from_reader::<Ipld, _>(&mut cursor) {
69
-
Err(serde_ipld_dagcbor::DecodeError::TrailingData) => {
70
-
value.split_at(cursor.position() as usize)
73
-
return Err(anyhow!("invalid frame type"));
76
-
let header = FrameHeader::try_from(serde_ipld_dagcbor::from_slice::<Ipld>(left)?)?;
77
-
if let FrameHeader::Message(t) = &header {
78
-
Ok(Frame::Message(t.clone(), MessageFrame { body: right.to_vec() }))
80
-
Ok(Frame::Error(ErrorFrame {}))
85
-
/// Subscribe to a repo's firehose and update inodes on changes
86
-
pub async fn subscribe_to_repo<R>(
89
-
inodes: Arc<Mutex<IndexSet<PdsFsEntry>>>,
90
-
sizes: Arc<Mutex<IndexMap<usize, u64>>>,
91
-
content_cache: Arc<Mutex<IndexMap<String, String>>>,
92
-
notifier: fuser::Notifier,
95
-
R: atrium_repo::blockstore::AsyncBlockStoreRead,
97
-
// Strip https:// or http:// prefix from PDS URL if present
98
-
let pds_host = pds.trim_start_matches("https://").trim_start_matches("http://");
99
-
let url = format!("wss://{}/xrpc/{}", pds_host, NSID);
100
-
println!("Connecting to firehose: {}", url);
102
-
let (mut stream, _) = connect_async(url).await?;
103
-
println!("Connected to firehose for {}", did);
106
-
match stream.next().await {
107
-
Some(Ok(Message::Binary(data))) => {
108
-
if let Ok(Frame::Message(Some(t), msg)) = Frame::try_from(data.as_slice()) {
109
-
if t.as_str() == "#commit" {
110
-
if let Ok(commit) = serde_ipld_dagcbor::from_reader::<Commit, _>(msg.body.as_slice()) {
111
-
// Only process commits for our DID
112
-
if commit.repo.as_str() == did {
113
-
if let Err(e) = handle_commit(&commit, &inodes, &sizes, &content_cache, &did, &pds, ¬ifier).await {
114
-
eprintln!("Error handling commit: {:?}", e);
121
-
Some(Ok(_)) => {} // Ignore other message types
123
-
eprintln!("WebSocket error: {}", e);
127
-
eprintln!("WebSocket closed");
136
-
/// Handle a commit by updating the inode tree and notifying Finder
137
-
async fn handle_commit(
139
-
inodes: &Arc<Mutex<IndexSet<PdsFsEntry>>>,
140
-
sizes: &Arc<Mutex<IndexMap<usize, u64>>>,
141
-
content_cache: &Arc<Mutex<IndexMap<String, String>>>,
144
-
notifier: &fuser::Notifier,
146
-
// Find the DID inode
147
-
let did_entry = PdsFsEntry::Did(did.to_string());
149
-
let inodes_lock = inodes.lock().unwrap();
150
-
inodes_lock.get_index_of(&did_entry)
153
-
let Some(did_inode) = did_inode else {
154
-
return Err(anyhow!("DID not found in inodes"));
157
-
for op in &commit.ops {
158
-
let Some((collection, rkey)) = op.path.split_once('/') else {
162
-
match op.action.as_str() {
164
-
// Fetch the record from PDS
165
-
let record_key = format!("{}/{}", collection, rkey);
166
-
let cache_key = format!("{}/{}", did, record_key);
168
-
// Fetch record content from PDS
169
-
match fetch_record(pds, did, collection, rkey).await {
171
-
let content_len = content.len() as u64;
173
-
// Add the record to inodes
174
-
let (collection_inode, record_inode) = {
175
-
let mut inodes_lock = inodes.lock().unwrap();
177
-
// Ensure collection exists
178
-
let collection_entry = PdsFsEntry::Collection(PdsFsCollection {
180
-
nsid: collection.to_string(),
182
-
let (collection_inode, _) = inodes_lock.insert_full(collection_entry);
185
-
let record_entry = PdsFsEntry::Record(PdsFsRecord {
186
-
parent: collection_inode,
187
-
rkey: rkey.to_string(),
189
-
let (record_inode, _) = inodes_lock.insert_full(record_entry);
190
-
(collection_inode, record_inode)
193
-
// Cache the content and size
194
-
content_cache.lock().unwrap().insert(cache_key, content);
195
-
sizes.lock().unwrap().insert(record_inode, content_len);
197
-
// Notify Finder about the new file (release lock first)
198
-
let filename = format!("{}.json", rkey);
199
-
if let Err(e) = notifier.inval_entry(collection_inode as u64, filename.as_ref()) {
200
-
eprintln!("Failed to invalidate entry for {}: {}", filename, e);
203
-
println!("Created: {}/{}", collection, rkey);
206
-
eprintln!("Failed to fetch record {}/{}: {}", collection, rkey, e);
211
-
// Get inodes before removing
212
-
let (collection_inode_opt, child_inode_opt) = {
213
-
let mut inodes_lock = inodes.lock().unwrap();
215
-
// Find the collection
216
-
let collection_entry = PdsFsEntry::Collection(PdsFsCollection {
218
-
nsid: collection.to_string(),
220
-
let collection_inode = inodes_lock.get_index_of(&collection_entry);
222
-
// Find and remove the record
223
-
let child_inode = if let Some(coll_ino) = collection_inode {
224
-
let record_entry = PdsFsEntry::Record(PdsFsRecord {
226
-
rkey: rkey.to_string(),
228
-
let child_ino = inodes_lock.get_index_of(&record_entry);
229
-
inodes_lock.shift_remove(&record_entry);
235
-
(collection_inode, child_inode)
238
-
// Notify Finder about the deletion (release lock first)
239
-
if let (Some(coll_ino), Some(child_ino)) = (collection_inode_opt, child_inode_opt) {
240
-
// Remove from caches
241
-
sizes.lock().unwrap().shift_remove(&child_ino);
242
-
let cache_key = format!("{}/{}/{}", did, collection, rkey);
243
-
content_cache.lock().unwrap().shift_remove(&cache_key);
245
-
let filename = format!("{}.json", rkey);
246
-
if let Err(e) = notifier.delete(coll_ino as u64, child_ino as u64, filename.as_ref()) {
247
-
eprintln!("Failed to notify deletion for {}: {}", filename, e);
251
-
println!("Deleted: {}/{}", collection, rkey);
254
-
// For updates, invalidate the inode so content is re-fetched
255
-
let record_inode_opt = {
256
-
let inodes_lock = inodes.lock().unwrap();
257
-
let collection_entry = PdsFsEntry::Collection(PdsFsCollection {
259
-
nsid: collection.to_string(),
262
-
if let Some(collection_inode) = inodes_lock.get_index_of(&collection_entry) {
263
-
let record_entry = PdsFsEntry::Record(PdsFsRecord {
264
-
parent: collection_inode,
265
-
rkey: rkey.to_string(),
267
-
inodes_lock.get_index_of(&record_entry)
273
-
// Notify Finder to invalidate the inode (release lock first)
274
-
if let Some(record_ino) = record_inode_opt {
275
-
// Clear caches so content is recalculated
276
-
sizes.lock().unwrap().shift_remove(&record_ino);
277
-
let cache_key = format!("{}/{}/{}", did, collection, rkey);
278
-
content_cache.lock().unwrap().shift_remove(&cache_key);
280
-
// Invalidate the entire inode (metadata and all data)
281
-
if let Err(e) = notifier.inval_inode(record_ino as u64, 0, 0) {
282
-
eprintln!("Failed to invalidate inode for {}/{}: {}", collection, rkey, e);
286
-
println!("Updated: {}/{}", collection, rkey);
295
-
/// Fetch a record from the PDS
296
-
async fn fetch_record(pds: &str, did: &str, collection: &str, rkey: &str) -> Result<String> {
297
-
let client = AtpServiceClient::new(IsahcClient::new(pds));
298
-
let did = types::string::Did::new(did.to_string()).map_err(|e| anyhow!(e))?;
299
-
let collection_nsid = types::string::Nsid::new(collection.to_string()).map_err(|e| anyhow!(e))?;
300
-
let record_key = types::string::RecordKey::new(rkey.to_string()).map_err(|e| anyhow!(e))?;
302
-
let response = client
307
-
.get_record(com::atproto::repo::get_record::Parameters::from(
308
-
com::atproto::repo::get_record::ParametersData {
310
-
collection: collection_nsid,
311
-
repo: types::string::AtIdentifier::Did(did),
317
-
Ok(serde_json::to_string_pretty(&response.value)?)
danabra.mov