patch of test · round #0 · pull #2 · parakeet.at/parakeet

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;

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();

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 {

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
       +
       }

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;