GitHub
commit
1047320c7c
@ -0,0 +1,5 @@
|
||||
CREATE TABLE job_states (
|
||||
job_type INTEGER NOT NULL,
|
||||
value BLOB,
|
||||
PRIMARY KEY (job_type)
|
||||
);
|
||||
@ -0,0 +1,45 @@
|
||||
use sea_orm::prelude::*;
|
||||
use sea_orm::TryFromU64;
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, DeriveEntityModel)]
|
||||
#[sea_orm(table_name = "job_states")]
|
||||
pub struct Model {
|
||||
#[sea_orm(primary_key)]
|
||||
pub job_type: JobType,
|
||||
pub value: Vec<u8>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, PartialEq, EnumIter, DeriveActiveEnum)]
|
||||
#[sea_orm(rs_type = "u32", db_type = "Integer")]
|
||||
pub enum JobType {
|
||||
#[sea_orm(num_value = 10)]
|
||||
MigrateCDs,
|
||||
#[sea_orm(num_value = 20)]
|
||||
CalculateSizes,
|
||||
#[sea_orm(num_value = 30)]
|
||||
GenerateThumbs,
|
||||
#[sea_orm(num_value = 40)]
|
||||
CheckIntegrity,
|
||||
#[sea_orm(num_value = 50)]
|
||||
Vacuum,
|
||||
}
|
||||
|
||||
impl TryFromU64 for JobType {
|
||||
fn try_from_u64(n: u64) -> Result<Self, DbErr> {
|
||||
let value = match n {
|
||||
10 => Self::MigrateCDs,
|
||||
20 => Self::CalculateSizes,
|
||||
30 => Self::GenerateThumbs,
|
||||
40 => Self::CheckIntegrity,
|
||||
50 => Self::Vacuum,
|
||||
_ => return Err(DbErr::Custom(String::from("Invalid job type"))),
|
||||
};
|
||||
|
||||
Ok(value)
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
|
||||
pub enum Relation {}
|
||||
|
||||
impl ActiveModelBehavior for ActiveModel {}
|
||||
@ -0,0 +1,58 @@
|
||||
use crate::dao::job::JobDao;
|
||||
use crate::dto::{JobStateDto, UpsertJobStateDto};
|
||||
use mediarepo_core::error::RepoResult;
|
||||
use mediarepo_database::entities::job_state;
|
||||
use mediarepo_database::entities::job_state::JobType;
|
||||
use sea_orm::prelude::*;
|
||||
use sea_orm::ActiveValue::Set;
|
||||
use sea_orm::{Condition, TransactionTrait};
|
||||
|
||||
impl JobDao {
|
||||
/// Returns all job states for a given job id
|
||||
pub async fn state_for_job_type(&self, job_type: JobType) -> RepoResult<Option<JobStateDto>> {
|
||||
let state = job_state::Entity::find()
|
||||
.filter(job_state::Column::JobType.eq(job_type))
|
||||
.one(&self.ctx.db)
|
||||
.await?
|
||||
.map(JobStateDto::new);
|
||||
|
||||
Ok(state)
|
||||
}
|
||||
|
||||
pub async fn upsert_state(&self, state: UpsertJobStateDto) -> RepoResult<()> {
|
||||
self.upsert_multiple_states(vec![state]).await
|
||||
}
|
||||
|
||||
pub async fn upsert_multiple_states(&self, states: Vec<UpsertJobStateDto>) -> RepoResult<()> {
|
||||
let trx = self.ctx.db.begin().await?;
|
||||
|
||||
job_state::Entity::delete_many()
|
||||
.filter(build_state_filters(&states))
|
||||
.exec(&trx)
|
||||
.await?;
|
||||
job_state::Entity::insert_many(build_active_state_models(states))
|
||||
.exec(&trx)
|
||||
.await?;
|
||||
|
||||
trx.commit().await?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
fn build_state_filters(states: &Vec<UpsertJobStateDto>) -> Condition {
|
||||
states
|
||||
.iter()
|
||||
.map(|s| Condition::all().add(job_state::Column::JobType.eq(s.job_type)))
|
||||
.fold(Condition::any(), |acc, cond| acc.add(cond))
|
||||
}
|
||||
|
||||
fn build_active_state_models(states: Vec<UpsertJobStateDto>) -> Vec<job_state::ActiveModel> {
|
||||
states
|
||||
.into_iter()
|
||||
.map(|s| job_state::ActiveModel {
|
||||
job_type: Set(s.job_type),
|
||||
value: Set(s.value),
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
@ -0,0 +1,31 @@
|
||||
use mediarepo_database::entities::job_state;
|
||||
use mediarepo_database::entities::job_state::JobType;
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct JobStateDto {
|
||||
model: job_state::Model,
|
||||
}
|
||||
|
||||
impl JobStateDto {
|
||||
pub(crate) fn new(model: job_state::Model) -> Self {
|
||||
Self { model }
|
||||
}
|
||||
|
||||
pub fn job_type(&self) -> JobType {
|
||||
self.model.job_type
|
||||
}
|
||||
|
||||
pub fn value(&self) -> &[u8] {
|
||||
&self.model.value
|
||||
}
|
||||
|
||||
pub fn into_value(self) -> Vec<u8> {
|
||||
self.model.value
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct UpsertJobStateDto {
|
||||
pub job_type: JobType,
|
||||
pub value: Vec<u8>,
|
||||
}
|
||||
@ -0,0 +1,31 @@
|
||||
[package]
|
||||
name = "mediarepo-worker"
|
||||
version = "0.1.0"
|
||||
edition = "2021"
|
||||
|
||||
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
|
||||
|
||||
[dependencies]
|
||||
async-trait = "0.1.52"
|
||||
tracing = "0.1.32"
|
||||
|
||||
[dependencies.mediarepo-core]
|
||||
path = "../mediarepo-core"
|
||||
|
||||
[dependencies.mediarepo-logic]
|
||||
path = "../mediarepo-logic"
|
||||
|
||||
[dependencies.mediarepo-database]
|
||||
path = "../mediarepo-database"
|
||||
|
||||
[dependencies.tokio]
|
||||
version = "1.17.0"
|
||||
features = ["macros"]
|
||||
|
||||
[dependencies.chrono]
|
||||
version = "0.4.19"
|
||||
features = ["serde"]
|
||||
|
||||
[dependencies.serde]
|
||||
version = "1.0.136"
|
||||
features = ["derive"]
|
||||
@ -0,0 +1,101 @@
|
||||
use mediarepo_core::error::{RepoError, RepoResult};
|
||||
use std::ops::{Deref, DerefMut};
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::broadcast::{Receiver, Sender};
|
||||
use tokio::sync::RwLock;
|
||||
|
||||
pub struct JobHandle<T: Send + Sync, R: Send + Sync> {
|
||||
status: Arc<RwLock<T>>,
|
||||
state: Arc<RwLock<JobState>>,
|
||||
result_receiver: CloneableReceiver<Arc<RwLock<Option<RepoResult<R>>>>>,
|
||||
}
|
||||
|
||||
impl<T: Send + Sync, R: Send + Sync> Clone for JobHandle<T, R> {
|
||||
fn clone(&self) -> Self {
|
||||
Self {
|
||||
status: self.status.clone(),
|
||||
state: self.state.clone(),
|
||||
result_receiver: self.result_receiver.clone(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: Send + Sync, R: Send + Sync> JobHandle<T, R> {
|
||||
pub fn new(
|
||||
status: Arc<RwLock<T>>,
|
||||
state: Arc<RwLock<JobState>>,
|
||||
result_receiver: CloneableReceiver<Arc<RwLock<Option<RepoResult<R>>>>>,
|
||||
) -> Self {
|
||||
Self {
|
||||
status,
|
||||
state,
|
||||
result_receiver,
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn state(&self) -> JobState {
|
||||
*self.state.read().await
|
||||
}
|
||||
|
||||
pub fn status(&self) -> &Arc<RwLock<T>> {
|
||||
&self.status
|
||||
}
|
||||
|
||||
pub async fn result(&mut self) -> Arc<RwLock<Option<RepoResult<R>>>> {
|
||||
match self.result_receiver.recv().await {
|
||||
Ok(v) => v,
|
||||
Err(e) => Arc::new(RwLock::new(Some(Err(RepoError::from(&*e.to_string()))))),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn take_result(&mut self) -> Option<RepoResult<R>> {
|
||||
let shared_result = self.result().await;
|
||||
let mut result = shared_result.write().await;
|
||||
result.take()
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, Ord, PartialOrd, Eq, PartialEq)]
|
||||
pub enum JobState {
|
||||
Queued,
|
||||
Scheduled,
|
||||
Running,
|
||||
Finished,
|
||||
}
|
||||
|
||||
pub struct CloneableReceiver<T: Clone> {
|
||||
receiver: Receiver<T>,
|
||||
sender: Sender<T>,
|
||||
}
|
||||
|
||||
impl<T: Clone> CloneableReceiver<T> {
|
||||
pub fn new(sender: Sender<T>) -> Self {
|
||||
Self {
|
||||
receiver: sender.subscribe(),
|
||||
sender,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: Clone> Clone for CloneableReceiver<T> {
|
||||
fn clone(&self) -> Self {
|
||||
Self {
|
||||
sender: self.sender.clone(),
|
||||
receiver: self.sender.subscribe(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: Clone> Deref for CloneableReceiver<T> {
|
||||
type Target = Receiver<T>;
|
||||
|
||||
fn deref(&self) -> &Self::Target {
|
||||
&self.receiver
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: Clone> DerefMut for CloneableReceiver<T> {
|
||||
fn deref_mut(&mut self) -> &mut Self::Target {
|
||||
&mut self.receiver
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,137 @@
|
||||
use crate::handle::{CloneableReceiver, JobHandle, JobState};
|
||||
use crate::jobs::{Job, JobTypeKey};
|
||||
use mediarepo_core::tokio_graceful_shutdown::SubsystemHandle;
|
||||
use mediarepo_core::trait_bound_typemap::{SendSyncTypeMap, TypeMap, TypeMapKey};
|
||||
use mediarepo_logic::dao::repo::Repo;
|
||||
use mediarepo_logic::dao::DaoProvider;
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
use tokio::sync::broadcast::channel;
|
||||
use tokio::sync::RwLock;
|
||||
use tokio::time::Instant;
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct JobDispatcher {
|
||||
subsystem: SubsystemHandle,
|
||||
job_handle_map: Arc<RwLock<SendSyncTypeMap>>,
|
||||
repo: Arc<Repo>,
|
||||
}
|
||||
|
||||
impl JobDispatcher {
|
||||
pub fn new(subsystem: SubsystemHandle, repo: Repo) -> Self {
|
||||
Self {
|
||||
job_handle_map: Arc::new(RwLock::new(SendSyncTypeMap::new())),
|
||||
subsystem,
|
||||
repo: Arc::new(repo),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn dispatch<T: 'static + Job>(&self, job: T) -> JobHandle<T::JobStatus, T::Result> {
|
||||
self._dispatch(job, None).await
|
||||
}
|
||||
|
||||
pub async fn dispatch_periodically<T: 'static + Job>(
|
||||
&self,
|
||||
job: T,
|
||||
interval: Duration,
|
||||
) -> JobHandle<T::JobStatus, T::Result> {
|
||||
self._dispatch(job, Some(interval)).await
|
||||
}
|
||||
|
||||
#[tracing::instrument(level = "debug", skip_all)]
|
||||
async fn _dispatch<T: 'static + Job>(
|
||||
&self,
|
||||
job: T,
|
||||
interval: Option<Duration>,
|
||||
) -> JobHandle<T::JobStatus, T::Result> {
|
||||
let status = job.status();
|
||||
let state = Arc::new(RwLock::new(JobState::Queued));
|
||||
let (sender, mut receiver) = channel(1);
|
||||
self.subsystem
|
||||
.start("channel-consumer", move |subsystem| async move {
|
||||
tokio::select! {
|
||||
_ = receiver.recv() => (),
|
||||
_ = subsystem.on_shutdown_requested() => (),
|
||||
}
|
||||
Ok(())
|
||||
});
|
||||
let receiver = CloneableReceiver::new(sender.clone());
|
||||
let handle = JobHandle::new(status.clone(), state.clone(), receiver);
|
||||
self.add_handle::<T>(handle.clone()).await;
|
||||
|
||||
let repo = self.repo.clone();
|
||||
|
||||
self.subsystem
|
||||
.start("worker-job", move |subsystem| async move {
|
||||
loop {
|
||||
let start = Instant::now();
|
||||
let job_2 = job.clone();
|
||||
{
|
||||
let mut state = state.write().await;
|
||||
*state = JobState::Running;
|
||||
}
|
||||
if let Err(e) = job.load_state(repo.job()).await {
|
||||
tracing::error!("failed to load the jobs state: {}", e);
|
||||
}
|
||||
let result = tokio::select! {
|
||||
_ = subsystem.on_shutdown_requested() => {
|
||||
job_2.save_state(repo.job()).await
|
||||
}
|
||||
r = job.run(repo.clone()) => {
|
||||
match r {
|
||||
Err(e) => Err(e),
|
||||
Ok(v) => {
|
||||
let _ = sender.send(Arc::new(RwLock::new(Some(Ok(v)))));
|
||||
job.save_state(repo.job()).await
|
||||
}
|
||||
}
|
||||
}
|
||||
};
|
||||
if let Err(e) = result {
|
||||
tracing::error!("job failed with error: {}", e);
|
||||
let _ = sender.send(Arc::new(RwLock::new(Some(Err(e)))));
|
||||
}
|
||||
if let Some(interval) = interval {
|
||||
{
|
||||
let mut state = state.write().await;
|
||||
*state = JobState::Scheduled;
|
||||
}
|
||||
let sleep_duration = interval - start.elapsed();
|
||||
tokio::select! {
|
||||
_ = tokio::time::sleep(sleep_duration) => {},
|
||||
_ = subsystem.on_shutdown_requested() => {break}
|
||||
}
|
||||
} else {
|
||||
let mut state = state.write().await;
|
||||
*state = JobState::Finished;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
});
|
||||
|
||||
handle
|
||||
}
|
||||
|
||||
#[inline]
|
||||
async fn add_handle<T: 'static + Job>(&self, handle: JobHandle<T::JobStatus, T::Result>) {
|
||||
let mut status_map = self.job_handle_map.write().await;
|
||||
status_map.insert::<JobTypeKey<T>>(handle);
|
||||
}
|
||||
|
||||
#[inline]
|
||||
pub async fn get_handle<T: 'static + Job>(&self) -> Option<JobHandle<T::JobStatus, T::Result>> {
|
||||
let map = self.job_handle_map.read().await;
|
||||
map.get::<JobTypeKey<T>>().cloned()
|
||||
}
|
||||
}
|
||||
|
||||
pub struct DispatcherKey;
|
||||
|
||||
impl TypeMapKey for DispatcherKey {
|
||||
type Value = JobDispatcher;
|
||||
}
|
||||
|
||||
unsafe impl Send for JobDispatcher {}
|
||||
unsafe impl Sync for JobDispatcher {}
|
||||
@ -0,0 +1,91 @@
|
||||
use crate::jobs::Job;
|
||||
use crate::status_utils::SimpleProgress;
|
||||
use async_trait::async_trait;
|
||||
use mediarepo_core::error::{RepoError, RepoResult};
|
||||
use mediarepo_core::mediarepo_api::types::repo::SizeType;
|
||||
use mediarepo_core::settings::Settings;
|
||||
use mediarepo_core::utils::get_folder_size;
|
||||
use mediarepo_logic::dao::repo::Repo;
|
||||
use std::path::PathBuf;
|
||||
use std::sync::Arc;
|
||||
use tokio::fs;
|
||||
use tokio::sync::broadcast::{self, Sender};
|
||||
use tokio::sync::RwLock;
|
||||
|
||||
pub struct CalculateSizesState {
|
||||
pub progress: SimpleProgress,
|
||||
pub sizes_channel: Sender<(SizeType, u64)>,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct CalculateSizesJob {
|
||||
repo_path: PathBuf,
|
||||
settings: Settings,
|
||||
state: Arc<RwLock<CalculateSizesState>>,
|
||||
}
|
||||
|
||||
impl CalculateSizesJob {
|
||||
pub fn new(repo_path: PathBuf, settings: Settings) -> Self {
|
||||
let (tx, _) = broadcast::channel(4);
|
||||
Self {
|
||||
repo_path,
|
||||
settings,
|
||||
state: Arc::new(RwLock::new(CalculateSizesState {
|
||||
sizes_channel: tx,
|
||||
progress: SimpleProgress::new(4),
|
||||
})),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl Job for CalculateSizesJob {
|
||||
type JobStatus = CalculateSizesState;
|
||||
type Result = ();
|
||||
|
||||
fn status(&self) -> Arc<RwLock<Self::JobStatus>> {
|
||||
self.state.clone()
|
||||
}
|
||||
|
||||
#[tracing::instrument(level = "debug", skip_all)]
|
||||
async fn run(&self, repo: Arc<Repo>) -> RepoResult<()> {
|
||||
let size_types = vec![
|
||||
SizeType::Total,
|
||||
SizeType::FileFolder,
|
||||
SizeType::ThumbFolder,
|
||||
SizeType::DatabaseFile,
|
||||
];
|
||||
for size_type in size_types {
|
||||
let size = calculate_size(&size_type, &repo, &self.repo_path, &self.settings).await?;
|
||||
let mut state = self.state.write().await;
|
||||
state
|
||||
.sizes_channel
|
||||
.send((size_type, size))
|
||||
.map_err(|_| RepoError::from("failed to broadcast new size"))?;
|
||||
state.progress.tick();
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
async fn calculate_size(
|
||||
size_type: &SizeType,
|
||||
repo: &Repo,
|
||||
repo_path: &PathBuf,
|
||||
settings: &Settings,
|
||||
) -> RepoResult<u64> {
|
||||
let size = match &size_type {
|
||||
SizeType::Total => get_folder_size(repo_path.clone()).await?,
|
||||
SizeType::FileFolder => repo.get_main_store_size().await?,
|
||||
SizeType::ThumbFolder => repo.get_thumb_store_size().await?,
|
||||
SizeType::DatabaseFile => {
|
||||
let db_path = settings.paths.db_file_path(repo_path);
|
||||
|
||||
let database_metadata = fs::metadata(db_path).await?;
|
||||
database_metadata.len()
|
||||
}
|
||||
};
|
||||
|
||||
Ok(size)
|
||||
}
|
||||
@ -0,0 +1,32 @@
|
||||
use crate::jobs::Job;
|
||||
use crate::status_utils::SimpleProgress;
|
||||
use async_trait::async_trait;
|
||||
use mediarepo_core::error::RepoResult;
|
||||
use mediarepo_logic::dao::repo::Repo;
|
||||
use mediarepo_logic::dao::DaoProvider;
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::RwLock;
|
||||
|
||||
#[derive(Clone, Default)]
|
||||
pub struct CheckIntegrityJob {
|
||||
progress: Arc<RwLock<SimpleProgress>>,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl Job for CheckIntegrityJob {
|
||||
type JobStatus = SimpleProgress;
|
||||
type Result = ();
|
||||
|
||||
fn status(&self) -> Arc<RwLock<Self::JobStatus>> {
|
||||
self.progress.clone()
|
||||
}
|
||||
|
||||
async fn run(&self, repo: Arc<Repo>) -> RepoResult<Self::Result> {
|
||||
repo.job().check_integrity().await?;
|
||||
{
|
||||
let mut progress = self.progress.write().await;
|
||||
progress.set_total(100);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,109 @@
|
||||
use crate::jobs::{deserialize_state, serialize_state, Job};
|
||||
use crate::status_utils::SimpleProgress;
|
||||
use async_trait::async_trait;
|
||||
use mediarepo_core::error::RepoResult;
|
||||
use mediarepo_core::thumbnailer::ThumbnailSize;
|
||||
use mediarepo_database::entities::job_state::JobType;
|
||||
use mediarepo_logic::dao::job::JobDao;
|
||||
use mediarepo_logic::dao::repo::Repo;
|
||||
use mediarepo_logic::dao::DaoProvider;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::mem;
|
||||
use std::sync::Arc;
|
||||
use std::time::{Duration, SystemTime};
|
||||
use tokio::sync::RwLock;
|
||||
|
||||
#[derive(Clone, Default)]
|
||||
pub struct GenerateMissingThumbsJob {
|
||||
state: Arc<RwLock<SimpleProgress>>,
|
||||
inner_state: Arc<RwLock<GenerateThumbsState>>,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl Job for GenerateMissingThumbsJob {
|
||||
type JobStatus = SimpleProgress;
|
||||
type Result = ();
|
||||
|
||||
fn status(&self) -> Arc<RwLock<Self::JobStatus>> {
|
||||
self.state.clone()
|
||||
}
|
||||
|
||||
async fn load_state(&self, job_dao: JobDao) -> RepoResult<()> {
|
||||
if let Some(state) = job_dao.state_for_job_type(JobType::GenerateThumbs).await? {
|
||||
let mut inner_state = self.inner_state.write().await;
|
||||
let state = deserialize_state::<GenerateThumbsState>(state)?;
|
||||
let _ = mem::replace(&mut *inner_state, state);
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn run(&self, repo: Arc<Repo>) -> RepoResult<()> {
|
||||
if !self.needs_generation(&repo).await? {
|
||||
return Ok(());
|
||||
}
|
||||
let file_dao = repo.file();
|
||||
let all_files = file_dao.all().await?;
|
||||
{
|
||||
let mut progress = self.state.write().await;
|
||||
progress.set_total(all_files.len() as u64);
|
||||
}
|
||||
|
||||
for file in all_files {
|
||||
if file_dao.thumbnails(file.encoded_cd()).await?.is_empty() {
|
||||
let _ = file_dao
|
||||
.create_thumbnails(&file, vec![ThumbnailSize::Medium])
|
||||
.await;
|
||||
}
|
||||
{
|
||||
let mut progress = self.state.write().await;
|
||||
progress.tick();
|
||||
}
|
||||
}
|
||||
|
||||
self.refresh_state(&repo).await?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn save_state(&self, job_dao: JobDao) -> RepoResult<()> {
|
||||
let state = self.inner_state.read().await;
|
||||
let state = serialize_state(JobType::GenerateThumbs, &*state)?;
|
||||
job_dao.upsert_state(state).await
|
||||
}
|
||||
}
|
||||
|
||||
impl GenerateMissingThumbsJob {
|
||||
async fn needs_generation(&self, repo: &Repo) -> RepoResult<bool> {
|
||||
let repo_counts = repo.get_counts().await?;
|
||||
let file_count = repo_counts.file_count as u64;
|
||||
let state = self.inner_state.read().await;
|
||||
|
||||
Ok(state.file_count != file_count
|
||||
|| state.last_run.elapsed().unwrap() > Duration::from_secs(60 * 60))
|
||||
}
|
||||
|
||||
async fn refresh_state(&self, repo: &Repo) -> RepoResult<()> {
|
||||
let repo_counts = repo.get_counts().await?;
|
||||
let mut state = self.inner_state.write().await;
|
||||
state.last_run = SystemTime::now();
|
||||
state.file_count = repo_counts.file_count as u64;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Serialize, Deserialize)]
|
||||
struct GenerateThumbsState {
|
||||
file_count: u64,
|
||||
last_run: SystemTime,
|
||||
}
|
||||
|
||||
impl Default for GenerateThumbsState {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
file_count: 0,
|
||||
last_run: SystemTime::now(),
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,65 @@
|
||||
use crate::jobs::{deserialize_state, serialize_state, Job};
|
||||
use crate::status_utils::SimpleProgress;
|
||||
use async_trait::async_trait;
|
||||
use mediarepo_core::error::RepoResult;
|
||||
use mediarepo_database::entities::job_state::JobType;
|
||||
use mediarepo_logic::dao::job::JobDao;
|
||||
use mediarepo_logic::dao::repo::Repo;
|
||||
use mediarepo_logic::dao::DaoProvider;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::RwLock;
|
||||
|
||||
#[derive(Clone, Default)]
|
||||
pub struct MigrateCDsJob {
|
||||
progress: Arc<RwLock<SimpleProgress>>,
|
||||
migrated: Arc<AtomicBool>,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl Job for MigrateCDsJob {
|
||||
type JobStatus = SimpleProgress;
|
||||
type Result = ();
|
||||
|
||||
fn status(&self) -> Arc<tokio::sync::RwLock<Self::JobStatus>> {
|
||||
self.progress.clone()
|
||||
}
|
||||
|
||||
async fn load_state(&self, job_dao: JobDao) -> RepoResult<()> {
|
||||
if let Some(state) = job_dao.state_for_job_type(JobType::MigrateCDs).await? {
|
||||
let state = deserialize_state::<MigrationStatus>(state)?;
|
||||
self.migrated.store(state.migrated, Ordering::SeqCst);
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn run(&self, repo: Arc<Repo>) -> RepoResult<Self::Result> {
|
||||
if self.migrated.load(Ordering::SeqCst) {
|
||||
return Ok(());
|
||||
}
|
||||
let job_dao = repo.job();
|
||||
|
||||
job_dao.migrate_content_descriptors().await?;
|
||||
self.migrated.store(true, Ordering::Relaxed);
|
||||
{
|
||||
let mut progress = self.progress.write().await;
|
||||
progress.set_total(100);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn save_state(&self, job_dao: JobDao) -> RepoResult<()> {
|
||||
if self.migrated.load(Ordering::Relaxed) {
|
||||
let state = serialize_state(JobType::MigrateCDs, &MigrationStatus { migrated: true })?;
|
||||
job_dao.upsert_state(state).await?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Serialize, Deserialize)]
|
||||
struct MigrationStatus {
|
||||
pub migrated: bool,
|
||||
}
|
||||
@ -0,0 +1,71 @@
|
||||
mod calculate_sizes;
|
||||
mod check_integrity;
|
||||
mod generate_missing_thumbnails;
|
||||
mod migrate_content_descriptors;
|
||||
mod vacuum;
|
||||
|
||||
pub use calculate_sizes::*;
|
||||
pub use check_integrity::*;
|
||||
pub use generate_missing_thumbnails::*;
|
||||
pub use migrate_content_descriptors::*;
|
||||
use std::marker::PhantomData;
|
||||
use std::sync::Arc;
|
||||
pub use vacuum::*;
|
||||
|
||||
use crate::handle::JobHandle;
|
||||
use async_trait::async_trait;
|
||||
use mediarepo_core::bincode;
|
||||
use mediarepo_core::error::{RepoError, RepoResult};
|
||||
use mediarepo_core::trait_bound_typemap::TypeMapKey;
|
||||
use mediarepo_database::entities::job_state::JobType;
|
||||
use mediarepo_logic::dao::job::JobDao;
|
||||
use mediarepo_logic::dao::repo::Repo;
|
||||
use mediarepo_logic::dto::{JobStateDto, UpsertJobStateDto};
|
||||
use serde::de::DeserializeOwned;
|
||||
use serde::Serialize;
|
||||
use tokio::sync::RwLock;
|
||||
|
||||
type EmptyStatus = Arc<RwLock<()>>;
|
||||
|
||||
#[async_trait]
|
||||
pub trait Job: Clone + Send + Sync {
|
||||
type JobStatus: Send + Sync;
|
||||
type Result: Send + Sync;
|
||||
|
||||
fn status(&self) -> Arc<RwLock<Self::JobStatus>>;
|
||||
|
||||
async fn load_state(&self, _job_dao: JobDao) -> RepoResult<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn run(&self, repo: Arc<Repo>) -> RepoResult<Self::Result>;
|
||||
|
||||
async fn save_state(&self, _job_dao: JobDao) -> RepoResult<()> {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
pub struct JobTypeKey<T: Job>(PhantomData<T>);
|
||||
|
||||
impl<T: 'static> TypeMapKey for JobTypeKey<T>
|
||||
where
|
||||
T: Job,
|
||||
{
|
||||
type Value = JobHandle<T::JobStatus, T::Result>;
|
||||
}
|
||||
|
||||
pub fn deserialize_state<T: DeserializeOwned>(dto: JobStateDto) -> RepoResult<T> {
|
||||
bincode::deserialize(dto.value()).map_err(RepoError::from)
|
||||
}
|
||||
|
||||
pub fn serialize_state<T: Serialize>(
|
||||
job_type: JobType,
|
||||
state: &T,
|
||||
) -> RepoResult<UpsertJobStateDto> {
|
||||
let dto = UpsertJobStateDto {
|
||||
value: bincode::serialize(state)?,
|
||||
job_type,
|
||||
};
|
||||
|
||||
Ok(dto)
|
||||
}
|
||||
@ -0,0 +1,27 @@
|
||||
use crate::jobs::{EmptyStatus, Job};
|
||||
use async_trait::async_trait;
|
||||
use mediarepo_core::error::RepoResult;
|
||||
use mediarepo_logic::dao::repo::Repo;
|
||||
use mediarepo_logic::dao::DaoProvider;
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::RwLock;
|
||||
|
||||
#[derive(Default, Clone)]
|
||||
pub struct VacuumJob;
|
||||
|
||||
#[async_trait]
|
||||
impl Job for VacuumJob {
|
||||
type JobStatus = ();
|
||||
type Result = ();
|
||||
|
||||
fn status(&self) -> Arc<RwLock<Self::JobStatus>> {
|
||||
EmptyStatus::default()
|
||||
}
|
||||
|
||||
#[tracing::instrument(level = "debug", skip_all)]
|
||||
async fn run(&self, repo: Arc<Repo>) -> RepoResult<()> {
|
||||
repo.job().vacuum().await?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,39 @@
|
||||
use crate::job_dispatcher::JobDispatcher;
|
||||
use crate::jobs::{CheckIntegrityJob, MigrateCDsJob, VacuumJob};
|
||||
use mediarepo_core::error::RepoError;
|
||||
use mediarepo_core::tokio_graceful_shutdown::Toplevel;
|
||||
use mediarepo_logic::dao::repo::Repo;
|
||||
use std::time::Duration;
|
||||
use tokio::sync::oneshot::channel;
|
||||
|
||||
pub mod handle;
|
||||
pub mod job_dispatcher;
|
||||
pub mod jobs;
|
||||
pub mod status_utils;
|
||||
|
||||
pub async fn start(top_level: Toplevel, repo: Repo) -> (Toplevel, JobDispatcher) {
|
||||
let (tx, rx) = channel();
|
||||
|
||||
let top_level = top_level.start("mediarepo-worker", |subsystem| async move {
|
||||
let dispatcher = JobDispatcher::new(subsystem, repo);
|
||||
tx.send(dispatcher.clone())
|
||||
.map_err(|_| RepoError::from("failed to send dispatcher"))?;
|
||||
dispatcher
|
||||
.dispatch_periodically(VacuumJob::default(), Duration::from_secs(60 * 30))
|
||||
.await;
|
||||
dispatcher
|
||||
.dispatch_periodically(
|
||||
CheckIntegrityJob::default(),
|
||||
Duration::from_secs(60 * 60 * 24),
|
||||
)
|
||||
.await;
|
||||
dispatcher.dispatch(MigrateCDsJob::default()).await;
|
||||
|
||||
Ok(())
|
||||
});
|
||||
let receiver = rx
|
||||
.await
|
||||
.expect("failed to create background job dispatcher");
|
||||
|
||||
(top_level, receiver)
|
||||
}
|
||||
@ -0,0 +1,39 @@
|
||||
pub struct SimpleProgress {
|
||||
pub current: u64,
|
||||
pub total: u64,
|
||||
}
|
||||
|
||||
impl Default for SimpleProgress {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
total: 100,
|
||||
current: 0,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl SimpleProgress {
|
||||
pub fn new(total: u64) -> Self {
|
||||
Self { total, current: 0 }
|
||||
}
|
||||
|
||||
/// Sets the total count
|
||||
pub fn set_total(&mut self, total: u64) {
|
||||
self.total = total;
|
||||
}
|
||||
|
||||
/// Increments the current progress by 1
|
||||
pub fn tick(&mut self) {
|
||||
self.current += 1;
|
||||
}
|
||||
|
||||
/// Sets the current progress to a defined value
|
||||
pub fn set_current(&mut self, current: u64) {
|
||||
self.current = current;
|
||||
}
|
||||
|
||||
/// Returns the total progress in percent
|
||||
pub fn percent(&self) -> f64 {
|
||||
(self.current as f64) / (self.total as f64)
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue