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
chore: just a few formatting fixes
+49
-31
+13
-10
consumer/src/backfill/mod.rs
+13
-10
consumer/src/backfill/mod.rs
···
170
170
let status = repo_status
171
171
.status
172
172
.unwrap_or(crate::firehose::AtpAccountStatus::Deleted);
173
-
db::actor_upsert(
174
-
conn,
175
-
did,
176
-
status.into(),
177
-
ActorSyncState::Dirty,
178
-
Utc::now(),
179
-
)
180
-
.await?;
173
+
db::actor_upsert(conn, did, status.into(), ActorSyncState::Dirty, Utc::now()).await?;
181
174
return Ok(());
182
175
}
183
176
···
336
329
337
330
#[derive(Debug, Default)]
338
331
struct CopyStore {
339
-
likes: Vec<(String, records::StrongRef, Option<records::StrongRef>, DateTime<Utc>)>,
332
+
likes: Vec<(
333
+
String,
334
+
records::StrongRef,
335
+
Option<records::StrongRef>,
336
+
DateTime<Utc>,
337
+
)>,
340
338
posts: Vec<(String, Cid, records::AppBskyFeedPost)>,
341
-
reposts: Vec<(String, records::StrongRef, Option<records::StrongRef>, DateTime<Utc>)>,
339
+
reposts: Vec<(
340
+
String,
341
+
records::StrongRef,
342
+
Option<records::StrongRef>,
343
+
DateTime<Utc>,
344
+
)>,
342
345
blocks: Vec<(String, String, DateTime<Utc>)>,
343
346
follows: Vec<(String, String, DateTime<Utc>)>,
344
347
list_items: Vec<(String, records::AppBskyGraphListItem)>,
+6
-2
consumer/src/backfill/repo.rs
+6
-2
consumer/src/backfill/repo.rs
···
131
131
deltas.incr(&rec.subject.uri, AggregateType::Like).await;
132
132
133
133
copies.push_record(&at_uri, cid);
134
-
copies.likes.push((at_uri, rec.subject, rec.via, rec.created_at));
134
+
copies
135
+
.likes
136
+
.push((at_uri, rec.subject, rec.via, rec.created_at));
135
137
}
136
138
RecordTypes::AppBskyFeedPost(rec) => {
137
139
let maybe_reply = rec.reply.as_ref().map(|v| v.parent.uri.clone());
···
167
169
deltas.incr(&rec.subject.uri, AggregateType::Repost).await;
168
170
169
171
copies.push_record(&at_uri, cid);
170
-
copies.reposts.push((at_uri, rec.subject, rec.via, rec.created_at));
172
+
copies
173
+
.reposts
174
+
.push((at_uri, rec.subject, rec.via, rec.created_at));
171
175
}
172
176
RecordTypes::AppBskyGraphBlock(rec) => {
173
177
copies.push_record(&at_uri, cid);
+5
-1
consumer/src/db/backfill.rs
+5
-1
consumer/src/db/backfill.rs
···
13
13
pub indexed_at: NaiveDateTime,
14
14
}
15
15
16
-
pub async fn backfill_job_write<C: GenericClient>(conn: &mut C, did: &str, status: &str) -> PgExecResult {
16
+
pub async fn backfill_job_write<C: GenericClient>(
17
+
conn: &mut C,
18
+
did: &str,
19
+
status: &str,
20
+
) -> PgExecResult {
17
21
conn.execute(
18
22
"INSERT INTO backfill_jobs (did, status) VALUES ($1, $2)",
19
23
&[&did, &status],
+8
-3
consumer/src/db/copy.rs
+8
-3
consumer/src/db/copy.rs
···
18
18
Type::TEXT,
19
19
Type::TIMESTAMP,
20
20
];
21
-
type StrongRefRow = (String, records::StrongRef, Option<records::StrongRef>, DateTime<Utc>);
21
+
type StrongRefRow = (
22
+
String,
23
+
records::StrongRef,
24
+
Option<records::StrongRef>,
25
+
DateTime<Utc>,
26
+
);
22
27
23
28
// SubjectRefs are used in both blocks and follows
24
29
const SUBJECT_TYPES: &[Type] = &[Type::TEXT, Type::TEXT, Type::TEXT, Type::TIMESTAMP];
···
50
55
51
56
for row in data {
52
57
let (via_uri, via_cid) = strongref_to_parts(row.2.as_ref());
53
-
58
+
54
59
let writer = writer.as_mut();
55
60
writer
56
61
.write(&[
···
97
102
98
103
for row in data {
99
104
let (via_uri, via_cid) = strongref_to_parts(row.2.as_ref());
100
-
105
+
101
106
let writer = writer.as_mut();
102
107
writer
103
108
.write(&[
+4
-4
consumer/src/db/record.rs
+4
-4
consumer/src/db/record.rs
···
158
158
rec: AppBskyFeedLike,
159
159
) -> PgExecResult {
160
160
let (via_uri, via_cid) = strongref_to_parts(rec.via.as_ref());
161
-
161
+
162
162
conn.execute(
163
163
"INSERT INTO likes (at_uri, did, subject, subject_cid, via_uri, via_cid, created_at) VALUES ($1, $2, $3, $4, $5, $6, $7)",
164
164
&[&at_uri, &repo, &rec.subject.uri, &rec.subject.cid.to_string(), &via_uri, &via_cid, &rec.created_at]
···
204
204
],
205
205
)
206
206
.await
207
-
.map(|v| v == 0)
207
+
.map(|v| v == 0)
208
208
}
209
209
210
210
pub async fn list_delete<C: GenericClient>(conn: &mut C, at_uri: &str) -> PgExecResult {
···
528
528
rec: AppBskyFeedRepost,
529
529
) -> PgExecResult {
530
530
let (via_uri, via_cid) = strongref_to_parts(rec.via.as_ref());
531
-
531
+
532
532
conn.execute(
533
533
"INSERT INTO reposts (at_uri, did, post, post_cid, via_uri, via_cid, created_at) VALUES ($1, $2, $3, $4, $5, $6, $7)",
534
534
&[
···
587
587
],
588
588
)
589
589
.await
590
-
.map(|v| v == 0)
590
+
.map(|v| v == 0)
591
591
}
592
592
593
593
pub async fn starter_pack_delete<C: GenericClient>(conn: &mut C, at_uri: &str) -> PgExecResult {
+3
-6
consumer/src/main.rs
+3
-6
consumer/src/main.rs
···
59
59
if cli.labels {
60
60
let resume = resume.clone().unwrap();
61
61
62
-
let label_mgr = label_indexer::LabelServiceManager::new(
63
-
pool.clone(),
64
-
resume,
65
-
user_agent.clone(),
66
-
)
67
-
.await?;
62
+
let label_mgr =
63
+
label_indexer::LabelServiceManager::new(pool.clone(), resume, user_agent.clone())
64
+
.await?;
68
65
69
66
if let Some(label_source) = conf.label_source {
70
67
join_set.spawn(label_mgr.run(label_source));
+1
-1
lexica/src/app_bsky/actor.rs
+1
-1
lexica/src/app_bsky/actor.rs
+1
-2
parakeet-db/src/models.rs
+1
-2
parakeet-db/src/models.rs
+8
-2
parakeet/src/xrpc/cdn.rs
+8
-2
parakeet/src/xrpc/cdn.rs
···
6
6
7
7
impl BskyCdn {
8
8
pub fn new(cdn_base: String, video_base: String) -> Self {
9
-
BskyCdn {cdn_base, video_base}
9
+
BskyCdn {
10
+
cdn_base,
11
+
video_base,
12
+
}
10
13
}
11
14
12
15
pub fn avatar(&self, did: &str, cid: &str) -> String {
···
18
21
}
19
22
20
23
pub fn embed_thumb(&self, did: &str, cid: &str) -> String {
21
-
format!("{}/img/feed_thumbnail/plain/{did}/{cid}@jpeg", self.cdn_base)
24
+
format!(
25
+
"{}/img/feed_thumbnail/plain/{did}/{cid}@jpeg",
26
+
self.cdn_base
27
+
)
22
28
}
23
29
24
30
pub fn embed_fullsize(&self, did: &str, cid: &str) -> String {