Merge pull request #301 from euzu/feature/vod_series_wal_file_refactor

update wal file only when content could written.

src/processing/processor/xtream.rs

@@ -1,5 +1,5 @@
 use crate::tuliprox_error::{info_err, notify_err};
-use crate::tuliprox_error::{str_to_io_error, to_io_error, TuliproxError, TuliproxErrorKind};
+use crate::tuliprox_error::{str_to_io_error, TuliproxError, TuliproxErrorKind};
 use crate::model::{Config, ConfigInput};
 use crate::model::{FetchedPlaylist, PlaylistEntry, PlaylistItem, PlaylistItemType, XtreamCluster};
 use crate::model::normalize_release_date;
@@ -7,7 +7,6 @@ use crate::repository::storage::get_input_storage_path;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use std::fs::File;
-use std::io::{BufWriter, Write};
 use std::path::PathBuf;
 use std::sync::Arc;
 use crate::repository::bplustree::BPlusTree;
@@ -47,16 +46,6 @@ pub(in crate::processing) fn normalize_json_content(content: String) -> String {
     }
 }
 
-pub(in crate::processing) fn write_info_content_to_wal_file(writer: &mut BufWriter<&File>, provider_id: u32, content: &str) -> std::io::Result<()> {
-    let length = u32::try_from(content.len()).map_err(to_io_error)?;
-    if length > 0 {
-        writer.write_all(&provider_id.to_le_bytes())?;
-        writer.write_all(&length.to_le_bytes())?;
-        writer.write_all(content.as_bytes())?;
-    }
-    Ok(())
-}
-
 pub(in crate::processing) fn create_resolve_episode_wal_files(cfg: &Config, input: &ConfigInput) -> Option<(File, PathBuf)> {
     match get_input_storage_path(&input.name, &cfg.working_dir) {
         Ok(storage_path) => {

src/processing/processor/xtream_series.rs

@@ -3,9 +3,9 @@ use crate::model::{Config, ConfigTarget, InputType};
 use crate::model::{FetchedPlaylist, PlaylistGroup, PlaylistItem, PlaylistItemType, XtreamCluster};
 use crate::processing::processor::playlist::ProcessingPipe;
 use crate::processing::parser::xtream::parse_xtream_series_info;
-use crate::processing::processor::xtream::{create_resolve_episode_wal_files, create_resolve_info_wal_files, playlist_resolve_download_playlist_item, read_processed_info_ids, should_update_info, write_info_content_to_wal_file};
+use crate::processing::processor::xtream::{create_resolve_episode_wal_files, create_resolve_info_wal_files, playlist_resolve_download_playlist_item, read_processed_info_ids, should_update_info};
 use crate::repository::storage::get_input_storage_path;
-use crate::repository::xtream_repository::{xtream_get_info_file_paths, xtream_update_input_info_file, xtream_update_input_series_episodes_record_from_wal_file, xtream_update_input_series_record_from_wal_file};
+use crate::repository::xtream_repository::{write_series_info_to_wal_file, xtream_get_info_file_paths, xtream_update_input_info_file, xtream_update_input_series_episodes_record_from_wal_file, xtream_update_input_series_record_from_wal_file};
 use crate::repository::IndexedDocumentReader;
 use crate::tuliprox_error::{notify_err, info_err};
 use crate::processing::processor::{handle_error, handle_error_and_return, create_resolve_options_function_for_xtream_target};
@@ -26,16 +26,6 @@ async fn read_processed_series_info_ids(cfg: &Config, errors: &mut Vec<TuliproxE
     read_processed_info_ids(cfg, errors, fpl, PlaylistItemType::SeriesInfo, |ts: &u64| *ts).await
 }
 
-fn write_series_info_record_to_wal_file(
-    writer: &mut BufWriter<&File>,
-    provider_id: u32,
-    ts: u64,
-) -> std::io::Result<()> {
-    writer.write_all(&provider_id.to_le_bytes())?;
-    writer.write_all(&ts.to_le_bytes())?;
-    Ok(())
-}
-
 fn write_series_episode_record_to_wal_file(
     writer: &mut BufWriter<&File>,
     provider_id: u32,
@@ -89,11 +79,9 @@ async fn playlist_resolve_series_info(client: Arc<reqwest::Client>, cfg: &Config
         if should_update {
             if let Some(content) = playlist_resolve_download_playlist_item(Arc::clone(&client), pli, fpl.input, errors, resolve_delay, XtreamCluster::Series).await {
                 let normalized_content = normalize_json_content(content);
-                handle_error_and_return!(write_info_content_to_wal_file(&mut content_writer, provider_id, &normalized_content),
-                    |err| errors.push(notify_err!(format!("Failed to resolve series, could not write to content wal file {err}"))));
+                handle_error_and_return!(write_series_info_to_wal_file(provider_id, ts, &normalized_content, &mut content_writer, &mut record_writer),
+                        |err| errors.push(notify_err!(format!("Failed to resolve series, could not write to wal file {err}"))));
                 processed_info_ids.insert(provider_id, ts);
-                handle_error_and_return!(write_series_info_record_to_wal_file(&mut record_writer, provider_id, ts),
-                    |err| errors.push(notify_err!(format!("Failed to resolve series wal, could not write to record wal file {err}"))));
                 content_updated = true;
             }
         }
@@ -114,10 +102,12 @@ async fn playlist_resolve_series_info(client: Arc<reqwest::Client>, cfg: &Config
     if content_updated {
         handle_error!(content_writer.flush(),
             |err| errors.push(notify_err!(format!("Failed to resolve vod, could not write to wal file {err}"))));
-        drop(content_writer);
-        drop(wal_content_file);
         handle_error!(record_writer.flush(),
             |err| errors.push(notify_err!(format!("Failed to resolve vod tmdb, could not write to wal file {err}"))));
+        handle_error!(content_writer.get_ref().sync_all(), |err| errors.push(notify_err!(format!("Failed to sync series info to wal file {err}"))));
+        handle_error!(record_writer.get_ref().sync_all(), |err| errors.push(notify_err!(format!("Failed to sync series info record to wal file {err}"))));
+        drop(content_writer);
+        drop(wal_content_file);
         drop(record_writer);
         drop(wal_record_file);
         handle_error!(xtream_update_input_info_file(cfg, fpl.input, &wal_content_path, XtreamCluster::Series).await,
@@ -126,6 +116,7 @@ async fn playlist_resolve_series_info(client: Arc<reqwest::Client>, cfg: &Config
             |err| errors.push(err));
     }
 
+    // TODO better approach for transactional updates is multiplexed WAL file.
     // we updated now
     // - series_info.db  which contains the original series_info json
     // - series_record.db which contains the series_info provider_id and timestamp

src/processing/processor/xtream_vod.rs

@@ -1,22 +1,20 @@
 use crate::tuliprox_error::{TuliproxError, TuliproxErrorKind};
 use crate::model::{Config, ConfigTarget, InputType};
 use crate::model::{FetchedPlaylist, PlaylistItem, PlaylistItemType, XtreamCluster};
-use crate::processing::processor::xtream::{create_resolve_info_wal_files, playlist_resolve_download_playlist_item, read_processed_info_ids, should_update_info, write_info_content_to_wal_file};
-use crate::repository::xtream_repository::{xtream_update_input_info_file, xtream_update_input_vod_record_from_wal_file, InputVodInfoRecord};
+use crate::processing::processor::xtream::{create_resolve_info_wal_files, playlist_resolve_download_playlist_item, read_processed_info_ids, should_update_info};
+use crate::repository::xtream_repository::{write_vod_info_to_wal_file, xtream_update_input_info_file, xtream_update_input_vod_record_from_wal_file, InputVodInfoRecord};
 use crate::tuliprox_error::{notify_err};
 use crate::processing::processor::{handle_error, handle_error_and_return, create_resolve_options_function_for_xtream_target};
 use crate::utils::{get_u32_from_serde_value, get_u64_from_serde_value, get_string_from_serde_value};
 use crate::repository::xtream_repository::xtream_get_input_info;
 use serde_json::{from_str, Map, Value};
 use std::collections::HashMap;
-use std::fs::File;
-use std::io::{BufWriter, Write};
+use std::io::{Write};
 use std::sync::Arc;
 use std::time::Instant;
 use log::{info, log_enabled, Level};
 use crate::utils;
 use crate::processing::processor::xtream::normalize_json_content;
-use crate::tuliprox_error::to_io_error;
 
 create_resolve_options_function_for_xtream_target!(vod);
 
@@ -55,30 +53,6 @@ fn extract_info_record_from_vod_info(content: &str) -> Option<(u32, InputVodInfo
     }))
 }
 
-fn write_vod_info_record_to_wal_file(
-    writer: &mut BufWriter<&File>,
-    provider_id: u32,
-    record: &InputVodInfoRecord,
-) -> std::io::Result<()> {
-    writer.write_all(&provider_id.to_le_bytes())?;
-    writer.write_all(&record.tmdb_id.to_le_bytes())?;
-    writer.write_all(&record.ts.to_le_bytes())?;
-
-    match &record.release_date {
-        Some(date_str) => {
-            // Write the length as a 4-byte u32 to prevent overflow.
-            let len = u32::try_from(date_str.len()).map_err(to_io_error)?;
-            writer.write_all(&len.to_le_bytes())?;
-            writer.write_all(date_str.as_bytes())?;
-        }
-        None => {
-            // Write a length of 0 as 4 bytes.
-            writer.write_all(&0u32.to_le_bytes())?;
-        }
-    }
-    Ok(())
-}
-
 fn should_update_vod_info(pli: &mut PlaylistItem, processed_provider_ids: &HashMap<u32, u64>) -> (bool, u32, u64) {
     should_update_info(pli, processed_provider_ids, crate::model::XC_TAG_VOD_INFO_ADDED)
 }
@@ -119,11 +93,9 @@ pub async fn playlist_resolve_vod(client: Arc<reqwest::Client>, cfg: &Config, ta
                 let normalized_content = normalize_json_content(content);
                 if let Some((provider_id, info_record)) = extract_info_record_from_vod_info(&normalized_content) {
                     let ts = info_record.ts;
-                    handle_error_and_return!(write_info_content_to_wal_file(&mut content_writer, provider_id, &normalized_content),
-                        |err| errors.push(notify_err!(format!("Failed to resolve vod, could not write to content wal file {err}"))));
+                    handle_error_and_return!(write_vod_info_to_wal_file(provider_id, &normalized_content, &info_record, &mut content_writer, &mut record_writer),
+                        |err| errors.push(notify_err!(format!("Failed to resolve vod, could not write to wal file {err}"))));
                     processed_info_ids.insert(provider_id, ts);
-                    handle_error_and_return!(write_vod_info_record_to_wal_file(&mut record_writer, provider_id, &info_record),
-                        |err| errors.push(notify_err!(format!("Failed to resolve vod wal, could not write to record wal file {err}"))));
                     content_updated = true;
                 }
             }
@@ -141,10 +113,14 @@ pub async fn playlist_resolve_vod(client: Arc<reqwest::Client>, cfg: &Config, ta
         info!("resolved {processed_vod_info_count}/{vod_info_count} vod info");
     }
     if content_updated {
+        // TODO better approach for transactional updates is multiplexed WAL file.
+
         handle_error!(content_writer.flush(),
             |err| errors.push(notify_err!(format!("Failed to resolve vod, could not write to wal file {err}"))));
         handle_error!(record_writer.flush(),
             |err| errors.push(notify_err!(format!("Failed to resolve vod tmdb, could not write to wal file {err}"))));
+        handle_error!(content_writer.get_ref().sync_all(), |err| errors.push(notify_err!(format!("Failed to sync vod info to wal file {err}"))));
+        handle_error!(record_writer.get_ref().sync_all(), |err| errors.push(notify_err!(format!("Failed to sync vod info record to wal file {err}"))));
         drop(content_writer);
         drop(record_writer);
         drop(wal_content_file);

src/repository/xtream_repository.rs

@@ -1,4 +1,4 @@
-use crate::tuliprox_error::{create_tuliprox_error, create_tuliprox_error_result, info_err, notify_err, str_to_io_error, TuliproxError, TuliproxErrorKind};
+use crate::tuliprox_error::{create_tuliprox_error, create_tuliprox_error_result, info_err, notify_err, str_to_io_error, to_io_error, TuliproxError, TuliproxErrorKind};
 use crate::model::{ProxyUserCredentials};
 use crate::model::{Config, ConfigInput, ConfigTarget, XtreamTargetOutput};
 use crate::model::{PlaylistEntry, PlaylistGroup, PlaylistItem, PlaylistItemType, XtreamCluster, XtreamPlaylistItem};
@@ -26,11 +26,10 @@ use serde_json::{json, Map, Value};
 use std::collections::HashMap;
 use std::fs;
 use std::fs::File;
-use std::io::{BufReader, Error, ErrorKind, Read};
+use std::io::{BufReader, BufWriter, Error, ErrorKind, Read, Write};
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
 
-
 macro_rules! cant_write_result {
     ($path:expr, $err:expr) => {
         create_tuliprox_error!(
@@ -982,3 +981,67 @@ pub(crate) async fn xtream_get_playlist_categories(config: &Config, target_name:
     }
     None
 }
+
+pub fn write_series_info_to_wal_file(provider_id: u32, ts: u64, content: &str, content_write: &mut BufWriter<&File>, record_writer: &mut BufWriter<&File>) -> std::io::Result<()> {
+
+    let encoded_content = encode_info_content_for_wal_file(provider_id, content)?;
+    let encoded_record = encode_series_info_record_for_wal_file(provider_id, ts);
+    content_write.write_all(&encoded_content)?;
+    record_writer.write_all(&encoded_record)?;
+    Ok(())
+}
+
+
+pub fn write_vod_info_to_wal_file(provider_id: u32, content: &str, info_record: &InputVodInfoRecord, content_write: &mut BufWriter<&File>, record_writer: &mut BufWriter<&File>) -> std::io::Result<()> {
+
+    let encoded_content = encode_info_content_for_wal_file(provider_id, content)?;
+    let encoded_record = encode_vod_info_record_for_wal_file(provider_id, info_record)?;
+    content_write.write_all(&encoded_content)?;
+    record_writer.write_all(&encoded_record)?;
+    Ok(())
+}
+
+
+fn encode_info_content_for_wal_file(provider_id: u32, content: &str) -> std::io::Result<Vec<u8>> {
+    let length = u32::try_from(content.len()).map_err(to_io_error)?;
+    let mut buffer = Vec::with_capacity(8 + content.len());
+
+    buffer.extend_from_slice(&provider_id.to_le_bytes());
+    buffer.extend_from_slice(&length.to_le_bytes());
+    buffer.extend_from_slice(content.as_bytes());
+
+    Ok(buffer)
+}
+
+
+fn encode_vod_info_record_for_wal_file(provider_id: u32, record: &InputVodInfoRecord) -> std::io::Result<Vec<u8>> {
+    let estimated_capacity = 30;
+    let mut buffer = Vec::with_capacity(estimated_capacity);
+
+    buffer.extend_from_slice(&provider_id.to_le_bytes());
+    buffer.extend_from_slice(&record.tmdb_id.to_le_bytes());
+    buffer.extend_from_slice(&record.ts.to_le_bytes());
+
+    match &record.release_date {
+        Some(date_str) => {
+            // Write the length as a 4-byte u32 to prevent overflow.
+            let len = u32::try_from(date_str.len()).map_err(to_io_error)?;
+            buffer.extend_from_slice(&len.to_le_bytes());
+            buffer.extend_from_slice(date_str.as_bytes());
+        }
+        None => {
+            // Write a length of 0 as 4 bytes.
+            buffer.extend_from_slice(&0u32.to_le_bytes());
+        }
+    }
+
+    Ok(buffer)
+}
+
+
+fn encode_series_info_record_for_wal_file(provider_id: u32, ts: u64) -> Vec<u8> {
+    let mut buffer = Vec::with_capacity(12);
+    buffer.extend_from_slice(&provider_id.to_le_bytes());
+    buffer.extend_from_slice(&ts.to_le_bytes());
+    buffer
+}