From 260ff5f6c72620de959ebf1b910423753aa4f451 Mon Sep 17 00:00:00 2001 From: euzu Date: Mon, 16 Jun 2025 10:48:01 +0200 Subject: [PATCH] update wal file only when content could written in once. --- src/processing/processor/xtream.rs | 13 +---- src/processing/processor/xtream_series.rs | 27 +++------ src/processing/processor/xtream_vod.rs | 42 +++----------- src/repository/xtream_repository.rs | 69 ++++++++++++++++++++++- 4 files changed, 85 insertions(+), 66 deletions(-) diff --git a/src/processing/processor/xtream.rs b/src/processing/processor/xtream.rs index 25682b685..010d42152 100644 --- a/src/processing/processor/xtream.rs +++ b/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) => { diff --git a/src/processing/processor/xtream_series.rs b/src/processing/processor/xtream_series.rs index 6754a7d62..f8f3b00a9 100644 --- a/src/processing/processor/xtream_series.rs +++ b/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, - 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, 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, 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, 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 diff --git a/src/processing/processor/xtream_vod.rs b/src/processing/processor/xtream_vod.rs index d4bd87411..eced6e37e 100644 --- a/src/processing/processor/xtream_vod.rs +++ b/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) -> (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, 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, 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); diff --git a/src/repository/xtream_repository.rs b/src/repository/xtream_repository.rs index c2e7d1d86..08897287b 100644 --- a/src/repository/xtream_repository.rs +++ b/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> { + 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> { + 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 { + 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 +}