use crate::{ config::ImageFormat, error::Error, repo::{ Alias, AliasRepo, DeleteToken, FullRepo, HashRepo, IdentifierRepo, QueueRepo, UploadId, }, serde_str::Serde, store::{Identifier, Store}, }; use std::{future::Future, path::PathBuf, pin::Pin}; use tracing::Instrument; mod cleanup; mod process; const CLEANUP_QUEUE: &str = "cleanup"; const PROCESS_QUEUE: &str = "process"; #[derive(Debug, serde::Deserialize, serde::Serialize)] enum Cleanup { Hash { hash: Vec, }, Identifier { identifier: Vec, }, Alias { alias: Serde, token: Serde, }, } #[derive(Debug, serde::Deserialize, serde::Serialize)] enum Process { Ingest { identifier: Vec, upload_id: Serde, declared_alias: Option>, should_validate: bool, is_cached: bool, }, Generate { target_format: ImageFormat, source: Serde, process_path: PathBuf, process_args: Vec, }, } pub(crate) async fn cleanup_alias( repo: &R, alias: Alias, token: DeleteToken, ) -> Result<(), Error> { let job = serde_json::to_vec(&Cleanup::Alias { alias: Serde::new(alias), token: Serde::new(token), })?; repo.push(CLEANUP_QUEUE, job.into()).await?; Ok(()) } pub(crate) async fn cleanup_hash(repo: &R, hash: R::Bytes) -> Result<(), Error> { let job = serde_json::to_vec(&Cleanup::Hash { hash: hash.as_ref().to_vec(), })?; repo.push(CLEANUP_QUEUE, job.into()).await?; Ok(()) } pub(crate) async fn cleanup_identifier( repo: &R, identifier: I, ) -> Result<(), Error> { let job = serde_json::to_vec(&Cleanup::Identifier { identifier: identifier.to_bytes()?, })?; repo.push(CLEANUP_QUEUE, job.into()).await?; Ok(()) } pub(crate) async fn queue_ingest( repo: &R, identifier: Vec, upload_id: UploadId, declared_alias: Option, should_validate: bool, is_cached: bool, ) -> Result<(), Error> { let job = serde_json::to_vec(&Process::Ingest { identifier, declared_alias: declared_alias.map(Serde::new), upload_id: Serde::new(upload_id), should_validate, is_cached, })?; repo.push(PROCESS_QUEUE, job.into()).await?; Ok(()) } pub(crate) async fn queue_generate( repo: &R, target_format: ImageFormat, source: Alias, process_path: PathBuf, process_args: Vec, ) -> Result<(), Error> { let job = serde_json::to_vec(&Process::Generate { target_format, source: Serde::new(source), process_path, process_args, })?; repo.push(PROCESS_QUEUE, job.into()).await?; Ok(()) } pub(crate) async fn process_cleanup(repo: R, store: S, worker_id: String) { process_jobs(&repo, &store, worker_id, CLEANUP_QUEUE, cleanup::perform).await } pub(crate) async fn process_images( repo: R, store: S, worker_id: String, ) { process_jobs(&repo, &store, worker_id, PROCESS_QUEUE, process::perform).await } type LocalBoxFuture<'a, T> = Pin + 'a>>; async fn process_jobs( repo: &R, store: &S, worker_id: String, queue: &'static str, callback: F, ) where R: QueueRepo + HashRepo + IdentifierRepo + AliasRepo, R::Bytes: Clone, S: Store, for<'a> F: Fn(&'a R, &'a S, &'a [u8]) -> LocalBoxFuture<'a, Result<(), Error>> + Copy, { loop { let res = job_loop(repo, store, worker_id.clone(), queue, callback).await; if let Err(e) = res { tracing::warn!("Error processing jobs: {}", e); tracing::warn!("{:?}", e); continue; } break; } } async fn job_loop( repo: &R, store: &S, worker_id: String, queue: &'static str, callback: F, ) -> Result<(), Error> where R: QueueRepo + HashRepo + IdentifierRepo + AliasRepo, R::Bytes: Clone, S: Store, for<'a> F: Fn(&'a R, &'a S, &'a [u8]) -> LocalBoxFuture<'a, Result<(), Error>> + Copy, { loop { let bytes = repo.pop(queue, worker_id.as_bytes().to_vec()).await?; let span = tracing::info_span!("Running Job", worker_id = ?worker_id); span.in_scope(|| (callback)(repo, store, bytes.as_ref())) .instrument(span) .await?; } }