Install remove-dup-image into Docker image

Reviewed-on: https://git.joinplu.me/Plume/Plume/pulls/912

CHANGELOG.md

@@ -24,6 +24,8 @@
 
 - Percent-encode URI for remote_interact (#866, #857)
 - Menu animation not opening on iOS (#876, #897)
+- Make actors subscribe to channel once (#913)
+- Upsert posts and media instead of trying to insert and fail (#912)
 
 ## [[0.6.0]] - 2020-12-29
 

Cargo.lock

@@ -1,5 +1,7 @@
 # This file is automatically @generated by Cargo.
 # It is not intended for manual editing.
+version = 3
+
 [[package]]
 name = "activitypub"
 version = "0.1.6"
@@ -2037,13 +2039,13 @@ checksum = "f44db4199cdb049b494a92d105acbfa43c25b3925e33803923ba9580b7bc9e1a"
 
 [[package]]
 name = "lexical-core"
-version = "0.7.4"
+version = "0.7.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "db65c6da02e61f55dae90a0ae427b2a5f6b3e8db09f58d10efab23af92592616"
+checksum = "21f866863575d0e1d654fbeeabdc927292fdf862873dc3c96c6f753357e13374"
 dependencies = [
  "arrayvec",
  "bitflags 1.2.1",
- "cfg-if 0.1.10",
+ "cfg-if 1.0.0",
  "ryu",
  "static_assertions",
 ]

Dockerfile

@@ -40,6 +40,7 @@ WORKDIR /app
 COPY --from=builder /app /app
 COPY --from=builder /usr/local/cargo/bin/plm /bin/
 COPY --from=builder /usr/local/cargo/bin/plume /bin/
+COPY --from=builder /usr/local/cargo/bin/remove-dup-images /bin/
 
 CMD ["plume"]
 

migrations/postgres/2021-02-23-153402_medias_index_file_path/down.sql

@@ -0,0 +1 @@
+DROP INDEX medias_index_file_path;

migrations/postgres/2021-02-23-153402_medias_index_file_path/up.sql

@@ -0,0 +1 @@
+CREATE INDEX medias_index_file_path ON medias (file_path);

migrations/sqlite/2021-02-23-153402_medias_index_file_path/down.sql

@@ -0,0 +1 @@
+DROP INDEX medias_index_file_path;

migrations/sqlite/2021-02-23-153402_medias_index_file_path/up.sql

@@ -0,0 +1 @@
+CREATE INDEX medias_index_file_path ON medias (file_path);

plume-cli/Cargo.toml

@@ -8,6 +8,10 @@ edition = "2018"
 name = "plm"
 path = "src/main.rs"
 
+[[bin]]
+name = "remove-dup-images"
+path = "src/remove-dup-images.rs"
+
 [dependencies]
 clap = "2.33"
 dotenv = "0.14"

plume-cli/src/remove-dup-images.rs

@@ -0,0 +1,149 @@
+use diesel::{
+    BoolExpressionMethods, Connection, ExpressionMethods, JoinOnDsl, NullableExpressionMethods,
+    QueryDsl, RunQueryDsl,
+};
+use plume_models::{
+    blogs::Blog, instance::Instance, medias::Media, posts::Post, Connection as Conn, CONFIG,
+};
+use std::collections::hash_map::{DefaultHasher, HashMap};
+use std::fs::File;
+use std::hash::Hasher;
+use std::io::{BufReader, Read};
+use std::path::Path;
+
+fn main() {
+    match dotenv::dotenv() {
+        Ok(path) => eprintln!("Configuration read from {}", path.display()),
+        Err(ref e) if e.not_found() => eprintln!("no .env was found"),
+        e => e.map(|_| ()).unwrap(),
+    }
+    let conn = Conn::establish(CONFIG.database_url.as_str()).expect("extablish connection");
+    Instance::cache_local(&conn);
+    let covers = get_remote_post_covers(&conn);
+    let remote_media_hashes = calculate_remote_media_hashes(covers);
+    eprintln!("remote medias: {:?}", remote_media_hashes);
+    let orphan_medias = get_orphan_medias(&conn);
+    eprintln!("{:?} orphan media(s)", orphan_medias.len());
+    for media in orphan_medias {
+        match calculate_file_hash(&Path::new(&media.file_path)) {
+            Some(hash) => {
+                match remote_media_hashes.get(&hash) {
+                    Some(file_path) => {
+                        eprintln!(
+                            "File already referred. Removes only medias record. {}",
+                            &file_path
+                        );
+                        // Remove medias record
+                        diesel::delete(&media)
+                            .execute(&conn)
+                            .expect("Delete medias record");
+                    }
+                    None => {
+                        eprintln!("Removes {}", &media.file_path);
+                        // Remove file and medias record
+                        media.delete(&conn).expect("Delete media record and file");
+                    }
+                }
+            }
+            None => {
+                eprintln!(
+                    "File doesn't exist. Removes medias record. medias.id: {}, path: {}",
+                    &media.id, &media.file_path
+                );
+                diesel::delete(&media)
+                    .execute(&conn)
+                    .expect("Delete medias record");
+            }
+        }
+    }
+}
+
+fn get_remote_post_covers(conn: &Conn) -> Vec<Media> {
+    use plume_models::schema::blogs;
+    use plume_models::schema::posts;
+
+    let remote_instances = Instance::get_remotes(&conn).expect("get remote instances");
+    let remote_instance_ids = remote_instances.iter().map(|instance| instance.id);
+    let remote_blogs = blogs::table
+        .filter(blogs::instance_id.eq_any(remote_instance_ids))
+        .load::<Blog>(conn)
+        .expect("remote blogs");
+    let remote_blog_ids = remote_blogs.iter().map(|blog| blog.id);
+    let remote_posts = posts::table
+        .filter(posts::blog_id.eq_any(remote_blog_ids))
+        .load::<Post>(conn)
+        .expect("remote posts");
+    remote_posts
+        .iter()
+        .filter_map(|post| post.cover_id)
+        .map(|cover_id| Media::get(conn, cover_id).expect("Media"))
+        .collect()
+}
+
+fn calculate_remote_media_hashes(medias: Vec<Media>) -> HashMap<u64, String> {
+    let mut media_hashes = HashMap::new();
+    for media in medias.iter() {
+        if let Some(hash) = calculate_file_hash(Path::new(&media.file_path)) {
+            let _ = media_hashes.insert(hash, media.file_path.clone());
+        }
+    }
+    media_hashes
+}
+
+fn calculate_file_hash(path: &Path) -> Option<u64> {
+    if !path.exists() {
+        return None;
+    }
+    let file = File::open(path).expect("open file");
+    let mut reader = BufReader::new(file);
+    let mut hasher = DefaultHasher::new();
+    let mut buffer = [0; 2048];
+
+    while let Ok(n) = reader.read(&mut buffer) {
+        hasher.write(&buffer);
+
+        if n == 0 {
+            break;
+        }
+    }
+    Some(hasher.finish())
+}
+
+fn get_orphan_medias(conn: &Conn) -> Vec<Media> {
+    use plume_models::schema::{self, medias};
+    use plume_models::schema::{blogs::dsl::blogs, posts::dsl::posts, users::dsl::users};
+    let query = medias::table
+        .select((
+            medias::id,
+            medias::file_path,
+            medias::alt_text,
+            medias::is_remote,
+            medias::remote_url,
+            medias::sensitive,
+            medias::content_warning,
+            medias::owner_id,
+        ))
+        .left_outer_join(users.on(schema::users::avatar_id.eq(medias::id.nullable())))
+        .left_outer_join(
+            blogs.on(schema::blogs::icon_id
+                .eq(medias::id.nullable())
+                .or(schema::blogs::banner_id.eq(medias::id.nullable()))),
+        )
+        .left_outer_join(posts.on(schema::posts::cover_id.eq(medias::id.nullable())))
+        .filter(
+            schema::users::avatar_id.is_null().and(
+                schema::blogs::icon_id.is_null().and(
+                    schema::blogs::banner_id.is_null().and(
+                        schema::posts::cover_id
+                            .is_null()
+                            .and(medias::is_remote.eq(false)),
+                    ),
+                ),
+            ),
+        );
+    eprintln!(
+        "query for orphan medias: {}",
+        diesel::debug_query::<_, _>(&query)
+    );
+    query.load::<Media>(conn).expect("Load orphan medias")
+}

plume-models/src/comments.rs

@@ -255,9 +255,7 @@ impl FromId<DbConn> for Comment {
                         .and_then(|m| {
                             let author = &Post::get(conn, comm.post_id)?.get_authors(conn)?[0];
                             let not_author = m.link_props.href_string()? != author.ap_url.clone();
-                            Ok(Mention::from_activity(
-                                conn, &m, comm.id, false, not_author,
-                            )?)
+                            Mention::from_activity(conn, &m, comm.id, false, not_author)
                         })
                         .ok();
                 }

plume-models/src/config.rs

@@ -164,11 +164,8 @@ impl Default for LogoConfig {
             };
             let mut custom_icons = env::vars()
                 .filter_map(|(var, val)| {
-                    if let Some(size) = var.strip_prefix("PLUME_LOGO_") {
-                        Some((size.to_owned(), val))
-                    } else {
-                        None
-                    }
+                    var.strip_prefix("PLUME_LOGO_")
+                        .map(|size| (size.to_owned(), val))
                 })
                 .filter_map(|(var, val)| var.parse::<u64>().ok().map(|var| (var, val)))
                 .map(|(dim, src)| Icon {

plume-models/src/lists.rs

@@ -285,7 +285,8 @@ impl List {
             .select(list_elems::word)
             .load::<Option<String>>(conn)
             .map_err(Error::from)
-            .map(|r| r.into_iter().filter_map(|o| o).collect::<Vec<String>>())
+            // .map(|r| r.into_iter().filter_map(|o| o).collect::<Vec<String>>())
+            .map(|r| r.into_iter().flatten().collect::<Vec<String>>())
     }
 
     pub fn clear(&self, conn: &Connection) -> Result<()> {

plume-models/src/medias.rs

@@ -12,14 +12,14 @@ use plume_common::{
 };
 use std::{
     fs::{self, DirBuilder},
-    path::{Path, PathBuf},
+    path::{self, Path, PathBuf},
 };
 use tracing::warn;
 use url::Url;
 
 const REMOTE_MEDIA_DIRECTORY: &str = "remote";
 
-#[derive(Clone, Identifiable, Queryable)]
+#[derive(Clone, Identifiable, Queryable, AsChangeset)]
 pub struct Media {
     pub id: i32,
     pub file_path: String,
@@ -65,6 +65,7 @@ impl MediaCategory {
 impl Media {
     insert!(medias, NewMedia);
     get!(medias);
+    find_by!(medias, find_by_file_path, file_path as &str);
 
     pub fn for_user(conn: &Connection, owner: i32) -> Result<Vec<Media>> {
         medias::table
@@ -155,12 +156,11 @@ impl Media {
         if self.is_remote {
             Ok(self.remote_url.clone().unwrap_or_default())
         } else {
-            let p = Path::new(&self.file_path);
-            let filename: String = p.file_name().unwrap().to_str().unwrap().to_owned();
+            let file_path = self.file_path.replace(path::MAIN_SEPARATOR, "/");
             Ok(ap_url(&format!(
-                "{}/static/media/{}",
+                "{}/{}",
                 Instance::get_local()?.public_domain,
-                &filename
+                &file_path
             )))
         }
     }
@@ -224,32 +224,65 @@ impl Media {
         .copy_to(&mut dest)
         .ok()?;
 
-        // TODO: upsert
-        Media::insert(
-            conn,
-            NewMedia {
-                file_path: path.to_str()?.to_string(),
-                alt_text: image.object_props.content_string().ok()?,
-                is_remote: false,
-                remote_url: None,
-                sensitive: image.object_props.summary_string().is_ok(),
-                content_warning: image.object_props.summary_string().ok(),
-                owner_id: User::from_id(
+        Media::find_by_file_path(conn, &path.to_str()?)
+            .and_then(|mut media| {
+                let mut updated = false;
+
+                let alt_text = image.object_props.content_string().ok()?;
+                let sensitive = image.object_props.summary_string().is_ok();
+                let content_warning = image.object_props.summary_string().ok();
+                if media.alt_text != alt_text {
+                    media.alt_text = alt_text;
+                    updated = true;
+                }
+                if media.is_remote {
+                    media.is_remote = false;
+                    updated = true;
+                }
+                if media.remote_url.is_some() {
+                    media.remote_url = None;
+                    updated = true;
+                }
+                if media.sensitive != sensitive {
+                    media.sensitive = sensitive;
+                    updated = true;
+                }
+                if media.content_warning != content_warning {
+                    media.content_warning = content_warning;
+                    updated = true;
+                }
+                if updated {
+                    diesel::update(&media).set(&media).execute(&**conn)?;
+                }
+                Ok(media)
+            })
+            .or_else(|_| {
+                Media::insert(
                     conn,
-                    image
-                        .object_props
-                        .attributed_to_link_vec::<Id>()
-                        .ok()?
-                        .into_iter()
-                        .next()?
-                        .as_ref(),
-                    None,
-                    CONFIG.proxy(),
+                    NewMedia {
+                        file_path: path.to_str()?.to_string(),
+                        alt_text: image.object_props.content_string().ok()?,
+                        is_remote: false,
+                        remote_url: None,
+                        sensitive: image.object_props.summary_string().is_ok(),
+                        content_warning: image.object_props.summary_string().ok(),
+                        owner_id: User::from_id(
+                            conn,
+                            image
+                                .object_props
+                                .attributed_to_link_vec::<Id>()
+                                .ok()?
+                                .into_iter()
+                                .next()?
+                                .as_ref(),
+                            None,
+                            CONFIG.proxy(),
+                        )
+                        .map_err(|(_, e)| e)?
+                        .id,
+                    },
                 )
-                .map_err(|(_, e)| e)?
-                .id,
-            },
-        )
+            })
     }
 
     pub fn get_media_processor<'a>(conn: &'a Connection, user: Vec<&User>) -> MediaProcessor<'a> {

plume-models/src/posts.rs

@@ -15,7 +15,7 @@ use heck::KebabCase;
 use once_cell::sync::Lazy;
 use plume_common::{
     activity_pub::{
-        inbox::{AsObject, FromId},
+        inbox::{AsActor, AsObject, FromId},
         Hashtag, Id, IntoId, Licensed, Source, PUBLIC_VISIBILITY,
     },
     utils::md_to_html,
@@ -94,7 +94,10 @@ impl Post {
         let post = Self::get(conn, self.id)?;
         // TODO: Call publish_published() when newly published
         if post.published {
-            self.publish_updated();
+            let blog = post.get_blog(conn);
+            if blog.is_ok() && blog.unwrap().is_local() {
+                self.publish_updated();
+            }
         }
         Ok(post)
     }
@@ -443,13 +446,7 @@ impl Post {
                     m,
                 )
             })
-            .filter_map(|(id, m)| {
-                if let Some(id) = id {
-                    Some((m, id))
-                } else {
-                    None
-                }
-            })
+            .filter_map(|(id, m)| id.map(|id| (m, id)))
             .collect::<Vec<_>>();
 
         let old_mentions = Mention::list_for_post(&conn, self.id)?;
@@ -644,37 +641,85 @@ impl FromId<DbConn> for Post {
             .and_then(|img| Media::from_activity(conn, &img).ok().map(|m| m.id));
 
         let title = article.object_props.name_string()?;
-        // TODO: upsert
-        let post = Post::insert(
-            conn,
-            NewPost {
-                blog_id: blog?.id,
-                slug: title.to_kebab_case(),
-                title,
-                content: SafeString::new(&article.object_props.content_string()?),
-                published: true,
-                license,
-                // FIXME: This is wrong: with this logic, we may use the display URL as the AP ID. We need two different fields
-                ap_url: article
-                    .object_props
-                    .url_string()
-                    .or_else(|_| article.object_props.id_string())?,
-                creation_date: Some(article.object_props.published_utctime()?.naive_utc()),
-                subtitle: article.object_props.summary_string()?,
-                source: article.ap_object_props.source_object::<Source>()?.content,
-                cover_id: cover,
-            },
-        )?;
+        let ap_url = article
+            .object_props
+            .url_string()
+            .or_else(|_| article.object_props.id_string())?;
+        let post = Post::from_db(conn, &ap_url)
+            .and_then(|mut post| {
+                let mut updated = false;
 
-        for author in authors {
-            PostAuthor::insert(
-                conn,
-                NewPostAuthor {
-                    post_id: post.id,
-                    author_id: author.id,
-                },
-            )?;
-        }
+                let slug = title.to_kebab_case();
+                let content = SafeString::new(&article.object_props.content_string()?);
+                let subtitle = article.object_props.summary_string()?;
+                let source = article.ap_object_props.source_object::<Source>()?.content;
+                if post.slug != slug {
+                    post.slug = slug;
+                    updated = true;
+                }
+                if post.title != title {
+                    post.title = title.clone();
+                    updated = true;
+                }
+                if post.content != content {
+                    post.content = content;
+                    updated = true;
+                }
+                if post.license != license {
+                    post.license = license.clone();
+                    updated = true;
+                }
+                if post.subtitle != subtitle {
+                    post.subtitle = subtitle;
+                    updated = true;
+                }
+                if post.source != source {
+                    post.source = source;
+                    updated = true;
+                }
+                if post.cover_id != cover {
+                    post.cover_id = cover;
+                    updated = true;
+                }
+
+                if updated {
+                    post.update(conn)?;
+                }
+
+                Ok(post)
+            })
+            .or_else(|_| {
+                Post::insert(
+                    conn,
+                    NewPost {
+                        blog_id: blog?.id,
+                        slug: title.to_kebab_case(),
+                        title,
+                        content: SafeString::new(&article.object_props.content_string()?),
+                        published: true,
+                        license,
+                        // FIXME: This is wrong: with this logic, we may use the display URL as the AP ID. We need two different fields
+                        ap_url,
+                        creation_date: Some(article.object_props.published_utctime()?.naive_utc()),
+                        subtitle: article.object_props.summary_string()?,
+                        source: article.ap_object_props.source_object::<Source>()?.content,
+                        cover_id: cover,
+                    },
+                )
+                .and_then(|post| {
+                    for author in authors {
+                        PostAuthor::insert(
+                            conn,
+                            NewPostAuthor {
+                                post_id: post.id,
+                                author_id: author.id,
+                            },
+                        )?;
+                    }
+
+                    Ok(post)
+                })
+            })?;
 
         // save mentions and tags
         let mut hashtags = md_to_html(&post.source, None, false, None)

plume-models/src/remote_fetch_actor.rs

@@ -17,24 +17,22 @@ pub struct RemoteFetchActor {
 
 impl RemoteFetchActor {
     pub fn init(conn: DbPool) {
-        ACTOR_SYS
+        let actor = ACTOR_SYS
             .actor_of_args::<RemoteFetchActor, _>("remote-fetch", conn)
             .expect("Failed to initialize remote fetch actor");
-    }
-}
 
-impl Actor for RemoteFetchActor {
-    type Msg = UserEvent;
-
-    fn pre_start(&mut self, ctx: &Context<Self::Msg>) {
         USER_CHAN.tell(
             Subscribe {
-                actor: Box::new(ctx.myself()),
+                actor: Box::new(actor),
                 topic: "*".into(),
             },
             None,
         )
     }
+}
+
+impl Actor for RemoteFetchActor {
+    type Msg = UserEvent;
 
     fn recv(&mut self, _ctx: &Context<Self::Msg>, msg: Self::Msg, _sender: Sender) {
         use UserEvent::*;

plume-models/src/safe_string.rs

@@ -156,7 +156,7 @@ impl<'de> Deserialize<'de> for SafeString {
     where
         D: Deserializer<'de>,
     {
-        Ok(deserializer.deserialize_string(SafeStringVisitor)?)
+        deserializer.deserialize_string(SafeStringVisitor)
     }
 }
 

plume-models/src/search/actor.rs

@@ -13,24 +13,22 @@ pub struct SearchActor {
 
 impl SearchActor {
     pub fn init(searcher: Arc<Searcher>, conn: DbPool) {
-        ACTOR_SYS
+        let actor = ACTOR_SYS
             .actor_of_args::<SearchActor, _>("search", (searcher, conn))
             .expect("Failed to initialize searcher actor");
-    }
-}
 
-impl Actor for SearchActor {
-    type Msg = PostEvent;
-
-    fn pre_start(&mut self, ctx: &Context<Self::Msg>) {
         POST_CHAN.tell(
             Subscribe {
-                actor: Box::new(ctx.myself()),
+                actor: Box::new(actor),
                 topic: "*".into(),
             },
             None,
         )
     }
+}
+
+impl Actor for SearchActor {
+    type Msg = PostEvent;
 
     fn recv(&mut self, _ctx: &Context<Self::Msg>, msg: Self::Msg, _sender: Sender) {
         use PostEvent::*;

plume-models/src/timeline/query.rs

@@ -601,11 +601,10 @@ fn parse_l<'a, 'b>(stream: &'b [Token<'a>]) -> QueryResult<(&'b [Token<'a>], Lis
 }
 
 fn parse_m<'a, 'b>(mut stream: &'b [Token<'a>]) -> QueryResult<(&'b [Token<'a>], Vec<&'a str>)> {
-    let mut res: Vec<&str> = Vec::new();
-    res.push(match stream.get(0)? {
+    let mut res: Vec<&str> = vec![match stream.get(0)? {
         Token::Word(_, _, w) => w,
         t => return t.get_error(Token::Word(0, 0, "any word")),
-    });
+    }];
     stream = &stream[1..];
     while let Token::Comma(_) = stream[0] {
         res.push(match stream.get(1)? {

plume-models/src/users.rs

@@ -486,10 +486,7 @@ impl User {
             .filter_map(|j| serde_json::from_value(j.clone()).ok())
             .collect::<Vec<T>>();
 
-        let next = match json.get("next") {
-            Some(x) => Some(x.as_str().unwrap().to_owned()),
-            None => None,
-        };
+        let next = json.get("next").map(|x| x.as_str().unwrap().to_owned());
         Ok((items, next))
     }
     pub fn fetch_outbox<T: Activity>(&self) -> Result<Vec<T>> {

rust-toolchain

@@ -1 +1 @@
-nightly-2021-01-15
+nightly-2021-03-25

snap/snapcraft.yaml

@@ -25,7 +25,7 @@ parts:
   plume:
     plugin: rust
     source: .
-    rust-revision: nightly-2020-01-15
+    rust-revision: nightly-2021-03-25
     build-packages:
       - libssl-dev
       - pkg-config

src/routes/blogs.rs

@@ -348,7 +348,7 @@ pub fn update(
 #[get("/~/<name>/outbox")]
 pub fn outbox(name: String, conn: DbConn) -> Option<ActivityStream<OrderedCollection>> {
     let blog = Blog::find_by_fqn(&conn, &name).ok()?;
-    Some(blog.outbox(&conn).ok()?)
+    blog.outbox(&conn).ok()
 }
 #[allow(unused_variables)]
 #[get("/~/<name>/outbox?<page>")]
@@ -358,7 +358,7 @@ pub fn outbox_page(
     conn: DbConn,
 ) -> Option<ActivityStream<OrderedCollectionPage>> {
     let blog = Blog::find_by_fqn(&conn, &name).ok()?;
-    Some(blog.outbox_page(&conn, page.limits()).ok()?)
+    blog.outbox_page(&conn, page.limits()).ok()
 }
 #[get("/~/<name>/atom.xml")]
 pub fn atom_feed(name: String, conn: DbConn) -> Option<Content<String>> {