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
feat: app.bsky.notification.declaration
mia.omg.lol
+16
consumer/src/db/record.rs
+16
consumer/src/db/record.rs
···
256
.await
257
}
258
259
+
pub async fn notif_decl_upsert<C: GenericClient>(
260
+
conn: &mut C,
261
+
repo: &str,
262
+
rec: AppBskyNotificationDeclaration,
263
+
) -> PgExecResult {
264
+
conn.execute(
265
+
"INSERT INTO notif_decl (did, allow_subscriptions) VALUES ($1, $2) ON CONFLICT (did) DO UPDATE SET allow_subscriptions=EXCLUDED.allow_subscriptions",
266
+
&[&repo, &rec.allow_subscriptions.to_string()],
267
+
).await
268
+
}
269
+
270
+
pub async fn notif_decl_delete<C: GenericClient>(conn: &mut C, repo: &str) -> PgExecResult {
271
+
conn.execute("DELETE FROM notif_decl WHERE did=$1", &[&repo])
272
+
.await
273
+
}
274
+
275
pub async fn post_insert<C: GenericClient>(
276
conn: &mut C,
277
at_uri: &str,
+6
consumer/src/indexer/mod.rs
+6
consumer/src/indexer/mod.rs
···
688
}
689
}
690
}
691
RecordTypes::ChatBskyActorDeclaration(record) => {
692
if rkey == "self" {
693
db::chat_decl_upsert(conn, repo, record).await?;
···
775
}
776
CollectionType::BskyVerification => db::verification_delete(conn, at_uri).await?,
777
CollectionType::BskyLabelerService => db::labeler_delete(conn, at_uri).await?,
778
CollectionType::ChatActorDecl => db::chat_decl_delete(conn, repo).await?,
779
_ => unreachable!(),
780
};
···
688
}
689
}
690
}
691
+
RecordTypes::AppBskyNotificationDeclaration(record) => {
692
+
if rkey == "self" {
693
+
db::notif_decl_upsert(conn, repo, record).await?;
694
+
}
695
+
}
696
RecordTypes::ChatBskyActorDeclaration(record) => {
697
if rkey == "self" {
698
db::chat_decl_upsert(conn, repo, record).await?;
···
780
}
781
CollectionType::BskyVerification => db::verification_delete(conn, at_uri).await?,
782
CollectionType::BskyLabelerService => db::labeler_delete(conn, at_uri).await?,
783
+
CollectionType::BskyNotificationDeclaration => db::notif_decl_delete(conn, repo).await?,
784
CollectionType::ChatActorDecl => db::chat_decl_delete(conn, repo).await?,
785
_ => unreachable!(),
786
};
+7
-1
consumer/src/indexer/records.rs
+7
-1
consumer/src/indexer/records.rs
···
1
use crate::utils;
2
use chrono::{DateTime, Utc};
3
use ipld_core::cid::Cid;
4
-
use lexica::app_bsky::actor::{ChatAllowIncoming, Status};
5
use lexica::app_bsky::embed::AspectRatio;
6
use lexica::app_bsky::labeler::LabelerPolicy;
7
use lexica::app_bsky::richtext::FacetMain;
···
405
pub subject_types: Option<Vec<SubjectType>>,
406
pub subject_collections: Option<Vec<String>>,
407
pub created_at: DateTime<Utc>,
408
}
409
410
#[derive(Debug, Deserialize, Serialize)]
···
1
use crate::utils;
2
use chrono::{DateTime, Utc};
3
use ipld_core::cid::Cid;
4
+
use lexica::app_bsky::actor::{ChatAllowIncoming, ProfileAllowSubscriptions, Status};
5
use lexica::app_bsky::embed::AspectRatio;
6
use lexica::app_bsky::labeler::LabelerPolicy;
7
use lexica::app_bsky::richtext::FacetMain;
···
405
pub subject_types: Option<Vec<SubjectType>>,
406
pub subject_collections: Option<Vec<String>>,
407
pub created_at: DateTime<Utc>,
408
+
}
409
+
410
+
#[derive(Debug, Deserialize, Serialize)]
411
+
#[serde(rename_all = "camelCase")]
412
+
pub struct AppBskyNotificationDeclaration {
413
+
pub allow_subscriptions: ProfileAllowSubscriptions,
414
}
415
416
#[derive(Debug, Deserialize, Serialize)]
+5
consumer/src/indexer/types.rs
+5
consumer/src/indexer/types.rs
···
37
AppBskyGraphVerification(records::AppBskyGraphVerification),
38
#[serde(rename = "app.bsky.labeler.service")]
39
AppBskyLabelerService(records::AppBskyLabelerService),
40
#[serde(rename = "chat.bsky.actor.declaration")]
41
ChatBskyActorDeclaration(records::ChatBskyActorDeclaration),
42
}
···
59
BskyStarterPack,
60
BskyVerification,
61
BskyLabelerService,
62
ChatActorDecl,
63
Unsupported,
64
}
···
82
"app.bsky.graph.starterpack" => CollectionType::BskyStarterPack,
83
"app.bsky.graph.verification" => CollectionType::BskyVerification,
84
"app.bsky.labeler.service" => CollectionType::BskyLabelerService,
85
"chat.bsky.actor.declaration" => CollectionType::ChatActorDecl,
86
_ => CollectionType::Unsupported,
87
}
···
106
CollectionType::BskyStarterPack => true,
107
CollectionType::BskyVerification => false,
108
CollectionType::BskyLabelerService => true,
109
CollectionType::Unsupported => false,
110
}
111
}
···
37
AppBskyGraphVerification(records::AppBskyGraphVerification),
38
#[serde(rename = "app.bsky.labeler.service")]
39
AppBskyLabelerService(records::AppBskyLabelerService),
40
+
#[serde(rename = "app.bsky.notification.declaration")]
41
+
AppBskyNotificationDeclaration(records::AppBskyNotificationDeclaration),
42
#[serde(rename = "chat.bsky.actor.declaration")]
43
ChatBskyActorDeclaration(records::ChatBskyActorDeclaration),
44
}
···
61
BskyStarterPack,
62
BskyVerification,
63
BskyLabelerService,
64
+
BskyNotificationDeclaration,
65
ChatActorDecl,
66
Unsupported,
67
}
···
85
"app.bsky.graph.starterpack" => CollectionType::BskyStarterPack,
86
"app.bsky.graph.verification" => CollectionType::BskyVerification,
87
"app.bsky.labeler.service" => CollectionType::BskyLabelerService,
88
+
"app.bsky.notification.declaration" => CollectionType::BskyNotificationDeclaration,
89
"chat.bsky.actor.declaration" => CollectionType::ChatActorDecl,
90
_ => CollectionType::Unsupported,
91
}
···
110
CollectionType::BskyStarterPack => true,
111
CollectionType::BskyVerification => false,
112
CollectionType::BskyLabelerService => true,
113
+
CollectionType::BskyNotificationDeclaration => true,
114
CollectionType::Unsupported => false,
115
}
116
}
+39
lexica/src/app_bsky/actor.rs
+39
lexica/src/app_bsky/actor.rs
···
14
pub labeler: bool,
15
#[serde(skip_serializing_if = "Option::is_none")]
16
pub chat: Option<ProfileAssociatedChat>,
17
}
18
19
#[derive(Clone, Debug, Serialize)]
···
48
"all" => Ok(ChatAllowIncoming::All),
49
"none" => Ok(ChatAllowIncoming::None),
50
"following" => Ok(ChatAllowIncoming::Following),
51
x => Err(format!("Unrecognized variant {x}")),
52
}
53
}
···
14
pub labeler: bool,
15
#[serde(skip_serializing_if = "Option::is_none")]
16
pub chat: Option<ProfileAssociatedChat>,
17
+
#[serde(skip_serializing_if = "Option::is_none")]
18
+
pub activity_subscription: Option<ProfileAssociatedActivitySubscription>,
19
}
20
21
#[derive(Clone, Debug, Serialize)]
···
50
"all" => Ok(ChatAllowIncoming::All),
51
"none" => Ok(ChatAllowIncoming::None),
52
"following" => Ok(ChatAllowIncoming::Following),
53
+
x => Err(format!("Unrecognized variant {x}")),
54
+
}
55
+
}
56
+
}
57
+
58
+
#[derive(Clone, Debug, Serialize)]
59
+
#[serde(rename_all = "camelCase")]
60
+
pub struct ProfileAssociatedActivitySubscription {
61
+
pub allow_subscriptions: ProfileAllowSubscriptions,
62
+
}
63
+
64
+
#[derive(Copy, Clone, Debug, Deserialize, Serialize)]
65
+
#[serde(rename_all = "lowercase")]
66
+
pub enum ProfileAllowSubscriptions {
67
+
Followers,
68
+
Mutuals,
69
+
None,
70
+
}
71
+
72
+
impl Display for ProfileAllowSubscriptions {
73
+
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
74
+
match self {
75
+
ProfileAllowSubscriptions::Followers => write!(f, "followers"),
76
+
ProfileAllowSubscriptions::Mutuals => write!(f, "mutuals"),
77
+
ProfileAllowSubscriptions::None => write!(f, "none"),
78
+
}
79
+
}
80
+
}
81
+
82
+
impl FromStr for ProfileAllowSubscriptions {
83
+
type Err = String;
84
+
85
+
fn from_str(s: &str) -> Result<Self, Self::Err> {
86
+
match s {
87
+
"none" => Ok(ProfileAllowSubscriptions::None),
88
+
"mutuals" => Ok(ProfileAllowSubscriptions::Mutuals),
89
+
"following" => Ok(ProfileAllowSubscriptions::Followers),
90
x => Err(format!("Unrecognized variant {x}")),
91
}
92
}
+1
migrations/2025-07-21-173906_notif-decl/down.sql
+1
migrations/2025-07-21-173906_notif-decl/down.sql
···
···
1
+
drop table notif_decl;
+6
migrations/2025-07-21-173906_notif-decl/up.sql
+6
migrations/2025-07-21-173906_notif-decl/up.sql
+10
parakeet-db/src/schema.rs
+10
parakeet-db/src/schema.rs
···
166
}
167
168
diesel::table! {
169
post_embed_ext (post_uri) {
170
post_uri -> Text,
171
uri -> Text,
···
359
diesel::joinable!(likes -> actors (did));
360
diesel::joinable!(list_blocks -> actors (did));
361
diesel::joinable!(lists -> actors (owner));
362
diesel::joinable!(post_embed_ext -> posts (post_uri));
363
diesel::joinable!(post_embed_images -> posts (post_uri));
364
diesel::joinable!(post_embed_record -> posts (post_uri));
···
388
list_blocks,
389
list_items,
390
lists,
391
post_embed_ext,
392
post_embed_images,
393
post_embed_record,
···
166
}
167
168
diesel::table! {
169
+
notif_decl (did) {
170
+
did -> Text,
171
+
allow_subscriptions -> Nullable<Text>,
172
+
indexed_at -> Timestamp,
173
+
}
174
+
}
175
+
176
+
diesel::table! {
177
post_embed_ext (post_uri) {
178
post_uri -> Text,
179
uri -> Text,
···
367
diesel::joinable!(likes -> actors (did));
368
diesel::joinable!(list_blocks -> actors (did));
369
diesel::joinable!(lists -> actors (owner));
370
+
diesel::joinable!(notif_decl -> actors (did));
371
diesel::joinable!(post_embed_ext -> posts (post_uri));
372
diesel::joinable!(post_embed_images -> posts (post_uri));
373
diesel::joinable!(post_embed_record -> posts (post_uri));
···
397
list_blocks,
398
list_items,
399
lists,
400
+
notif_decl,
401
post_embed_ext,
402
post_embed_images,
403
post_embed_record,
+10
-6
parakeet/src/hydration/profile.rs
+10
-6
parakeet/src/hydration/profile.rs
···
17
chat: Option<ChatAllowIncoming>,
18
labeler: bool,
19
stats: Option<ProfileStats>,
20
) -> Option<ProfileAssociated> {
21
if chat.is_some() || labeler || stats.is_some() {
22
let stats = stats.unwrap_or_default();
···
27
starter_packs: stats.starterpacks as i64,
28
labeler,
29
chat: chat.map(|v| ProfileAssociatedChat { allow_incoming: v }),
30
})
31
} else {
32
None
···
148
}
149
150
fn build_basic(
151
-
(handle, profile, chat_decl, is_labeler, stats, status): ProfileLoaderRet,
152
labels: Vec<models::Label>,
153
verifications: Option<Vec<models::VerificationEntry>>,
154
cdn: &BskyCdn,
155
) -> ProfileViewBasic {
156
-
let associated = build_associated(chat_decl, is_labeler, stats);
157
let verification = build_verification(&profile, &handle, verifications);
158
let status = status.and_then(|status| build_status(status, cdn));
159
let avatar = profile.avatar_cid.map(|cid| cdn.avatar(&profile.did, &cid));
···
172
}
173
174
fn build_profile(
175
-
(handle, profile, chat_decl, is_labeler, stats, status): ProfileLoaderRet,
176
labels: Vec<models::Label>,
177
verifications: Option<Vec<models::VerificationEntry>>,
178
cdn: &BskyCdn,
179
) -> ProfileView {
180
-
let associated = build_associated(chat_decl, is_labeler, stats);
181
let verification = build_verification(&profile, &handle, verifications);
182
let status = status.and_then(|status| build_status(status, cdn));
183
let avatar = profile.avatar_cid.map(|cid| cdn.avatar(&profile.did, &cid));
···
198
}
199
200
fn build_detailed(
201
-
(handle, profile, chat_decl, is_labeler, stats, status): ProfileLoaderRet,
202
labels: Vec<models::Label>,
203
verifications: Option<Vec<models::VerificationEntry>>,
204
cdn: &BskyCdn,
205
) -> ProfileViewDetailed {
206
-
let associated = build_associated(chat_decl, is_labeler, stats);
207
let verification = build_verification(&profile, &handle, verifications);
208
let status = status.and_then(|status| build_status(status, cdn));
209
let avatar = profile.avatar_cid.map(|cid| cdn.avatar(&profile.did, &cid));
···
17
chat: Option<ChatAllowIncoming>,
18
labeler: bool,
19
stats: Option<ProfileStats>,
20
+
notif: Option<ProfileAllowSubscriptions>,
21
) -> Option<ProfileAssociated> {
22
if chat.is_some() || labeler || stats.is_some() {
23
let stats = stats.unwrap_or_default();
···
28
starter_packs: stats.starterpacks as i64,
29
labeler,
30
chat: chat.map(|v| ProfileAssociatedChat { allow_incoming: v }),
31
+
activity_subscription: notif.map(|v| ProfileAssociatedActivitySubscription {
32
+
allow_subscriptions: v,
33
+
}),
34
})
35
} else {
36
None
···
152
}
153
154
fn build_basic(
155
+
(handle, profile, chat_decl, is_labeler, stats, status, notif_decl): ProfileLoaderRet,
156
labels: Vec<models::Label>,
157
verifications: Option<Vec<models::VerificationEntry>>,
158
cdn: &BskyCdn,
159
) -> ProfileViewBasic {
160
+
let associated = build_associated(chat_decl, is_labeler, stats, notif_decl);
161
let verification = build_verification(&profile, &handle, verifications);
162
let status = status.and_then(|status| build_status(status, cdn));
163
let avatar = profile.avatar_cid.map(|cid| cdn.avatar(&profile.did, &cid));
···
176
}
177
178
fn build_profile(
179
+
(handle, profile, chat_decl, is_labeler, stats, status, notif_decl): ProfileLoaderRet,
180
labels: Vec<models::Label>,
181
verifications: Option<Vec<models::VerificationEntry>>,
182
cdn: &BskyCdn,
183
) -> ProfileView {
184
+
let associated = build_associated(chat_decl, is_labeler, stats, notif_decl);
185
let verification = build_verification(&profile, &handle, verifications);
186
let status = status.and_then(|status| build_status(status, cdn));
187
let avatar = profile.avatar_cid.map(|cid| cdn.avatar(&profile.did, &cid));
···
202
}
203
204
fn build_detailed(
205
+
(handle, profile, chat_decl, is_labeler, stats, status, notif_decl): ProfileLoaderRet,
206
labels: Vec<models::Label>,
207
verifications: Option<Vec<models::VerificationEntry>>,
208
cdn: &BskyCdn,
209
) -> ProfileViewDetailed {
210
+
let associated = build_associated(chat_decl, is_labeler, stats, notif_decl);
211
let verification = build_verification(&profile, &handle, verifications);
212
let status = status.and_then(|status| build_status(status, cdn));
213
let avatar = profile.avatar_cid.map(|cid| cdn.avatar(&profile.did, &cid));
+19
-3
parakeet/src/loaders.rs
+19
-3
parakeet/src/loaders.rs
···
5
use diesel_async::pooled_connection::deadpool::Pool;
6
use diesel_async::{AsyncPgConnection, RunQueryDsl};
7
use itertools::Itertools;
8
-
use lexica::app_bsky::actor::ChatAllowIncoming;
9
use parakeet_db::{models, schema};
10
use std::collections::HashMap;
11
use std::str::FromStr;
···
74
bool,
75
Option<parakeet_index::ProfileStats>,
76
Option<models::Status>,
77
);
78
impl BatchFn<String, ProfileLoaderRet> for ProfileLoader {
79
async fn load(&mut self, keys: &[String]) -> HashMap<String, ProfileLoaderRet> {
···
86
)
87
.left_join(schema::labelers::table.on(schema::labelers::did.eq(schema::actors::did)))
88
.left_join(schema::statuses::table.on(schema::statuses::did.eq(schema::actors::did)))
89
.select((
90
schema::actors::did,
91
schema::actors::handle,
···
93
schema::chat_decls::allow_incoming.nullable(),
94
schema::labelers::cid.nullable(),
95
Option::<models::Status>::as_select(),
96
))
97
.filter(
98
schema::actors::did
···
106
Option<String>,
107
Option<String>,
108
Option<models::Status>,
109
)>(&mut conn)
110
.await;
111
···
122
123
match res {
124
Ok(res) => HashMap::from_iter(res.into_iter().map(
125
-
|(did, handle, profile, chat_decl, labeler_cid, status)| {
126
let chat_decl = chat_decl.and_then(|v| ChatAllowIncoming::from_str(&v).ok());
127
let is_labeler = labeler_cid.is_some();
128
let maybe_stats = stats.remove(&did);
129
130
-
let val = (handle, profile, chat_decl, is_labeler, maybe_stats, status);
131
132
(did, val)
133
},
···
5
use diesel_async::pooled_connection::deadpool::Pool;
6
use diesel_async::{AsyncPgConnection, RunQueryDsl};
7
use itertools::Itertools;
8
+
use lexica::app_bsky::actor::{ChatAllowIncoming, ProfileAllowSubscriptions};
9
use parakeet_db::{models, schema};
10
use std::collections::HashMap;
11
use std::str::FromStr;
···
74
bool,
75
Option<parakeet_index::ProfileStats>,
76
Option<models::Status>,
77
+
Option<ProfileAllowSubscriptions>,
78
);
79
impl BatchFn<String, ProfileLoaderRet> for ProfileLoader {
80
async fn load(&mut self, keys: &[String]) -> HashMap<String, ProfileLoaderRet> {
···
87
)
88
.left_join(schema::labelers::table.on(schema::labelers::did.eq(schema::actors::did)))
89
.left_join(schema::statuses::table.on(schema::statuses::did.eq(schema::actors::did)))
90
+
.left_join(
91
+
schema::notif_decl::table.on(schema::notif_decl::did.eq(schema::actors::did)),
92
+
)
93
.select((
94
schema::actors::did,
95
schema::actors::handle,
···
97
schema::chat_decls::allow_incoming.nullable(),
98
schema::labelers::cid.nullable(),
99
Option::<models::Status>::as_select(),
100
+
schema::notif_decl::allow_subscriptions.nullable(),
101
))
102
.filter(
103
schema::actors::did
···
111
Option<String>,
112
Option<String>,
113
Option<models::Status>,
114
+
Option<String>,
115
)>(&mut conn)
116
.await;
117
···
128
129
match res {
130
Ok(res) => HashMap::from_iter(res.into_iter().map(
131
+
|(did, handle, profile, chat_decl, labeler_cid, status, notif_decl)| {
132
let chat_decl = chat_decl.and_then(|v| ChatAllowIncoming::from_str(&v).ok());
133
+
let notif_decl =
134
+
notif_decl.and_then(|v| ProfileAllowSubscriptions::from_str(&v).ok());
135
let is_labeler = labeler_cid.is_some();
136
let maybe_stats = stats.remove(&did);
137
138
+
let val = (
139
+
handle,
140
+
profile,
141
+
chat_decl,
142
+
is_labeler,
143
+
maybe_stats,
144
+
status,
145
+
notif_decl,
146
+
);
147
148
(did, val)
149
},