Add support for video posts

rsky-feedgen/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "rsky-feedgen"
-version = "0.0.1"
+version = "0.1.0"
 authors = ["Rudy Fraser <[email protected]>"]
 description = "A framework for building AT Protocol feed generators, in Rust."
 license = "Apache-2.0"

rsky-feedgen/migrations/2024-09-12-133947_018/down.sql

@@ -0,0 +1,2 @@
+-- This file should undo anything in `up.sql`
+DROP TABLE public.video;

rsky-feedgen/migrations/2024-09-12-133947_018/up.sql

@@ -0,0 +1,15 @@
+-- Your SQL goes here
+CREATE TABLE IF NOT EXISTS public.video (
+    cid character varying NOT NULL,
+    alt character varying,
+    "postCid" character varying NOT NULL,
+    "postUri" character varying NOT NULL,
+    "createdAt" character varying NOT NULL,
+    "indexedAt" character varying NOT NULL,
+    labels TEXT []
+);
+
+ALTER TABLE ONLY public.video
+    DROP CONSTRAINT IF EXISTS video_pkey;
+ALTER TABLE ONLY public.video
+    ADD CONSTRAINT video_pkey PRIMARY KEY (cid);

rsky-feedgen/src/apis/mod.rs

@@ -496,6 +496,7 @@ pub async fn queue_creation(
 ) -> Result<(), String> {
     use crate::schema::follow::dsl as FollowSchema;
     use crate::schema::image::dsl as ImageSchema;
+    use crate::schema::video::dsl as VideoSchema;
     use crate::schema::like::dsl as LikeSchema;
     use crate::schema::membership::dsl as MembershipSchema;
     use crate::schema::post::dsl as PostSchema;
@@ -505,6 +506,7 @@ pub async fn queue_creation(
             let mut new_posts = Vec::new();
             let mut new_members = Vec::new();
             let mut new_images = Vec::new();
+            let mut new_videos = Vec::new();
             let mut members_to_rm = Vec::new();
 
             body
@@ -518,6 +520,7 @@ pub async fn queue_creation(
                     let mut post_text = String::new();
                     let mut post_text_original = String::new();
                     let mut post_images = Vec::new();
+                    let mut post_videos = Vec::new();
                     let mut new_post = Post {
                         uri: req.uri,
                         cid: req.cid,
@@ -570,6 +573,36 @@ pub async fn queue_creation(
                                         };
                                     }
                                 },
+                                Embeds::Video(ref e) => {
+                                    let labels: Vec<Option<String>> = vec![];
+                                    match (&e.video.cid, e.video.r#ref) {
+                                        (Some(video_cid), _) => {
+                                            let new_video = (
+                                                VideoSchema::cid.eq(video_cid.clone()),
+                                                VideoSchema::alt.eq(e.alt.clone()),
+                                                VideoSchema::postCid.eq(new_post.cid.clone()),
+                                                VideoSchema::postUri.eq(new_post.uri.clone()),
+                                                VideoSchema::indexedAt.eq(new_post.indexed_at.clone()),
+                                                VideoSchema::createdAt.eq(post_created_at.clone()),
+                                                VideoSchema::labels.eq(labels),
+                                            );
+                                            post_videos.push(new_video);
+                                        },
+                                        (_, Some(video_ref)) => {
+                                            let new_video = (
+                                                VideoSchema::cid.eq(video_ref.to_string()),
+                                                VideoSchema::alt.eq(e.alt.clone()),
+                                                VideoSchema::postCid.eq(new_post.cid.clone()),
+                                                VideoSchema::postUri.eq(new_post.uri.clone()),
+                                                VideoSchema::indexedAt.eq(new_post.indexed_at.clone()),
+                                                VideoSchema::createdAt.eq(post_created_at.clone()),
+                                                VideoSchema::labels.eq(labels),
+                                            );
+                                            post_videos.push(new_video);
+                                        },
+                                        _ => eprintln!("Unknown video type: {e:?}")
+                                    };
+                                }
                                 Embeds::RecordWithMedia(e) => {
                                     match e.media {
                                         MediaUnion::Images(m) => {
@@ -591,6 +624,36 @@ pub async fn queue_creation(
                                                 };
                                             }
                                         },
+                                        MediaUnion::Video(ref v) => {
+                                            let labels: Vec<Option<String>> = vec![];
+                                            match (&v.video.cid, v.video.r#ref) {
+                                                (Some(video_cid), _) => {
+                                                    let new_video = (
+                                                        VideoSchema::cid.eq(video_cid.clone()),
+                                                        VideoSchema::alt.eq(v.alt.clone()),
+                                                        VideoSchema::postCid.eq(new_post.cid.clone()),
+                                                        VideoSchema::postUri.eq(new_post.uri.clone()),
+                                                        VideoSchema::indexedAt.eq(new_post.indexed_at.clone()),
+                                                        VideoSchema::createdAt.eq(post_created_at.clone()),
+                                                        VideoSchema::labels.eq(labels),
+                                                    );
+                                                    post_videos.push(new_video);
+                                                },
+                                                (_, Some(video_ref)) => {
+                                                    let new_video = (
+                                                        VideoSchema::cid.eq(video_ref.to_string()),
+                                                        VideoSchema::alt.eq(v.alt.clone()),
+                                                        VideoSchema::postCid.eq(new_post.cid.clone()),
+                                                        VideoSchema::postUri.eq(new_post.uri.clone()),
+                                                        VideoSchema::indexedAt.eq(new_post.indexed_at.clone()),
+                                                        VideoSchema::createdAt.eq(post_created_at.clone()),
+                                                        VideoSchema::labels.eq(labels),
+                                                    );
+                                                    post_videos.push(new_video);
+                                                },
+                                                _ => eprintln!("Unknown video type: {v:?}")
+                                            };
+                                        }
                                         MediaUnion::External(e) => {
                                             new_post.external_uri = Some(e.external.uri);
                                             new_post.external_title = Some(e.external.title);
@@ -657,6 +720,7 @@ pub async fn queue_creation(
                         );
                         new_posts.push(new_post);
                         new_images.extend(post_images);
+                        new_videos.extend(post_videos);
 
                         if hashtags.contains("#addtoblacksky") && !is_member {
                             println!("New member: {:?}", &req.author);
@@ -704,6 +768,13 @@ pub async fn queue_creation(
                 .execute(conn)
                 .expect("Error inserting image records");
 
+            diesel::insert_into(VideoSchema::video)
+                .values(&new_videos)
+                .on_conflict(VideoSchema::cid)
+                .do_nothing()
+                .execute(conn)
+                .expect("Error inserting video records");
+
             diesel::insert_into(MembershipSchema::membership)
                 .values(&new_members)
                 .on_conflict((MembershipSchema::did,MembershipSchema::list))

rsky-feedgen/src/schema.rs

@@ -93,6 +93,18 @@ diesel::table! {
     }
 }
 
+diesel::table! {
+    video (cid) {
+        cid -> Varchar,
+        alt -> Nullable<Varchar>,
+        postCid -> Varchar,
+        postUri -> Varchar,
+        createdAt -> Varchar,
+        indexedAt -> Varchar,
+        labels -> Nullable<Array<Nullable<Text>>>,
+    }
+}
+
 diesel::table! {
     visitor (id) {
         id -> Int4,
@@ -112,5 +124,6 @@ diesel::allow_tables_to_appear_in_same_query!(
     membership,
     post,
     sub_state,
+    video,
     visitor,
 );

rsky-firehose/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "rsky-firehose"
-version = "0.0.1"
+version = "0.1.0"
 authors = ["Rudy Fraser <[email protected]>"]
 description = "A framework for subscribing to the AT Protocol firehose, in Rust."
 license = "Apache-2.0"

rsky-firehose/src/main.rs

@@ -282,7 +282,7 @@ async fn main() {
         match tokio_tungstenite::connect_async(
             Url::parse(
                 format!(
-                    "{}/xrpc/com.atproto.sync.subscribeRepos",
+                    "{}/xrpc/com.atproto.sync.subscribeRepos?cursor=1670190430",
                     default_subscriber_path
                 )
                 .as_str(),
@@ -298,6 +298,7 @@ async fn main() {
                     tokio::spawn(async move {
                         process(message, &client).await;
                     });
+                    thread::sleep(Duration::from_millis(8));
                 }
             }
             Err(error) => {

rsky-lexicon/src/app/bsky/embed/mod.rs

@@ -2,17 +2,21 @@ pub mod external;
 pub mod images;
 pub mod record;
 pub mod record_with_media;
+pub mod video;
 
 use crate::app::bsky::embed::external::{External, View as ExternalView};
 use crate::app::bsky::embed::images::{Images, View as ImagesView};
 use crate::app::bsky::embed::record::{Record, View as RecordView};
 use crate::app::bsky::embed::record_with_media::{RecordWithMedia, View as RecordWithMediaView};
+use crate::app::bsky::embed::video::{Video, View as VideoView};
 
 #[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
 #[serde(tag = "$type")]
 pub enum MediaUnion {
     #[serde(rename = "app.bsky.embed.images")]
     Images(Images),
+    #[serde(rename = "app.bsky.embed.video")]
+    Video(Video),
     #[serde(rename = "app.bsky.embed.external")]
     External(External),
 }
@@ -22,6 +26,8 @@ pub enum MediaUnion {
 pub enum MediaViewUnion {
     #[serde(rename = "app.bsky.embed.images#view")]
     ImagesView(ImagesView),
+    #[serde(rename = "app.bsky.embed.video#view")]
+    VideoView(VideoView),
     #[serde(rename = "app.bsky.embed.external#view")]
     ExternalView(ExternalView),
 }
@@ -31,7 +37,10 @@ pub enum MediaViewUnion {
 pub enum Embeds {
     #[serde(rename = "app.bsky.embed.images")]
     Images(Images),
-
+
+    #[serde(rename = "app.bsky.embed.video")]
+    Video(Video),
+
     #[serde(
         alias = "app.bsky.embed.external",
         alias = "app.bsky.embed.external#main"
@@ -50,6 +59,7 @@ pub enum Embeds {
 pub enum EmbedViews {
     ImagesView(ImagesView),
     ExternalView(ExternalView),
+    VideoView(VideoView),
     RecordView(RecordView),
     RecordWithMediaView(RecordWithMediaView),
 }

rsky-lexicon/src/app/bsky/embed/video.rs

@@ -0,0 +1,31 @@
+use crate::app::bsky::embed::images::AspectRatio;
+use crate::com::atproto::repo::Blob;
+
+#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct Video {
+    pub video: Blob,
+    pub captions: Option<Vec<Caption>>,
+    /// Alt text description of video image, for accessibility
+    pub alt: Option<String>,
+    pub aspect_ratio: Option<AspectRatio>,
+}
+
+
+#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
+pub struct Caption {
+    pub lang: String,
+    pub file: Blob,
+}
+
+#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
+#[serde(tag = "$type")]
+#[serde(rename = "app.bsky.embed.video#view")]
+#[serde(rename_all = "camelCase")]
+pub struct View {
+    pub cid: String,
+    pub playlist: String,
+    pub thumbnail: Option<String>,
+    pub alt: Option<String>,
+    pub aspect_ratio: Option<AspectRatio>
+}