2
0
Fork 0
mirror of https://git.asonix.dog/asonix/pict-rs synced 2024-12-22 19:31:35 +00:00
pict-rs/src/lib.rs

2424 lines
76 KiB
Rust
Raw Normal View History

2022-12-23 18:56:15 +00:00
mod backgrounded;
mod bytes_stream;
mod concurrent_processor;
mod config;
mod details;
mod discover;
2022-12-23 18:56:15 +00:00
mod either;
mod error;
2023-09-02 01:50:10 +00:00
mod error_code;
2022-12-23 18:56:15 +00:00
mod exiftool;
mod ffmpeg;
mod file;
mod formats;
2023-09-05 02:51:27 +00:00
mod future;
2022-12-23 18:56:15 +00:00
mod generate;
mod ingest;
mod init_tracing;
mod magick;
mod middleware;
2023-07-17 03:07:42 +00:00
mod migrate_store;
2022-12-23 18:56:15 +00:00
mod process;
mod processor;
mod queue;
mod range;
mod read;
2022-12-23 18:56:15 +00:00
mod repo;
mod repo_04;
2022-12-23 18:56:15 +00:00
mod serde_str;
mod store;
mod stream;
mod sync;
2022-12-23 18:56:15 +00:00
mod tmp_file;
mod validate;
use actix_form_data::{Field, Form, FormData, Multipart, Value};
use actix_web::{
guard,
http::header::{CacheControl, CacheDirective, LastModified, Range, ACCEPT_RANGES},
web, App, HttpRequest, HttpResponse, HttpResponseBuilder, HttpServer,
};
use details::{ApiDetails, HumanDate};
2023-09-05 02:58:57 +00:00
use future::WithTimeout;
2023-08-23 16:59:42 +00:00
use futures_core::Stream;
2023-07-22 21:47:59 +00:00
use metrics_exporter_prometheus::PrometheusBuilder;
use middleware::{Metrics, Payload};
use repo::ArcRepo;
2023-07-21 21:58:31 +00:00
use reqwest_middleware::{ClientBuilder, ClientWithMiddleware};
use reqwest_tracing::TracingMiddleware;
2022-12-23 18:56:15 +00:00
use rusty_s3::UrlStyle;
use std::{
marker::PhantomData,
2022-12-23 18:56:15 +00:00
path::Path,
path::PathBuf,
2023-09-24 20:32:00 +00:00
sync::{Arc, OnceLock},
2023-07-17 03:07:42 +00:00
time::{Duration, SystemTime},
2022-12-23 18:56:15 +00:00
};
use streem::IntoStreamer;
use tmp_file::{ArcTmpDir, TmpDir};
2022-12-23 18:56:15 +00:00
use tokio::sync::Semaphore;
2023-09-05 02:51:27 +00:00
use tracing::Instrument;
2022-12-23 18:56:15 +00:00
use tracing_actix_web::TracingLogger;
use self::{
backgrounded::Backgrounded,
2023-07-22 17:47:32 +00:00
concurrent_processor::ProcessMap,
config::{Configuration, Operation},
2022-12-23 18:56:15 +00:00
details::Details,
either::Either,
error::{Error, UploadError},
2023-07-22 17:47:32 +00:00
formats::InputProcessableFormat,
2022-12-23 18:56:15 +00:00
ingest::Session,
init_tracing::init_tracing,
middleware::{Deadline, Internal},
2023-07-17 03:07:42 +00:00
migrate_store::migrate_store,
2022-12-23 18:56:15 +00:00
queue::queue_generate,
2023-08-16 21:09:40 +00:00
repo::{sled::SledRepo, Alias, DeleteToken, Hash, Repo, UploadId, UploadResult},
2022-12-23 18:56:15 +00:00
serde_str::Serde,
store::{file_store::FileStore, object_store::ObjectStore, Store},
stream::{empty, once},
2022-12-23 18:56:15 +00:00
};
pub use self::config::{ConfigSource, PictRsConfiguration};
2022-12-23 18:56:15 +00:00
const MEGABYTES: usize = 1024 * 1024;
const MINUTES: u32 = 60;
const HOURS: u32 = 60 * MINUTES;
const DAYS: u32 = 24 * HOURS;
const NOT_FOUND_KEY: &str = "404-alias";
2023-09-24 20:32:00 +00:00
static PROCESS_SEMAPHORE: OnceLock<Semaphore> = OnceLock::new();
2023-09-24 20:32:00 +00:00
fn process_semaphore() -> &'static Semaphore {
PROCESS_SEMAPHORE.get_or_init(|| {
let permits = std::thread::available_parallelism()
.map(usize::from)
.unwrap_or(1)
.saturating_sub(1)
.max(1);
crate::sync::bare_semaphore(permits)
})
}
2022-12-23 18:56:15 +00:00
async fn ensure_details<S: Store + 'static>(
2023-10-07 16:32:36 +00:00
tmp_dir: &TmpDir,
repo: &ArcRepo,
2022-12-23 18:56:15 +00:00
store: &S,
config: &Configuration,
2022-12-23 18:56:15 +00:00
alias: &Alias,
) -> Result<Details, Error> {
let Some(identifier) = repo.identifier_from_alias(alias).await? else {
return Err(UploadError::MissingAlias.into());
};
2023-11-11 20:22:12 +00:00
ensure_details_identifier(tmp_dir, repo, store, config, &identifier).await
}
async fn ensure_details_identifier<S: Store + 'static>(
tmp_dir: &TmpDir,
repo: &ArcRepo,
store: &S,
config: &Configuration,
identifier: &Arc<str>,
) -> Result<Details, Error> {
2023-11-11 20:28:05 +00:00
let details = repo.details(identifier).await?;
2022-12-23 18:56:15 +00:00
if let Some(details) = details {
tracing::debug!("details exist");
Ok(details)
} else {
if config.server.read_only {
2023-07-17 19:24:49 +00:00
return Err(UploadError::ReadOnly.into());
2023-11-11 20:22:12 +00:00
} else if config.server.danger_dummy_mode {
return Ok(Details::danger_dummy(formats::InternalFormat::Image(
formats::ImageFormat::Png,
)));
2023-07-17 19:24:49 +00:00
}
2022-12-23 18:56:15 +00:00
tracing::debug!("generating new details from {:?}", identifier);
2023-11-11 20:22:12 +00:00
let bytes_stream = store.to_bytes(identifier, None, None).await?;
2023-10-07 16:32:36 +00:00
let new_details = Details::from_bytes(
tmp_dir,
config.media.process_timeout,
bytes_stream.into_bytes(),
)
.await?;
2022-12-23 18:56:15 +00:00
tracing::debug!("storing details for {:?}", identifier);
2023-11-11 20:22:12 +00:00
repo.relate_details(identifier, &new_details).await?;
2022-12-23 18:56:15 +00:00
tracing::debug!("stored");
Ok(new_details)
}
}
struct Upload<S>(Value<Session>, PhantomData<S>);
2022-12-23 18:56:15 +00:00
impl<S: Store + 'static> FormData for Upload<S> {
type Item = Session;
2022-12-23 18:56:15 +00:00
type Error = Error;
fn form(req: &HttpRequest) -> Form<Self::Item, Self::Error> {
let tmp_dir = req
.app_data::<web::Data<ArcTmpDir>>()
.expect("No TmpDir in request")
.clone();
2022-12-23 18:56:15 +00:00
let repo = req
.app_data::<web::Data<ArcRepo>>()
2022-12-23 18:56:15 +00:00
.expect("No repo in request")
.clone();
let store = req
.app_data::<web::Data<S>>()
.expect("No store in request")
.clone();
2023-09-06 01:45:07 +00:00
let client = req
.app_data::<web::Data<ClientWithMiddleware>>()
.expect("No client in request")
.clone();
let config = req
.app_data::<web::Data<Configuration>>()
.expect("No configuration in request")
.clone();
2022-12-23 18:56:15 +00:00
// Create a new Multipart Form validator
//
// This form is expecting a single array field, 'images' with at most 10 files in it
2022-12-23 18:56:15 +00:00
Form::new()
.max_files(config.server.max_file_count)
.max_file_size(config.media.max_file_size * MEGABYTES)
2022-12-23 18:56:15 +00:00
.transform_error(transform_error)
.field(
"images",
Field::array(Field::file(move |filename, _, stream| {
let tmp_dir = tmp_dir.clone();
2022-12-23 18:56:15 +00:00
let repo = repo.clone();
let store = store.clone();
2023-09-06 01:45:07 +00:00
let client = client.clone();
let config = config.clone();
2022-12-23 18:56:15 +00:00
metrics::counter!("pict-rs.files", "upload" => "inline").increment(1);
2023-07-22 21:47:59 +00:00
2022-12-23 18:56:15 +00:00
let span = tracing::info_span!("file-upload", ?filename);
Box::pin(
2023-07-13 22:42:21 +00:00
async move {
if config.server.read_only {
2023-07-17 19:24:49 +00:00
return Err(UploadError::ReadOnly.into());
}
let stream = crate::stream::from_err(stream);
ingest::ingest(
2023-11-11 20:22:12 +00:00
&tmp_dir, &repo, &**store, &client, stream, None, &config,
)
.await
2023-07-13 22:42:21 +00:00
}
.instrument(span),
2022-12-23 18:56:15 +00:00
)
})),
)
}
fn extract(value: Value<Self::Item>) -> Result<Self, Self::Error> {
Ok(Upload(value, PhantomData))
2022-12-23 18:56:15 +00:00
}
}
struct Import<S: Store + 'static>(Value<Session>, PhantomData<S>);
2022-12-23 18:56:15 +00:00
impl<S: Store + 'static> FormData for Import<S> {
type Item = Session;
2022-12-23 18:56:15 +00:00
type Error = Error;
fn form(req: &actix_web::HttpRequest) -> Form<Self::Item, Self::Error> {
let tmp_dir = req
.app_data::<web::Data<ArcTmpDir>>()
.expect("No TmpDir in request")
.clone();
2022-12-23 18:56:15 +00:00
let repo = req
.app_data::<web::Data<ArcRepo>>()
2022-12-23 18:56:15 +00:00
.expect("No repo in request")
.clone();
let store = req
.app_data::<web::Data<S>>()
.expect("No store in request")
.clone();
2023-09-06 01:45:07 +00:00
let client = req
.app_data::<ClientWithMiddleware>()
.expect("No client in request")
.clone();
let config = req
.app_data::<web::Data<Configuration>>()
.expect("No configuration in request")
.clone();
2022-12-23 18:56:15 +00:00
// Create a new Multipart Form validator for internal imports
//
// This form is expecting a single array field, 'images' with at most 10 files in it
Form::new()
.max_files(config.server.max_file_count)
.max_file_size(config.media.max_file_size * MEGABYTES)
2022-12-23 18:56:15 +00:00
.transform_error(transform_error)
.field(
"images",
Field::array(Field::file(move |filename, _, stream| {
let tmp_dir = tmp_dir.clone();
2022-12-23 18:56:15 +00:00
let repo = repo.clone();
let store = store.clone();
2023-09-06 01:45:07 +00:00
let client = client.clone();
let config = config.clone();
2022-12-23 18:56:15 +00:00
metrics::counter!("pict-rs.files", "import" => "inline").increment(1);
2023-07-22 21:47:59 +00:00
2022-12-23 18:56:15 +00:00
let span = tracing::info_span!("file-import", ?filename);
Box::pin(
async move {
if config.server.read_only {
2023-07-17 19:24:49 +00:00
return Err(UploadError::ReadOnly.into());
}
let stream = crate::stream::from_err(stream);
2022-12-23 18:56:15 +00:00
ingest::ingest(
&tmp_dir,
&repo,
2022-12-23 18:56:15 +00:00
&**store,
2023-09-06 01:45:07 +00:00
&client,
2022-12-23 18:56:15 +00:00
stream,
Some(Alias::from_existing(&filename)),
2023-11-11 20:22:12 +00:00
&config,
2022-12-23 18:56:15 +00:00
)
.await
}
.instrument(span),
)
})),
)
}
fn extract(value: Value<Self::Item>) -> Result<Self, Self::Error>
where
Self: Sized,
{
Ok(Import(value, PhantomData))
2022-12-23 18:56:15 +00:00
}
}
2023-06-29 16:45:06 +00:00
/// Handle responding to successful uploads
#[tracing::instrument(name = "Uploaded files", skip(value, repo, store, config))]
async fn upload<S: Store + 'static>(
Multipart(Upload(value, _)): Multipart<Upload<S>>,
2023-10-07 16:32:36 +00:00
tmp_dir: web::Data<ArcTmpDir>,
repo: web::Data<ArcRepo>,
2022-12-23 18:56:15 +00:00
store: web::Data<S>,
config: web::Data<Configuration>,
2022-12-23 18:56:15 +00:00
) -> Result<HttpResponse, Error> {
2023-10-07 16:32:36 +00:00
handle_upload(value, tmp_dir, repo, store, config).await
2022-12-23 18:56:15 +00:00
}
2023-06-29 16:45:06 +00:00
/// Handle responding to successful uploads
#[tracing::instrument(name = "Imported files", skip(value, repo, store, config))]
async fn import<S: Store + 'static>(
Multipart(Import(value, _)): Multipart<Import<S>>,
2023-10-07 16:32:36 +00:00
tmp_dir: web::Data<ArcTmpDir>,
repo: web::Data<ArcRepo>,
2022-12-23 18:56:15 +00:00
store: web::Data<S>,
config: web::Data<Configuration>,
2022-12-23 18:56:15 +00:00
) -> Result<HttpResponse, Error> {
2023-10-07 16:32:36 +00:00
handle_upload(value, tmp_dir, repo, store, config).await
2022-12-23 18:56:15 +00:00
}
2023-06-29 16:45:06 +00:00
/// Handle responding to successful uploads
#[tracing::instrument(name = "Uploaded files", skip(value, repo, store, config))]
async fn handle_upload<S: Store + 'static>(
value: Value<Session>,
2023-10-07 16:32:36 +00:00
tmp_dir: web::Data<ArcTmpDir>,
repo: web::Data<ArcRepo>,
2022-12-23 18:56:15 +00:00
store: web::Data<S>,
config: web::Data<Configuration>,
2022-12-23 18:56:15 +00:00
) -> Result<HttpResponse, Error> {
let images = value
.map()
.and_then(|mut m| m.remove("images"))
.and_then(|images| images.array())
.ok_or(UploadError::NoFiles)?;
let mut files = Vec::new();
let images = images
.into_iter()
.filter_map(|i| i.file())
.collect::<Vec<_>>();
for image in &images {
if let Some(alias) = image.result.alias() {
tracing::debug!("Uploaded {} as {:?}", image.filename, alias);
let delete_token = image.result.delete_token();
2022-12-23 18:56:15 +00:00
2023-10-07 16:32:36 +00:00
let details = ensure_details(&tmp_dir, &repo, &store, &config, alias).await?;
2022-12-23 18:56:15 +00:00
files.push(serde_json::json!({
"file": alias.to_string(),
"delete_token": delete_token.to_string(),
"details": details.into_api_details(),
2022-12-23 18:56:15 +00:00
}));
}
}
for image in images {
2022-12-23 18:56:15 +00:00
image.result.disarm();
}
Ok(HttpResponse::Created().json(&serde_json::json!({
"msg": "ok",
"files": files
})))
}
struct BackgroundedUpload<S: Store + 'static>(Value<Backgrounded>, PhantomData<S>);
2022-12-23 18:56:15 +00:00
impl<S: Store + 'static> FormData for BackgroundedUpload<S> {
type Item = Backgrounded;
2022-12-23 18:56:15 +00:00
type Error = Error;
fn form(req: &actix_web::HttpRequest) -> Form<Self::Item, Self::Error> {
// Create a new Multipart Form validator for backgrounded uploads
//
// This form is expecting a single array field, 'images' with at most 10 files in it
let repo = req
.app_data::<web::Data<ArcRepo>>()
2022-12-23 18:56:15 +00:00
.expect("No repo in request")
.clone();
let store = req
.app_data::<web::Data<S>>()
.expect("No store in request")
.clone();
let config = req
.app_data::<web::Data<Configuration>>()
.expect("No configuration in request")
.clone();
let read_only = config.server.read_only;
2022-12-23 18:56:15 +00:00
Form::new()
.max_files(config.server.max_file_count)
.max_file_size(config.media.max_file_size * MEGABYTES)
2022-12-23 18:56:15 +00:00
.transform_error(transform_error)
.field(
"images",
Field::array(Field::file(move |filename, _, stream| {
let repo = (**repo).clone();
let store = (**store).clone();
metrics::counter!("pict-rs.files", "upload" => "background").increment(1);
2023-07-22 21:47:59 +00:00
2022-12-23 18:56:15 +00:00
let span = tracing::info_span!("file-proxy", ?filename);
Box::pin(
2023-07-17 19:24:49 +00:00
async move {
if read_only {
2023-07-17 19:24:49 +00:00
return Err(UploadError::ReadOnly.into());
}
let stream = crate::stream::from_err(stream);
2023-07-17 19:24:49 +00:00
Backgrounded::proxy(repo, store, stream).await
}
.instrument(span),
2022-12-23 18:56:15 +00:00
)
})),
)
}
fn extract(value: Value<Self::Item>) -> Result<Self, Self::Error>
where
Self: Sized,
{
Ok(BackgroundedUpload(value, PhantomData))
2022-12-23 18:56:15 +00:00
}
}
#[tracing::instrument(name = "Uploaded files", skip(value, repo))]
async fn upload_backgrounded<S: Store>(
Multipart(BackgroundedUpload(value, _)): Multipart<BackgroundedUpload<S>>,
repo: web::Data<ArcRepo>,
2022-12-23 18:56:15 +00:00
) -> Result<HttpResponse, Error> {
let images = value
.map()
.and_then(|mut m| m.remove("images"))
.and_then(|images| images.array())
.ok_or(UploadError::NoFiles)?;
let mut files = Vec::new();
let images = images
.into_iter()
.filter_map(|i| i.file())
.collect::<Vec<_>>();
for image in &images {
let upload_id = image.result.upload_id().expect("Upload ID exists");
let identifier = image.result.identifier().expect("Identifier exists");
2022-12-23 18:56:15 +00:00
queue::queue_ingest(&repo, identifier, upload_id, None).await?;
2022-12-23 18:56:15 +00:00
files.push(serde_json::json!({
"upload_id": upload_id.to_string(),
}));
}
for image in images {
image.result.disarm();
}
Ok(HttpResponse::Accepted().json(&serde_json::json!({
"msg": "ok",
"uploads": files
})))
}
#[derive(Debug, serde::Deserialize)]
struct ClaimQuery {
upload_id: Serde<UploadId>,
}
/// Claim a backgrounded upload
#[tracing::instrument(name = "Waiting on upload", skip_all)]
async fn claim_upload<S: Store + 'static>(
2023-10-07 16:32:36 +00:00
tmp_dir: web::Data<ArcTmpDir>,
repo: web::Data<ArcRepo>,
2022-12-23 18:56:15 +00:00
store: web::Data<S>,
config: web::Data<Configuration>,
2022-12-23 18:56:15 +00:00
query: web::Query<ClaimQuery>,
) -> Result<HttpResponse, Error> {
let upload_id = Serde::into_inner(query.into_inner().upload_id);
2023-09-05 02:58:57 +00:00
match repo
.wait(upload_id)
.with_timeout(Duration::from_secs(10))
.await
{
2022-12-23 18:56:15 +00:00
Ok(wait_res) => {
let upload_result = wait_res?;
repo.claim(upload_id).await?;
metrics::counter!("pict-rs.background.upload.claim").increment(1);
2022-12-23 18:56:15 +00:00
match upload_result {
UploadResult::Success { alias, token } => {
2023-10-07 16:32:36 +00:00
let details = ensure_details(&tmp_dir, &repo, &store, &config, &alias).await?;
2022-12-23 18:56:15 +00:00
Ok(HttpResponse::Ok().json(&serde_json::json!({
"msg": "ok",
"files": [{
"file": alias.to_string(),
"delete_token": token.to_string(),
"details": details.into_api_details(),
2022-12-23 18:56:15 +00:00
}]
})))
}
2023-09-02 01:50:10 +00:00
UploadResult::Failure { message, code } => Ok(HttpResponse::UnprocessableEntity()
.json(&serde_json::json!({
2022-12-23 18:56:15 +00:00
"msg": message,
2023-09-02 01:50:10 +00:00
"code": code,
}))),
2022-12-23 18:56:15 +00:00
}
}
Err(_) => Ok(HttpResponse::NoContent().finish()),
}
}
#[derive(Debug, serde::Deserialize)]
struct UrlQuery {
url: String,
#[serde(default)]
backgrounded: bool,
}
async fn ingest_inline<S: Store + 'static>(
stream: impl Stream<Item = Result<web::Bytes, Error>> + 'static,
tmp_dir: &TmpDir,
repo: &ArcRepo,
2023-07-23 20:45:52 +00:00
store: &S,
2023-09-06 01:45:07 +00:00
client: &ClientWithMiddleware,
2023-07-23 20:45:52 +00:00
config: &Configuration,
) -> Result<(Alias, DeleteToken, Details), Error> {
2023-11-11 20:28:05 +00:00
let session = ingest::ingest(tmp_dir, repo, store, client, stream, None, config).await?;
2023-07-23 20:45:52 +00:00
let alias = session.alias().expect("alias should exist").to_owned();
2023-10-07 16:32:36 +00:00
let details = ensure_details(tmp_dir, repo, store, config, &alias).await?;
2023-07-23 20:45:52 +00:00
let delete_token = session.disarm();
2023-07-23 20:45:52 +00:00
Ok((alias, delete_token, details))
}
2022-12-23 18:56:15 +00:00
/// download an image from a URL
#[tracing::instrument(name = "Downloading file", skip(client, repo, store, config))]
async fn download<S: Store + 'static>(
2023-07-21 21:58:31 +00:00
client: web::Data<ClientWithMiddleware>,
tmp_dir: web::Data<ArcTmpDir>,
repo: web::Data<ArcRepo>,
2022-12-23 18:56:15 +00:00
store: web::Data<S>,
config: web::Data<Configuration>,
2022-12-23 18:56:15 +00:00
query: web::Query<UrlQuery>,
) -> Result<HttpResponse, Error> {
2023-09-06 01:45:07 +00:00
let stream = download_stream(&client, &query.url, &config).await?;
2023-07-23 20:45:52 +00:00
if query.backgrounded {
do_download_backgrounded(stream, repo, store).await
} else {
do_download_inline(stream, &tmp_dir, repo, store, &client, config).await
2023-07-23 20:45:52 +00:00
}
}
async fn download_stream(
2023-09-06 01:45:07 +00:00
client: &ClientWithMiddleware,
2023-07-23 20:45:52 +00:00
url: &str,
config: &Configuration,
) -> Result<impl Stream<Item = Result<web::Bytes, Error>> + 'static, Error> {
if config.server.read_only {
2023-07-17 19:24:49 +00:00
return Err(UploadError::ReadOnly.into());
}
2023-07-23 20:45:52 +00:00
let res = client.get(url).send().await?;
2022-12-23 18:56:15 +00:00
if !res.status().is_success() {
return Err(UploadError::Download(res.status()).into());
}
let stream = crate::stream::limit(
config.media.max_file_size * MEGABYTES,
crate::stream::from_err(res.bytes_stream()),
);
2022-12-23 18:56:15 +00:00
2023-07-23 20:45:52 +00:00
Ok(stream)
2022-12-23 18:56:15 +00:00
}
2023-09-06 01:45:07 +00:00
#[tracing::instrument(
name = "Downloading file inline",
skip(stream, repo, store, client, config)
)]
async fn do_download_inline<S: Store + 'static>(
stream: impl Stream<Item = Result<web::Bytes, Error>> + 'static,
tmp_dir: &TmpDir,
repo: web::Data<ArcRepo>,
2022-12-23 18:56:15 +00:00
store: web::Data<S>,
2023-09-06 01:45:07 +00:00
client: &ClientWithMiddleware,
config: web::Data<Configuration>,
2022-12-23 18:56:15 +00:00
) -> Result<HttpResponse, Error> {
metrics::counter!("pict-rs.files", "download" => "inline").increment(1);
2023-07-22 21:47:59 +00:00
2023-09-06 01:45:07 +00:00
let (alias, delete_token, details) =
ingest_inline(stream, tmp_dir, &repo, &store, client, &config).await?;
2022-12-23 18:56:15 +00:00
Ok(HttpResponse::Created().json(&serde_json::json!({
"msg": "ok",
"files": [{
"file": alias.to_string(),
"delete_token": delete_token.to_string(),
"details": details.into_api_details(),
2022-12-23 18:56:15 +00:00
}]
})))
}
#[tracing::instrument(name = "Downloading file in background", skip(stream, repo, store))]
async fn do_download_backgrounded<S: Store + 'static>(
stream: impl Stream<Item = Result<web::Bytes, Error>> + 'static,
repo: web::Data<ArcRepo>,
2022-12-23 18:56:15 +00:00
store: web::Data<S>,
) -> Result<HttpResponse, Error> {
metrics::counter!("pict-rs.files", "download" => "background").increment(1);
2023-07-22 21:47:59 +00:00
2022-12-23 18:56:15 +00:00
let backgrounded = Backgrounded::proxy((**repo).clone(), (**store).clone(), stream).await?;
let upload_id = backgrounded.upload_id().expect("Upload ID exists");
let identifier = backgrounded.identifier().expect("Identifier exists");
2022-12-23 18:56:15 +00:00
queue::queue_ingest(&repo, identifier, upload_id, None).await?;
2022-12-23 18:56:15 +00:00
backgrounded.disarm();
Ok(HttpResponse::Accepted().json(&serde_json::json!({
"msg": "ok",
"uploads": [{
"upload_id": upload_id.to_string(),
}]
})))
}
2023-08-28 21:02:11 +00:00
#[derive(Debug, serde::Deserialize)]
struct PageQuery {
slug: Option<String>,
2023-08-29 18:27:18 +00:00
timestamp: Option<HumanDate>,
2023-08-28 21:02:11 +00:00
limit: Option<usize>,
}
#[derive(serde::Serialize)]
struct PageJson {
limit: usize,
#[serde(skip_serializing_if = "Option::is_none")]
current: Option<String>,
2023-08-28 21:02:11 +00:00
#[serde(skip_serializing_if = "Option::is_none")]
prev: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
next: Option<String>,
hashes: Vec<HashJson>,
}
#[derive(serde::Serialize)]
struct HashJson {
hex: String,
aliases: Vec<String>,
details: Option<ApiDetails>,
2023-08-28 21:02:11 +00:00
}
/// Get a page of hashes
#[tracing::instrument(name = "Hash Page", skip(repo))]
async fn page(
repo: web::Data<ArcRepo>,
2023-08-29 18:27:18 +00:00
web::Query(PageQuery {
slug,
timestamp,
limit,
}): web::Query<PageQuery>,
2023-08-28 21:02:11 +00:00
) -> Result<HttpResponse, Error> {
let limit = limit.unwrap_or(20);
2023-08-29 18:27:18 +00:00
let page = if let Some(timestamp) = timestamp {
repo.hash_page_by_date(timestamp.timestamp, limit).await?
} else {
repo.hash_page(slug, limit).await?
};
2023-08-28 21:02:11 +00:00
let mut hashes = Vec::with_capacity(page.hashes.len());
for hash in &page.hashes {
let hex = hash.to_hex();
let aliases = repo
.aliases_for_hash(hash.clone())
2023-08-28 21:02:11 +00:00
.await?
.into_iter()
.map(|a| a.to_string())
.collect();
let identifier = repo.identifier(hash.clone()).await?;
let details = if let Some(identifier) = identifier {
repo.details(&identifier)
.await?
.map(|d| d.into_api_details())
2023-08-28 21:02:11 +00:00
} else {
None
};
hashes.push(HashJson {
hex,
aliases,
details,
});
}
let page = PageJson {
limit: page.limit,
2023-08-29 17:53:29 +00:00
current: page.current(),
2023-08-28 21:02:11 +00:00
prev: page.prev(),
next: page.next(),
hashes,
};
Ok(HttpResponse::Ok().json(serde_json::json!({
"msg": "ok",
"page": page,
})))
}
2022-12-23 18:56:15 +00:00
/// Delete aliases and files
2023-07-26 01:30:22 +00:00
#[tracing::instrument(name = "Deleting file", skip(repo, config))]
async fn delete(
repo: web::Data<ArcRepo>,
config: web::Data<Configuration>,
2022-12-23 18:56:15 +00:00
path_entries: web::Path<(String, String)>,
) -> Result<HttpResponse, Error> {
if config.server.read_only {
2023-07-17 19:24:49 +00:00
return Err(UploadError::ReadOnly.into());
}
2022-12-23 18:56:15 +00:00
let (token, alias) = path_entries.into_inner();
let token = DeleteToken::from_existing(&token);
let alias = Alias::from_existing(&alias);
// delete alias inline
queue::cleanup::alias(&repo, alias, token).await?;
2022-12-23 18:56:15 +00:00
Ok(HttpResponse::NoContent().finish())
}
#[derive(Debug, serde::Deserialize, serde::Serialize, PartialEq, Eq, PartialOrd, Ord)]
#[serde(untagged)]
enum ProcessSource {
Source { src: Serde<Alias> },
Alias { alias: Serde<Alias> },
Proxy { proxy: url::Url },
}
#[derive(Debug, serde::Deserialize, serde::Serialize, PartialEq, Eq, PartialOrd, Ord)]
struct ProcessQuery {
#[serde(flatten)]
source: ProcessSource,
#[serde(with = "tuple_vec_map", flatten)]
operations: Vec<(String, String)>,
}
2022-12-23 18:56:15 +00:00
fn prepare_process(
config: &Configuration,
operations: Vec<(String, String)>,
2022-12-23 18:56:15 +00:00
ext: &str,
) -> Result<(InputProcessableFormat, PathBuf, Vec<String>), Error> {
2022-12-23 18:56:15 +00:00
let operations = operations
.into_iter()
.filter(|(k, _)| config.media.filters.contains(&k.to_lowercase()))
2022-12-23 18:56:15 +00:00
.collect::<Vec<_>>();
let format = ext
.parse::<InputProcessableFormat>()
.map_err(|_| UploadError::UnsupportedProcessExtension)?;
2022-12-23 18:56:15 +00:00
let (thumbnail_path, thumbnail_args) =
self::processor::build_chain(&operations, &format.to_string())?;
2022-12-23 18:56:15 +00:00
Ok((format, thumbnail_path, thumbnail_args))
2022-12-23 18:56:15 +00:00
}
#[tracing::instrument(name = "Fetching derived details", skip(repo, config))]
async fn process_details<S: Store>(
web::Query(ProcessQuery { source, operations }): web::Query<ProcessQuery>,
2022-12-23 18:56:15 +00:00
ext: web::Path<String>,
repo: web::Data<ArcRepo>,
config: web::Data<Configuration>,
2022-12-23 18:56:15 +00:00
) -> Result<HttpResponse, Error> {
let alias = match source {
ProcessSource::Alias { alias } | ProcessSource::Source { src: alias } => {
Serde::into_inner(alias)
}
2023-07-23 17:10:36 +00:00
ProcessSource::Proxy { proxy } => {
let Some(alias) = repo.related(proxy).await? else {
return Ok(HttpResponse::NotFound().json(&serde_json::json!({
"msg": "No images associated with provided proxy url"
})));
};
alias
}
};
let (_, thumbnail_path, _) = prepare_process(&config, operations, ext.as_str())?;
2022-12-23 18:56:15 +00:00
let Some(hash) = repo.hash(&alias).await? else {
// Invalid alias
return Ok(HttpResponse::NotFound().json(&serde_json::json!({
"msg": "No images associated with provided alias",
})));
};
2023-07-22 23:50:04 +00:00
let thumbnail_string = thumbnail_path.to_string_lossy().to_string();
if !config.server.read_only {
2023-08-16 21:09:40 +00:00
repo.accessed_variant(hash.clone(), thumbnail_string.clone())
.await?;
2023-07-22 23:50:04 +00:00
}
2022-12-23 18:56:15 +00:00
let identifier = repo
.variant_identifier(hash, thumbnail_string)
2022-12-23 18:56:15 +00:00
.await?
.ok_or(UploadError::MissingAlias)?;
let details = repo.details(&identifier).await?;
let details = details.ok_or(UploadError::NoFiles)?;
Ok(HttpResponse::Ok().json(&details.into_api_details()))
2022-12-23 18:56:15 +00:00
}
async fn not_found_hash(repo: &ArcRepo) -> Result<Option<(Alias, Hash)>, Error> {
let Some(not_found) = repo.get(NOT_FOUND_KEY).await? else {
return Ok(None);
};
let Some(alias) = Alias::from_slice(not_found.as_ref()) else {
tracing::warn!("Couldn't parse not-found alias");
return Ok(None);
};
let Some(hash) = repo.hash(&alias).await? else {
tracing::warn!("No hash found for not-found alias");
return Ok(None);
};
Ok(Some((alias, hash)))
}
2022-12-23 18:56:15 +00:00
/// Process files
2023-07-23 21:04:17 +00:00
#[allow(clippy::too_many_arguments)]
#[tracing::instrument(
name = "Serving processed image",
skip(tmp_dir, repo, store, client, config, process_map)
)]
async fn process<S: Store + 'static>(
2022-12-23 18:56:15 +00:00
range: Option<web::Header<Range>>,
web::Query(ProcessQuery { source, operations }): web::Query<ProcessQuery>,
2022-12-23 18:56:15 +00:00
ext: web::Path<String>,
tmp_dir: web::Data<ArcTmpDir>,
repo: web::Data<ArcRepo>,
2022-12-23 18:56:15 +00:00
store: web::Data<S>,
2023-07-23 20:45:52 +00:00
client: web::Data<ClientWithMiddleware>,
config: web::Data<Configuration>,
2023-07-22 16:15:30 +00:00
process_map: web::Data<ProcessMap>,
2022-12-23 18:56:15 +00:00
) -> Result<HttpResponse, Error> {
let alias = match source {
ProcessSource::Alias { alias } | ProcessSource::Source { src: alias } => {
Serde::into_inner(alias)
}
2023-07-23 17:10:36 +00:00
ProcessSource::Proxy { proxy } => {
2023-07-23 20:45:52 +00:00
let alias = if let Some(alias) = repo.related(proxy.clone()).await? {
alias
} else if !config.server.read_only {
2023-09-06 01:45:07 +00:00
let stream = download_stream(&client, proxy.as_str(), &config).await?;
2023-07-23 20:45:52 +00:00
let (alias, _, _) =
ingest_inline(stream, &tmp_dir, &repo, &store, &client, &config).await?;
2023-07-23 20:45:52 +00:00
repo.relate_url(proxy, alias.clone()).await?;
2023-07-23 17:10:36 +00:00
alias
} else {
2023-07-23 20:45:52 +00:00
return Err(UploadError::ReadOnly.into());
};
if !config.server.read_only {
2023-08-16 21:09:40 +00:00
repo.accessed_alias(alias.clone()).await?;
2023-07-23 17:10:36 +00:00
}
2023-07-23 20:45:52 +00:00
alias
2023-07-23 17:10:36 +00:00
}
};
let (format, thumbnail_path, thumbnail_args) =
prepare_process(&config, operations, ext.as_str())?;
2022-12-23 18:56:15 +00:00
let path_string = thumbnail_path.to_string_lossy().to_string();
let (hash, alias, not_found) = if let Some(hash) = repo.hash(&alias).await? {
(hash, alias, false)
} else {
let Some((alias, hash)) = not_found_hash(&repo).await? else {
return Ok(HttpResponse::NotFound().finish());
};
(hash, alias, true)
};
2022-12-23 18:56:15 +00:00
2023-07-22 23:50:04 +00:00
if !config.server.read_only {
2023-08-16 21:09:40 +00:00
repo.accessed_variant(hash.clone(), path_string.clone())
.await?;
2023-07-22 23:50:04 +00:00
}
let identifier_opt = repo.variant_identifier(hash.clone(), path_string).await?;
2022-12-23 18:56:15 +00:00
if let Some(identifier) = identifier_opt {
2023-11-11 20:22:12 +00:00
let details =
ensure_details_identifier(&tmp_dir, &repo, &store, &config, &identifier).await?;
2022-12-23 18:56:15 +00:00
if let Some(public_url) = store.public_url(&identifier) {
return Ok(HttpResponse::SeeOther()
.insert_header((actix_web::http::header::LOCATION, public_url.as_str()))
.finish());
}
return ranged_file_resp(&store, identifier, range, details, not_found).await;
2022-12-23 18:56:15 +00:00
}
if config.server.read_only {
2023-07-17 19:24:49 +00:00
return Err(UploadError::ReadOnly.into());
}
2023-10-07 16:32:36 +00:00
let original_details = ensure_details(&tmp_dir, &repo, &store, &config, &alias).await?;
2022-12-23 18:56:15 +00:00
let (details, bytes) = generate::generate(
2023-10-07 16:32:36 +00:00
&tmp_dir,
2022-12-23 18:56:15 +00:00
&repo,
&store,
2023-07-22 16:15:30 +00:00
&process_map,
2022-12-23 18:56:15 +00:00
format,
thumbnail_path,
thumbnail_args,
&original_details,
2023-11-11 20:22:12 +00:00
&config,
2022-12-23 18:56:15 +00:00
hash,
)
.await?;
let (builder, stream) = if let Some(web::Header(range_header)) = range {
if let Some(range) = range::single_bytes_range(&range_header) {
let len = bytes.len() as u64;
if let Some(content_range) = range::to_content_range(range, len) {
let mut builder = HttpResponse::PartialContent();
builder.insert_header(content_range);
let stream = range::chop_bytes(range, bytes, len)?;
(builder, Either::left(Either::left(stream)))
} else {
(
HttpResponse::RangeNotSatisfiable(),
Either::left(Either::right(empty())),
)
}
} else {
return Err(UploadError::Range.into());
}
} else if not_found {
2023-08-23 16:59:42 +00:00
(HttpResponse::NotFound(), Either::right(once(Ok(bytes))))
2022-12-23 18:56:15 +00:00
} else {
2023-08-23 16:59:42 +00:00
(HttpResponse::Ok(), Either::right(once(Ok(bytes))))
2022-12-23 18:56:15 +00:00
};
Ok(srv_response(
builder,
stream,
details.media_type(),
2022-12-23 18:56:15 +00:00
7 * DAYS,
details.system_time(),
))
}
2023-12-22 19:30:17 +00:00
#[tracing::instrument(
name = "Serving processed image headers",
skip(tmp_dir, repo, store, config)
)]
async fn process_head<S: Store + 'static>(
2022-12-23 18:56:15 +00:00
range: Option<web::Header<Range>>,
web::Query(ProcessQuery { source, operations }): web::Query<ProcessQuery>,
2022-12-23 18:56:15 +00:00
ext: web::Path<String>,
2023-10-07 16:32:36 +00:00
tmp_dir: web::Data<ArcTmpDir>,
repo: web::Data<ArcRepo>,
2022-12-23 18:56:15 +00:00
store: web::Data<S>,
config: web::Data<Configuration>,
2022-12-23 18:56:15 +00:00
) -> Result<HttpResponse, Error> {
let alias = match source {
ProcessSource::Alias { alias } | ProcessSource::Source { src: alias } => {
Serde::into_inner(alias)
}
2023-07-23 17:10:36 +00:00
ProcessSource::Proxy { proxy } => {
let Some(alias) = repo.related(proxy).await? else {
return Ok(HttpResponse::NotFound().finish());
};
alias
}
};
let (_, thumbnail_path, _) = prepare_process(&config, operations, ext.as_str())?;
2022-12-23 18:56:15 +00:00
let path_string = thumbnail_path.to_string_lossy().to_string();
let Some(hash) = repo.hash(&alias).await? else {
// Invalid alias
return Ok(HttpResponse::NotFound().finish());
};
2023-07-22 23:50:04 +00:00
if !config.server.read_only {
2023-08-16 21:09:40 +00:00
repo.accessed_variant(hash.clone(), path_string.clone())
.await?;
2023-07-22 23:50:04 +00:00
}
let identifier_opt = repo.variant_identifier(hash.clone(), path_string).await?;
2022-12-23 18:56:15 +00:00
if let Some(identifier) = identifier_opt {
2023-11-11 20:22:12 +00:00
let details =
ensure_details_identifier(&tmp_dir, &repo, &store, &config, &identifier).await?;
2022-12-23 18:56:15 +00:00
if let Some(public_url) = store.public_url(&identifier) {
return Ok(HttpResponse::SeeOther()
.insert_header((actix_web::http::header::LOCATION, public_url.as_str()))
.finish());
}
2022-12-23 18:56:15 +00:00
return ranged_file_head_resp(&store, identifier, range, details).await;
}
Ok(HttpResponse::NotFound().finish())
}
/// Process files
#[tracing::instrument(name = "Spawning image process", skip(repo))]
async fn process_backgrounded<S: Store>(
web::Query(ProcessQuery { source, operations }): web::Query<ProcessQuery>,
2022-12-23 18:56:15 +00:00
ext: web::Path<String>,
repo: web::Data<ArcRepo>,
config: web::Data<Configuration>,
2022-12-23 18:56:15 +00:00
) -> Result<HttpResponse, Error> {
let source = match source {
ProcessSource::Alias { alias } | ProcessSource::Source { src: alias } => {
Serde::into_inner(alias)
}
2023-07-23 17:10:36 +00:00
ProcessSource::Proxy { proxy } => {
let Some(alias) = repo.related(proxy).await? else {
return Ok(HttpResponse::NotFound().finish());
};
alias
}
};
let (target_format, process_path, process_args) =
prepare_process(&config, operations, ext.as_str())?;
2022-12-23 18:56:15 +00:00
let path_string = process_path.to_string_lossy().to_string();
let Some(hash) = repo.hash(&source).await? else {
// Invalid alias
return Ok(HttpResponse::BadRequest().finish());
};
let identifier_opt = repo.variant_identifier(hash.clone(), path_string).await?;
2022-12-23 18:56:15 +00:00
if identifier_opt.is_some() {
return Ok(HttpResponse::Accepted().finish());
}
if config.server.read_only {
2023-07-17 19:24:49 +00:00
return Err(UploadError::ReadOnly.into());
}
2022-12-23 18:56:15 +00:00
queue_generate(&repo, target_format, source, process_path, process_args).await?;
Ok(HttpResponse::Accepted().finish())
}
/// Fetch file details
#[tracing::instrument(name = "Fetching query details", skip(repo, store, config))]
async fn details_query<S: Store + 'static>(
web::Query(alias_query): web::Query<AliasQuery>,
2023-10-07 16:32:36 +00:00
tmp_dir: web::Data<ArcTmpDir>,
repo: web::Data<ArcRepo>,
store: web::Data<S>,
config: web::Data<Configuration>,
) -> Result<HttpResponse, Error> {
let alias = match alias_query {
2023-07-23 17:10:36 +00:00
AliasQuery::Alias { alias } => Serde::into_inner(alias),
AliasQuery::Proxy { proxy } => {
2023-07-23 17:10:36 +00:00
let Some(alias) = repo.related(proxy).await? else {
return Ok(HttpResponse::NotFound().json(&serde_json::json!({
"msg": "Provided proxy URL has not been cached",
})));
2023-07-23 17:10:36 +00:00
};
alias
}
};
2023-10-07 16:32:36 +00:00
do_details(alias, tmp_dir, repo, store, config).await
}
2022-12-23 18:56:15 +00:00
/// Fetch file details
#[tracing::instrument(name = "Fetching details", skip(tmp_dir, repo, store, config))]
async fn details<S: Store + 'static>(
2022-12-23 18:56:15 +00:00
alias: web::Path<Serde<Alias>>,
2023-10-07 16:32:36 +00:00
tmp_dir: web::Data<ArcTmpDir>,
repo: web::Data<ArcRepo>,
2022-12-23 18:56:15 +00:00
store: web::Data<S>,
config: web::Data<Configuration>,
2022-12-23 18:56:15 +00:00
) -> Result<HttpResponse, Error> {
2023-10-07 16:32:36 +00:00
do_details(
Serde::into_inner(alias.into_inner()),
tmp_dir,
repo,
store,
config,
)
.await
}
2022-12-23 18:56:15 +00:00
async fn do_details<S: Store + 'static>(
2023-07-23 17:10:36 +00:00
alias: Alias,
2023-10-07 16:32:36 +00:00
tmp_dir: web::Data<ArcTmpDir>,
repo: web::Data<ArcRepo>,
store: web::Data<S>,
config: web::Data<Configuration>,
) -> Result<HttpResponse, Error> {
2023-10-07 16:32:36 +00:00
let details = ensure_details(&tmp_dir, &repo, &store, &config, &alias).await?;
2022-12-23 18:56:15 +00:00
Ok(HttpResponse::Ok().json(&details.into_api_details()))
2022-12-23 18:56:15 +00:00
}
/// Serve files based on alias query
2023-07-23 20:45:52 +00:00
#[tracing::instrument(name = "Serving file query", skip(repo, store, client, config))]
async fn serve_query<S: Store + 'static>(
range: Option<web::Header<Range>>,
web::Query(alias_query): web::Query<AliasQuery>,
tmp_dir: web::Data<ArcTmpDir>,
repo: web::Data<ArcRepo>,
store: web::Data<S>,
2023-07-23 20:45:52 +00:00
client: web::Data<ClientWithMiddleware>,
config: web::Data<Configuration>,
) -> Result<HttpResponse, Error> {
let alias = match alias_query {
2023-07-23 17:10:36 +00:00
AliasQuery::Alias { alias } => Serde::into_inner(alias),
AliasQuery::Proxy { proxy } => {
2023-07-23 20:45:52 +00:00
let alias = if let Some(alias) = repo.related(proxy.clone()).await? {
alias
} else if !config.server.read_only {
2023-09-06 01:45:07 +00:00
let stream = download_stream(&client, proxy.as_str(), &config).await?;
2023-07-23 20:45:52 +00:00
let (alias, _, _) =
ingest_inline(stream, &tmp_dir, &repo, &store, &client, &config).await?;
2023-07-23 20:45:52 +00:00
repo.relate_url(proxy, alias.clone()).await?;
2023-07-23 17:10:36 +00:00
alias
} else {
2023-07-23 20:45:52 +00:00
return Err(UploadError::ReadOnly.into());
};
if !config.server.read_only {
2023-08-16 21:09:40 +00:00
repo.accessed_alias(alias.clone()).await?;
2023-07-23 17:10:36 +00:00
}
2023-07-23 20:45:52 +00:00
alias
}
};
2023-10-07 16:32:36 +00:00
do_serve(range, alias, tmp_dir, repo, store, config).await
}
2022-12-23 18:56:15 +00:00
/// Serve files
#[tracing::instrument(name = "Serving file", skip(repo, store, config))]
async fn serve<S: Store + 'static>(
2022-12-23 18:56:15 +00:00
range: Option<web::Header<Range>>,
alias: web::Path<Serde<Alias>>,
2023-10-07 16:32:36 +00:00
tmp_dir: web::Data<ArcTmpDir>,
repo: web::Data<ArcRepo>,
2022-12-23 18:56:15 +00:00
store: web::Data<S>,
config: web::Data<Configuration>,
2022-12-23 18:56:15 +00:00
) -> Result<HttpResponse, Error> {
2023-07-23 17:10:36 +00:00
do_serve(
range,
Serde::into_inner(alias.into_inner()),
2023-10-07 16:32:36 +00:00
tmp_dir,
2023-07-23 17:10:36 +00:00
repo,
store,
config,
)
.await
}
2022-12-23 18:56:15 +00:00
async fn do_serve<S: Store + 'static>(
range: Option<web::Header<Range>>,
2023-07-23 17:10:36 +00:00
alias: Alias,
2023-10-07 16:32:36 +00:00
tmp_dir: web::Data<ArcTmpDir>,
repo: web::Data<ArcRepo>,
store: web::Data<S>,
config: web::Data<Configuration>,
) -> Result<HttpResponse, Error> {
let (hash, alias, not_found) = if let Some(hash) = repo.hash(&alias).await? {
2023-07-23 17:10:36 +00:00
(hash, alias, false)
} else {
let Some((alias, hash)) = not_found_hash(&repo).await? else {
return Ok(HttpResponse::NotFound().finish());
};
(hash, alias, true)
};
2022-12-23 18:56:15 +00:00
let Some(identifier) = repo.identifier(hash.clone()).await? else {
tracing::warn!("Original File identifier for hash {hash:?} is missing, queue cleanup task",);
crate::queue::cleanup_hash(&repo, hash).await?;
return Ok(HttpResponse::NotFound().finish());
};
2023-10-07 16:32:36 +00:00
let details = ensure_details(&tmp_dir, &repo, &store, &config, &alias).await?;
2022-12-23 18:56:15 +00:00
if let Some(public_url) = store.public_url(&identifier) {
return Ok(HttpResponse::SeeOther()
.insert_header((actix_web::http::header::LOCATION, public_url.as_str()))
.finish());
}
ranged_file_resp(&store, identifier, range, details, not_found).await
2022-12-23 18:56:15 +00:00
}
#[tracing::instrument(name = "Serving query file headers", skip(repo, store, config))]
async fn serve_query_head<S: Store + 'static>(
range: Option<web::Header<Range>>,
web::Query(alias_query): web::Query<AliasQuery>,
2023-10-07 16:32:36 +00:00
tmp_dir: web::Data<ArcTmpDir>,
repo: web::Data<ArcRepo>,
store: web::Data<S>,
config: web::Data<Configuration>,
) -> Result<HttpResponse, Error> {
let alias = match alias_query {
2023-07-23 17:10:36 +00:00
AliasQuery::Alias { alias } => Serde::into_inner(alias),
AliasQuery::Proxy { proxy } => {
2023-07-23 17:10:36 +00:00
let Some(alias) = repo.related(proxy).await? else {
return Ok(HttpResponse::NotFound().finish());
};
alias
}
};
2023-10-07 16:32:36 +00:00
do_serve_head(range, alias, tmp_dir, repo, store, config).await
}
#[tracing::instrument(name = "Serving file headers", skip(repo, store, config))]
async fn serve_head<S: Store + 'static>(
2022-12-23 18:56:15 +00:00
range: Option<web::Header<Range>>,
alias: web::Path<Serde<Alias>>,
2023-10-07 16:32:36 +00:00
tmp_dir: web::Data<ArcTmpDir>,
repo: web::Data<ArcRepo>,
2022-12-23 18:56:15 +00:00
store: web::Data<S>,
config: web::Data<Configuration>,
2022-12-23 18:56:15 +00:00
) -> Result<HttpResponse, Error> {
2023-07-23 17:10:36 +00:00
do_serve_head(
range,
Serde::into_inner(alias.into_inner()),
2023-10-07 16:32:36 +00:00
tmp_dir,
2023-07-23 17:10:36 +00:00
repo,
store,
config,
)
.await
}
2022-12-23 18:56:15 +00:00
async fn do_serve_head<S: Store + 'static>(
range: Option<web::Header<Range>>,
2023-07-23 17:10:36 +00:00
alias: Alias,
2023-10-07 16:32:36 +00:00
tmp_dir: web::Data<ArcTmpDir>,
repo: web::Data<ArcRepo>,
store: web::Data<S>,
config: web::Data<Configuration>,
) -> Result<HttpResponse, Error> {
let Some(identifier) = repo.identifier_from_alias(&alias).await? else {
// Invalid alias
return Ok(HttpResponse::NotFound().finish());
};
2022-12-23 18:56:15 +00:00
2023-10-07 16:32:36 +00:00
let details = ensure_details(&tmp_dir, &repo, &store, &config, &alias).await?;
2022-12-23 18:56:15 +00:00
if let Some(public_url) = store.public_url(&identifier) {
return Ok(HttpResponse::SeeOther()
.insert_header((actix_web::http::header::LOCATION, public_url.as_str()))
.finish());
}
2022-12-23 18:56:15 +00:00
ranged_file_head_resp(&store, identifier, range, details).await
}
async fn ranged_file_head_resp<S: Store + 'static>(
store: &S,
identifier: Arc<str>,
2022-12-23 18:56:15 +00:00
range: Option<web::Header<Range>>,
details: Details,
) -> Result<HttpResponse, Error> {
let builder = if let Some(web::Header(range_header)) = range {
//Range header exists - return as ranged
if let Some(range) = range::single_bytes_range(&range_header) {
let len = store.len(&identifier).await?;
if let Some(content_range) = range::to_content_range(range, len) {
let mut builder = HttpResponse::PartialContent();
builder.insert_header(content_range);
builder
} else {
HttpResponse::RangeNotSatisfiable()
}
} else {
return Err(UploadError::Range.into());
}
} else {
// no range header
HttpResponse::Ok()
};
Ok(srv_head(
builder,
details.media_type(),
2022-12-23 18:56:15 +00:00
7 * DAYS,
details.system_time(),
)
.finish())
}
async fn ranged_file_resp<S: Store + 'static>(
store: &S,
identifier: Arc<str>,
2022-12-23 18:56:15 +00:00
range: Option<web::Header<Range>>,
details: Details,
not_found: bool,
2022-12-23 18:56:15 +00:00
) -> Result<HttpResponse, Error> {
let (builder, stream) = if let Some(web::Header(range_header)) = range {
//Range header exists - return as ranged
if let Some(range) = range::single_bytes_range(&range_header) {
let len = store.len(&identifier).await?;
if let Some(content_range) = range::to_content_range(range, len) {
let mut builder = HttpResponse::PartialContent();
builder.insert_header(content_range);
(
builder,
Either::left(Either::left(
range::chop_store(range, store, &identifier, len).await?,
2022-12-23 18:56:15 +00:00
)),
)
} else {
(
HttpResponse::RangeNotSatisfiable(),
Either::left(Either::right(empty())),
)
}
} else {
return Err(UploadError::Range.into());
}
} else {
//No Range header in the request - return the entire document
let stream = crate::stream::from_err(store.to_stream(&identifier, None, None).await?);
if not_found {
(HttpResponse::NotFound(), Either::right(stream))
} else {
(HttpResponse::Ok(), Either::right(stream))
}
2022-12-23 18:56:15 +00:00
};
Ok(srv_response(
builder,
stream,
details.media_type(),
2022-12-23 18:56:15 +00:00
7 * DAYS,
details.system_time(),
))
}
// A helper method to produce responses with proper cache headers
fn srv_response<S, E>(
builder: HttpResponseBuilder,
stream: S,
ext: mime::Mime,
expires: u32,
modified: SystemTime,
) -> HttpResponse
where
S: Stream<Item = Result<web::Bytes, E>> + 'static,
E: std::error::Error + 'static,
actix_web::Error: From<E>,
{
let stream = crate::stream::timeout(Duration::from_secs(5), stream);
let stream = streem::try_from_fn(|yielder| async move {
let stream = std::pin::pin!(stream);
let mut streamer = stream.into_streamer();
while let Some(res) = streamer.next().await {
tracing::trace!("srv_response: looping");
let item = res.map_err(Error::from)??;
yielder.yield_ok(item).await;
}
Ok(()) as Result<(), actix_web::Error>
2022-12-23 18:56:15 +00:00
});
srv_head(builder, ext, expires, modified).streaming(stream)
}
// A helper method to produce responses with proper cache headers
fn srv_head(
mut builder: HttpResponseBuilder,
ext: mime::Mime,
expires: u32,
modified: SystemTime,
) -> HttpResponseBuilder {
builder
.insert_header(LastModified(modified.into()))
.insert_header(CacheControl(vec![
CacheDirective::Public,
CacheDirective::MaxAge(expires),
CacheDirective::Extension("immutable".to_owned(), None),
]))
.insert_header((ACCEPT_RANGES, "bytes"))
.content_type(ext.to_string());
builder
}
2023-12-12 22:54:41 +00:00
#[derive(serde::Serialize)]
struct PruneResponse {
complete: bool,
progress: u64,
total: u64,
}
#[derive(Debug, serde::Deserialize)]
struct PruneQuery {
force: bool,
}
#[tracing::instrument(name = "Prune missing identifiers", skip(repo))]
async fn prune_missing(
repo: web::Data<ArcRepo>,
query: Option<web::Query<PruneQuery>>,
) -> Result<HttpResponse, Error> {
let total = repo.size().await?;
let progress = if let Some(progress) = repo.get("prune-missing-queued").await? {
progress
.as_ref()
.try_into()
.map(u64::from_be_bytes)
.unwrap_or(0)
} else {
0
};
let complete = repo.get("prune-missing-complete").await?.is_some();
let started = repo.get("prune-missing-started").await?.is_some();
if !started || query.is_some_and(|q| q.force) {
queue::prune_missing(&repo).await?;
}
Ok(HttpResponse::Ok().json(PruneResponse {
complete,
progress,
total,
}))
}
#[tracing::instrument(name = "Spawning variant cleanup", skip(repo, config))]
async fn clean_variants(
repo: web::Data<ArcRepo>,
config: web::Data<Configuration>,
) -> Result<HttpResponse, Error> {
if config.server.read_only {
2023-07-17 19:24:49 +00:00
return Err(UploadError::ReadOnly.into());
}
2022-12-23 18:56:15 +00:00
queue::cleanup_all_variants(&repo).await?;
Ok(HttpResponse::NoContent().finish())
}
#[derive(Debug, serde::Deserialize)]
2023-07-23 20:45:52 +00:00
#[serde(untagged)]
enum AliasQuery {
Proxy { proxy: url::Url },
Alias { alias: Serde<Alias> },
2022-12-23 18:56:15 +00:00
}
#[tracing::instrument(name = "Setting 404 Image", skip(repo, config))]
async fn set_not_found(
json: web::Json<AliasQuery>,
repo: web::Data<ArcRepo>,
2023-07-23 20:45:52 +00:00
client: web::Data<ClientWithMiddleware>,
config: web::Data<Configuration>,
) -> Result<HttpResponse, Error> {
if config.server.read_only {
2023-07-17 19:24:49 +00:00
return Err(UploadError::ReadOnly.into());
}
let alias = match json.into_inner() {
2023-07-23 17:10:36 +00:00
AliasQuery::Alias { alias } => Serde::into_inner(alias),
2023-07-23 20:45:52 +00:00
AliasQuery::Proxy { .. } => {
return Ok(HttpResponse::BadRequest().json(serde_json::json!({
"msg": "Cannot use proxied media as Not Found image",
})));
}
};
if repo.hash(&alias).await?.is_none() {
return Ok(HttpResponse::BadRequest().json(serde_json::json!({
"msg": "No hash associated with provided alias"
})));
}
repo.set(NOT_FOUND_KEY, alias.to_bytes().into()).await?;
Ok(HttpResponse::Created().json(serde_json::json!({
"msg": "ok",
})))
}
#[tracing::instrument(name = "Purging file", skip(repo, config))]
async fn purge(
web::Query(alias_query): web::Query<AliasQuery>,
repo: web::Data<ArcRepo>,
config: web::Data<Configuration>,
2022-12-23 18:56:15 +00:00
) -> Result<HttpResponse, Error> {
if config.server.read_only {
2023-07-17 19:24:49 +00:00
return Err(UploadError::ReadOnly.into());
}
let alias = match alias_query {
2023-07-23 17:10:36 +00:00
AliasQuery::Alias { alias } => Serde::into_inner(alias),
AliasQuery::Proxy { proxy } => {
2023-07-23 17:10:36 +00:00
let Some(alias) = repo.related(proxy).await? else {
return Ok(HttpResponse::NotFound().finish());
};
alias
}
};
2022-12-23 18:56:15 +00:00
let aliases = repo.aliases_from_alias(&alias).await?;
let Some(hash) = repo.hash(&alias).await? else {
return Ok(HttpResponse::BadRequest().json(&serde_json::json!({
"msg": "No images associated with provided alias",
})));
};
2022-12-23 18:56:15 +00:00
queue::cleanup_hash(&repo, hash).await?;
Ok(HttpResponse::Ok().json(&serde_json::json!({
"msg": "ok",
"aliases": aliases.iter().map(|a| a.to_string()).collect::<Vec<_>>()
})))
}
#[tracing::instrument(name = "Deleting alias", skip(repo, config))]
async fn delete_alias(
web::Query(alias_query): web::Query<AliasQuery>,
repo: web::Data<ArcRepo>,
config: web::Data<Configuration>,
) -> Result<HttpResponse, Error> {
if config.server.read_only {
return Err(UploadError::ReadOnly.into());
}
let alias = match alias_query {
AliasQuery::Alias { alias } => Serde::into_inner(alias),
AliasQuery::Proxy { proxy } => {
let Some(alias) = repo.related(proxy).await? else {
return Ok(HttpResponse::NotFound().finish());
};
alias
}
};
if let Some(token) = repo.delete_token(&alias).await? {
queue::cleanup_alias(&repo, alias, token).await?;
} else {
return Ok(HttpResponse::NotFound().finish());
}
Ok(HttpResponse::Ok().json(&serde_json::json!({
"msg": "ok",
})))
}
2022-12-23 18:56:15 +00:00
#[tracing::instrument(name = "Fetching aliases", skip(repo))]
async fn aliases(
web::Query(alias_query): web::Query<AliasQuery>,
repo: web::Data<ArcRepo>,
2022-12-23 18:56:15 +00:00
) -> Result<HttpResponse, Error> {
let alias = match alias_query {
2023-07-23 17:10:36 +00:00
AliasQuery::Alias { alias } => Serde::into_inner(alias),
AliasQuery::Proxy { proxy } => {
2023-07-23 17:10:36 +00:00
let Some(alias) = repo.related(proxy).await? else {
return Ok(HttpResponse::NotFound().finish());
};
alias
}
};
2022-12-23 18:56:15 +00:00
let aliases = repo.aliases_from_alias(&alias).await?;
Ok(HttpResponse::Ok().json(&serde_json::json!({
"msg": "ok",
"aliases": aliases.iter().map(|a| a.to_string()).collect::<Vec<_>>()
})))
}
#[tracing::instrument(name = "Fetching identifier", skip(repo))]
async fn identifier<S: Store>(
web::Query(alias_query): web::Query<AliasQuery>,
repo: web::Data<ArcRepo>,
2022-12-23 18:56:15 +00:00
) -> Result<HttpResponse, Error> {
let alias = match alias_query {
2023-07-23 17:10:36 +00:00
AliasQuery::Alias { alias } => Serde::into_inner(alias),
AliasQuery::Proxy { proxy } => {
2023-07-23 17:10:36 +00:00
let Some(alias) = repo.related(proxy).await? else {
return Ok(HttpResponse::NotFound().finish());
};
alias
}
};
let Some(identifier) = repo.identifier_from_alias(&alias).await? else {
// Invalid alias
return Ok(HttpResponse::NotFound().json(serde_json::json!({
"msg": "No identifiers associated with provided alias"
})));
};
2022-12-23 18:56:15 +00:00
Ok(HttpResponse::Ok().json(&serde_json::json!({
"msg": "ok",
"identifier": identifier.as_ref(),
2022-12-23 18:56:15 +00:00
})))
}
2023-09-05 02:51:27 +00:00
#[tracing::instrument(skip(repo, store))]
async fn healthz<S: Store>(
repo: web::Data<ArcRepo>,
2023-07-07 17:05:42 +00:00
store: web::Data<S>,
) -> Result<HttpResponse, Error> {
2023-01-29 17:36:09 +00:00
repo.health_check().await?;
2023-07-07 17:05:42 +00:00
store.health_check().await?;
2023-01-29 17:36:09 +00:00
Ok(HttpResponse::Ok().finish())
}
2022-12-23 18:56:15 +00:00
fn transform_error(error: actix_form_data::Error) -> actix_web::Error {
let error: Error = error.into();
let error: actix_web::Error = error.into();
error
}
fn build_client() -> Result<ClientWithMiddleware, Error> {
2023-07-21 21:58:31 +00:00
let client = reqwest::Client::builder()
.user_agent("pict-rs v0.5.0-main")
.use_rustls_tls()
.build()
.map_err(UploadError::BuildClient)?;
Ok(ClientBuilder::new(client)
.with(TracingMiddleware::default())
.build())
2022-12-23 18:56:15 +00:00
}
fn configure_endpoints<S: Store + 'static, F: Fn(&mut web::ServiceConfig)>(
config: &mut web::ServiceConfig,
repo: ArcRepo,
store: S,
configuration: Configuration,
2023-07-21 21:58:31 +00:00
client: ClientWithMiddleware,
extra_config: F,
) {
config
.app_data(web::Data::new(repo))
.app_data(web::Data::new(store))
.app_data(web::Data::new(client))
.app_data(web::Data::new(configuration.clone()))
.route("/healthz", web::get().to(healthz::<S>))
.service(
web::scope("/image")
.service(
web::resource("")
.guard(guard::Post())
.route(web::post().to(upload::<S>)),
)
.service(
web::scope("/backgrounded")
.service(
web::resource("")
.guard(guard::Post())
.route(web::post().to(upload_backgrounded::<S>)),
)
.service(web::resource("/claim").route(web::get().to(claim_upload::<S>))),
)
.service(web::resource("/download").route(web::get().to(download::<S>)))
.service(
web::resource("/delete/{delete_token}/{filename}")
.route(web::delete().to(delete))
.route(web::get().to(delete)),
)
.service(
web::scope("/original")
.service(
web::resource("")
.route(web::get().to(serve_query::<S>))
.route(web::head().to(serve_query_head::<S>)),
)
.service(
web::resource("/{filename}")
.route(web::get().to(serve::<S>))
.route(web::head().to(serve_head::<S>)),
),
)
.service(
web::resource("/process.{ext}")
.route(web::get().to(process::<S>))
.route(web::head().to(process_head::<S>)),
)
.service(
web::resource("/process_backgrounded.{ext}")
.route(web::get().to(process_backgrounded::<S>)),
)
.service(
web::scope("/details")
.service(
web::scope("/original")
.service(web::resource("").route(web::get().to(details_query::<S>)))
.service(
web::resource("/{filename}").route(web::get().to(details::<S>)),
),
)
.service(
web::resource("/process.{ext}")
.route(web::get().to(process_details::<S>)),
),
),
)
.service(
web::scope("/internal")
.wrap(Internal(
configuration.server.api_key.as_ref().map(|s| s.to_owned()),
))
.service(web::resource("/import").route(web::post().to(import::<S>)))
.service(web::resource("/variants").route(web::delete().to(clean_variants)))
.service(web::resource("/purge").route(web::post().to(purge)))
.service(web::resource("/delete").route(web::post().to(delete_alias)))
.service(web::resource("/aliases").route(web::get().to(aliases)))
.service(web::resource("/identifier").route(web::get().to(identifier::<S>)))
.service(web::resource("/set_not_found").route(web::post().to(set_not_found)))
2023-08-28 21:02:11 +00:00
.service(web::resource("/hashes").route(web::get().to(page)))
2023-12-12 22:54:41 +00:00
.service(web::resource("/prune_missing").route(web::post().to(prune_missing)))
.configure(extra_config),
);
}
2022-12-23 18:56:15 +00:00
fn spawn_cleanup(repo: ArcRepo, config: &Configuration) {
2023-07-23 01:26:28 +00:00
if config.server.read_only {
return;
}
crate::sync::spawn("queue-cleanup", async move {
let mut interval = tokio::time::interval(Duration::from_secs(30));
loop {
tracing::trace!("queue_cleanup: looping");
interval.tick().await;
if let Err(e) = queue::cleanup_outdated_variants(&repo).await {
tracing::warn!(
"Failed to spawn cleanup for outdated variants:{}",
format!("\n{e}\n{e:?}")
);
}
2023-07-23 20:45:52 +00:00
if let Err(e) = queue::cleanup_outdated_proxies(&repo).await {
tracing::warn!(
"Failed to spawn cleanup for outdated proxies:{}",
format!("\n{e}\n{e:?}")
);
}
}
});
}
2023-09-06 01:45:07 +00:00
fn spawn_workers<S>(
tmp_dir: ArcTmpDir,
2023-09-06 01:45:07 +00:00
repo: ArcRepo,
store: S,
client: ClientWithMiddleware,
config: Configuration,
process_map: ProcessMap,
) where
S: Store + 'static,
{
crate::sync::spawn(
"cleanup-worker",
queue::process_cleanup(repo.clone(), store.clone(), config.clone()),
);
crate::sync::spawn(
"process-worker",
queue::process_images(tmp_dir, repo, store, client, process_map, config),
);
}
async fn launch_file_store<F: Fn(&mut web::ServiceConfig) + Send + Clone + 'static>(
tmp_dir: ArcTmpDir,
repo: ArcRepo,
store: FileStore,
2023-07-21 21:58:31 +00:00
client: ClientWithMiddleware,
config: Configuration,
extra_config: F,
) -> std::io::Result<()> {
2023-07-22 16:15:30 +00:00
let process_map = ProcessMap::new();
let address = config.server.address;
2023-07-23 01:26:28 +00:00
spawn_cleanup(repo.clone(), &config);
2022-12-23 18:56:15 +00:00
HttpServer::new(move || {
let tmp_dir = tmp_dir.clone();
2023-07-21 21:58:31 +00:00
let client = client.clone();
let store = store.clone();
2022-12-23 18:56:15 +00:00
let repo = repo.clone();
let config = config.clone();
let extra_config = extra_config.clone();
2022-12-23 18:56:15 +00:00
spawn_workers(
tmp_dir.clone(),
repo.clone(),
store.clone(),
2023-09-06 01:45:07 +00:00
client.clone(),
config.clone(),
process_map.clone(),
);
2022-12-23 18:56:15 +00:00
App::new()
.wrap(TracingLogger::default())
.wrap(Deadline)
2023-07-22 21:47:59 +00:00
.wrap(Metrics)
.wrap(Payload::new())
2023-07-22 16:15:30 +00:00
.app_data(web::Data::new(process_map.clone()))
.app_data(web::Data::new(tmp_dir))
.configure(move |sc| configure_endpoints(sc, repo, store, config, client, extra_config))
2022-12-23 18:56:15 +00:00
})
.bind(address)?
2022-12-23 18:56:15 +00:00
.run()
.await
}
2022-12-23 18:56:15 +00:00
async fn launch_object_store<F: Fn(&mut web::ServiceConfig) + Send + Clone + 'static>(
tmp_dir: ArcTmpDir,
repo: ArcRepo,
store: ObjectStore,
2023-07-21 21:58:31 +00:00
client: ClientWithMiddleware,
config: Configuration,
extra_config: F,
) -> std::io::Result<()> {
2023-07-22 16:15:30 +00:00
let process_map = ProcessMap::new();
let address = config.server.address;
2023-07-23 01:26:28 +00:00
spawn_cleanup(repo.clone(), &config);
HttpServer::new(move || {
let tmp_dir = tmp_dir.clone();
2023-07-21 21:58:31 +00:00
let client = client.clone();
let store = store.clone();
let repo = repo.clone();
let config = config.clone();
let extra_config = extra_config.clone();
spawn_workers(
tmp_dir.clone(),
repo.clone(),
store.clone(),
2023-09-06 01:45:07 +00:00
client.clone(),
config.clone(),
process_map.clone(),
);
App::new()
.wrap(TracingLogger::default())
.wrap(Deadline)
2023-07-22 21:47:59 +00:00
.wrap(Metrics)
.wrap(Payload::new())
2023-07-22 16:15:30 +00:00
.app_data(web::Data::new(process_map.clone()))
.app_data(web::Data::new(tmp_dir))
.configure(move |sc| configure_endpoints(sc, repo, store, config, client, extra_config))
})
.bind(address)?
.run()
.await
2022-12-23 18:56:15 +00:00
}
2023-10-07 16:32:36 +00:00
#[allow(clippy::too_many_arguments)]
async fn migrate_inner<S1>(
2023-10-07 16:32:36 +00:00
tmp_dir: ArcTmpDir,
2023-09-02 16:52:55 +00:00
repo: ArcRepo,
2023-07-21 21:58:31 +00:00
client: ClientWithMiddleware,
from: S1,
to: config::primitives::Store,
skip_missing_files: bool,
2023-08-05 17:41:06 +00:00
timeout: u64,
concurrency: usize,
) -> color_eyre::Result<()>
2022-12-23 18:56:15 +00:00
where
S1: Store + 'static,
2022-12-23 18:56:15 +00:00
{
match to {
config::primitives::Store::Filesystem(config::Filesystem { path }) => {
let to = FileStore::build(path.clone(), repo.clone()).await?;
2022-12-23 18:56:15 +00:00
2023-10-07 16:32:36 +00:00
migrate_store(
tmp_dir,
repo,
from,
to,
skip_missing_files,
timeout,
concurrency,
)
.await?
2022-12-23 18:56:15 +00:00
}
config::primitives::Store::ObjectStorage(config::primitives::ObjectStorage {
2022-12-23 18:56:15 +00:00
endpoint,
bucket_name,
use_path_style,
region,
access_key,
secret_key,
session_token,
signature_duration,
client_timeout,
public_endpoint,
2022-12-23 18:56:15 +00:00
}) => {
let to = ObjectStore::build(
endpoint.clone(),
bucket_name,
if use_path_style {
UrlStyle::Path
} else {
UrlStyle::VirtualHost
},
region,
access_key,
secret_key,
session_token,
signature_duration.unwrap_or(15),
client_timeout.unwrap_or(30),
public_endpoint,
2022-12-23 18:56:15 +00:00
repo.clone(),
)
.await?
.build(client);
2022-12-23 18:56:15 +00:00
2023-10-07 16:32:36 +00:00
migrate_store(
tmp_dir,
repo,
from,
to,
skip_missing_files,
timeout,
concurrency,
)
.await?
2022-12-23 18:56:15 +00:00
}
}
Ok(())
}
impl<P: AsRef<Path>, T: serde::Serialize> ConfigSource<P, T> {
/// Initialize the pict-rs configuration
///
/// This takes an optional config_file path which is a valid pict-rs configuration file, and an
/// optional save_to path, which the generated configuration will be saved into. Since many
/// parameters have defaults, it can be useful to dump a valid configuration with default values to
/// see what is available for tweaking.
///
/// When running pict-rs as a library, configuration is limited to environment variables and
/// configuration files. Commandline options are not available.
///
/// ```rust
/// fn main() -> Result<(), Box<dyn std::error::Error>> {
/// let configuration = pict_rs::ConfigSource::memory(serde_json::json!({
/// "server": {
/// "address": "127.0.0.1:8080"
/// },
/// "repo": {
/// "type": "sled",
/// "path": "./sled-repo"
/// },
/// "store": {
/// "type": "filesystem",
/// "path": "./files"
/// }
/// })).init::<&str>(None)?;
///
/// Ok(())
/// }
/// ```
pub fn init<Q: AsRef<Path>>(
self,
save_to: Option<Q>,
) -> color_eyre::Result<PictRsConfiguration> {
config::configure_without_clap(self, save_to)
}
2022-12-23 18:56:15 +00:00
}
async fn export_handler(repo: web::Data<SledRepo>) -> Result<HttpResponse, Error> {
repo.export().await?;
Ok(HttpResponse::Created().json(&serde_json::json!({
"msg": "ok"
})))
}
fn sled_extra_config(sc: &mut web::ServiceConfig, repo: SledRepo) {
sc.app_data(web::Data::new(repo))
.service(web::resource("/export").route(web::post().to(export_handler)));
}
impl PictRsConfiguration {
/// Build the pict-rs configuration from commandline arguments
///
/// This is probably not useful for 3rd party applications that handle their own commandline
pub fn build_default() -> color_eyre::Result<Self> {
config::configure()
}
/// Install the default pict-rs tracer
///
/// This is probably not useful for 3rd party applications that install their own tracing
/// subscribers.
2023-07-22 17:39:39 +00:00
pub fn install_tracing(self) -> color_eyre::Result<Self> {
init_tracing(&self.config.tracing)?;
Ok(self)
}
/// Install the configured pict-rs metrics collector
///
/// This is a no-op if pict-rs is not configured to export metrics. Applications that register
/// their own metrics collectors shouldn't call this method.
2023-07-22 21:47:59 +00:00
pub fn install_metrics(self) -> color_eyre::Result<Self> {
if let Some(addr) = self.config.metrics.prometheus_address {
PrometheusBuilder::new()
.with_http_listener(addr)
.install()?;
tracing::info!("Starting prometheus endpoint on {addr}");
2023-07-22 21:47:59 +00:00
}
Ok(self)
}
2023-10-07 16:48:58 +00:00
/// Run the pict-rs application on a tokio `LocalSet`
///
/// This must be called from within `tokio::main` directly
///
/// Example:
/// ```rust
/// #[tokio::main]
/// async fn main() -> color_eyre::Result<()> {
/// let pict_rs_server = pict_rs::ConfigSource::memory(serde_json::json!({
/// "repo": {
/// "type": "sled",
/// "path": "/tmp/pict-rs/run-on-localset/sled-repo",
/// },
/// "store": {
/// "type": "filesystem",
/// "path": "/tmp/pict-rs/run-on-localset/files",
/// },
/// }))
/// .init::<&str>(None)?
/// .run_on_localset();
///
/// let _ = tokio::time::timeout(std::time::Duration::from_secs(1), pict_rs_server).await;
///
/// Ok(())
/// }
/// ```
2023-10-07 16:48:58 +00:00
pub async fn run_on_localset(self) -> color_eyre::Result<()> {
tokio::task::LocalSet::new().run_until(self.run()).await
}
/// Run the pict-rs application
///
/// This must be called from within a tokio `LocalSet`, which is created by default for
/// actix-rt runtimes, and by tokio_uring
///
/// Example:
2024-01-24 23:41:42 +00:00
/// ```rust,ignore
/// fn main() -> color_eyre::Result<()> {
/// tokio_uring::start(async move {
/// let pict_rs_server = pict_rs::ConfigSource::memory(serde_json::json!({
/// "repo": {
/// "type": "sled",
/// "path": "/tmp/pict-rs/run/sled-repo",
/// },
/// "store": {
/// "type": "filesystem",
/// "path": "/tmp/pict-rs/run/files",
/// },
/// }))
/// .init::<&str>(None)?
/// .run();
///
/// let _ = tokio::time::timeout(std::time::Duration::from_secs(1), pict_rs_server).await;
///
/// Ok(())
/// })
/// }
/// ```
pub async fn run(self) -> color_eyre::Result<()> {
let PictRsConfiguration { config, operation } = self;
let tmp_dir = TmpDir::init(&config.server.temporary_directory).await?;
2023-10-07 16:32:36 +00:00
let client = build_client()?;
match operation {
Operation::Run => (),
Operation::MigrateStore {
skip_missing_files,
concurrency,
from,
to,
} => {
2023-09-02 16:52:55 +00:00
let repo = Repo::open(config.repo.clone()).await?.to_arc();
2023-08-16 21:39:41 +00:00
match from {
config::primitives::Store::Filesystem(config::Filesystem { path }) => {
let from = FileStore::build(path.clone(), repo.clone()).await?;
2023-08-05 17:41:06 +00:00
migrate_inner(
2023-10-07 16:32:36 +00:00
tmp_dir,
2023-08-05 17:41:06 +00:00
repo,
client,
from,
to,
skip_missing_files,
config.media.process_timeout,
concurrency,
2023-08-05 17:41:06 +00:00
)
.await?;
}
config::primitives::Store::ObjectStorage(
config::primitives::ObjectStorage {
endpoint,
bucket_name,
use_path_style,
region,
access_key,
secret_key,
session_token,
signature_duration,
client_timeout,
public_endpoint,
2022-12-23 18:56:15 +00:00
},
) => {
let from = ObjectStore::build(
endpoint,
bucket_name,
if use_path_style {
UrlStyle::Path
} else {
UrlStyle::VirtualHost
},
region,
access_key,
secret_key,
session_token,
signature_duration.unwrap_or(15),
client_timeout.unwrap_or(30),
public_endpoint,
repo.clone(),
)
.await?
.build(client.clone());
2022-12-23 18:56:15 +00:00
2023-08-05 17:41:06 +00:00
migrate_inner(
2023-10-07 16:32:36 +00:00
tmp_dir,
2023-08-05 17:41:06 +00:00
repo,
client,
from,
to,
skip_missing_files,
config.media.process_timeout,
concurrency,
2023-08-05 17:41:06 +00:00
)
.await?;
}
2022-12-23 18:56:15 +00:00
}
return Ok(());
}
2023-08-16 21:32:19 +00:00
Operation::MigrateRepo { from, to } => {
2023-09-02 16:52:55 +00:00
let from = Repo::open(from).await?.to_arc();
let to = Repo::open(to).await?.to_arc();
2023-08-16 21:32:19 +00:00
repo::migrate_repo(from, to).await?;
2023-08-16 21:34:42 +00:00
return Ok(());
2023-08-16 21:32:19 +00:00
}
2022-12-23 18:56:15 +00:00
}
2023-09-02 16:52:55 +00:00
let repo = Repo::open(config.repo.clone()).await?;
2023-08-16 21:39:41 +00:00
if config.server.read_only {
tracing::warn!("Launching in READ ONLY mode");
}
2023-07-17 19:24:49 +00:00
match config.store.clone() {
config::Store::Filesystem(config::Filesystem { path }) => {
let arc_repo = repo.to_arc();
2023-09-02 16:52:55 +00:00
let store = FileStore::build(path, arc_repo.clone()).await?;
if arc_repo.get("migrate-0.4").await?.is_none() {
if let Some(path) = config.old_repo_path() {
if let Some(old_repo) = repo_04::open(path)? {
repo::migrate_04(
2023-10-07 16:32:36 +00:00
tmp_dir.clone(),
old_repo,
arc_repo.clone(),
store.clone(),
config.clone(),
)
.await?;
arc_repo
.set("migrate-0.4", Arc::from(b"migrated".to_vec()))
.await?;
}
}
}
match repo {
Repo::Sled(sled_repo) => {
2023-10-07 16:36:49 +00:00
launch_file_store(
tmp_dir.clone(),
arc_repo,
store,
client,
config,
move |sc| sled_extra_config(sc, sled_repo.clone()),
)
.await?;
}
2023-09-03 22:11:34 +00:00
Repo::Postgres(_) => {
2023-10-07 16:36:49 +00:00
launch_file_store(tmp_dir.clone(), arc_repo, store, client, config, |_| {})
.await?;
2023-09-03 22:11:34 +00:00
}
}
2022-12-23 18:56:15 +00:00
}
config::Store::ObjectStorage(config::ObjectStorage {
2022-12-23 18:56:15 +00:00
endpoint,
bucket_name,
use_path_style,
2022-12-23 18:56:15 +00:00
region,
access_key,
secret_key,
session_token,
signature_duration,
client_timeout,
public_endpoint,
}) => {
2023-09-02 16:52:55 +00:00
let arc_repo = repo.to_arc();
let store = ObjectStore::build(
endpoint,
bucket_name,
if use_path_style {
UrlStyle::Path
} else {
UrlStyle::VirtualHost
},
region,
access_key,
secret_key,
session_token,
signature_duration,
client_timeout,
public_endpoint,
2023-09-02 16:52:55 +00:00
arc_repo.clone(),
)
.await?
.build(client.clone());
2022-12-23 18:56:15 +00:00
if arc_repo.get("migrate-0.4").await?.is_none() {
if let Some(path) = config.old_repo_path() {
if let Some(old_repo) = repo_04::open(path)? {
repo::migrate_04(
2023-10-07 16:32:36 +00:00
tmp_dir.clone(),
old_repo,
arc_repo.clone(),
store.clone(),
config.clone(),
)
.await?;
arc_repo
.set("migrate-0.4", Arc::from(b"migrated".to_vec()))
.await?;
}
}
}
match repo {
Repo::Sled(sled_repo) => {
2023-10-07 16:36:49 +00:00
launch_object_store(
tmp_dir.clone(),
arc_repo,
store,
client,
config,
move |sc| sled_extra_config(sc, sled_repo.clone()),
)
.await?;
}
2023-09-03 22:11:34 +00:00
Repo::Postgres(_) => {
2023-10-07 16:36:49 +00:00
launch_object_store(
tmp_dir.clone(),
arc_repo,
store,
client,
config,
|_| {},
)
.await?;
2023-09-03 22:11:34 +00:00
}
}
2022-12-23 18:56:15 +00:00
}
}
2023-10-07 16:36:49 +00:00
tmp_dir.cleanup().await?;
Ok(())
}
2022-12-23 18:56:15 +00:00
}
#[cfg(test)]
mod tests {
#[test]
fn source() {
let query = super::ProcessQuery {
source: super::ProcessSource::Source {
src: super::Serde::new(super::Alias::from_existing("example.png")),
},
operations: vec![("resize".into(), "200".into())],
};
let encoded = serde_urlencoded::to_string(&query).expect("Encoded");
let new_query: super::ProcessQuery = serde_urlencoded::from_str(&encoded).expect("Decoded");
// Don't compare entire query - "src" gets deserialized twice
assert_eq!(new_query.source, query.source);
assert!(new_query
.operations
.contains(&("resize".into(), "200".into())));
}
#[test]
fn alias() {
let query = super::ProcessQuery {
source: super::ProcessSource::Alias {
alias: super::Serde::new(super::Alias::from_existing("example.png")),
},
operations: vec![("resize".into(), "200".into())],
};
let encoded = serde_urlencoded::to_string(&query).expect("Encoded");
let new_query: super::ProcessQuery = serde_urlencoded::from_str(&encoded).expect("Decoded");
// Don't compare entire query - "alias" gets deserialized twice
assert_eq!(new_query.source, query.source);
assert!(new_query
.operations
.contains(&("resize".into(), "200".into())));
}
#[test]
fn url() {
let query = super::ProcessQuery {
source: super::ProcessSource::Proxy {
proxy: "http://example.com/image.png".parse().expect("valid url"),
},
operations: vec![("resize".into(), "200".into())],
};
let encoded = serde_urlencoded::to_string(&query).expect("Encoded");
let new_query: super::ProcessQuery = serde_urlencoded::from_str(&encoded).expect("Decoded");
// Don't compare entire query - "proxy" gets deserialized twice
assert_eq!(new_query.source, query.source);
assert!(new_query
.operations
.contains(&("resize".into(), "200".into())));
}
}