feat: community moderation, invitations, notifications, and cursor pagination · colibri.social/appview@83a005a

+1

Cargo.lock

··· 821 821 "migration", 822 822 "p256", 823 823 "pretty_env_logger", 824 + "rand 0.8.5", 824 825 "regex", 825 826 "reqwest", 826 827 "rocket",

+1

Cargo.toml

··· 34 34 async-stream = "0.3.6" 35 35 rocket_cors = "0.6.0" 36 36 regex = "1.12.3" 37 + rand = "0.8"

+4

migration/src/lib.rs

··· 2 2 3 3 mod m20260426_145024_create_user_states; 4 4 mod m20260504_071943_complete_user_stats; 5 + mod m20260513_120000_create_community_invitations; 6 + mod m20260514_090000_create_notifications; 5 7 6 8 pub struct Migrator; 7 9 ··· 11 13 vec![ 12 14 Box::new(m20260426_145024_create_user_states::Migration), 13 15 Box::new(m20260504_071943_complete_user_stats::Migration), 16 + Box::new(m20260513_120000_create_community_invitations::Migration), 17 + Box::new(m20260514_090000_create_notifications::Migration), 14 18 ] 15 19 } 16 20 }

+39

migration/src/m20260513_120000_create_community_invitations.rs

··· 1 + use sea_orm_migration::{prelude::*, schema::*}; 2 + 3 + #[derive(DeriveMigrationName)] 4 + pub struct Migration; 5 + 6 + #[async_trait::async_trait] 7 + impl MigrationTrait for Migration { 8 + async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { 9 + manager 10 + .create_table( 11 + Table::create() 12 + .table("community_invitations") 13 + .if_not_exists() 14 + .col(string("code").primary_key()) 15 + .col(string("community_uri")) 16 + .col(string("created_by")) 17 + .col(boolean("active").default(true)) 18 + .col(string("created_at")) 19 + .to_owned(), 20 + ) 21 + .await?; 22 + 23 + manager 24 + .create_index( 25 + Index::create() 26 + .name("idx_community_invitations_community_uri") 27 + .table("community_invitations") 28 + .col("community_uri") 29 + .to_owned(), 30 + ) 31 + .await 32 + } 33 + 34 + async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> { 35 + manager 36 + .drop_table(Table::drop().table("community_invitations").to_owned()) 37 + .await 38 + } 39 + }

+56

migration/src/m20260514_090000_create_notifications.rs

··· 1 + use sea_orm_migration::{prelude::*, schema::*}; 2 + 3 + #[derive(DeriveMigrationName)] 4 + pub struct Migration; 5 + 6 + #[async_trait::async_trait] 7 + impl MigrationTrait for Migration { 8 + async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { 9 + manager 10 + .create_table( 11 + Table::create() 12 + .table("notifications") 13 + .if_not_exists() 14 + .col(big_integer("id").primary_key().auto_increment()) 15 + .col(string("recipient_did")) 16 + .col(string("kind")) 17 + .col(string("message_uri")) 18 + .col(string("author_did")) 19 + .col(string("channel_rkey")) 20 + .col(string("indexed_at")) 21 + .col(string_null("seen_at")) 22 + .to_owned(), 23 + ) 24 + .await?; 25 + 26 + manager 27 + .create_index( 28 + Index::create() 29 + .name("idx_notifications_recipient_did_id") 30 + .table("notifications") 31 + .col("recipient_did") 32 + .col("id") 33 + .to_owned(), 34 + ) 35 + .await?; 36 + 37 + manager 38 + .create_index( 39 + Index::create() 40 + .name("idx_notifications_dedupe") 41 + .table("notifications") 42 + .col("recipient_did") 43 + .col("message_uri") 44 + .col("kind") 45 + .unique() 46 + .to_owned(), 47 + ) 48 + .await 49 + } 50 + 51 + async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> { 52 + manager 53 + .drop_table(Table::drop().table("notifications").to_owned()) 54 + .await 55 + } 56 + }

+65

src/lib/at_uri.rs

··· 1 + /// A parsed AT URI of the form `at://<authority>/<collection>/<rkey>`. 2 + #[derive(Debug, Clone, PartialEq, Eq)] 3 + pub struct AtUri { 4 + pub authority: String, 5 + pub collection: String, 6 + pub rkey: String, 7 + } 8 + 9 + impl AtUri { 10 + /// Parses a fully qualified AT URI. Returns `None` if the URI does not have 11 + /// the `at://<authority>/<collection>/<rkey>` shape. 12 + pub fn parse(uri: &str) -> Option<Self> { 13 + let rest = uri.strip_prefix("at://")?; 14 + let mut parts = rest.splitn(3, '/'); 15 + let authority = parts.next()?; 16 + let collection = parts.next()?; 17 + let rkey = parts.next()?; 18 + 19 + if authority.is_empty() || collection.is_empty() || rkey.is_empty() { 20 + return None; 21 + } 22 + 23 + Some(AtUri { 24 + authority: authority.to_string(), 25 + collection: collection.to_string(), 26 + rkey: rkey.to_string(), 27 + }) 28 + } 29 + } 30 + 31 + #[cfg(test)] 32 + mod tests { 33 + use super::*; 34 + 35 + #[test] 36 + fn parses_well_formed_uri() { 37 + let uri = AtUri::parse("at://did:plc:abc/social.colibri.community/r1").unwrap(); 38 + assert_eq!(uri.authority, "did:plc:abc"); 39 + assert_eq!(uri.collection, "social.colibri.community"); 40 + assert_eq!(uri.rkey, "r1"); 41 + } 42 + 43 + #[test] 44 + fn rejects_missing_scheme() { 45 + assert!(AtUri::parse("did:plc:abc/social.colibri.community/r1").is_none()); 46 + } 47 + 48 + #[test] 49 + fn rejects_missing_rkey() { 50 + assert!(AtUri::parse("at://did:plc:abc/social.colibri.community").is_none()); 51 + } 52 + 53 + #[test] 54 + fn rejects_empty_segments() { 55 + assert!(AtUri::parse("at:///social.colibri.community/r1").is_none()); 56 + assert!(AtUri::parse("at://did:plc:abc//r1").is_none()); 57 + assert!(AtUri::parse("at://did:plc:abc/social.colibri.community/").is_none()); 58 + } 59 + 60 + #[test] 61 + fn keeps_extra_path_segments_with_rkey() { 62 + let uri = AtUri::parse("at://did:plc:abc/social.colibri.community/r1/extra").unwrap(); 63 + assert_eq!(uri.rkey, "r1/extra"); 64 + } 65 + }

+104

src/lib/colibri.rs

··· 184 184 /// Format: datetime 185 185 pub created_at: String, 186 186 } 187 + 188 + #[derive(Serialize, Deserialize, Debug, Clone)] 189 + #[serde(rename_all = "camelCase")] 190 + pub struct ColibriRoleChannelOverride { 191 + /// The channel rkey this override applies to. 192 + pub channel: String, 193 + /// Permissions explicitly granted in this channel. 194 + #[serde(default)] 195 + pub allow: Vec<String>, 196 + /// Permissions explicitly denied in this channel. 197 + #[serde(default)] 198 + pub deny: Vec<String>, 199 + } 200 + 201 + #[derive(Serialize, Deserialize, Debug, Clone)] 202 + #[serde(rename_all = "camelCase")] 203 + pub struct ColibriRole { 204 + #[serde(rename = "$type", skip_serializing_if = "Option::is_none")] 205 + pub record_type: Option<String>, 206 + 207 + /// Display name of the role. 1-32 characters. 208 + pub name: String, 209 + 210 + /// Optional hex color (`#rrggbb`). 211 + #[serde(skip_serializing_if = "Option::is_none")] 212 + pub color: Option<String>, 213 + 214 + /// Granted permissions. See `permissions::Permission` for the enumerated set. 215 + #[serde(default)] 216 + pub permissions: Vec<String>, 217 + 218 + /// Hierarchy position. Higher values sit higher in the hierarchy. 219 + pub position: i64, 220 + 221 + /// Whether members of this role are displayed separately in the member list. 222 + #[serde(default, skip_serializing_if = "Option::is_none")] 223 + pub hoisted: Option<bool>, 224 + 225 + /// Whether `@role` style mentions resolve to this role. 226 + #[serde(default, skip_serializing_if = "Option::is_none")] 227 + pub mentionable: Option<bool>, 228 + 229 + /// Per-channel permission overrides for this role. 230 + #[serde(default, skip_serializing_if = "Vec::is_empty")] 231 + pub channel_overrides: Vec<ColibriRoleChannelOverride>, 232 + } 233 + 234 + #[derive(Serialize, Deserialize, Debug, Clone)] 235 + #[serde(rename_all = "camelCase")] 236 + pub struct ColibriMember { 237 + #[serde(rename = "$type", skip_serializing_if = "Option::is_none")] 238 + pub record_type: Option<String>, 239 + 240 + /// The member's DID. 241 + pub subject: String, 242 + 243 + /// Role rkeys assigned to this member, stored on the same community repo. 244 + #[serde(default)] 245 + pub roles: Vec<String>, 246 + 247 + /// Format: datetime 248 + pub joined_at: String, 249 + 250 + /// Optional per-community display-name override. 251 + #[serde(skip_serializing_if = "Option::is_none")] 252 + pub nickname: Option<String>, 253 + 254 + /// Optional AT-URI of the user-side `social.colibri.membership` declaration 255 + /// that this member record was admitted from. Audit trail. 256 + #[serde(skip_serializing_if = "Option::is_none")] 257 + pub from_membership: Option<String>, 258 + } 259 + 260 + #[derive(Serialize, Deserialize, Debug, Clone)] 261 + #[serde(rename_all = "camelCase")] 262 + pub struct ColibriModerationSubject { 263 + /// DID of the subject for user-targeted actions. 264 + #[serde(skip_serializing_if = "Option::is_none")] 265 + pub did: Option<String>, 266 + /// AT-URI of the subject for content-targeted actions. 267 + #[serde(skip_serializing_if = "Option::is_none")] 268 + pub uri: Option<String>, 269 + } 270 + 271 + #[derive(Serialize, Deserialize, Debug, Clone)] 272 + #[serde(rename_all = "camelCase")] 273 + pub struct ColibriModeration { 274 + #[serde(rename = "$type", skip_serializing_if = "Option::is_none")] 275 + pub record_type: Option<String>, 276 + 277 + /// `ban` | `unban` | `hideMessage` | `unhideMessage` | `kick` 278 + pub action: String, 279 + 280 + pub subject: ColibriModerationSubject, 281 + 282 + #[serde(skip_serializing_if = "Option::is_none")] 283 + pub reason: Option<String>, 284 + 285 + /// DID of the issuer. 286 + pub created_by: String, 287 + 288 + /// Format: datetime 289 + pub created_at: String, 290 + }

+299

src/lib/community_authz.rs

··· 1 + use sea_orm::{ColumnTrait, Condition, DatabaseConnection, DbErr, EntityTrait, QueryFilter}; 2 + 3 + use crate::lib::at_uri::AtUri; 4 + use crate::lib::colibri::{ColibriMember, ColibriRole}; 5 + use crate::lib::permissions::Permission; 6 + use crate::models::record_data; 7 + 8 + const MEMBER_NSID: &str = "social.colibri.member"; 9 + const ROLE_NSID: &str = "social.colibri.role"; 10 + 11 + /// Aggregated authz state for a single (actor, community) pair. 12 + #[derive(Debug, Clone)] 13 + pub struct ActorAuthz { 14 + pub is_owner: bool, 15 + pub member: Option<ColibriMember>, 16 + pub roles: Vec<ColibriRole>, 17 + } 18 + 19 + impl ActorAuthz { 20 + /// Highest role position held by the actor, or `None` if the actor has no 21 + /// roles. The owner is considered to outrank every role and is represented 22 + /// as `Some(i64::MAX)`. 23 + pub fn highest_position(&self) -> Option<i64> { 24 + if self.is_owner { 25 + return Some(i64::MAX); 26 + } 27 + self.roles.iter().map(|r| r.position).max() 28 + } 29 + 30 + /// Whether the actor holds the given permission, optionally in the context 31 + /// of a specific channel (so per-channel overrides apply). 32 + /// 33 + /// Evaluation order, matching Discord: 34 + /// 1. Owner short-circuits to allow. 35 + /// 2. Channel-override `deny` wins over any allow. 36 + /// 3. Channel-override `allow` grants without needing a base permission. 37 + /// 4. Otherwise the base `permissions` list of any role is consulted. 38 + pub fn has(&self, permission: Permission, channel_rkey: Option<&str>) -> bool { 39 + if self.is_owner { 40 + return true; 41 + } 42 + 43 + let needle = permission.as_str(); 44 + let mut has_base = false; 45 + let mut has_override_allow = false; 46 + let mut has_override_deny = false; 47 + 48 + for role in &self.roles { 49 + if role.permissions.iter().any(|p| p == needle) { 50 + has_base = true; 51 + } 52 + if let Some(channel) = channel_rkey { 53 + for override_entry in role.channel_overrides.iter() { 54 + if override_entry.channel != channel { 55 + continue; 56 + } 57 + if override_entry.deny.iter().any(|p| p == needle) { 58 + has_override_deny = true; 59 + } 60 + if override_entry.allow.iter().any(|p| p == needle) { 61 + has_override_allow = true; 62 + } 63 + } 64 + } 65 + } 66 + 67 + if has_override_deny { 68 + return false; 69 + } 70 + has_override_allow || has_base 71 + } 72 + 73 + /// Hierarchy guard for actions targeting another user. The acting party 74 + /// must outrank the target (strictly greater highest-role position). The 75 + /// owner outranks everyone. 76 + pub fn outranks(&self, target: &ActorAuthz) -> bool { 77 + if self.is_owner { 78 + return true; 79 + } 80 + if target.is_owner { 81 + return false; 82 + } 83 + match (self.highest_position(), target.highest_position()) { 84 + (Some(a), Some(b)) => a > b, 85 + (Some(_), None) => true, 86 + (None, _) => false, 87 + } 88 + } 89 + } 90 + 91 + /// Loads the authz state for `actor_did` in the community identified by 92 + /// `community_uri`. Returns the assembled state, doing per-table queries. 93 + pub async fn load_actor_authz( 94 + db: &DatabaseConnection, 95 + community_uri: &str, 96 + actor_did: &str, 97 + ) -> Result<ActorAuthz, DbErr> { 98 + let community = AtUri::parse(community_uri).ok_or_else(|| { 99 + DbErr::Custom(format!("invalid community AT-URI: {community_uri}")) 100 + })?; 101 + 102 + let is_owner = community.authority == actor_did; 103 + 104 + let member_record = record_data::Entity::find() 105 + .filter(record_data::Column::Did.eq(&community.authority)) 106 + .filter(record_data::Column::Nsid.eq(MEMBER_NSID)) 107 + .filter(sea_orm::prelude::Expr::cust_with_values( 108 + r#""record_data"."data"->>'subject' = $1"#, 109 + vec![sea_orm::Value::from(actor_did.to_string())], 110 + )) 111 + .one(db) 112 + .await?; 113 + 114 + let member = member_record 115 + .and_then(|m| serde_json::from_value::<ColibriMember>(m.data).ok()); 116 + 117 + let role_rkeys: Vec<String> = member.as_ref().map(|m| m.roles.clone()).unwrap_or_default(); 118 + 119 + let roles = if role_rkeys.is_empty() { 120 + Vec::new() 121 + } else { 122 + let role_records = record_data::Entity::find() 123 + .filter( 124 + Condition::all() 125 + .add(record_data::Column::Did.eq(&community.authority)) 126 + .add(record_data::Column::Nsid.eq(ROLE_NSID)) 127 + .add(record_data::Column::Rkey.is_in(role_rkeys.clone())), 128 + ) 129 + .all(db) 130 + .await?; 131 + role_records 132 + .into_iter() 133 + .filter_map(|r| serde_json::from_value::<ColibriRole>(r.data).ok()) 134 + .collect() 135 + }; 136 + 137 + Ok(ActorAuthz { 138 + is_owner, 139 + member, 140 + roles, 141 + }) 142 + } 143 + 144 + #[cfg(test)] 145 + mod tests { 146 + use super::*; 147 + use crate::lib::colibri::{ColibriMember, ColibriRole, ColibriRoleChannelOverride}; 148 + 149 + fn role(name: &str, position: i64, permissions: Vec<Permission>) -> ColibriRole { 150 + ColibriRole { 151 + record_type: None, 152 + name: name.to_string(), 153 + color: None, 154 + permissions: permissions.into_iter().map(|p| p.as_str().to_string()).collect(), 155 + position, 156 + hoisted: None, 157 + mentionable: None, 158 + channel_overrides: vec![], 159 + } 160 + } 161 + 162 + fn role_with_override( 163 + name: &str, 164 + position: i64, 165 + permissions: Vec<Permission>, 166 + channel: &str, 167 + allow: Vec<Permission>, 168 + deny: Vec<Permission>, 169 + ) -> ColibriRole { 170 + let mut r = role(name, position, permissions); 171 + r.channel_overrides.push(ColibriRoleChannelOverride { 172 + channel: channel.to_string(), 173 + allow: allow.into_iter().map(|p| p.as_str().to_string()).collect(), 174 + deny: deny.into_iter().map(|p| p.as_str().to_string()).collect(), 175 + }); 176 + r 177 + } 178 + 179 + fn member(subject: &str, roles: Vec<&str>) -> ColibriMember { 180 + ColibriMember { 181 + record_type: None, 182 + subject: subject.to_string(), 183 + roles: roles.into_iter().map(String::from).collect(), 184 + joined_at: String::from("2026-05-13T00:00:00Z"), 185 + nickname: None, 186 + from_membership: None, 187 + } 188 + } 189 + 190 + #[test] 191 + fn owner_has_every_permission() { 192 + let authz = ActorAuthz { 193 + is_owner: true, 194 + member: None, 195 + roles: vec![], 196 + }; 197 + assert!(authz.has(Permission::MemberBan, None)); 198 + assert!(authz.has(Permission::MessageDelete, Some("chan-a"))); 199 + assert_eq!(authz.highest_position(), Some(i64::MAX)); 200 + } 201 + 202 + #[test] 203 + fn non_member_has_no_permissions() { 204 + let authz = ActorAuthz { 205 + is_owner: false, 206 + member: None, 207 + roles: vec![], 208 + }; 209 + assert!(!authz.has(Permission::MemberBan, None)); 210 + assert!(authz.highest_position().is_none()); 211 + } 212 + 213 + #[test] 214 + fn role_permission_grants_action() { 215 + let authz = ActorAuthz { 216 + is_owner: false, 217 + member: Some(member("did:plc:alice", vec!["mod"])), 218 + roles: vec![role("Moderator", 10, vec![Permission::MemberBan])], 219 + }; 220 + assert!(authz.has(Permission::MemberBan, None)); 221 + assert!(!authz.has(Permission::CommunityDelete, None)); 222 + } 223 + 224 + #[test] 225 + fn channel_override_deny_beats_base_permission() { 226 + let authz = ActorAuthz { 227 + is_owner: false, 228 + member: Some(member("did:plc:alice", vec!["mod"])), 229 + roles: vec![role_with_override( 230 + "Moderator", 231 + 10, 232 + vec![Permission::MessageDelete], 233 + "chan-a", 234 + vec![], 235 + vec![Permission::MessageDelete], 236 + )], 237 + }; 238 + assert!(authz.has(Permission::MessageDelete, None)); 239 + assert!(authz.has(Permission::MessageDelete, Some("chan-b"))); 240 + assert!(!authz.has(Permission::MessageDelete, Some("chan-a"))); 241 + } 242 + 243 + #[test] 244 + fn channel_override_allow_grants_without_base() { 245 + let authz = ActorAuthz { 246 + is_owner: false, 247 + member: Some(member("did:plc:alice", vec!["mod"])), 248 + roles: vec![role_with_override( 249 + "Helper", 250 + 5, 251 + vec![], 252 + "chan-a", 253 + vec![Permission::MessageDelete], 254 + vec![], 255 + )], 256 + }; 257 + assert!(authz.has(Permission::MessageDelete, Some("chan-a"))); 258 + assert!(!authz.has(Permission::MessageDelete, Some("chan-b"))); 259 + assert!(!authz.has(Permission::MessageDelete, None)); 260 + } 261 + 262 + #[test] 263 + fn hierarchy_owner_outranks_everyone() { 264 + let owner = ActorAuthz { 265 + is_owner: true, 266 + member: None, 267 + roles: vec![], 268 + }; 269 + let mod_user = ActorAuthz { 270 + is_owner: false, 271 + member: Some(member("did:plc:alice", vec!["mod"])), 272 + roles: vec![role("Moderator", 10, vec![])], 273 + }; 274 + assert!(owner.outranks(&mod_user)); 275 + assert!(!mod_user.outranks(&owner)); 276 + } 277 + 278 + #[test] 279 + fn hierarchy_requires_strictly_higher_position() { 280 + let high = ActorAuthz { 281 + is_owner: false, 282 + member: Some(member("did:plc:a", vec!["r1"])), 283 + roles: vec![role("High", 20, vec![])], 284 + }; 285 + let mid = ActorAuthz { 286 + is_owner: false, 287 + member: Some(member("did:plc:b", vec!["r2"])), 288 + roles: vec![role("Mid", 10, vec![])], 289 + }; 290 + let other_mid = ActorAuthz { 291 + is_owner: false, 292 + member: Some(member("did:plc:c", vec!["r3"])), 293 + roles: vec![role("Other Mid", 10, vec![])], 294 + }; 295 + assert!(high.outranks(&mid)); 296 + assert!(!mid.outranks(&other_mid)); 297 + assert!(!mid.outranks(&high)); 298 + } 299 + }

+31

src/lib/events.rs

··· 117 117 } 118 118 119 119 #[derive(Serialize, Deserialize, Debug)] 120 + pub struct NotificationEventMessage { 121 + pub text: String, 122 + #[serde(default)] 123 + pub facets: Vec<Value>, 124 + #[serde(rename = "createdAt")] 125 + pub created_at: String, 126 + #[serde(skip_serializing_if = "Option::is_none")] 127 + pub parent: Option<String>, 128 + #[serde(default)] 129 + pub attachments: Vec<Value>, 130 + #[serde(default, skip_serializing_if = "Option::is_none")] 131 + pub edited: Option<bool>, 132 + } 133 + 134 + #[derive(Serialize, Deserialize, Debug)] 135 + pub struct NotificationEventData { 136 + pub id: i64, 137 + pub kind: String, 138 + #[serde(rename = "messageUri")] 139 + pub message_uri: String, 140 + #[serde(rename = "authorDid")] 141 + pub author_did: String, 142 + #[serde(rename = "channelRkey")] 143 + pub channel_rkey: String, 144 + #[serde(rename = "indexedAt")] 145 + pub indexed_at: String, 146 + pub message: NotificationEventMessage, 147 + } 148 + 149 + #[derive(Serialize, Deserialize, Debug)] 120 150 #[serde(untagged)] 121 151 pub enum ColibriServerEventData { 122 152 Community(CommunityEventData), ··· 127 157 Reaction(ReactionEventData), 128 158 User(UserEventData), 129 159 Typing(TypingEventData), 160 + Notification(NotificationEventData), 130 161 } 131 162 132 163 #[derive(Serialize, Deserialize, Debug)]

+136 -62

src/lib/list_atproto_records.rs

··· 1 1 use sea_orm::{ 2 - ColumnTrait, Condition, DatabaseConnection, DbErr, EntityTrait, QueryFilter, QuerySelect, 2 + ColumnTrait, Condition, DatabaseConnection, DbErr, EntityTrait, QueryFilter, QueryOrder, 3 + QuerySelect, 3 4 }; 4 5 5 6 use crate::models::record_data::{self, Entity as RecordData, Model}; 6 7 8 + const DEFAULT_LIMIT: u64 = 100; 9 + 10 + pub struct PagedRecords<T> { 11 + pub records: Vec<T>, 12 + pub cursor: Option<String>, 13 + } 14 + 15 + /// Lists records for a given (did, nsid) with rkey-based cursor pagination. 16 + /// 17 + /// The cursor is the rkey of the last record returned. Default order is 18 + /// descending by rkey (newest TID-encoded records first); when `reverse` is 19 + /// true, ascending order is used. The returned cursor is only set when the 20 + /// page is full, matching `com.atproto.repo.listRecords` semantics. 7 21 pub async fn list_atproto_records<T: for<'de> serde::Deserialize<'de>>( 8 22 did: String, 9 23 nsid: String, 10 24 limit: Option<u64>, 11 - _cursor: Option<&str>, // TODO: Cursor support 25 + cursor: Option<&str>, 12 26 reverse: Option<bool>, 13 27 db: &DatabaseConnection, 14 - ) -> Result<Vec<T>, DbErr> { 15 - let default_limit = 100; 16 - let records = RecordData::find() 17 - .filter( 18 - Condition::all() 19 - .add(record_data::Column::Did.eq(&did)) 20 - .add(record_data::Column::Nsid.eq(&nsid)), 21 - ) 22 - .limit(limit.unwrap_or(default_limit).to_owned()); 28 + ) -> Result<PagedRecords<T>, DbErr> { 29 + let effective_limit = limit.unwrap_or(DEFAULT_LIMIT); 30 + let ascending = reverse.is_some_and(|v| v); 31 + 32 + let mut condition = Condition::all() 33 + .add(record_data::Column::Did.eq(&did)) 34 + .add(record_data::Column::Nsid.eq(&nsid)); 35 + 36 + if let Some(c) = cursor { 37 + condition = if ascending { 38 + condition.add(record_data::Column::Rkey.gt(c)) 39 + } else { 40 + condition.add(record_data::Column::Rkey.lt(c)) 41 + }; 42 + } 43 + 44 + let query = RecordData::find().filter(condition).limit(effective_limit); 45 + 46 + let result: Vec<Model> = if ascending { 47 + query 48 + .order_by_asc(record_data::Column::Rkey) 49 + .all(db) 50 + .await? 51 + } else { 52 + query 53 + .order_by_desc(record_data::Column::Rkey) 54 + .all(db) 55 + .await? 56 + }; 23 57 24 - let result: Vec<Model> = if reverse.is_some_and(|v| v) { 25 - records.order_by_id_asc().all(db).await? 58 + let next_cursor = if (result.len() as u64) == effective_limit { 59 + result.last().map(|r| r.rkey.clone()) 26 60 } else { 27 - records.order_by_id_desc().all(db).await? 61 + None 28 62 }; 29 63 30 - let mut json_records: Vec<T> = result 31 - .iter() 32 - .map(|r| serde_json::from_value::<T>(r.to_owned().data).unwrap()) 64 + let records: Vec<T> = result 65 + .into_iter() 66 + .map(|r| serde_json::from_value::<T>(r.data).unwrap()) 33 67 .collect(); 34 68 35 - if reverse.is_some_and(|v| v) { 36 - json_records.reverse(); 37 - } 38 - 39 - Ok(json_records) 69 + Ok(PagedRecords { 70 + records, 71 + cursor: next_cursor, 72 + }) 40 73 } 41 74 42 75 #[cfg(test)] ··· 49 82 #[derive(Deserialize)] 50 83 struct SampleRecord { 51 84 text: String, 85 + } 86 + 87 + fn model(id: i64, rkey: &str, text: &str) -> crate::models::record_data::Model { 88 + crate::models::record_data::Model { 89 + id, 90 + did: String::from("did:plc:abc"), 91 + nsid: String::from("social.colibri.message"), 92 + rkey: String::from(rkey), 93 + data: serde_json::json!({ "text": text }), 94 + } 52 95 } 53 96 54 97 #[tokio::test] 55 - async fn returns_records_in_default_order() { 98 + async fn returns_records_in_descending_rkey_order_by_default() { 56 99 let db = MockDatabase::new(DatabaseBackend::Postgres) 57 - .append_query_results([vec![ 58 - crate::models::record_data::Model { 59 - id: 2, 60 - did: String::from("did:plc:abc"), 61 - nsid: String::from("social.colibri.message"), 62 - rkey: String::from("r2"), 63 - data: serde_json::json!({ "text": "second" }), 64 - }, 65 - crate::models::record_data::Model { 66 - id: 1, 67 - did: String::from("did:plc:abc"), 68 - nsid: String::from("social.colibri.message"), 69 - rkey: String::from("r1"), 70 - data: serde_json::json!({ "text": "first" }), 71 - }, 72 - ]]) 100 + .append_query_results([vec![model(2, "r2", "second"), model(1, "r1", "first")]]) 73 101 .into_connection(); 74 102 75 - let records = list_atproto_records::<SampleRecord>( 103 + let result = list_atproto_records::<SampleRecord>( 76 104 String::from("did:plc:abc"), 77 105 String::from("social.colibri.message"), 78 106 Some(10), ··· 83 111 .await 84 112 .unwrap(); 85 113 86 - assert_eq!(records[0].text, "second"); 87 - assert_eq!(records[1].text, "first"); 114 + assert_eq!(result.records[0].text, "second"); 115 + assert_eq!(result.records[1].text, "first"); 88 116 } 89 117 90 118 #[tokio::test] 91 - async fn reverses_records_when_reverse_flag_is_true() { 119 + async fn returns_records_in_ascending_rkey_order_when_reverse_is_true() { 92 120 let db = MockDatabase::new(DatabaseBackend::Postgres) 93 - .append_query_results([vec![ 94 - crate::models::record_data::Model { 95 - id: 1, 96 - did: String::from("did:plc:abc"), 97 - nsid: String::from("social.colibri.message"), 98 - rkey: String::from("r1"), 99 - data: serde_json::json!({ "text": "first" }), 100 - }, 101 - crate::models::record_data::Model { 102 - id: 2, 103 - did: String::from("did:plc:abc"), 104 - nsid: String::from("social.colibri.message"), 105 - rkey: String::from("r2"), 106 - data: serde_json::json!({ "text": "second" }), 107 - }, 108 - ]]) 121 + .append_query_results([vec![model(1, "r1", "first"), model(2, "r2", "second")]]) 109 122 .into_connection(); 110 123 111 - let records = list_atproto_records::<SampleRecord>( 124 + let result = list_atproto_records::<SampleRecord>( 112 125 String::from("did:plc:abc"), 113 126 String::from("social.colibri.message"), 114 127 Some(10), ··· 119 132 .await 120 133 .unwrap(); 121 134 122 - assert_eq!(records[0].text, "second"); 123 - assert_eq!(records[1].text, "first"); 135 + assert_eq!(result.records[0].text, "first"); 136 + assert_eq!(result.records[1].text, "second"); 137 + } 138 + 139 + #[tokio::test] 140 + async fn returns_cursor_when_page_is_full() { 141 + let db = MockDatabase::new(DatabaseBackend::Postgres) 142 + .append_query_results([vec![model(2, "r2", "second"), model(1, "r1", "first")]]) 143 + .into_connection(); 144 + 145 + let result = list_atproto_records::<SampleRecord>( 146 + String::from("did:plc:abc"), 147 + String::from("social.colibri.message"), 148 + Some(2), 149 + None, 150 + None, 151 + &db, 152 + ) 153 + .await 154 + .unwrap(); 155 + 156 + assert_eq!(result.cursor.as_deref(), Some("r1")); 157 + } 158 + 159 + #[tokio::test] 160 + async fn omits_cursor_when_page_is_not_full() { 161 + let db = MockDatabase::new(DatabaseBackend::Postgres) 162 + .append_query_results([vec![model(1, "r1", "first")]]) 163 + .into_connection(); 164 + 165 + let result = list_atproto_records::<SampleRecord>( 166 + String::from("did:plc:abc"), 167 + String::from("social.colibri.message"), 168 + Some(10), 169 + None, 170 + None, 171 + &db, 172 + ) 173 + .await 174 + .unwrap(); 175 + 176 + assert!(result.cursor.is_none()); 177 + } 178 + 179 + #[tokio::test] 180 + async fn returns_no_cursor_when_no_records_match() { 181 + let db = MockDatabase::new(DatabaseBackend::Postgres) 182 + .append_query_results([Vec::<crate::models::record_data::Model>::new()]) 183 + .into_connection(); 184 + 185 + let result = list_atproto_records::<SampleRecord>( 186 + String::from("did:plc:abc"), 187 + String::from("social.colibri.message"), 188 + Some(10), 189 + Some("r9"), 190 + None, 191 + &db, 192 + ) 193 + .await 194 + .unwrap(); 195 + 196 + assert!(result.records.is_empty()); 197 + assert!(result.cursor.is_none()); 124 198 } 125 199 }

+7

src/lib/mod.rs

··· 1 + pub mod at_uri; 1 2 pub mod bsky; 2 3 pub mod colibri; 4 + pub mod community_authz; 3 5 pub mod db; 4 6 pub mod did_document; 5 7 pub mod events; ··· 7 9 pub mod get_state; 8 10 pub mod list_atproto_records; 9 11 pub mod map_tap_event; 12 + pub mod moderation; 13 + pub mod notifications; 14 + pub mod permissions; 15 + pub mod reactions; 10 16 pub mod responses; 11 17 pub mod sentry; 12 18 pub mod service_auth; 13 19 pub mod state; 14 20 pub mod tap; 21 + pub mod time; 15 22 pub mod validate_state;

+199

src/lib/moderation.rs

··· 1 + use std::collections::HashSet; 2 + 3 + use sea_orm::{ColumnTrait, DatabaseConnection, DbErr, EntityTrait, QueryFilter, QueryOrder}; 4 + 5 + use crate::lib::at_uri::AtUri; 6 + use crate::lib::colibri::{ColibriModeration, ColibriModerationSubject}; 7 + use crate::models::record_data; 8 + 9 + pub const MODERATION_NSID: &str = "social.colibri.moderation"; 10 + 11 + pub const ACTION_BAN: &str = "ban"; 12 + pub const ACTION_UNBAN: &str = "unban"; 13 + pub const ACTION_HIDE_MESSAGE: &str = "hideMessage"; 14 + pub const ACTION_UNHIDE_MESSAGE: &str = "unhideMessage"; 15 + pub const ACTION_KICK: &str = "kick"; 16 + 17 + /// Generates a TID-like rkey (13 base32-sortable chars) based on the current 18 + /// system time. Sufficient for cursor ordering; the AppView is the only writer. 19 + pub fn generate_tid() -> String { 20 + use std::time::{SystemTime, UNIX_EPOCH}; 21 + 22 + const ALPHABET: &[u8] = b"234567abcdefghijklmnopqrstuvwxyz"; 23 + let micros = SystemTime::now() 24 + .duration_since(UNIX_EPOCH) 25 + .unwrap_or_default() 26 + .as_micros() as u64; 27 + // Pack microseconds into 13 base32 chars (65 bits, we drop the high bit). 28 + let mut out = [0u8; 13]; 29 + let mut value = micros & ((1u64 << 53) - 1); 30 + for slot in out.iter_mut().rev() { 31 + *slot = ALPHABET[(value & 31) as usize]; 32 + value >>= 5; 33 + } 34 + String::from_utf8(out.to_vec()).expect("ASCII characters") 35 + } 36 + 37 + /// Writes a `social.colibri.moderation` record onto the community repo. 38 + pub async fn write_moderation_record( 39 + db: &DatabaseConnection, 40 + community: &AtUri, 41 + record: &ColibriModeration, 42 + ) -> Result<record_data::Model, DbErr> { 43 + let rkey = generate_tid(); 44 + let data = serde_json::to_value(record).map_err(|e| DbErr::Custom(e.to_string()))?; 45 + 46 + let active = record_data::ActiveModel { 47 + did: sea_orm::ActiveValue::Set(community.authority.clone()), 48 + nsid: sea_orm::ActiveValue::Set(MODERATION_NSID.to_string()), 49 + rkey: sea_orm::ActiveValue::Set(rkey.clone()), 50 + data: sea_orm::ActiveValue::Set(data), 51 + ..Default::default() 52 + }; 53 + 54 + record_data::Entity::insert(active).exec(db).await?; 55 + 56 + let written = record_data::Entity::find() 57 + .filter(record_data::Column::Did.eq(&community.authority)) 58 + .filter(record_data::Column::Nsid.eq(MODERATION_NSID)) 59 + .filter(record_data::Column::Rkey.eq(&rkey)) 60 + .one(db) 61 + .await? 62 + .ok_or_else(|| DbErr::Custom(format!("inserted moderation record missing: {rkey}")))?; 63 + 64 + Ok(written) 65 + } 66 + 67 + /// Computes the set of DIDs currently banned in a community by replaying every 68 + /// `ban`/`unban` moderation event in createdAt order. 69 + pub async fn currently_banned_dids( 70 + db: &DatabaseConnection, 71 + community: &AtUri, 72 + ) -> Result<Vec<String>, DbErr> { 73 + let records = record_data::Entity::find() 74 + .filter(record_data::Column::Did.eq(&community.authority)) 75 + .filter(record_data::Column::Nsid.eq(MODERATION_NSID)) 76 + .order_by_asc(record_data::Column::Rkey) 77 + .all(db) 78 + .await?; 79 + 80 + let mut banned: HashSet<String> = HashSet::new(); 81 + for record in records { 82 + let Ok(mod_record) = serde_json::from_value::<ColibriModeration>(record.data) else { 83 + continue; 84 + }; 85 + let Some(did) = mod_record.subject.did else { 86 + continue; 87 + }; 88 + match mod_record.action.as_str() { 89 + ACTION_BAN => { 90 + banned.insert(did); 91 + } 92 + ACTION_UNBAN => { 93 + banned.remove(&did); 94 + } 95 + _ => {} 96 + } 97 + } 98 + let mut result: Vec<String> = banned.into_iter().collect(); 99 + result.sort(); 100 + Ok(result) 101 + } 102 + 103 + /// Returns true if the given user is currently banned in the community. 104 + pub async fn is_user_banned( 105 + db: &DatabaseConnection, 106 + community: &AtUri, 107 + did: &str, 108 + ) -> Result<bool, DbErr> { 109 + let banned = currently_banned_dids(db, community).await?; 110 + Ok(banned.iter().any(|b| b == did)) 111 + } 112 + 113 + /// Convenience constructor for building a moderation record payload. 114 + pub fn moderation_record( 115 + action: &str, 116 + subject: ColibriModerationSubject, 117 + created_by: String, 118 + created_at: String, 119 + reason: Option<String>, 120 + ) -> ColibriModeration { 121 + ColibriModeration { 122 + record_type: Some(MODERATION_NSID.to_string()), 123 + action: action.to_string(), 124 + subject, 125 + reason, 126 + created_by, 127 + created_at, 128 + } 129 + } 130 + 131 + /// Computes the latest moderation action targeting a given (subject) DID. Used 132 + /// for tests and finer-grained checks beyond the boolean `is_user_banned`. 133 + #[cfg(test)] 134 + pub fn latest_action_for_did(records: &[ColibriModeration], did: &str) -> Option<String> { 135 + records 136 + .iter() 137 + .filter(|r| { 138 + r.subject 139 + .did 140 + .as_ref() 141 + .map(|d| d == did) 142 + .unwrap_or(false) 143 + }) 144 + .map(|r| r.action.clone()) 145 + .next_back() 146 + } 147 + 148 + #[cfg(test)] 149 + mod tests { 150 + use super::*; 151 + 152 + fn record(action: &str, did: &str) -> ColibriModeration { 153 + ColibriModeration { 154 + record_type: Some(MODERATION_NSID.to_string()), 155 + action: action.to_string(), 156 + subject: ColibriModerationSubject { 157 + did: Some(did.to_string()), 158 + uri: None, 159 + }, 160 + reason: None, 161 + created_by: String::from("did:plc:owner"), 162 + created_at: String::from("2026-05-13T00:00:00Z"), 163 + } 164 + } 165 + 166 + #[test] 167 + fn ban_then_unban_results_in_not_banned() { 168 + let log = vec![record("ban", "did:plc:alice"), record("unban", "did:plc:alice")]; 169 + assert_eq!( 170 + latest_action_for_did(&log, "did:plc:alice"), 171 + Some(String::from("unban")) 172 + ); 173 + } 174 + 175 + #[test] 176 + fn moderation_record_helper_sets_type() { 177 + let rec = moderation_record( 178 + ACTION_BAN, 179 + ColibriModerationSubject { 180 + did: Some(String::from("did:plc:alice")), 181 + uri: None, 182 + }, 183 + String::from("did:plc:owner"), 184 + String::from("2026-05-13T00:00:00Z"), 185 + Some(String::from("spam")), 186 + ); 187 + assert_eq!(rec.action, "ban"); 188 + assert_eq!(rec.created_by, "did:plc:owner"); 189 + assert_eq!(rec.record_type.as_deref(), Some(MODERATION_NSID)); 190 + assert_eq!(rec.reason.as_deref(), Some("spam")); 191 + } 192 + 193 + #[test] 194 + fn generate_tid_yields_13_char_lowercase_ascii() { 195 + let t = generate_tid(); 196 + assert_eq!(t.len(), 13); 197 + assert!(t.chars().all(|c| c.is_ascii_alphanumeric())); 198 + } 199 + }

+420

src/lib/notifications.rs

··· 1 + use std::collections::{HashMap, HashSet}; 2 + 3 + use sea_orm::{ 4 + ActiveValue, ColumnTrait, Condition, DatabaseConnection, DbErr, EntityTrait, PaginatorTrait, 5 + QueryFilter, QueryOrder, QuerySelect, sea_query, 6 + }; 7 + use serde::{Deserialize, Serialize}; 8 + use serde_json::Value; 9 + 10 + use crate::lib::at_uri::AtUri; 11 + use crate::lib::colibri::ColibriMessage; 12 + use crate::lib::time::current_iso8601_utc; 13 + use crate::models::{notifications, record_data}; 14 + 15 + pub const KIND_MENTION: &str = "mention"; 16 + pub const KIND_REPLY: &str = "reply"; 17 + pub const FACET_MENTION_TYPE: &str = "social.colibri.richtext.facet#mention"; 18 + 19 + /// Bundle carried over the WS-broadcast channel so each subscriber can render 20 + /// a notification without having to fetch the message body separately. 21 + #[derive(Debug, Clone)] 22 + pub struct IndexedNotification { 23 + pub row: notifications::Model, 24 + pub message: NotificationMessage, 25 + } 26 + 27 + /// Inline message body rendered alongside a notification so clients can show 28 + /// the contents without a follow-up fetch. Mirrors the subset of `ColibriMessage` 29 + /// useful for notification previews. 30 + #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)] 31 + pub struct NotificationMessage { 32 + pub text: String, 33 + #[serde(default)] 34 + pub facets: Vec<Value>, 35 + #[serde(rename = "createdAt")] 36 + pub created_at: String, 37 + #[serde(skip_serializing_if = "Option::is_none")] 38 + pub parent: Option<String>, 39 + #[serde(default)] 40 + pub attachments: Vec<Value>, 41 + #[serde(default, skip_serializing_if = "Option::is_none")] 42 + pub edited: Option<bool>, 43 + } 44 + 45 + impl NotificationMessage { 46 + pub fn from_colibri_message(message: ColibriMessage) -> Self { 47 + Self { 48 + text: message.text, 49 + facets: message.facets.unwrap_or_default(), 50 + created_at: message.created_at, 51 + parent: message.parent, 52 + attachments: message.attachments.unwrap_or_default(), 53 + edited: message.edited, 54 + } 55 + } 56 + } 57 + 58 + /// One notification awaiting delivery, in the shape we send to clients. 59 + #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)] 60 + pub struct NotificationView { 61 + pub id: i64, 62 + #[serde(rename = "recipientDid")] 63 + pub recipient_did: String, 64 + pub kind: String, 65 + #[serde(rename = "messageUri")] 66 + pub message_uri: String, 67 + #[serde(rename = "authorDid")] 68 + pub author_did: String, 69 + #[serde(rename = "channelRkey")] 70 + pub channel_rkey: String, 71 + #[serde(rename = "indexedAt")] 72 + pub indexed_at: String, 73 + #[serde(rename = "seenAt", skip_serializing_if = "Option::is_none")] 74 + pub seen_at: Option<String>, 75 + /// Inline message body. `None` if the underlying message has been deleted 76 + /// from the AppView cache. 77 + #[serde(skip_serializing_if = "Option::is_none")] 78 + pub message: Option<NotificationMessage>, 79 + } 80 + 81 + impl NotificationView { 82 + pub fn from_row(row: notifications::Model, message: Option<NotificationMessage>) -> Self { 83 + Self { 84 + id: row.id, 85 + recipient_did: row.recipient_did, 86 + kind: row.kind, 87 + message_uri: row.message_uri, 88 + author_did: row.author_did, 89 + channel_rkey: row.channel_rkey, 90 + indexed_at: row.indexed_at, 91 + seen_at: row.seen_at, 92 + message, 93 + } 94 + } 95 + } 96 + 97 + /// Extracts every mentioned DID from the facets of a message. Returns DIDs in 98 + /// the order they first appear, deduplicated. 99 + pub fn extract_mentioned_dids(facets: &[Value]) -> Vec<String> { 100 + let mut out: Vec<String> = Vec::new(); 101 + let mut seen: HashSet<String> = HashSet::new(); 102 + for facet in facets { 103 + let Some(features) = facet.get("features").and_then(|f| f.as_array()) else { 104 + continue; 105 + }; 106 + for feature in features { 107 + let Some(kind) = feature.get("$type").and_then(|t| t.as_str()) else { 108 + continue; 109 + }; 110 + if kind != FACET_MENTION_TYPE { 111 + continue; 112 + } 113 + let Some(did) = feature.get("did").and_then(|d| d.as_str()) else { 114 + continue; 115 + }; 116 + if seen.insert(did.to_string()) { 117 + out.push(did.to_string()); 118 + } 119 + } 120 + } 121 + out 122 + } 123 + 124 + /// Looks up the author DID of a parent message inside the same channel. 125 + /// Returns `None` if the parent is not in the cache. 126 + pub async fn fetch_parent_author( 127 + db: &DatabaseConnection, 128 + channel_rkey: &str, 129 + parent_rkey: &str, 130 + ) -> Result<Option<String>, DbErr> { 131 + let record = record_data::Entity::find() 132 + .filter(record_data::Column::Nsid.eq("social.colibri.message")) 133 + .filter(record_data::Column::Rkey.eq(parent_rkey)) 134 + .filter(sea_orm::prelude::Expr::cust_with_values( 135 + r#""record_data"."data"->>'channel' = $1"#, 136 + vec![sea_orm::Value::from(channel_rkey.to_string())], 137 + )) 138 + .one(db) 139 + .await?; 140 + 141 + Ok(record.map(|r| r.did)) 142 + } 143 + 144 + /// Inserts notification rows for a freshly indexed message. Mentions of the 145 + /// author themself are skipped, as are duplicate insertions (the table has a 146 + /// unique index on `(recipient_did, message_uri, kind)`). 147 + pub async fn index_message_notifications( 148 + db: &DatabaseConnection, 149 + author_did: &str, 150 + message_uri: &str, 151 + message: &ColibriMessage, 152 + ) -> Result<Vec<IndexedNotification>, DbErr> { 153 + let mut recipients: Vec<(String, &'static str)> = Vec::new(); 154 + let mut seen: HashSet<(String, &'static str)> = HashSet::new(); 155 + 156 + if let Some(facets) = message.facets.as_ref() { 157 + for did in extract_mentioned_dids(facets) { 158 + if did == author_did { 159 + continue; 160 + } 161 + let key = (did.clone(), KIND_MENTION); 162 + if seen.insert(key.clone()) { 163 + recipients.push(key); 164 + } 165 + } 166 + } 167 + 168 + if let Some(parent_rkey) = message.parent.as_ref() 169 + && let Some(parent_author) = fetch_parent_author(db, &message.channel, parent_rkey).await? 170 + && parent_author != author_did 171 + { 172 + let key = (parent_author, KIND_REPLY); 173 + if seen.insert(key.clone()) { 174 + recipients.push(key); 175 + } 176 + } 177 + 178 + if recipients.is_empty() { 179 + return Ok(Vec::new()); 180 + } 181 + 182 + let now = current_iso8601_utc(); 183 + let mut inserted = Vec::new(); 184 + for (recipient_did, kind) in recipients { 185 + let row = notifications::ActiveModel { 186 + recipient_did: ActiveValue::Set(recipient_did.clone()), 187 + kind: ActiveValue::Set(kind.to_string()), 188 + message_uri: ActiveValue::Set(message_uri.to_string()), 189 + author_did: ActiveValue::Set(author_did.to_string()), 190 + channel_rkey: ActiveValue::Set(message.channel.clone()), 191 + indexed_at: ActiveValue::Set(now.clone()), 192 + seen_at: ActiveValue::Set(None), 193 + ..Default::default() 194 + }; 195 + 196 + let result = notifications::Entity::insert(row) 197 + .on_conflict( 198 + sea_query::OnConflict::columns([ 199 + notifications::Column::RecipientDid, 200 + notifications::Column::MessageUri, 201 + notifications::Column::Kind, 202 + ]) 203 + .do_nothing() 204 + .to_owned(), 205 + ) 206 + .exec(db) 207 + .await; 208 + 209 + // Conflict (= already indexed) is a no-op; surface other failures. 210 + match result { 211 + Ok(_) => {} 212 + Err(DbErr::RecordNotInserted) => continue, 213 + Err(e) => return Err(e), 214 + } 215 + 216 + if let Some(row) = notifications::Entity::find() 217 + .filter(notifications::Column::RecipientDid.eq(&recipient_did)) 218 + .filter(notifications::Column::MessageUri.eq(message_uri)) 219 + .filter(notifications::Column::Kind.eq(kind)) 220 + .one(db) 221 + .await? 222 + { 223 + inserted.push(IndexedNotification { 224 + row, 225 + message: NotificationMessage::from_colibri_message(message.clone()), 226 + }); 227 + } 228 + } 229 + Ok(inserted) 230 + } 231 + 232 + /// Lists notifications for a recipient, newest first. The optional cursor is 233 + /// the stringified id of the oldest notification on the previous page; rows 234 + /// with `id < cursor` are returned. 235 + pub async fn list_notifications( 236 + db: &DatabaseConnection, 237 + recipient_did: &str, 238 + limit: u64, 239 + cursor: Option<&str>, 240 + ) -> Result<Vec<notifications::Model>, DbErr> { 241 + let mut condition = Condition::all().add(notifications::Column::RecipientDid.eq(recipient_did)); 242 + 243 + if let Some(c) = cursor { 244 + let parsed: i64 = c 245 + .parse() 246 + .map_err(|_| DbErr::Custom(format!("invalid cursor: {c}")))?; 247 + condition = condition.add(notifications::Column::Id.lt(parsed)); 248 + } 249 + 250 + notifications::Entity::find() 251 + .filter(condition) 252 + .order_by_desc(notifications::Column::Id) 253 + .limit(limit) 254 + .all(db) 255 + .await 256 + } 257 + 258 + /// Fetches the cached `social.colibri.message` records referenced by the given 259 + /// AT-URIs and returns them keyed by URI. URIs that don't resolve to a 260 + /// matching record (deleted, never indexed, malformed URI) are simply absent 261 + /// from the result. 262 + pub async fn hydrate_messages( 263 + db: &DatabaseConnection, 264 + message_uris: &[String], 265 + ) -> Result<HashMap<String, NotificationMessage>, DbErr> { 266 + if message_uris.is_empty() { 267 + return Ok(HashMap::new()); 268 + } 269 + 270 + let mut by_uri: HashMap<String, (String, String)> = HashMap::new(); 271 + let mut condition = Condition::any(); 272 + for uri in message_uris { 273 + let Some(parsed) = AtUri::parse(uri) else { 274 + continue; 275 + }; 276 + if parsed.collection != "social.colibri.message" { 277 + continue; 278 + } 279 + let did = parsed.authority.clone(); 280 + let rkey = parsed.rkey.clone(); 281 + by_uri.insert(uri.clone(), (did.clone(), rkey.clone())); 282 + condition = condition.add( 283 + Condition::all() 284 + .add(record_data::Column::Did.eq(did)) 285 + .add(record_data::Column::Rkey.eq(rkey)), 286 + ); 287 + } 288 + 289 + if by_uri.is_empty() { 290 + return Ok(HashMap::new()); 291 + } 292 + 293 + let records = record_data::Entity::find() 294 + .filter(record_data::Column::Nsid.eq("social.colibri.message")) 295 + .filter(condition) 296 + .all(db) 297 + .await?; 298 + 299 + let mut by_key: HashMap<(String, String), NotificationMessage> = HashMap::new(); 300 + for record in records { 301 + if let Ok(message) = serde_json::from_value::<ColibriMessage>(record.data) { 302 + by_key.insert( 303 + (record.did, record.rkey), 304 + NotificationMessage::from_colibri_message(message), 305 + ); 306 + } 307 + } 308 + 309 + let mut out = HashMap::new(); 310 + for (uri, key) in by_uri { 311 + if let Some(msg) = by_key.remove(&key) { 312 + out.insert(uri, msg); 313 + } 314 + } 315 + Ok(out) 316 + } 317 + 318 + /// Counts unseen notifications for a recipient. 319 + pub async fn unseen_count(db: &DatabaseConnection, recipient_did: &str) -> Result<u64, DbErr> { 320 + notifications::Entity::find() 321 + .filter(notifications::Column::RecipientDid.eq(recipient_did)) 322 + .filter(notifications::Column::SeenAt.is_null()) 323 + .count(db) 324 + .await 325 + } 326 + 327 + /// Marks every unseen notification for the recipient with `indexed_at <= now` 328 + /// as seen at `seen_at`. Returns the number of rows updated. 329 + pub async fn mark_seen_up_to( 330 + db: &DatabaseConnection, 331 + recipient_did: &str, 332 + seen_at: &str, 333 + cutoff_indexed_at: &str, 334 + ) -> Result<u64, DbErr> { 335 + let res = notifications::Entity::update_many() 336 + .col_expr( 337 + notifications::Column::SeenAt, 338 + sea_query::Expr::value(seen_at), 339 + ) 340 + .filter(notifications::Column::RecipientDid.eq(recipient_did)) 341 + .filter(notifications::Column::SeenAt.is_null()) 342 + .filter(notifications::Column::IndexedAt.lte(cutoff_indexed_at)) 343 + .exec(db) 344 + .await?; 345 + Ok(res.rows_affected) 346 + } 347 + 348 + #[cfg(test)] 349 + mod tests { 350 + use super::*; 351 + use serde_json::json; 352 + 353 + fn message_with_facets(facets: Value, parent: Option<&str>) -> ColibriMessage { 354 + ColibriMessage { 355 + r#type: "social.colibri.message".to_string(), 356 + text: "hello".to_string(), 357 + facets: Some(facets.as_array().cloned().unwrap_or_default()), 358 + created_at: "2026-05-14T00:00:00Z".to_string(), 359 + channel: "chan-a".to_string(), 360 + edited: None, 361 + parent: parent.map(String::from), 362 + attachments: None, 363 + } 364 + } 365 + 366 + #[test] 367 + fn extract_mentioned_dids_pulls_did_from_mention_features() { 368 + let facets = json!([ 369 + { 370 + "index": {"byteStart": 0, "byteEnd": 5}, 371 + "features": [ 372 + {"$type": "social.colibri.richtext.facet#mention", "did": "did:plc:alice"} 373 + ] 374 + }, 375 + { 376 + "index": {"byteStart": 10, "byteEnd": 13}, 377 + "features": [ 378 + {"$type": "social.colibri.richtext.facet#bold"} 379 + ] 380 + }, 381 + { 382 + "index": {"byteStart": 20, "byteEnd": 25}, 383 + "features": [ 384 + {"$type": "social.colibri.richtext.facet#mention", "did": "did:plc:bob"}, 385 + {"$type": "social.colibri.richtext.facet#mention", "did": "did:plc:alice"} 386 + ] 387 + } 388 + ]); 389 + let dids = extract_mentioned_dids(facets.as_array().unwrap()); 390 + assert_eq!( 391 + dids, 392 + vec![String::from("did:plc:alice"), String::from("did:plc:bob")] 393 + ); 394 + } 395 + 396 + #[test] 397 + fn extract_mentioned_dids_ignores_non_mention_features_and_missing_dids() { 398 + let facets = json!([ 399 + { 400 + "features": [ 401 + {"$type": "social.colibri.richtext.facet#link", "uri": "https://example"} 402 + ] 403 + }, 404 + { 405 + "features": [ 406 + {"$type": "social.colibri.richtext.facet#mention"} 407 + ] 408 + } 409 + ]); 410 + assert!(extract_mentioned_dids(facets.as_array().unwrap()).is_empty()); 411 + } 412 + 413 + #[test] 414 + fn message_with_facets_is_constructible() { 415 + // Just guarantees that ColibriMessage stays compatible with this helper. 416 + let msg = message_with_facets(json!([]), Some("parent")); 417 + assert_eq!(msg.channel, "chan-a"); 418 + assert_eq!(msg.parent.as_deref(), Some("parent")); 419 + } 420 + }

+91

src/lib/permissions.rs

··· 1 + /// Enumerated permissions for the Colibri community role system. 2 + /// 3 + /// Permissions are stored as strings on `social.colibri.role` records, so this 4 + /// enum exists primarily to give the Rust code a single source of truth for 5 + /// the namespaced permission identifiers. 6 + #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] 7 + pub enum Permission { 8 + CommunityUpdate, 9 + CommunityDelete, 10 + CategoryCreate, 11 + CategoryUpdate, 12 + CategoryDelete, 13 + ChannelCreate, 14 + ChannelUpdate, 15 + ChannelDelete, 16 + MessageDelete, 17 + MemberKick, 18 + MemberBan, 19 + MemberUnban, 20 + RoleManage, 21 + InvitationCreate, 22 + InvitationDelete, 23 + ModerationViewLog, 24 + ApprovalManage, 25 + } 26 + 27 + impl Permission { 28 + pub fn as_str(&self) -> &'static str { 29 + match self { 30 + Permission::CommunityUpdate => "community.update", 31 + Permission::CommunityDelete => "community.delete", 32 + Permission::CategoryCreate => "category.create", 33 + Permission::CategoryUpdate => "category.update", 34 + Permission::CategoryDelete => "category.delete", 35 + Permission::ChannelCreate => "channel.create", 36 + Permission::ChannelUpdate => "channel.update", 37 + Permission::ChannelDelete => "channel.delete", 38 + Permission::MessageDelete => "message.delete", 39 + Permission::MemberKick => "member.kick", 40 + Permission::MemberBan => "member.ban", 41 + Permission::MemberUnban => "member.unban", 42 + Permission::RoleManage => "role.manage", 43 + Permission::InvitationCreate => "invitation.create", 44 + Permission::InvitationDelete => "invitation.delete", 45 + Permission::ModerationViewLog => "moderation.viewLog", 46 + Permission::ApprovalManage => "approval.manage", 47 + } 48 + } 49 + } 50 + 51 + impl std::fmt::Display for Permission { 52 + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { 53 + write!(f, "{}", self.as_str()) 54 + } 55 + } 56 + 57 + #[cfg(test)] 58 + mod tests { 59 + use super::*; 60 + 61 + #[test] 62 + fn permission_strings_are_namespaced_and_unique() { 63 + use std::collections::HashSet; 64 + 65 + let all = [ 66 + Permission::CommunityUpdate, 67 + Permission::CommunityDelete, 68 + Permission::CategoryCreate, 69 + Permission::CategoryUpdate, 70 + Permission::CategoryDelete, 71 + Permission::ChannelCreate, 72 + Permission::ChannelUpdate, 73 + Permission::ChannelDelete, 74 + Permission::MessageDelete, 75 + Permission::MemberKick, 76 + Permission::MemberBan, 77 + Permission::MemberUnban, 78 + Permission::RoleManage, 79 + Permission::InvitationCreate, 80 + Permission::InvitationDelete, 81 + Permission::ModerationViewLog, 82 + Permission::ApprovalManage, 83 + ]; 84 + 85 + let strings: HashSet<&str> = all.iter().map(|p| p.as_str()).collect(); 86 + assert_eq!(strings.len(), all.len()); 87 + for s in &strings { 88 + assert!(s.contains('.'), "{s} should be namespaced"); 89 + } 90 + } 91 + }

+171

src/lib/reactions.rs

··· 1 + use std::collections::HashMap; 2 + 3 + use sea_orm::prelude::Expr; 4 + use sea_orm::{ColumnTrait, DatabaseConnection, DbErr, EntityTrait, QueryFilter}; 5 + use serde::{Deserialize, Serialize}; 6 + 7 + use crate::models::record_data; 8 + 9 + #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)] 10 + pub struct ReactionSummary { 11 + pub emoji: String, 12 + pub count: u32, 13 + #[serde(rename = "reactorDIDs")] 14 + pub reactor_dids: Vec<String>, 15 + } 16 + 17 + #[derive(Deserialize)] 18 + struct StoredReaction { 19 + emoji: String, 20 + #[serde(rename = "targetMessage")] 21 + target_message: String, 22 + } 23 + 24 + /// Loads every reaction targeting the provided message rkeys and returns them 25 + /// keyed by the target message rkey, with reactions of the same emoji folded 26 + /// into a single summary entry. 27 + pub async fn group_reactions_for_messages( 28 + db: &DatabaseConnection, 29 + message_rkeys: &[String], 30 + ) -> Result<HashMap<String, Vec<ReactionSummary>>, DbErr> { 31 + if message_rkeys.is_empty() { 32 + return Ok(HashMap::new()); 33 + } 34 + 35 + let values: Vec<sea_orm::Value> = message_rkeys 36 + .iter() 37 + .map(|k| sea_orm::Value::from(k.clone())) 38 + .collect(); 39 + 40 + let placeholders: Vec<String> = (1..=values.len()).map(|i| format!("${i}")).collect(); 41 + let in_clause = format!( 42 + r#""record_data"."data"->>'targetMessage' IN ({})"#, 43 + placeholders.join(", ") 44 + ); 45 + 46 + let records = record_data::Entity::find() 47 + .filter(record_data::Column::Nsid.eq("social.colibri.reaction")) 48 + .filter(Expr::cust_with_values(in_clause, values)) 49 + .all(db) 50 + .await?; 51 + 52 + Ok(group_reaction_records(records)) 53 + } 54 + 55 + /// Loads every reaction targeting a single message rkey. 56 + pub async fn list_reactions_for_message( 57 + db: &DatabaseConnection, 58 + message_rkey: &str, 59 + ) -> Result<Vec<ReactionSummary>, DbErr> { 60 + let records = record_data::Entity::find() 61 + .filter(record_data::Column::Nsid.eq("social.colibri.reaction")) 62 + .filter(Expr::cust_with_values( 63 + r#""record_data"."data"->>'targetMessage' = $1"#, 64 + vec![sea_orm::Value::from(message_rkey.to_string())], 65 + )) 66 + .all(db) 67 + .await?; 68 + 69 + let mut grouped = group_reaction_records(records); 70 + Ok(grouped.remove(message_rkey).unwrap_or_default()) 71 + } 72 + 73 + fn group_reaction_records( 74 + records: Vec<record_data::Model>, 75 + ) -> HashMap<String, Vec<ReactionSummary>> { 76 + // (target_message_rkey, emoji) -> Vec<reactor_did> 77 + let mut buckets: HashMap<(String, String), Vec<String>> = HashMap::new(); 78 + 79 + for record in records { 80 + let Ok(stored) = serde_json::from_value::<StoredReaction>(record.data) else { 81 + continue; 82 + }; 83 + buckets 84 + .entry((stored.target_message, stored.emoji)) 85 + .or_default() 86 + .push(record.did); 87 + } 88 + 89 + let mut grouped: HashMap<String, Vec<ReactionSummary>> = HashMap::new(); 90 + for ((target, emoji), mut dids) in buckets { 91 + dids.sort(); 92 + dids.dedup(); 93 + grouped.entry(target).or_default().push(ReactionSummary { 94 + emoji, 95 + count: dids.len() as u32, 96 + reactor_dids: dids, 97 + }); 98 + } 99 + 100 + grouped 101 + } 102 + 103 + #[cfg(test)] 104 + mod tests { 105 + use super::*; 106 + 107 + fn reaction(did: &str, rkey: &str, target: &str, emoji: &str) -> record_data::Model { 108 + record_data::Model { 109 + id: 0, 110 + did: did.to_string(), 111 + nsid: String::from("social.colibri.reaction"), 112 + rkey: rkey.to_string(), 113 + data: serde_json::json!({ 114 + "emoji": emoji, 115 + "targetMessage": target, 116 + }), 117 + } 118 + } 119 + 120 + #[test] 121 + fn groups_reactions_by_message_and_emoji() { 122 + let records = vec![ 123 + reaction("did:plc:alice", "r1", "msg-1", "🦜"), 124 + reaction("did:plc:bob", "r2", "msg-1", "🦜"), 125 + reaction("did:plc:alice", "r3", "msg-1", "🔥"), 126 + reaction("did:plc:carol", "r4", "msg-2", "🔥"), 127 + ]; 128 + let grouped = group_reaction_records(records); 129 + 130 + let msg1 = grouped.get("msg-1").unwrap(); 131 + let parrot = msg1.iter().find(|r| r.emoji == "🦜").unwrap(); 132 + assert_eq!(parrot.count, 2); 133 + assert_eq!( 134 + parrot.reactor_dids, 135 + vec![String::from("did:plc:alice"), String::from("did:plc:bob")] 136 + ); 137 + 138 + let fire = msg1.iter().find(|r| r.emoji == "🔥").unwrap(); 139 + assert_eq!(fire.count, 1); 140 + 141 + let msg2 = grouped.get("msg-2").unwrap(); 142 + assert_eq!(msg2.len(), 1); 143 + assert_eq!(msg2[0].emoji, "🔥"); 144 + assert_eq!(msg2[0].reactor_dids, vec![String::from("did:plc:carol")]); 145 + } 146 + 147 + #[test] 148 + fn deduplicates_reactor_dids_for_same_emoji() { 149 + let records = vec![ 150 + reaction("did:plc:alice", "r1", "msg-1", "🦜"), 151 + reaction("did:plc:alice", "r2", "msg-1", "🦜"), 152 + ]; 153 + let grouped = group_reaction_records(records); 154 + let parrot = &grouped.get("msg-1").unwrap()[0]; 155 + assert_eq!(parrot.count, 1); 156 + assert_eq!(parrot.reactor_dids, vec![String::from("did:plc:alice")]); 157 + } 158 + 159 + #[test] 160 + fn skips_records_with_invalid_payload() { 161 + let records = vec![record_data::Model { 162 + id: 0, 163 + did: String::from("did:plc:alice"), 164 + nsid: String::from("social.colibri.reaction"), 165 + rkey: String::from("r1"), 166 + data: serde_json::json!({ "irrelevant": true }), 167 + }]; 168 + let grouped = group_reaction_records(records); 169 + assert!(grouped.is_empty()); 170 + } 171 + }

+66

src/lib/time.rs

··· 1 + /// Returns the current UTC timestamp in ISO 8601 form 2 + /// (`YYYY-MM-DDTHH:MM:SS.mmmZ`). Used wherever we mint AppView-side records. 3 + pub fn current_iso8601_utc() -> String { 4 + use std::time::{SystemTime, UNIX_EPOCH}; 5 + let now = SystemTime::now() 6 + .duration_since(UNIX_EPOCH) 7 + .unwrap_or_default(); 8 + iso8601_from_epoch(now.as_secs(), now.subsec_millis()) 9 + } 10 + 11 + /// Formats a (seconds_since_epoch, millis) pair as ISO 8601 UTC. 12 + pub fn iso8601_from_epoch(secs: u64, millis: u32) -> String { 13 + let (year, month, day, hour, minute, second) = epoch_to_ymdhms(secs); 14 + format!( 15 + "{:04}-{:02}-{:02}T{:02}:{:02}:{:02}.{:03}Z", 16 + year, month, day, hour, minute, second, millis 17 + ) 18 + } 19 + 20 + /// Howard Hinnant's "civil_from_days" algorithm — converts seconds since the 21 + /// Unix epoch to a (year, month, day, hour, minute, second) tuple in UTC. 22 + fn epoch_to_ymdhms(secs: u64) -> (u32, u32, u32, u32, u32, u32) { 23 + let days = (secs / 86_400) as i64; 24 + let seconds_of_day = (secs % 86_400) as u32; 25 + let hour = seconds_of_day / 3600; 26 + let minute = (seconds_of_day % 3600) / 60; 27 + let second = seconds_of_day % 60; 28 + 29 + let z = days + 719_468; 30 + let era = if z >= 0 { z } else { z - 146_096 } / 146_097; 31 + let doe = (z - era * 146_097) as u64; 32 + let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365; 33 + let y = yoe as i64 + era * 400; 34 + let doy = doe - (365 * yoe + yoe / 4 - yoe / 100); 35 + let mp = (5 * doy + 2) / 153; 36 + let d = (doy - (153 * mp + 2) / 5 + 1) as u32; 37 + let m = if mp < 10 { mp + 3 } else { mp - 9 } as u32; 38 + let y = if m <= 2 { y + 1 } else { y }; 39 + 40 + (y as u32, m, d, hour, minute, second) 41 + } 42 + 43 + #[cfg(test)] 44 + mod tests { 45 + use super::*; 46 + 47 + #[test] 48 + fn formats_epoch_zero_as_unix_epoch_string() { 49 + assert_eq!( 50 + iso8601_from_epoch(0, 0), 51 + "1970-01-01T00:00:00.000Z" 52 + ); 53 + } 54 + 55 + #[test] 56 + fn formats_known_datetime() { 57 + // 2026-05-13T12:34:56.789Z 58 + // = (2026-1970) years -> days(2026-01-01) - days(1970-01-01) + 132 days 59 + // Easier: compute via known offset. 60 + // 1747139696 secs is 2025-05-13T12:34:56Z; +365.25*86400 ~ 2026-05-13T... 61 + // We just verify shape rather than hand-computing. 62 + let s = iso8601_from_epoch(1_778_675_696, 789); 63 + assert!(s.ends_with(".789Z")); 64 + assert_eq!(s.len(), "1970-01-01T00:00:00.000Z".len()); 65 + } 66 + }

+55 -1

src/main.rs

··· 2 2 3 3 extern crate pretty_env_logger; 4 4 5 + use crate::lib::colibri::ColibriMessage; 5 6 use crate::lib::db::init_db; 7 + use crate::lib::notifications::{IndexedNotification, index_message_notifications}; 6 8 use crate::lib::sentry::init_sentry; 7 9 use crate::lib::tap::{self, TapMessage, TapMessageRecord, TapStream, ack_tap_msg}; 8 10 use futures::{SinkExt, StreamExt}; ··· 30 32 #[allow(dead_code)] 31 33 channel: mpsc::Sender<String>, 32 34 broadcast: broadcast::Sender<TapMessageRecord>, 35 + notifications: broadcast::Sender<IndexedNotification>, 33 36 } 34 37 35 38 #[derive(Clone)] ··· 44 47 mut rx_outbound: mpsc::Receiver<String>, 45 48 tx_inbound: broadcast::Sender<TapMessageRecord>, 46 49 mut to_tap: mpsc::Sender<String>, 50 + notifications_tx: broadcast::Sender<IndexedNotification>, 47 51 ) { 48 52 loop { 49 53 tokio::select! { ··· 69 73 return; 70 74 } 71 75 72 - let _ = tx_inbound.send(tap_msg.record.unwrap()); 76 + let record = tap_msg.record.unwrap(); 77 + 78 + // Centralized notification indexing for newly authored 79 + // Colibri messages. Persists rows + broadcasts to live WS 80 + // subscribers, so each notification is written exactly 81 + // once regardless of how many clients are connected. 82 + if record.collection == "social.colibri.message" 83 + && record.action != "delete" 84 + && let Some(payload) = record.record.clone() 85 + && let Ok(message) = serde_json::from_value::<ColibriMessage>(payload) 86 + { 87 + let message_uri = format!( 88 + "at://{}/{}/{}", 89 + record.did, record.collection, record.rkey 90 + ); 91 + match index_message_notifications(&db, &record.did, &message_uri, &message) 92 + .await 93 + { 94 + Ok(rows) => { 95 + for row in rows { 96 + let _ = notifications_tx.send(row); 97 + } 98 + } 99 + Err(e) => { 100 + log::error!("notification indexing failed for {message_uri}: {e}"); 101 + } 102 + } 103 + } 104 + 105 + let _ = tx_inbound.send(record); 73 106 ack_tap_msg(&db, &mut to_tap, text.to_string()).await; 74 107 } 75 108 } ··· 122 155 123 156 let c2c_broadcast_channel = broadcast::channel::<EventNotification>(128); 124 157 158 + let (notification_broadcast, _) = broadcast::channel::<IndexedNotification>(128); 159 + 125 160 let tap_bridge = CommsBridge { 126 161 channel: to_tap.clone(), 127 162 broadcast: from_tap_broadcast.clone(), 163 + notifications: notification_broadcast.clone(), 128 164 }; 129 165 130 166 let db = match init_db().await { ··· 149 185 from_tap_channel, 150 186 from_tap_broadcast, 151 187 to_tap, 188 + notification_broadcast, 152 189 )); 153 190 154 191 rocket::build() ··· 165 202 xrpc::social::colibri::actor::get_data, 166 203 xrpc::social::colibri::actor::list_communities, 167 204 xrpc::social::colibri::actor::set_state, 205 + xrpc::social::colibri::channel::get_read_cursor, 206 + xrpc::social::colibri::channel::list_messages, 207 + xrpc::social::colibri::channel::list_reactions, 208 + xrpc::social::colibri::community::block_message, 209 + xrpc::social::colibri::community::block_user, 210 + xrpc::social::colibri::community::create_invitation, 211 + xrpc::social::colibri::community::delete_invitation, 212 + xrpc::social::colibri::community::get_invitation, 213 + xrpc::social::colibri::community::list_blocked_users, 214 + xrpc::social::colibri::community::list_categories, 215 + xrpc::social::colibri::community::list_channels, 216 + xrpc::social::colibri::community::list_invitations, 217 + xrpc::social::colibri::community::list_members, 218 + xrpc::social::colibri::community::unblock_user, 219 + xrpc::social::colibri::notification::get_unread_count, 220 + xrpc::social::colibri::notification::list_notifications, 221 + xrpc::social::colibri::notification::update_seen, 168 222 xrpc::social::colibri::sync::subscribe_events, 169 223 ], 170 224 )

+19

src/models/community_invitations.rs

··· 1 + use sea_orm::entity::prelude::*; 2 + use serde::Serialize; 3 + 4 + #[sea_orm::model] 5 + #[derive(Clone, Debug, PartialEq, Eq, DeriveEntityModel, Serialize)] 6 + #[sea_orm(table_name = "community_invitations")] 7 + pub struct Model { 8 + #[sea_orm(primary_key, auto_increment = false, column_type = "Text")] 9 + pub code: String, 10 + #[sea_orm(column_type = "Text")] 11 + pub community_uri: String, 12 + #[sea_orm(column_type = "Text")] 13 + pub created_by: String, 14 + pub active: bool, 15 + #[sea_orm(column_type = "Text")] 16 + pub created_at: String, 17 + } 18 + 19 + impl ActiveModelBehavior for ActiveModel {}

+2

src/models/mod.rs

··· 3 3 pub mod prelude; 4 4 5 5 pub mod collection_cursors; 6 + pub mod community_invitations; 6 7 pub mod firehose_cursors; 7 8 pub mod list_repos_cursors; 9 + pub mod notifications; 8 10 pub mod outbox_buffers; 9 11 pub mod record_data; 10 12 pub mod repo_records;

+26

src/models/notifications.rs

··· 1 + use sea_orm::entity::prelude::*; 2 + use serde::Serialize; 3 + 4 + #[sea_orm::model] 5 + #[derive(Clone, Debug, PartialEq, Eq, DeriveEntityModel, Serialize)] 6 + #[sea_orm(table_name = "notifications")] 7 + pub struct Model { 8 + #[sea_orm(primary_key)] 9 + pub id: i64, 10 + #[sea_orm(column_type = "Text")] 11 + pub recipient_did: String, 12 + #[sea_orm(column_type = "Text")] 13 + pub kind: String, 14 + #[sea_orm(column_type = "Text")] 15 + pub message_uri: String, 16 + #[sea_orm(column_type = "Text")] 17 + pub author_did: String, 18 + #[sea_orm(column_type = "Text")] 19 + pub channel_rkey: String, 20 + #[sea_orm(column_type = "Text")] 21 + pub indexed_at: String, 22 + #[sea_orm(column_type = "Text", nullable)] 23 + pub seen_at: Option<String>, 24 + } 25 + 26 + impl ActiveModelBehavior for ActiveModel {}

+40 -15

src/xrpc/com/atproto/sync/list_records_handler.rs

··· 9 9 10 10 #[derive(Serialize, Debug)] 11 11 pub struct ListRecordsResponse { 12 + #[serde(skip_serializing_if = "Option::is_none")] 12 13 pub cursor: Option<String>, 13 14 pub records: Vec<Value>, 14 15 } ··· 21 22 cursor: Option<&str>, 22 23 reverse: Option<bool>, 23 24 ) -> Result<Json<ListRecordsResponse>, ErrorResponse> { 24 - let records = list_atproto_records::<Value>( 25 + let result = list_atproto_records::<Value>( 25 26 repo.to_string(), 26 27 collection.to_string(), 27 28 limit, ··· 31 32 ) 32 33 .await; 33 34 34 - if records.is_err() { 35 + if result.is_err() { 35 36 return Err(ErrorResponse { 36 37 body: Json(ErrorBody { 37 38 error: String::from("InternalServerError"), ··· 40 41 }); 41 42 } 42 43 43 - let safe_records = records.unwrap(); 44 + let paged = result.unwrap(); 44 45 45 46 Ok(Json(ListRecordsResponse { 46 - cursor: Some(String::from(cursor.unwrap_or(""))), 47 - records: safe_records, 47 + cursor: paged.cursor, 48 + records: paged.records, 48 49 })) 49 50 } 50 51 51 52 #[get("/xrpc/com.atproto.sync.listRecords?<repo>&<collection>&<limit>&<cursor>&<reverse>")] 52 - /// Returns a cached record from the database. 53 + /// Returns a paginated list of cached records from the database. 53 54 pub async fn list_records( 54 55 repo: &str, 55 56 collection: &str, ··· 68 69 use sea_orm::{DatabaseBackend, MockDatabase}; 69 70 use serde_json::json; 70 71 72 + fn model(id: i64, rkey: &str, text: &str) -> crate::models::record_data::Model { 73 + crate::models::record_data::Model { 74 + id, 75 + did: String::from("did:plc:abc"), 76 + nsid: String::from("social.colibri.message"), 77 + rkey: String::from(rkey), 78 + data: json!({ "text": text }), 79 + } 80 + } 81 + 71 82 #[tokio::test] 72 83 async fn list_records_returns_records() { 73 84 let db = MockDatabase::new(DatabaseBackend::Postgres) 74 - .append_query_results([vec![crate::models::record_data::Model { 75 - id: 1, 76 - did: String::from("did:plc:abc"), 77 - nsid: String::from("social.colibri.message"), 78 - rkey: String::from("r1"), 79 - data: json!({"text":"hello"}), 80 - }]]) 85 + .append_query_results([vec![model(1, "r1", "hello")]]) 81 86 .into_connection(); 82 87 83 88 let result = list_records_with_db( ··· 85 90 "did:plc:abc", 86 91 "social.colibri.message", 87 92 Some(10), 88 - Some("cursor"), 93 + None, 89 94 None, 90 95 ) 91 96 .await 92 97 .unwrap(); 93 98 94 - assert_eq!(result.cursor.as_deref(), Some("cursor")); 95 99 assert_eq!(result.records.len(), 1); 96 100 assert_eq!(result.records[0]["text"], "hello"); 101 + assert!(result.cursor.is_none()); 102 + } 103 + 104 + #[tokio::test] 105 + async fn list_records_returns_cursor_when_page_is_full() { 106 + let db = MockDatabase::new(DatabaseBackend::Postgres) 107 + .append_query_results([vec![model(2, "r2", "second"), model(1, "r1", "first")]]) 108 + .into_connection(); 109 + 110 + let result = list_records_with_db( 111 + &db, 112 + "did:plc:abc", 113 + "social.colibri.message", 114 + Some(2), 115 + None, 116 + None, 117 + ) 118 + .await 119 + .unwrap(); 120 + 121 + assert_eq!(result.cursor.as_deref(), Some("r1")); 97 122 } 98 123 99 124 #[tokio::test]

+2 -2

src/xrpc/social/colibri/actor/get_data_handler.rs

···

+232

src/xrpc/social/colibri/channel/get_read_cursor_handler.rs

···

+418

src/xrpc/social/colibri/channel/list_messages_handler.rs

···

+118

src/xrpc/social/colibri/channel/list_reactions_handler.rs

···

+7

src/xrpc/social/colibri/channel/mod.rs

···

+238

src/xrpc/social/colibri/community/block_message_handler.rs

···

+401

src/xrpc/social/colibri/community/block_user_handler.rs

···

+640

src/xrpc/social/colibri/community/invitations_handler.rs

···

+99

src/xrpc/social/colibri/community/list_blocked_users_handler.rs

···

+216

src/xrpc/social/colibri/community/list_categories_handler.rs

···

+173

src/xrpc/social/colibri/community/list_channels_handler.rs

···

+333

src/xrpc/social/colibri/community/list_members_handler.rs

···

+17

src/xrpc/social/colibri/community/mod.rs

···

+4

src/xrpc/social/colibri/mod.rs

···

+109

src/xrpc/social/colibri/notification/get_unread_count_handler.rs

···

+228

src/xrpc/social/colibri/notification/list_notifications_handler.rs

···

+7

src/xrpc/social/colibri/notification/mod.rs

···

+162

src/xrpc/social/colibri/notification/update_seen_handler.rs

···

+59 -1

src/xrpc/social/colibri/sync/subscribe_events_handler.rs

··················

Configure Feed

Configure Feed