test #2

+1

lexica/src/app_bsky/mod.rs

··· 7 7 pub mod graph; 8 8 pub mod labeler; 9 9 pub mod richtext; 10 10 + pub mod unspecced; 10 11 11 12 #[derive(Clone, Default, Debug, Serialize)] 12 13 #[serde(rename_all = "camelCase")]

+33

lexica/src/app_bsky/unspecced.rs

··· 1 1 + use crate::app_bsky::feed::{BlockedAuthor, PostView}; 2 2 + use serde::Serialize; 3 3 + 4 4 + #[derive(Clone, Debug, Serialize)] 5 5 + pub struct ThreadV2Item { 6 6 + pub uri: String, 7 7 + pub depth: i32, 8 8 + pub value: ThreadV2ItemType, 9 9 + } 10 10 + 11 11 + #[derive(Clone, Debug, Serialize)] 12 12 + #[serde(tag = "$type")] 13 13 + pub enum ThreadV2ItemType { 14 14 + #[serde(rename = "app.bsky.unspecced.defs#threadItemPost")] 15 15 + Post(ThreadItemPost), 16 16 + #[serde(rename = "app.bsky.unspecced.defs#threadItemNoUnauthenticated")] 17 17 + NoUnauthenticated {}, 18 18 + #[serde(rename = "app.bsky.unspecced.defs#threadItemNotFound")] 19 19 + NotFound {}, 20 20 + #[serde(rename = "app.bsky.unspecced.defs#threadItemBlocked")] 21 21 + Blocked { author: BlockedAuthor }, 22 22 + } 23 23 + 24 24 + #[derive(Clone, Debug, Serialize)] 25 25 + #[serde(rename_all = "camelCase")] 26 26 + pub struct ThreadItemPost { 27 27 + pub post: PostView, 28 28 + pub more_parents: bool, 29 29 + pub more_replies: i32, 30 30 + pub op_thread: bool, 31 31 + pub hidden_by_threadgate: bool, 32 32 + pub muted_by_viewer: bool, 33 33 + }

+95 -1

parakeet/src/db.rs

··· 1 1 use diesel::prelude::*; 2 2 - use diesel::sql_types::{Array, Bool, Nullable, Text}; 2 2 + use diesel::sql_types::{Array, Bool, Integer, Nullable, Text}; 3 3 use diesel_async::{AsyncPgConnection, RunQueryDsl}; 4 4 use parakeet_db::{schema, types}; 5 5 + use parakeet_db::models::TextArray; 5 6 6 7 pub async fn get_actor_status( 7 8 conn: &mut AsyncPgConnection, ··· 196 197 .await 197 198 .optional() 198 199 } 200 200 + 201 201 + #[derive(Debug, QueryableByName)] 202 202 + #[diesel(check_for_backend(diesel::pg::Pg))] 203 203 + #[allow(unused)] 204 204 + pub struct ThreadItem { 205 205 + #[diesel(sql_type = Text)] 206 206 + pub at_uri: String, 207 207 + #[diesel(sql_type = Nullable<Text>)] 208 208 + pub parent_uri: Option<String>, 209 209 + #[diesel(sql_type = Nullable<Text>)] 210 210 + pub root_uri: Option<String>, 211 211 + #[diesel(sql_type = Integer)] 212 212 + pub depth: i32, 213 213 + } 214 214 + 215 215 + pub async fn get_thread_children( 216 216 + conn: &mut AsyncPgConnection, 217 217 + uri: &str, 218 218 + depth: i32, 219 219 + ) -> QueryResult<Vec<ThreadItem>> { 220 220 + diesel::sql_query(include_str!("sql/thread.sql")) 221 221 + .bind::<Text, _>(uri) 222 222 + .bind::<Integer, _>(depth) 223 223 + .load(conn) 224 224 + .await 225 225 + } 226 226 + 227 227 + pub async fn get_thread_children_branching( 228 228 + conn: &mut AsyncPgConnection, 229 229 + uri: &str, 230 230 + depth: i32, 231 231 + branching_factor: i32, 232 232 + ) -> QueryResult<Vec<ThreadItem>> { 233 233 + diesel::sql_query(include_str!("sql/thread_branching.sql")) 234 234 + .bind::<Text, _>(uri) 235 235 + .bind::<Integer, _>(depth) 236 236 + .bind::<Integer, _>(branching_factor) 237 237 + .load(conn) 238 238 + .await 239 239 + } 240 240 + 241 241 + #[derive(Debug, QueryableByName)] 242 242 + #[diesel(check_for_backend(diesel::pg::Pg))] 243 243 + pub struct HiddenThreadChildItem { 244 244 + #[diesel(sql_type = Text)] 245 245 + pub at_uri: String, 246 246 + } 247 247 + 248 248 + pub async fn get_thread_children_hidden( 249 249 + conn: &mut AsyncPgConnection, 250 250 + uri: &str, 251 251 + root: &str, 252 252 + ) -> QueryResult<Vec<HiddenThreadChildItem>> { 253 253 + diesel::sql_query(include_str!("sql/thread_v2_hidden_children.sql")) 254 254 + .bind::<Text, _>(uri) 255 255 + .bind::<Text, _>(root) 256 256 + .load(conn) 257 257 + .await 258 258 + } 259 259 + 260 260 + pub async fn get_thread_parents( 261 261 + conn: &mut AsyncPgConnection, 262 262 + uri: &str, 263 263 + height: i32, 264 264 + ) -> QueryResult<Vec<ThreadItem>> { 265 265 + diesel::sql_query(include_str!("sql/thread_parent.sql")) 266 266 + .bind::<Text, _>(uri) 267 267 + .bind::<Integer, _>(height) 268 268 + .load(conn) 269 269 + .await 270 270 + } 271 271 + 272 272 + pub async fn get_root_post(conn: &mut AsyncPgConnection, uri: &str) -> QueryResult<Option<String>> { 273 273 + schema::posts::table 274 274 + .select(schema::posts::root_uri) 275 275 + .find(&uri) 276 276 + .get_result(conn) 277 277 + .await 278 278 + .optional() 279 279 + .map(|v| v.flatten()) 280 280 + } 281 281 + 282 282 + pub async fn get_threadgate_hiddens( 283 283 + conn: &mut AsyncPgConnection, 284 284 + uri: &str, 285 285 + ) -> QueryResult<Option<TextArray>> { 286 286 + schema::threadgates::table 287 287 + .select(schema::threadgates::hidden_replies) 288 288 + .find(&uri) 289 289 + .get_result(conn) 290 290 + .await 291 291 + .optional() 292 292 + }

+1 -3

parakeet/src/hydration/profile.rs

··· 51 51 .followed 52 52 .map(|rkey| format!("at://{}/app.bsky.graph.follow/{rkey}", data.subject)); 53 53 54 54 - let blocking = data.list_block.or(data 55 55 - .blocking 56 56 - .map(|rkey| format!("at://{}/app.bsky.graph.block/{rkey}", data.did))); 54 54 + let blocking = data.list_block.or(data.blocking); 57 55 58 56 ProfileViewerState { 59 57 muted: data.muting.unwrap_or_default(),

+1 -1

parakeet/src/sql/thread.sql

··· 1 1 - with recursive thread as (select at_uri, parent_uri, root_uri, 0 as depth 1 1 + with recursive thread as (select at_uri, parent_uri, root_uri, 1 as depth 2 2 from posts 3 3 where parent_uri = $1 and violates_threadgate=FALSE 4 4 union all

+13

parakeet/src/sql/thread_branching.sql

··· 1 1 + with recursive thread as (select at_uri, parent_uri, root_uri, 1 as depth 2 2 + from posts 3 3 + where parent_uri = $1 4 4 + and violates_threadgate = FALSE 5 5 + union all 6 6 + (select p.at_uri, p.parent_uri, p.root_uri, thread.depth + 1 7 7 + from posts p 8 8 + join thread on p.parent_uri = thread.at_uri 9 9 + where thread.depth <= $2 10 10 + and violates_threadgate = FALSE 11 11 + LIMIT $3)) 12 12 + select * 13 13 + from thread;

+6

parakeet/src/sql/thread_v2_hidden_children.sql

··· 1 1 + select at_uri 2 2 + from posts 3 3 + where parent_uri = $1 4 4 + and at_uri = any (select unnest(hidden_replies) 5 5 + from threadgates 6 6 + where post_uri = $2)

+2 -24

parakeet/src/xrpc/app_bsky/feed/posts.rs

··· 361 361 pub threadgate: Option<ThreadgateView>, 362 362 } 363 363 364 364 - #[derive(Debug, QueryableByName)] 365 365 - #[diesel(check_for_backend(diesel::pg::Pg))] 366 366 - struct ThreadItem { 367 367 - #[diesel(sql_type = diesel::sql_types::Text)] 368 368 - at_uri: String, 369 369 - #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Text>)] 370 370 - parent_uri: Option<String>, 371 371 - // #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Text>)] 372 372 - // root_uri: Option<String>, 373 373 - #[diesel(sql_type = diesel::sql_types::Integer)] 374 374 - depth: i32, 375 375 - } 376 376 - 377 364 pub async fn get_post_thread( 378 365 State(state): State<GlobalState>, 379 366 AtpAcceptLabelers(labelers): AtpAcceptLabelers, ··· 409 396 } 410 397 } 411 398 412 412 - let replies = diesel::sql_query(include_str!("../../../sql/thread.sql")) 413 413 - .bind::<diesel::sql_types::Text, _>(&uri) 414 414 - .bind::<diesel::sql_types::Integer, _>(depth as i32) 415 415 - .load::<ThreadItem>(&mut conn) 416 416 - .await?; 417 417 - 418 418 - let parents = diesel::sql_query(include_str!("../../../sql/thread_parent.sql")) 419 419 - .bind::<diesel::sql_types::Text, _>(&uri) 420 420 - .bind::<diesel::sql_types::Integer, _>(parent_height as i32) 421 421 - .load::<ThreadItem>(&mut conn) 422 422 - .await?; 399 399 + let replies = crate::db::get_thread_children(&mut conn, &uri, depth as i32).await?; 400 400 + let parents = crate::db::get_thread_parents(&mut conn, &uri, parent_height as i32).await?; 423 401 424 402 let reply_uris = replies.iter().map(|item| item.at_uri.clone()).collect(); 425 403 let parent_uris = parents.iter().map(|item| item.at_uri.clone()).collect();

+3

parakeet/src/xrpc/app_bsky/mod.rs

··· 6 6 mod feed; 7 7 mod graph; 8 8 mod labeler; 9 9 + mod unspecced; 9 10 10 11 #[rustfmt::skip] 11 12 pub fn routes() -> Router<crate::GlobalState> { ··· 64 65 // TODO: app.bsky.notification.putActivitySubscriptions 65 66 // TODO: app.bsky.notification.putPreferences 66 67 // TODO: app.bsky.notification.putPreferencesV2 68 68 + .route("/app.bsky.unspecced.getPostThreadV2", get(unspecced::thread_v2::get_post_thread_v2)) 69 69 + .route("/app.bsky.unspecced.getPostThreadOtherV2", get(unspecced::thread_v2::get_post_thread_other_v2)) 67 70 } 68 71 69 72 async fn not_implemented() -> axum::http::StatusCode {

+1

parakeet/src/xrpc/app_bsky/unspecced/mod.rs

··· 1 1 + pub mod thread_v2;

+382

parakeet/src/xrpc/app_bsky/unspecced/thread_v2.rs

··· 1 1 + use crate::db::ThreadItem; 2 2 + use crate::hydration::StatefulHydrator; 3 3 + use crate::xrpc::error::{Error, XrpcResult}; 4 4 + use crate::xrpc::extract::{AtpAcceptLabelers, AtpAuth}; 5 5 + use crate::xrpc::normalise_at_uri; 6 6 + use crate::GlobalState; 7 7 + use axum::extract::{Query, State}; 8 8 + use axum::Json; 9 9 + use itertools::Itertools; 10 10 + use lexica::app_bsky::feed::{BlockedAuthor, PostView, ThreadgateView}; 11 11 + use lexica::app_bsky::unspecced::{ThreadItemPost, ThreadV2Item, ThreadV2ItemType}; 12 12 + use serde::{Deserialize, Serialize}; 13 13 + use std::cmp::Ordering; 14 14 + use std::collections::{HashMap, HashSet}; 15 15 + 16 16 + const THREAD_PARENTS: usize = 50; 17 17 + const DEFAULT_BRANCHING: u32 = 10; 18 18 + const DEFAULT_DEPTH: u32 = 6; 19 19 + 20 20 + #[derive(Copy, Clone, Debug, Default, Deserialize)] 21 21 + #[serde(rename_all = "lowercase")] 22 22 + pub enum PostThreadSort { 23 23 + Newest, 24 24 + #[default] 25 25 + Oldest, 26 26 + Top, 27 27 + } 28 28 + 29 29 + #[derive(Debug, Deserialize)] 30 30 + #[serde(rename_all = "camelCase")] 31 31 + pub struct GetPostThreadV2Req { 32 32 + pub anchor: String, 33 33 + pub above: Option<bool>, 34 34 + pub below: Option<u32>, 35 35 + pub branching_factor: Option<u32>, 36 36 + #[serde(default)] 37 37 + pub sort: PostThreadSort, 38 38 + } 39 39 + 40 40 + #[derive(Debug, Serialize)] 41 41 + #[serde(rename_all = "camelCase")] 42 42 + pub struct GetPostThreadV2Res { 43 43 + pub thread: Vec<ThreadV2Item>, 44 44 + #[serde(skip_serializing_if = "Option::is_none")] 45 45 + pub threadgate: Option<ThreadgateView>, 46 46 + pub has_other_replies: bool, 47 47 + } 48 48 + 49 49 + pub async fn get_post_thread_v2( 50 50 + State(state): State<GlobalState>, 51 51 + AtpAcceptLabelers(labelers): AtpAcceptLabelers, 52 52 + maybe_auth: Option<AtpAuth>, 53 53 + Query(query): Query<GetPostThreadV2Req>, 54 54 + ) -> XrpcResult<Json<GetPostThreadV2Res>> { 55 55 + let mut conn = state.pool.get().await?; 56 56 + let maybe_did = maybe_auth.clone().map(|v| v.0); 57 57 + let hyd = StatefulHydrator::new(&state.dataloaders, &state.cdn, &labelers, maybe_auth); 58 58 + 59 59 + let uri = normalise_at_uri(&state.dataloaders, &query.anchor).await?; 60 60 + let depth = query.below.unwrap_or(DEFAULT_DEPTH).clamp(0, 20) as i32; 61 61 + let branching_factor = query 62 62 + .branching_factor 63 63 + .unwrap_or(DEFAULT_BRANCHING) 64 64 + .clamp(0, 100) as i32; 65 65 + 66 66 + let anchor = hyd 67 67 + .hydrate_post(uri.clone()) 68 68 + .await 69 69 + .ok_or(Error::not_found())?; 70 70 + 71 71 + if let Some(v) = &anchor.author.viewer { 72 72 + if v.blocked_by || v.blocking.is_some() { 73 73 + let block = ThreadV2ItemType::Blocked { 74 74 + author: BlockedAuthor { 75 75 + did: anchor.author.did, 76 76 + viewer: anchor.author.viewer, 77 77 + }, 78 78 + }; 79 79 + 80 80 + return Ok(Json(GetPostThreadV2Res { 81 81 + thread: vec![ThreadV2Item { 82 82 + uri, 83 83 + depth: 0, 84 84 + value: block, 85 85 + }], 86 86 + threadgate: anchor.threadgate, 87 87 + has_other_replies: false, 88 88 + })); 89 89 + } 90 90 + } 91 91 + 92 92 + // get the root post URI (if there is one) and return its author's DID. 93 93 + let root_uri = crate::db::get_root_post(&mut conn, &uri) 94 94 + .await? 95 95 + .unwrap_or(uri.clone()); 96 96 + let root_did = root_uri[5..].split('/').collect::<Vec<_>>()[0]; 97 97 + 98 98 + let replies = 99 99 + crate::db::get_thread_children_branching(&mut conn, &uri, depth, branching_factor + 1) 100 100 + .await?; 101 101 + let reply_uris = replies 102 102 + .iter() 103 103 + .map(|item| item.at_uri.clone()) 104 104 + .collect::<Vec<_>>(); 105 105 + 106 106 + // bluesky seems to use -50 atm. we get 1 extra to know if to set more_parents. 107 107 + let parents = match query.above.unwrap_or(true) { 108 108 + true => crate::db::get_thread_parents(&mut conn, &uri, THREAD_PARENTS as i32 + 1).await?, 109 109 + false => vec![], 110 110 + }; 111 111 + let parent_uris = parents 112 112 + .iter() 113 113 + .map(|item| item.at_uri.clone()) 114 114 + .collect::<Vec<_>>(); 115 115 + 116 116 + let (mut replies_hyd, mut parents_hyd) = tokio::join!( 117 117 + hyd.hydrate_posts(reply_uris), 118 118 + hyd.hydrate_posts(parent_uris), 119 119 + ); 120 120 + 121 121 + let threadgate = anchor.threadgate.clone(); 122 122 + let hidden: HashSet<_, std::hash::RandomState> = match &threadgate { 123 123 + Some(tg) => crate::db::get_threadgate_hiddens(&mut conn, &tg.uri).await?, 124 124 + None => None, 125 125 + } 126 126 + .map(|hiddens| HashSet::from_iter(Vec::from(hiddens))) 127 127 + .unwrap_or_default(); 128 128 + 129 129 + let root_has_more = parents.len() > THREAD_PARENTS; 130 130 + let mut is_op_thread = true; 131 131 + 132 132 + let mut thread = Vec::with_capacity(1 + replies.len() + parents.len()); 133 133 + 134 134 + thread.extend( 135 135 + parents 136 136 + .into_iter() 137 137 + .tail(THREAD_PARENTS) 138 138 + .enumerate() 139 139 + .map(|(idx, item)| { 140 140 + let value = parents_hyd 141 141 + .remove(&item.at_uri) 142 142 + .map(|post| { 143 143 + if let Some(v) = &post.author.viewer { 144 144 + if v.blocked_by || v.blocking.is_some() { 145 145 + return ThreadV2ItemType::Blocked { 146 146 + author: BlockedAuthor { 147 147 + did: post.author.did, 148 148 + viewer: post.author.viewer, 149 149 + }, 150 150 + }; 151 151 + } 152 152 + } 153 153 + 154 154 + let op_thread = (is_op_thread 155 155 + || item.root_uri.is_none() && item.parent_uri.is_none()) 156 156 + && post.author.did == root_did; 157 157 + 158 158 + ThreadV2ItemType::Post(ThreadItemPost { 159 159 + post, 160 160 + more_parents: idx == 0 && root_has_more, 161 161 + more_replies: 0, 162 162 + op_thread, 163 163 + hidden_by_threadgate: false, 164 164 + muted_by_viewer: false, 165 165 + }) 166 166 + }) 167 167 + .unwrap_or(ThreadV2ItemType::NotFound {}); 168 168 + 169 169 + ThreadV2Item { 170 170 + uri: item.at_uri, 171 171 + depth: -item.depth - 1, 172 172 + value, 173 173 + } 174 174 + }), 175 175 + ); 176 176 + 177 177 + is_op_thread = is_op_thread && anchor.author.did == root_did; 178 178 + thread.push(ThreadV2Item { 179 179 + uri: uri.clone(), 180 180 + depth: 0, 181 181 + value: ThreadV2ItemType::Post(ThreadItemPost { 182 182 + post: anchor, 183 183 + more_parents: false, 184 184 + more_replies: 0, 185 185 + op_thread: is_op_thread, 186 186 + hidden_by_threadgate: false, 187 187 + muted_by_viewer: false, 188 188 + }), 189 189 + }); 190 190 + 191 191 + let mut replies_grouped = replies 192 192 + .into_iter() 193 193 + .into_group_map_by(|item| item.parent_uri.clone().unwrap_or_default()); 194 194 + 195 195 + // start with the anchor 196 196 + let (children, has_other_replies) = build_thread_children( 197 197 + &mut replies_grouped, 198 198 + &mut replies_hyd, 199 199 + &hidden, 200 200 + &uri, 201 201 + is_op_thread, 202 202 + 1, 203 203 + &BuildThreadChildrenOpts { 204 204 + root_did, 205 205 + sort: query.sort, 206 206 + maybe_did: &maybe_did, 207 207 + max_depth: depth, 208 208 + }, 209 209 + ); 210 210 + thread.extend(children); 211 211 + 212 212 + Ok(Json(GetPostThreadV2Res { 213 213 + thread, 214 214 + threadgate, 215 215 + has_other_replies, 216 216 + })) 217 217 + } 218 218 + 219 219 + #[derive(Debug, Deserialize)] 220 220 + #[serde(rename_all = "camelCase")] 221 221 + pub struct GetPostThreadOtherV2Req { 222 222 + pub anchor: String, 223 223 + } 224 224 + 225 225 + #[derive(Debug, Serialize)] 226 226 + #[serde(rename_all = "camelCase")] 227 227 + pub struct GetPostThreadOtherV2Res { 228 228 + pub thread: Vec<ThreadV2Item>, 229 229 + } 230 230 + 231 231 + pub async fn get_post_thread_other_v2( 232 232 + State(state): State<GlobalState>, 233 233 + AtpAcceptLabelers(labelers): AtpAcceptLabelers, 234 234 + maybe_auth: Option<AtpAuth>, 235 235 + Query(query): Query<GetPostThreadOtherV2Req>, 236 236 + ) -> XrpcResult<Json<GetPostThreadOtherV2Res>> { 237 237 + let mut conn = state.pool.get().await?; 238 238 + let hyd = StatefulHydrator::new(&state.dataloaders, &state.cdn, &labelers, maybe_auth); 239 239 + 240 240 + let uri = normalise_at_uri(&state.dataloaders, &query.anchor).await?; 241 241 + 242 242 + let root = crate::db::get_root_post(&mut conn, &uri) 243 243 + .await? 244 244 + .unwrap_or(uri.clone()); 245 245 + 246 246 + // this only returns immediate children (depth==1) where hiddenByThreadgate=TRUE 247 247 + let replies = crate::db::get_thread_children_hidden(&mut conn, &uri, &root).await?; 248 248 + let reply_uris = replies 249 249 + .into_iter() 250 250 + .map(|item| item.at_uri) 251 251 + .collect::<Vec<_>>(); 252 252 + let thread = hyd 253 253 + .hydrate_posts(reply_uris) 254 254 + .await 255 255 + .into_iter() 256 256 + .filter(|(_, post)| match &post.author.viewer { 257 257 + Some(viewer) if viewer.blocked_by || viewer.blocking.is_some() => false, 258 258 + _ => true, 259 259 + }) 260 260 + .map(|(uri, post)| { 261 261 + let post = ThreadItemPost { 262 262 + post, 263 263 + more_parents: false, 264 264 + more_replies: 0, 265 265 + op_thread: false, 266 266 + hidden_by_threadgate: true, 267 267 + muted_by_viewer: false, 268 268 + }; 269 269 + 270 270 + ThreadV2Item { 271 271 + uri, 272 272 + depth: 1, 273 273 + value: ThreadV2ItemType::Post(post), 274 274 + } 275 275 + }) 276 276 + .collect(); 277 277 + 278 278 + Ok(Json(GetPostThreadOtherV2Res { thread })) 279 279 + } 280 280 + 281 281 + #[derive(Debug)] 282 282 + struct BuildThreadChildrenOpts<'a> { 283 283 + root_did: &'a str, 284 284 + sort: PostThreadSort, 285 285 + maybe_did: &'a Option<String>, 286 286 + max_depth: i32, 287 287 + } 288 288 + 289 289 + fn build_thread_children( 290 290 + grouped_replies: &mut HashMap<String, Vec<ThreadItem>>, 291 291 + replies_hyd: &mut HashMap<String, PostView>, 292 292 + hidden: &HashSet<String>, 293 293 + parent: &str, 294 294 + is_op_thread: bool, 295 295 + depth: i32, 296 296 + opts: &BuildThreadChildrenOpts, 297 297 + ) -> (Vec<ThreadV2Item>, bool) { 298 298 + let mut has_other_replies = false; 299 299 + 300 300 + let Some(replies) = grouped_replies.remove(parent) else { 301 301 + return (Vec::default(), has_other_replies); 302 302 + }; 303 303 + 304 304 + let replies = replies 305 305 + .into_iter() 306 306 + .filter_map(|item| replies_hyd.remove(&item.at_uri)) 307 307 + .sorted_by(sort_replies(&opts.sort)); 308 308 + 309 309 + let mut out = Vec::new(); 310 310 + 311 311 + for post in replies { 312 312 + let reply_count = grouped_replies 313 313 + .get(&post.uri) 314 314 + .map(|v| v.len()) 315 315 + .unwrap_or_default(); 316 316 + let at_max = depth == opts.max_depth; 317 317 + let more_replies = if at_max { reply_count } else { 0 }; 318 318 + let op_thread = is_op_thread && post.author.did == opts.root_did; 319 319 + 320 320 + // shouldn't push to the thread if there's a block relation. Bsky doesn't push a type of Blocked for replies... 321 321 + if let Some(v) = &post.author.viewer { 322 322 + if v.blocked_by || v.blocking.is_some() { 323 323 + continue; 324 324 + } 325 325 + } 326 326 + 327 327 + // check if the post is hidden AND we're NOT the author (hidden posts still show for their author) 328 328 + if hidden.contains(&post.uri) && !did_is_cur(opts.maybe_did, &post.author.did) { 329 329 + // post is hidden - do not ~pass go~ push to the thread. 330 330 + if depth == 1 { 331 331 + has_other_replies = true; 332 332 + } 333 333 + continue; 334 334 + } 335 335 + 336 336 + let uri = post.uri.clone(); 337 337 + out.push(ThreadV2Item { 338 338 + uri: post.uri.clone(), 339 339 + depth, 340 340 + value: ThreadV2ItemType::Post(ThreadItemPost { 341 341 + post, 342 342 + more_parents: false, 343 343 + more_replies: more_replies as i32, 344 344 + op_thread, 345 345 + hidden_by_threadgate: false, 346 346 + muted_by_viewer: false, 347 347 + }), 348 348 + }); 349 349 + 350 350 + if !at_max { 351 351 + // we don't care about has_other_replies when recursing 352 352 + let (children, _) = build_thread_children( 353 353 + grouped_replies, 354 354 + replies_hyd, 355 355 + hidden, 356 356 + &uri, 357 357 + op_thread, 358 358 + depth + 1, 359 359 + opts, 360 360 + ); 361 361 + 362 362 + out.extend(children); 363 363 + } 364 364 + } 365 365 + 366 366 + (out, has_other_replies) 367 367 + } 368 368 + 369 369 + fn sort_replies(sort: &PostThreadSort) -> impl Fn(&PostView, &PostView) -> Ordering + use<'_> { 370 370 + move |a: &PostView, b: &PostView| match sort { 371 371 + PostThreadSort::Newest => b.indexed_at.cmp(&a.indexed_at), 372 372 + PostThreadSort::Oldest => a.indexed_at.cmp(&b.indexed_at), 373 373 + PostThreadSort::Top => b.stats.like_count.cmp(&a.stats.like_count), 374 374 + } 375 375 + } 376 376 + 377 377 + fn did_is_cur(cur: &Option<String>, did: &String) -> bool { 378 378 + match cur { 379 379 + Some(cur) => did == cur, 380 380 + None => false, 381 381 + } 382 382 + }

+1

parakeet-db/src/models.rs

··· 431 431 pub sort_at: DateTime<Utc>, 432 432 } 433 433 434 434 + pub use not_null_vec::TextArray; 434 435 mod not_null_vec { 435 436 use diesel::deserialize::FromSql; 436 437 use diesel::pg::Pg;