feat: Feeds

Cargo.lock

···
  "multibase",
  "parakeet-db",
  "parakeet-index",
+ "reqwest",
  "serde",
  "serde_json",
  "tokio",

lexica/src/app_bsky/feed.rs

···
 use crate::app_bsky::richtext::FacetMain;
 use crate::com_atproto::label::Label;
 use chrono::prelude::*;
-use serde::Serialize;
+use serde::{Deserialize, Serialize};
 use std::str::FromStr;
 
 #[derive(Clone, Debug, Serialize)]
···
     pub reply: Option<ReplyRef>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub reason: Option<FeedViewPostReason>,
-    // #[serde(skip_serializing_if = "Option::is_none")]
-    // pub feed_context: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub feed_context: Option<String>,
 }
 
 #[derive(Debug, Serialize)]
···
 #[serde(tag = "$type")]
 pub enum FeedViewPostReason {
     #[serde(rename = "app.bsky.feed.defs#reasonRepost")]
-    Repost {
-        by: ProfileViewBasic,
-        #[serde(rename = "indexedAt")]
-        indexed_at: DateTime<Utc>,
-    },
+    Repost(FeedReasonRepost),
     #[serde(rename = "app.bsky.feed.defs#reasonPin")]
     Pin,
+}
+
+#[derive(Debug, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct FeedReasonRepost {
+    pub by: ProfileViewBasic,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub uri: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub cid: Option<String>,
+    pub indexed_at: DateTime<Utc>,
 }
 
 #[derive(Debug, Serialize)]
···
     pub created_at: DateTime<Utc>,
     pub indexed_at: DateTime<Utc>,
 }
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct FeedSkeletonResponse {
+    pub feed: Vec<SkeletonFeedPost>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub cursor: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub req_id: Option<String>,
+}
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct SkeletonFeedPost {
+    pub post: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub reason: Option<SkeletonReason>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub feed_context: Option<String>,
+}
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+#[serde(tag = "$type")]
+pub enum SkeletonReason {
+    #[serde(rename = "app.bsky.feed.defs#skeletonReasonPin")]
+    Pin {},
+    #[serde(rename = "app.bsky.feed.defs#skeletonReasonRepost")]
+    Repost {
+        repost: String,
+    },
+}

parakeet/Cargo.toml

···
 multibase = "0.9.1"
 parakeet-db = { path = "../parakeet-db" }
 parakeet-index = { path = "../parakeet-index" }
+reqwest = { version = "0.12", features = ["json"] }
 serde = { version = "1.0.217", features = ["derive"] }
 serde_json = "1.0.134"
 tokio = { version = "1.42.0", features = ["full"] }

parakeet/src/hydration/posts.rs

···
                         post,
                         reply,
                         reason: None,
+                        feed_context: None,
                     },
                 ))
             })

parakeet/src/main.rs

···
 pub struct GlobalState {
     pub pool: Pool<AsyncPgConnection>,
     pub dataloaders: Arc<loaders::Dataloaders>,
+    pub resolver: Arc<did_resolver::Resolver>,
     pub index_client: parakeet_index::Client,
     pub jwt: Arc<xrpc::jwt::JwtVerifier>,
     pub cdn: Arc<xrpc::cdn::BskyCdn>,
···
         pool.clone(),
         index_client.clone(),
     ));
-    let resolver = did_resolver::Resolver::new(did_resolver::ResolverOpts {
+    let resolver = Arc::new(did_resolver::Resolver::new(did_resolver::ResolverOpts {
         plc_directory: conf.plc_directory,
         ..Default::default()
-    })?;
+    })?);
     let jwt = Arc::new(xrpc::jwt::JwtVerifier::new(
         conf.service.did.clone(),
-        resolver,
+        resolver.clone(),
     ));
 
     let cdn = Arc::new(xrpc::cdn::BskyCdn::new(conf.cdn.base, conf.cdn.video_base));
···
         .with_state(GlobalState {
             pool,
             dataloaders,
+            resolver,
             index_client,
             jwt,
             cdn,

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

···
 use crate::xrpc::{check_actor_status, datetime_cursor, get_actor_did, normalise_at_uri};
 use crate::GlobalState;
 use axum::extract::{Query, State};
+use axum::http::StatusCode;
 use axum::Json;
 use axum_extra::extract::Query as ExtraQuery;
+use axum_extra::headers::authorization::Bearer;
+use axum_extra::headers::Authorization;
+use axum_extra::TypedHeader;
+use chrono::prelude::*;
 use diesel::prelude::*;
-use diesel_async::RunQueryDsl;
+use diesel_async::{AsyncPgConnection, RunQueryDsl};
 use lexica::app_bsky::actor::ProfileView;
 use lexica::app_bsky::feed::{
-    FeedViewPost, PostView, ThreadViewPost, ThreadViewPostType, ThreadgateView,
+    FeedReasonRepost, FeedSkeletonResponse, FeedViewPost, FeedViewPostReason, PostView,
+    SkeletonReason, ThreadViewPost, ThreadViewPostType, ThreadgateView,
 };
 use parakeet_db::schema;
+use reqwest::Url;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 
-// TODO: getFeed: once we get auth!
+const FEEDGEN_SERVICE_ID: &str = "#bsky_fg";
+
+#[derive(Debug, Serialize)]
+pub struct FeedRes {
+    #[serde(skip_serializing_if = "Option::is_none")]
+    cursor: Option<String>,
+    feed: Vec<FeedViewPost>,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct GetFeedQuery {
+    pub feed: String,
+    pub limit: Option<u8>,
+    pub cursor: Option<String>,
+}
+
+pub async fn get_feed(
+    State(state): State<GlobalState>,
+    // we have to use Bearer because the tokens come with `aud` set to the feedgen did.
+    AtpAcceptLabelers(labelers): AtpAcceptLabelers,
+    maybe_tok: Option<TypedHeader<Authorization<Bearer>>>,
+    Query(query): Query<GetFeedQuery>,
+) -> XrpcResult<Json<FeedRes>> {
+    let mut conn = state.pool.get().await?;
+
+    // first, look up the feedgen
+    let service_did: String = schema::feedgens::table
+        .select(schema::feedgens::service_did)
+        .find(&query.feed)
+        .get_result(&mut conn)
+        .await?;
+
+    // resolve the did
+    let did_doc = match state.resolver.resolve_did(&service_did).await {
+        Ok(Some(did_doc)) => did_doc,
+        Ok(None) => return Err(Error::invalid_request(None)),
+        Err(err) => {
+            tracing::error!(
+                feedgen = service_did,
+                "failed to resolve feedgen service did: {err}"
+            );
+            return Err(Error::invalid_request(None));
+        }
+    };
+
+    // find the service
+    let Some(service) = did_doc.find_service_by_id(FEEDGEN_SERVICE_ID) else {
+        tracing::error!(
+            feedgen = service_did,
+            "DID doc didn't contain BskyFeedGenerator service"
+        );
+        return Err(Error::invalid_request(None));
+    };
+
+    let endpoint = service.service_endpoint.clone();
+    let skeleton = get_feed_skeleton(
+        &query.feed,
+        &endpoint,
+        maybe_tok.as_ref(),
+        query.limit,
+        query.cursor,
+    )
+    .await?;
+
+    let maybe_auth = match maybe_tok {
+        Some(hdr) => {
+            match state
+                .jwt
+                .resolve_and_verify_jwt(hdr.token(), Some(&service_did))
+                .await
+            {
+                Some(claims) => match &state.did_allowlist {
+                    Some(allowlist) if !allowlist.contains(&claims.iss) => {
+                        return Err(Error::new(
+                            StatusCode::FORBIDDEN,
+                            "forbidden".to_string(),
+                            None,
+                        ));
+                    }
+                    _ => Some(AtpAuth(claims.iss)),
+                },
+                None => None,
+            }
+        }
+        None => None,
+    };
+
+    let hyd = StatefulHydrator::new(&state.dataloaders, &state.cdn, &labelers, maybe_auth);
+
+    let at_uris = skeleton.feed.iter().map(|v| v.post.clone()).collect();
+    let repost_skeleton = skeleton
+        .feed
+        .iter()
+        .filter_map(|v| match &v.reason {
+            Some(SkeletonReason::Repost { repost }) => Some(repost.clone()),
+            _ => None,
+        })
+        .collect::<Vec<_>>();
+
+    let mut posts = hyd.hydrate_feed_posts(at_uris).await;
+    let mut repost_data = get_skeleton_repost_data(&mut conn, &hyd, repost_skeleton).await;
+
+    let feed = skeleton
+        .feed
+        .into_iter()
+        .filter_map(|item| {
+            let mut post = posts.remove(&item.post)?;
+            let reason = match item.reason {
+                Some(SkeletonReason::Repost { repost }) => {
+                    repost_data.remove(&repost).map(FeedViewPostReason::Repost)
+                }
+                Some(SkeletonReason::Pin {}) => Some(FeedViewPostReason::Pin),
+                _ => None,
+            };
+
+            post.reason = reason;
+            post.feed_context = item.feed_context;
+
+            Some(post)
+        })
+        .collect();
+
+    Ok(Json(FeedRes {
+        cursor: skeleton.cursor,
+        feed,
+    }))
+}
 
 #[derive(Debug, Deserialize)]
 #[serde(rename_all = "snake_case")]
···
     pub filter: GetAuthorFeedFilter,
     #[serde(default)]
     pub include_pins: bool,
-}
-
-#[derive(Debug, Serialize)]
-pub struct FeedRes {
-    #[serde(skip_serializing_if = "Option::is_none")]
-    cursor: Option<String>,
-    feed: Vec<FeedViewPost>,
 }
 
 pub async fn get_author_feed(
···
         .eq_any(filter)
         .or(schema::posts::embed_subtype.eq_any(filter))
 }
+
+async fn get_feed_skeleton(
+    feed: &str,
+    service: &str,
+    maybe_tok: Option<&TypedHeader<Authorization<Bearer>>>,
+    limit: Option<u8>,
+    cursor: Option<String>,
+) -> XrpcResult<FeedSkeletonResponse> {
+    let mut params = vec![("feed", feed.to_string())];
+
+    if let Some(cursor) = cursor {
+        params.push(("cursor", cursor));
+    }
+    if let Some(limit) = limit {
+        params.push(("limit", limit.to_string()));
+    }
+    let url = Url::parse_with_params(
+        &format!("{service}/xrpc/app.bsky.feed.getFeedSkeleton"),
+        params,
+    )
+    .unwrap();
+
+    let mut req = reqwest::Client::new().get(url);
+    if let Some(auth) = maybe_tok {
+        req = req.bearer_auth(auth.token());
+    }
+
+    match req.send().await {
+        Ok(skeleton) => match skeleton.json().await {
+            Ok(skeleton) => Ok(skeleton),
+            Err(err) => {
+                tracing::error!("Failed to parse feed skeleton: {err}");
+                Err(Error::server_error(Some("Failed to fetch feed skeleton")))
+            }
+        },
+        Err(err) => {
+            tracing::error!("Failed to fetch feed skeleton: {err}");
+            Err(Error::server_error(Some("Failed to fetch feed skeleton")))
+        }
+    }
+}
+
+async fn get_skeleton_repost_data<'a>(
+    conn: &mut AsyncPgConnection,
+    hyd: &StatefulHydrator<'a>,
+    reposts: Vec<String>,
+) -> HashMap<String, FeedReasonRepost> {
+    let Ok(repost_data) = schema::records::table
+        .select((
+            schema::records::at_uri,
+            schema::records::did,
+            schema::records::indexed_at,
+        ))
+        .filter(schema::records::at_uri.eq_any(&reposts))
+        .get_results::<(String, String, NaiveDateTime)>(conn)
+        .await
+    else {
+        return HashMap::new();
+    };
+
+    let profiles = repost_data.iter().map(|(_, did, _)| did.clone()).collect();
+    let profiles = hyd.hydrate_profiles_basic(profiles).await;
+
+    repost_data
+        .into_iter()
+        .filter_map(|(uri, did, indexed_at)| {
+            let by = profiles.get(&did).cloned()?;
+
+            let repost = FeedReasonRepost {
+                by,
+                uri: Some(uri.clone()),
+                cid: None, // okay, we do have this, but the app doesn't seem to be bothered about not setting it.
+                indexed_at: indexed_at.and_utc(),
+            };
+
+            Some((uri, repost))
+        })
+        .collect()
+}

parakeet/src/xrpc/app_bsky/mod.rs

···
         .route("/app.bsky.feed.getActorFeeds", get(feed::feedgen::get_actor_feeds))
         .route("/app.bsky.feed.getActorLikes", get(feed::likes::get_actor_likes))
         .route("/app.bsky.feed.getAuthorFeed", get(feed::posts::get_author_feed))
+        .route("/app.bsky.feed.getFeed", get(feed::posts::get_feed))
         .route("/app.bsky.feed.getLikes", get(feed::likes::get_likes))
         .route("/app.bsky.feed.getListFeed", get(feed::posts::get_list_feed))
         .route("/app.bsky.feed.getPostThread", get(feed::posts::get_post_thread))

parakeet/src/xrpc/extract.rs

···
         .map_err(|err| (StatusCode::BAD_REQUEST, err.to_string()))?
         .ok_or((StatusCode::UNAUTHORIZED, "missing JWT".to_string()))?;
 
-        match state.jwt.resolve_and_verify_jwt(hdr.token()).await {
+        match state.jwt.resolve_and_verify_jwt(hdr.token(), None).await {
             Some(claims) => match &state.did_allowlist {
                 Some(allowlist) if !allowlist.contains(&claims.iss) => {
                     Err((StatusCode::FORBIDDEN, "forbidden".to_string()))
···
             return Ok(None);
         };
 
-        match state.jwt.resolve_and_verify_jwt(hdr.token()).await {
+        match state.jwt.resolve_and_verify_jwt(hdr.token(), None).await {
             Some(claims) => match &state.did_allowlist {
                 Some(allowlist) if !allowlist.contains(&claims.iss) => {
                     Err((StatusCode::FORBIDDEN, "forbidden".to_string()))

parakeet/src/xrpc/jwt.rs

···
 use jsonwebtoken::{Algorithm, DecodingKey, Validation};
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
-use std::sync::LazyLock;
+use std::sync::{Arc, LazyLock};
 use tokio::sync::RwLock;
 
 static DUMMY_KEY: LazyLock<DecodingKey> = LazyLock::new(|| DecodingKey::from_secret(&[]));
···
 
 pub struct JwtVerifier {
     aud: String,
-    resolver: Resolver,
+    resolver: Arc<Resolver>,
     key_cache: RwLock<HashMap<String, String>>,
 }
 
 impl JwtVerifier {
-    pub fn new(aud: String, resolver: Resolver) -> Self {
+    pub fn new(aud: String, resolver: Arc<Resolver>) -> Self {
         JwtVerifier {
             aud,
             resolver,
···
         }
     }
 
-    pub async fn resolve_and_verify_jwt(&self, token: &str) -> Option<Claims> {
+    pub async fn resolve_and_verify_jwt(&self, token: &str, aud: Option<&str>) -> Option<Claims> {
         // first we need to decode without verifying, to get iss.
         let unsafe_data = jsonwebtoken::decode::<Claims>(token, &DUMMY_KEY, &NO_VERIFY).ok()?;
         let unsafe_iss = unsafe_data.claims.iss;
···
             None => self.resolve_key(&unsafe_iss).await?,
         };
 
-        self.verify_jwt_multibase_with_alg(token, &multibase_key, unsafe_data.header.alg)
+        let aud = aud.unwrap_or(&self.aud);
+        self.verify_jwt_multibase_with_alg(token, &multibase_key, unsafe_data.header.alg, aud)
     }
 
     async fn resolve_key(&self, did: &str) -> Option<String> {
···
     pub fn verify_jwt_multibase(&self, token: &str, multibase_key: &str) -> Option<Claims> {
         let alg = jsonwebtoken::decode_header(token).ok()?.alg;
 
-        self.verify_jwt_multibase_with_alg(token, multibase_key, alg)
+        self.verify_jwt_multibase_with_alg(token, multibase_key, alg, &self.aud)
     }
 
     pub fn verify_jwt_multibase_with_alg(
···
         token: &str,
         multibase_key: &str,
         alg: Algorithm,
+        aud: &str,
     ) -> Option<Claims> {
         // decode the multibase key
         let (_, key) = multibase::decode(multibase_key).ok()?;
···
         let key = DecodingKey::from_ec_der(&key[2..]);
 
         let mut validation = Validation::new(alg);
-        validation.set_audience(&[&self.aud]);
+        validation.set_audience(&[&aud]);
 
         let decoded = jsonwebtoken::decode::<Claims>(token, &key, &validation).ok()?;