parakeet.at
Parakeet is a Rust-based Bluesky AppServer aiming to implement most of the functionality required to support the Bluesky client
appview
atproto
bluesky
rust
appserver
start on commit processing
mia.omg.lol
+54
-5
Cargo.lock
+54
-5
Cargo.lock
···
83
83
]
84
84
85
85
[[package]]
86
+
name = "anyhow"
87
+
version = "1.0.95"
88
+
source = "registry+https://github.com/rust-lang/crates.io-index"
89
+
checksum = "34ac096ce696dc2fcabef30516bb13c0a68a11d30131d3df6f04711467681b04"
90
+
91
+
[[package]]
86
92
name = "async-trait"
87
93
version = "0.1.85"
88
94
source = "registry+https://github.com/rust-lang/crates.io-index"
···
251
257
"multihash",
252
258
"serde",
253
259
"serde_bytes",
254
-
"unsigned-varint",
260
+
"unsigned-varint 0.8.0",
255
261
]
256
262
257
263
[[package]]
···
312
318
"figment",
313
319
"futures",
314
320
"ipld-core",
321
+
"iroh-car",
315
322
"parakeet-db",
316
323
"serde",
317
324
"serde_bytes",
···
839
846
]
840
847
841
848
[[package]]
849
+
name = "iroh-car"
850
+
version = "0.5.1"
851
+
source = "registry+https://github.com/rust-lang/crates.io-index"
852
+
checksum = "cb7f8cd4cb9aa083fba8b52e921764252d0b4dcb1cd6d120b809dbfe1106e81a"
853
+
dependencies = [
854
+
"anyhow",
855
+
"cid",
856
+
"futures",
857
+
"serde",
858
+
"serde_ipld_dagcbor",
859
+
"thiserror 1.0.69",
860
+
"tokio",
861
+
"unsigned-varint 0.7.2",
862
+
]
863
+
864
+
[[package]]
842
865
name = "is_terminal_polyfill"
843
866
version = "1.70.1"
844
867
source = "registry+https://github.com/rust-lang/crates.io-index"
···
949
972
dependencies = [
950
973
"core2",
951
974
"serde",
952
-
"unsigned-varint",
975
+
"unsigned-varint 0.8.0",
953
976
]
954
977
955
978
[[package]]
···
1095
1118
"eyre",
1096
1119
"serde",
1097
1120
"serde_json",
1098
-
"thiserror",
1121
+
"thiserror 2.0.11",
1099
1122
"walkdir",
1100
1123
]
1101
1124
···
1571
1594
1572
1595
[[package]]
1573
1596
name = "thiserror"
1597
+
version = "1.0.69"
1598
+
source = "registry+https://github.com/rust-lang/crates.io-index"
1599
+
checksum = "b6aaf5339b578ea85b50e080feb250a3e8ae8cfcdff9a461c9ec2904bc923f52"
1600
+
dependencies = [
1601
+
"thiserror-impl 1.0.69",
1602
+
]
1603
+
1604
+
[[package]]
1605
+
name = "thiserror"
1574
1606
version = "2.0.11"
1575
1607
source = "registry+https://github.com/rust-lang/crates.io-index"
1576
1608
checksum = "d452f284b73e6d76dd36758a0c8684b1d5be31f92b89d07fd5822175732206fc"
1577
1609
dependencies = [
1578
-
"thiserror-impl",
1610
+
"thiserror-impl 2.0.11",
1611
+
]
1612
+
1613
+
[[package]]
1614
+
name = "thiserror-impl"
1615
+
version = "1.0.69"
1616
+
source = "registry+https://github.com/rust-lang/crates.io-index"
1617
+
checksum = "4fee6c4efc90059e10f81e6d42c60a18f76588c3d74cb83a0b242a2b6c7504c1"
1618
+
dependencies = [
1619
+
"proc-macro2",
1620
+
"quote",
1621
+
"syn",
1579
1622
]
1580
1623
1581
1624
[[package]]
···
1812
1855
"native-tls",
1813
1856
"rand",
1814
1857
"sha1",
1815
-
"thiserror",
1858
+
"thiserror 2.0.11",
1816
1859
"utf-8",
1817
1860
]
1818
1861
···
1857
1900
version = "0.1.3"
1858
1901
source = "registry+https://github.com/rust-lang/crates.io-index"
1859
1902
checksum = "e70f2a8b45122e719eb623c01822704c4e0907e7e426a05927e1a1cfff5b75d0"
1903
+
1904
+
[[package]]
1905
+
name = "unsigned-varint"
1906
+
version = "0.7.2"
1907
+
source = "registry+https://github.com/rust-lang/crates.io-index"
1908
+
checksum = "6889a77d49f1f013504cec6bf97a2c730394adedaeb1deb5ea08949a50541105"
1860
1909
1861
1910
[[package]]
1862
1911
name = "unsigned-varint"
+1
consumer/Cargo.toml
+1
consumer/Cargo.toml
+64
-4
consumer/src/indexer/mod.rs
+64
-4
consumer/src/indexer/mod.rs
···
1
+
use crate::firehose::{AtpAccountEvent, AtpCommitEvent, AtpIdentityEvent, FirehoseEvent};
2
+
use crate::indexer::types::{CollectionType, RecordTypes};
3
+
use diesel_async::pooled_connection::deadpool::Pool;
1
4
use diesel_async::AsyncPgConnection;
2
-
use diesel_async::pooled_connection::deadpool::Pool;
3
-
use crate::firehose::{AtpAccountEvent, AtpCommitEvent, AtpIdentityEvent, FirehoseEvent};
5
+
use futures::StreamExt;
6
+
use ipld_core::cid::Cid;
7
+
use std::collections::HashMap;
4
8
use tokio::sync::mpsc::Receiver;
9
+
use tracing::Instrument;
5
10
6
-
pub async fn relay_indexer(pool: Pool<AsyncPgConnection>, mut rx: Receiver<FirehoseEvent>) -> eyre::Result<()> {
11
+
mod types;
12
+
13
+
pub async fn relay_indexer(
14
+
pool: Pool<AsyncPgConnection>,
15
+
mut rx: Receiver<FirehoseEvent>,
16
+
) -> eyre::Result<()> {
7
17
let mut conn = pool.get().await?;
8
18
9
19
while let Some(event) = rx.recv().await {
10
20
let res = match event {
11
21
FirehoseEvent::Identity(identity) => index_identity(&mut conn, identity).await,
12
22
FirehoseEvent::Account(account) => index_account(&mut conn, account).await,
13
-
FirehoseEvent::Commit(commit) => index_commit(&mut conn, commit).await,
23
+
FirehoseEvent::Commit(commit) => {
24
+
let span = tracing::info_span!(
25
+
"commit",
26
+
seq = commit.seq,
27
+
repo = commit.repo,
28
+
rev = commit.rev
29
+
);
30
+
index_commit(&mut conn, commit).instrument(span).await
31
+
}
14
32
FirehoseEvent::Label(_) => {
15
33
// We handle all labels through direct connections to labelers
16
34
tracing::warn!("got #labels from the relay");
···
35
53
}
36
54
37
55
async fn index_commit(conn: &mut AsyncPgConnection, commit: AtpCommitEvent) -> eyre::Result<()> {
56
+
// turn the car slice into a map of cid:block
57
+
let car_reader = iroh_car::CarReader::new(commit.blocks.as_slice()).await?;
58
+
let blocks = car_reader
59
+
.stream()
60
+
.filter_map(|car| async move { car.ok() })
61
+
.collect::<HashMap<Cid, Vec<u8>>>()
62
+
.await;
63
+
64
+
for op in &commit.ops {
65
+
let Some((collection_raw, rkey)) = op.path.split_once("/") else {
66
+
tracing::warn!("op contained invalid path {}", op.path);
67
+
continue;
68
+
};
69
+
70
+
let collection = CollectionType::from_str(collection_raw);
71
+
if collection == CollectionType::Unsupported {
72
+
tracing::debug!("{} {collection_raw} is unsupported", op.action);
73
+
continue;
74
+
}
75
+
76
+
if op.action == "create" || op.action == "update" {
77
+
if let Some(block) = op.cid.and_then(|cid| blocks.get(&cid)) {
78
+
let decoded: RecordTypes = match serde_ipld_dagcbor::from_slice(block) {
79
+
Ok(decoded) => decoded,
80
+
Err(err) => {
81
+
tracing::error!("Failed to decode record: {err}");
82
+
continue;
83
+
}
84
+
};
85
+
86
+
dbg!(decoded);
87
+
} else {
88
+
tracing::error!("Missing Cid or the block was not found");
89
+
continue;
90
+
}
91
+
} else if op.action == "delete" {
92
+
//
93
+
} else {
94
+
tracing::warn!("op contained invalid action {}", op.action);
95
+
}
96
+
}
97
+
38
98
Ok(())
39
99
}
+24
consumer/src/indexer/types.rs
+24
consumer/src/indexer/types.rs
···
1
+
use serde::{Deserialize, Serialize};
2
+
3
+
#[derive(Debug, Deserialize, Serialize)]
4
+
#[serde(tag = "$type")]
5
+
pub enum RecordTypes {}
6
+
7
+
#[derive(Debug, PartialOrd, PartialEq)]
8
+
pub enum CollectionType {
9
+
Unsupported,
10
+
}
11
+
12
+
impl CollectionType {
13
+
pub(crate) fn from_str(input: &str) -> CollectionType {
14
+
match input {
15
+
_ => CollectionType::Unsupported,
16
+
}
17
+
}
18
+
19
+
pub fn can_update(&self) -> bool {
20
+
match self {
21
+
CollectionType::Unsupported => false,
22
+
}
23
+
}
24
+
}