Canonicalize watched ids and migrate history

Author: neckaros

src/model/episodes.rs

@@ -1,10 +1,10 @@
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 
 use async_recursion::async_recursion;
 use nanoid::nanoid;
 use query_external_ip::SourceError;
 use rs_plugin_common_interfaces::{
-    domain::rs_ids::RsIds,
+    domain::rs_ids::{ApplyRsIds, RsIds},
     lookup::{RsLookupEpisode, RsLookupMetadataResult, RsLookupQuery},
     ImageType, MediaType,
 };
@@ -32,6 +32,7 @@ use crate::{
 use super::{
     entity_search::merge_result_ids,
     error::{Error, Result},
+    history::{episode_history_id, history_id_rsids},
     medias::{RsSort, RsSortOrder},
     store::sql::SqlOrder,
     users::{ConnectedUser, HistoryQuery},
@@ -106,6 +107,22 @@ pub struct EpisodeForUpdate {
 }
 
 impl ModelController {
+    async fn episode_series_by_ref(
+        &self,
+        library_id: &str,
+        episodes: &[Episode],
+    ) -> RsResult<HashMap<String, Serie>> {
+        let store = self.store.get_library_store(library_id)?;
+        let serie_refs: HashSet<String> = episodes.iter().map(|episode| episode.serie.clone()).collect();
+        let mut series = HashMap::new();
+        for serie_ref in serie_refs {
+            if let Some(serie) = store.get_serie(&serie_ref).await? {
+                series.insert(serie_ref, serie.item);
+            }
+        }
+        Ok(series)
+    }
+
     async fn lookup_episodes_metadata(
         &self,
         library_id: &str,
@@ -245,27 +262,35 @@ impl ModelController {
         requesting_user: &ConnectedUser,
         library_id: Option<String>,
     ) -> RsResult<()> {
-        let ids: RsIds = episode.clone().into();
-        let watched = self
-            .get_watched(
-                HistoryQuery {
-                    types: vec![MediaType::Episode],
-                    id: Some(ids.clone()),
-                    ..Default::default()
-                },
-                requesting_user,
-                library_id.clone(),
-            )
-            .await?;
-        let progress = self
-            .get_view_progress(ids, requesting_user, library_id)
-            .await?;
-        if let Some(progress) = progress {
-            episode.progress = Some(progress.progress);
-        }
-        let watched = watched.first();
-        if let Some(watched) = watched {
-            episode.watched = Some(watched.date);
+        if let Some(library_id) = library_id {
+            let store = self.store.get_library_store(&library_id)?;
+            if let Some(serie) = store.get_serie(&episode.serie).await? {
+                let history_id = episode_history_id(&serie.item, episode);
+                let watched = self
+                    .get_watched(
+                        HistoryQuery {
+                            types: vec![MediaType::Episode],
+                            id: Some(history_id_rsids(history_id.clone())),
+                            ..Default::default()
+                        },
+                        requesting_user,
+                        Some(library_id.clone()),
+                    )
+                    .await?;
+                let progress = self
+                    .get_view_progress(
+                        history_id_rsids(history_id),
+                        requesting_user,
+                        Some(library_id),
+                    )
+                    .await?;
+                if let Some(progress) = progress {
+                    episode.progress = Some(progress.progress);
+                }
+                if let Some(watched) = watched.first() {
+                    episode.watched = Some(watched.date);
+                }
+            }
         }
         episode.fill_imdb_ratings(&self.imdb).await;
         Ok(())
@@ -276,6 +301,11 @@ impl ModelController {
         requesting_user: &ConnectedUser,
         library_id: Option<String>,
     ) -> RsResult<()> {
+        let series_by_ref = if let Some(library_id) = library_id.as_deref() {
+            self.episode_series_by_ref(library_id, episodes).await?
+        } else {
+            HashMap::new()
+        };
         let watched = self
             .get_watched(
                 HistoryQuery {
@@ -304,15 +334,12 @@ impl ModelController {
             .collect::<HashMap<_, _>>();
 
         for episode in episodes {
-            let ids = RsIds::from(episode.clone());
-            let ids_string: Vec<String> = ids.into();
-            for id in ids_string {
-                let watch = watched.get(&id);
-                if let Some(watch) = watch {
+            if let Some(serie) = series_by_ref.get(&episode.serie) {
+                let history_id = episode_history_id(serie, episode);
+                if let Some(watch) = watched.get(&history_id) {
                     episode.watched = Some(*watch);
                 }
-                let progress = progresses.get(&id);
-                if let Some(progress) = progress {
+                if let Some(progress) = progresses.get(&history_id) {
                     episode.progress = Some(*progress);
                 }
             }
@@ -520,6 +547,20 @@ impl ModelController {
         let ids = self
             .get_serie_ids(library_id, serie_id, requesting_user)
             .await?;
+        let existing_episodes = self
+            .get_episodes(
+                library_id,
+                EpisodeQuery {
+                    serie_ref: Some(serie_id.to_string()),
+                    ..Default::default()
+                },
+                requesting_user,
+            )
+            .await?;
+        let existing_ids_by_episode: HashMap<(u32, u32), RsIds> = existing_episodes
+            .into_iter()
+            .map(|episode| ((episode.season, episode.number), RsIds::from(episode)))
+            .collect();
         let mut all_episodes = self
             .lookup_episodes_metadata(library_id, serie_id, &ids, requesting_user)
             .await?;
@@ -530,6 +571,15 @@ impl ModelController {
             )))
             .into());
         }
+        for episode in &mut all_episodes {
+            if let Some(existing_ids) =
+                existing_ids_by_episode.get(&(episode.season, episode.number))
+            {
+                let mut merged_ids = RsIds::from(episode.clone());
+                merged_ids.merge(existing_ids);
+                episode.apply_rs_ids(&merged_ids);
+            }
+        }
         let store = self.store.get_library_store_optional(library_id).ok_or(
             Error::LibraryStoreNotFoundFor(
                 library_id.clone().to_string(),

src/model/history.rs

@@ -0,0 +1,184 @@
+use std::collections::{HashMap, HashSet};
+
+use rs_plugin_common_interfaces::{domain::rs_ids::RsIds, MediaType};
+
+use crate::domain::{episode::Episode, library::LibraryType, movie::Movie, serie::Serie};
+use crate::tools::log::{log_info, LogServiceType};
+
+use super::{
+    episodes::EpisodeQuery,
+    movies::MovieQuery,
+    series::SerieQuery,
+    store::sql::users::{HistoryIdRewrite, ProgressIdRewrite},
+    ModelController,
+};
+
+fn series_history_key(serie: &Serie) -> String {
+    if let Some(imdb) = &serie.imdb {
+        format!("imdb/{imdb}")
+    } else if let Some(tmdb) = serie.tmdb {
+        format!("tmdb/{tmdb}")
+    } else if let Some(tvdb) = serie.tvdb {
+        format!("tvdb/{tvdb}")
+    } else {
+        format!("redseat/{}", serie.id)
+    }
+}
+
+pub fn series_history_id(serie: &Serie) -> String {
+    format!("series:{}", series_history_key(serie))
+}
+
+pub fn movie_history_id(movie: &Movie) -> String {
+    if let Some(imdb) = &movie.imdb {
+        format!("movie:imdb/{imdb}")
+    } else if let Some(tmdb) = movie.tmdb {
+        format!("movie:tmdb/{tmdb}")
+    } else {
+        format!("movie:redseat/{}", movie.id)
+    }
+}
+
+pub fn episode_history_id(serie: &Serie, episode: &Episode) -> String {
+    format!(
+        "episode:{}/{}/{}",
+        series_history_key(serie),
+        episode.season,
+        episode.number
+    )
+}
+
+pub fn history_id_rsids(id: String) -> RsIds {
+    RsIds::try_from(id).expect("history ids must stay in key:value format")
+}
+
+fn dedup_ids(ids: Vec<String>) -> Vec<String> {
+    let mut seen = HashSet::new();
+    ids.into_iter().filter(|id| seen.insert(id.clone())).collect()
+}
+
+fn movie_legacy_ids(movie: &Movie) -> Vec<String> {
+    let ids: Vec<String> = RsIds::from(movie.clone()).into();
+    dedup_ids(ids)
+}
+
+fn episode_legacy_ids(episode: &Episode) -> Vec<String> {
+    let ids: Vec<String> = RsIds::from(episode.clone()).into();
+    dedup_ids(ids)
+}
+
+impl ModelController {
+    pub async fn migrate_history_ids(&self) -> crate::Result<()> {
+        let watched = self.store.get_all_watched().await?;
+        let progress = self.store.get_all_view_progress_rows().await?;
+        let has_history = watched
+            .iter()
+            .any(|row| matches!(row.kind, MediaType::Movie | MediaType::Episode))
+            || progress
+                .iter()
+                .any(|row| matches!(row.kind, MediaType::Movie | MediaType::Episode));
+        if !has_history {
+            return Ok(());
+        }
+
+        let libraries = self.store.get_libraries().await?;
+        let mut movie_targets = HashMap::<String, String>::new();
+        let mut episode_targets = HashMap::<String, (String, String)>::new();
+
+        for library in libraries {
+            if !matches!(library.kind, LibraryType::Movies | LibraryType::Shows) {
+                continue;
+            }
+            let store = self.store.get_library_store(&library.id)?;
+
+            for movie in store.get_movies(MovieQuery::default()).await? {
+                let canonical_id = movie_history_id(&movie);
+                for legacy_id in movie_legacy_ids(&movie) {
+                    if legacy_id != canonical_id {
+                        movie_targets.entry(legacy_id).or_insert_with(|| canonical_id.clone());
+                    }
+                }
+            }
+
+            let series = store.get_series(SerieQuery::default()).await?;
+            let series_by_ref: HashMap<String, Serie> = series
+                .into_iter()
+                .map(|serie| (serie.item.id.clone(), serie.item))
+                .collect();
+            for episode in store.get_episodes(EpisodeQuery::default()).await? {
+                let Some(serie) = series_by_ref.get(&episode.serie) else {
+                    continue;
+                };
+                let canonical_id = episode_history_id(serie, &episode);
+                let parent_id = series_history_id(serie);
+                for legacy_id in episode_legacy_ids(&episode) {
+                    if legacy_id != canonical_id {
+                        episode_targets
+                            .entry(legacy_id)
+                            .or_insert_with(|| (canonical_id.clone(), parent_id.clone()));
+                    }
+                }
+            }
+        }
+
+        let watched_rewrites = watched
+            .into_iter()
+            .filter_map(|watched| match watched.kind {
+                MediaType::Movie => movie_targets.get(&watched.id).map(|new_id| HistoryIdRewrite {
+                    kind: watched.kind,
+                    old_id: watched.id,
+                    new_id: new_id.clone(),
+                    user_ref: watched.user_ref.unwrap_or_default(),
+                }),
+                MediaType::Episode => {
+                    episode_targets
+                        .get(&watched.id)
+                        .map(|(new_id, _)| HistoryIdRewrite {
+                            kind: watched.kind,
+                            old_id: watched.id,
+                            new_id: new_id.clone(),
+                            user_ref: watched.user_ref.unwrap_or_default(),
+                        })
+                }
+                _ => None,
+            })
+            .collect::<Vec<_>>();
+        let progress_rewrites = progress
+            .into_iter()
+            .filter_map(|progress| match progress.kind {
+                MediaType::Movie => movie_targets.get(&progress.id).map(|new_id| ProgressIdRewrite {
+                    kind: progress.kind,
+                    old_id: progress.id,
+                    new_id: new_id.clone(),
+                    new_parent: None,
+                    user_ref: progress.user_ref,
+                }),
+                MediaType::Episode => {
+                    episode_targets
+                        .get(&progress.id)
+                        .map(|(new_id, parent_id)| ProgressIdRewrite {
+                            kind: progress.kind,
+                            old_id: progress.id,
+                            new_id: new_id.clone(),
+                            new_parent: Some(parent_id.clone()),
+                            user_ref: progress.user_ref,
+                        })
+                }
+                _ => None,
+            })
+            .collect::<Vec<_>>();
+
+        let (watched_count, progress_count) = self
+            .store
+            .apply_history_rewrites(watched_rewrites, progress_rewrites)
+            .await?;
+        log_info(
+            LogServiceType::Database,
+            format!(
+                "History migration complete: watched rewrites={}, progress rewrites={}",
+                watched_count, progress_count
+            ),
+        );
+        Ok(())
+    }
+}

src/model/mod.rs

@@ -13,6 +13,7 @@ pub mod deleted;
 pub mod entity_images;
 pub mod entity_search;
 pub mod episodes;
+pub mod history;
 pub mod media_progresses;
 pub mod media_ratings;
 pub mod medias;
@@ -188,6 +189,7 @@ impl ModelController {
         });
 
         mc.cache_update_all_libraries().await?;
+        mc.migrate_history_ids().await?;
 
         let scheduler = &mc.scheduler;
         scheduler.start(mc.clone()).await?;