Merge remote-tracking branch 'mia/main' into experiments

parakeet-db/src/models.rs

···
 
     pub content: String,
     pub facets: Option<serde_json::Value>,
-    pub languages: Vec<Option<String>>,
-    pub tags: Vec<Option<String>>,
+    pub languages: not_null_vec::TextArray,
+    pub tags: not_null_vec::TextArray,
 
     pub parent_uri: Option<String>,
     pub parent_cid: Option<String>,
···
     pub embed: Option<String>,
     pub embed_subtype: Option<String>,
 
-    pub mentions: Option<Vec<Option<String>>>,
+    pub mentions: Option<not_null_vec::TextArray>,
     pub violates_threadgate: bool,
 
     pub created_at: DateTime<Utc>,
···
     pub cid: String,
     pub post_uri: String,
 
-    pub detached: Vec<Option<String>>,
-    pub rules: Vec<Option<String>>,
+    pub detached: not_null_vec::TextArray,
+    pub rules: not_null_vec::TextArray,
 
     pub created_at: DateTime<Utc>,
     pub indexed_at: NaiveDateTime,
···
     pub cid: String,
     pub post_uri: String,
 
-    pub hidden_replies: Vec<Option<String>>,
-    pub allow: Option<Vec<Option<String>>>,
-    pub allowed_lists: Option<Vec<Option<String>>>,
+    pub hidden_replies: not_null_vec::TextArray,
+    pub allow: Option<not_null_vec::TextArray>,
+    pub allowed_lists: Option<not_null_vec::TextArray>,
 
     pub record: serde_json::Value,
 
···
     pub description: Option<String>,
     pub description_facets: Option<serde_json::Value>,
     pub list: String,
-    pub feeds: Option<Vec<Option<String>>>,
+    pub feeds: Option<not_null_vec::TextArray>,
 
     pub created_at: DateTime<Utc>,
     pub indexed_at: NaiveDateTime,
···
     pub did: String,
     pub cid: String,
 
-    pub reasons: Option<Vec<Option<String>>>,
-    pub subject_types: Option<Vec<Option<String>>>,
-    pub subject_collections: Option<Vec<Option<String>>>,
+    pub reasons: Option<not_null_vec::TextArray>,
+    pub subject_types: Option<not_null_vec::TextArray>,
+    pub subject_collections: Option<not_null_vec::TextArray>,
 
     pub created_at: NaiveDateTime,
     pub indexed_at: NaiveDateTime,
···
     pub subject: String,
     pub subject_cid: Option<String>,
     pub subject_type: String,
-    pub tags: Vec<Option<String>>,
+    pub tags: not_null_vec::TextArray,
     pub created_at: DateTime<Utc>,
 }
 
···
     pub typ: String,
     pub sort_at: DateTime<Utc>,
 }
+
+mod not_null_vec {
+    use diesel::deserialize::FromSql;
+    use diesel::pg::Pg;
+    use diesel::sql_types::{Array, Nullable, Text};
+    use diesel::{deserialize, FromSqlRow};
+    use serde::{Deserialize, Serialize};
+    use std::ops::{Deref, DerefMut};
+
+    #[derive(Clone, Debug, Default, Serialize, Deserialize, FromSqlRow)]
+    #[diesel(sql_type = Array<Nullable<Text>>)]
+    pub struct TextArray(pub Vec<String>);
+
+    impl FromSql<Array<Nullable<Text>>, Pg> for TextArray {
+        fn from_sql(bytes: diesel::pg::PgValue<'_>) -> deserialize::Result<Self> {
+            let vec_with_nulls =
+                <Vec<Option<String>> as FromSql<Array<Nullable<Text>>, Pg>>::from_sql(bytes)?;
+            Ok(TextArray(vec_with_nulls.into_iter().flatten().collect()))
+        }
+    }
+
+    impl Deref for TextArray {
+        type Target = Vec<String>;
+
+        fn deref(&self) -> &Self::Target {
+            &self.0
+        }
+    }
+
+    impl DerefMut for TextArray {
+        fn deref_mut(&mut self) -> &mut Self::Target {
+            &mut self.0
+        }
+    }
+
+    impl From<TextArray> for Vec<String> {
+        fn from(v: TextArray) -> Vec<String> {
+            v.0
+        }
+    }
+}

parakeet/src/hydration/labeler.rs

···
     likes: Option<i32>,
 ) -> LabelerViewDetailed {
     let reason_types = labeler.reasons.map(|v| {
-        v.into_iter()
-            .flatten()
-            .filter_map(|v| ReasonType::from_str(&v).ok())
+        v.iter()
+            .filter_map(|v| ReasonType::from_str(v).ok())
             .collect()
     });
 
···
         })
         .collect();
     let subject_types = labeler.subject_types.map(|v| {
-        v.into_iter()
-            .flatten()
-            .filter_map(|v| SubjectType::from_str(&v).ok())
+        v.iter()
+            .filter_map(|v| SubjectType::from_str(v).ok())
             .collect()
     });
-    let subject_collections = labeler
-        .subject_collections
-        .map(|v| v.into_iter().flatten().collect());
+    let subject_collections = labeler.subject_collections.map(Vec::from);
 
     LabelerViewDetailed {
         uri: format!("at://{}/app.bsky.labeler.service/self", labeler.did),

parakeet/src/hydration/posts/mod.rs

···
     ) -> Option<ThreadgateView> {
         let threadgate = threadgate?;
 
-        let lists = threadgate
-            .allowed_lists
-            .as_ref()
-            .map_or_else(Vec::new, |allowed_lists| {
-                allowed_lists.iter().flatten().cloned().collect()
-            });
+        let lists = match threadgate.allowed_lists.as_ref() {
+            Some(allowed_lists) => allowed_lists.clone().into(),
+            None => Vec::new(),
+        };
         let lists = self.hydrate_lists_basic(lists).await;
 
         Some(build_threadgate_view(
···
     ) -> HashMap<String, ThreadgateView> {
         let lists = threadgates.iter().fold(Vec::new(), |mut acc, c| {
             if let Some(lists) = &c.allowed_lists {
-                acc.extend(lists.iter().flatten().cloned());
+                acc.extend(lists.clone().0);
             }
             acc
         });
···
         threadgates
             .into_iter()
             .map(|threadgate| {
-                let this_lists =
-                    threadgate
-                        .allowed_lists
-                        .as_ref()
-                        .map_or_else(Vec::new, |allowed_lists| {
-                            allowed_lists
-                                .iter()
-                                .filter_map(|v| {
-                                    let v = v.clone()?;
-                                    lists.get(&v).cloned()
-                                })
-                                .collect()
-                        });
+                let this_lists = match &threadgate.allowed_lists {
+                    Some(allowed_lists) => allowed_lists
+                        .iter()
+                        .filter_map(|v| lists.get(v).cloned())
+                        .collect(),
+                    None => Vec::new(),
+                };
 
                 (
                     threadgate.at_uri.clone(),

parakeet/src/hydration/starter_packs.rs

···
         let feeds = sp
             .feeds
             .clone()
-            .unwrap_or_default()
-            .into_iter()
-            .flatten()
-            .collect();
-        let feeds = self.hydrate_feedgens(feeds).await.into_values().collect();
+            .unwrap_or_default();
+        let feeds = self.hydrate_feedgens(feeds.into()).await.into_values().collect();
 
         Some(build_spview(sp, creator, labels, list, feeds))
     }
···
         let feeds = packs
             .values()
             .filter_map(|pack| pack.feeds.clone())
-            .flat_map(|feeds| feeds.into_iter().flatten())
+            .flat_map(Vec::from)
             .collect();
 
         let creators = self.hydrate_profiles_basic(creators).await;
···
                 let list = lists.get(&pack.list).cloned();
                 let feeds = pack.feeds.as_ref().map(|v| {
                     v.iter()
-                        .flatten()
                         .filter_map(|feed| feeds.get(feed).cloned())
                         .collect()
                 });

parakeet/src/loaders/post.rs

···
 use crate::db;
 use dataloader::BatchFn;
+use diesel::dsl::sql;
 use diesel::prelude::*;
 use diesel_async::pooled_connection::deadpool::Pool;
 use diesel_async::AsyncPgConnection;
···
 
         let res = diesel_async::RunQueryDsl::load(
             schema::posts::table
-                .left_join(schema::threadgates::table)
+                .left_join(schema::threadgates::table.on(
+                    schema::threadgates::post_uri.eq(sql("coalesce(posts.root_uri, posts.at_uri)")),
+                ))
                 .select((
                     models::Post::as_select(),
                     Option::<models::Threadgate>::as_select(),

parakeet/src/xrpc/community_lexicon/bookmarks.rs

···
         .into_iter()
         .map(|bookmark| Bookmark {
             subject: bookmark.subject,
-            tags: bookmark.tags.into_iter().flatten().collect(),
+            tags: bookmark.tags.into(),
             created_at: bookmark.created_at,
         })
         .collect();