2
0
Fork 0
mirror of https://git.asonix.dog/asonix/pict-rs synced 2024-11-14 00:13:59 +00:00
pict-rs/src/main.rs

1297 lines
40 KiB
Rust
Raw Normal View History

use actix_form_data::{Field, Form, FormData, Multipart, Value};
2021-01-14 16:26:17 +00:00
use actix_web::{
guard,
2022-03-01 17:23:15 +00:00
http::header::{CacheControl, CacheDirective, LastModified, Range, ACCEPT_RANGES},
web, App, HttpRequest, HttpResponse, HttpResponseBuilder, HttpServer,
2021-01-14 16:26:17 +00:00
};
2021-03-10 02:51:03 +00:00
use awc::Client;
use futures_util::{
stream::{empty, once},
Stream, StreamExt, TryStreamExt,
};
2022-03-25 23:47:50 +00:00
use once_cell::sync::Lazy;
2022-09-24 19:18:49 +00:00
use rusty_s3::UrlStyle;
use std::{
2021-10-21 00:28:40 +00:00
future::ready,
path::PathBuf,
2022-03-29 18:18:47 +00:00
sync::atomic::{AtomicU64, Ordering},
time::{Duration, SystemTime},
};
use tokio::sync::Semaphore;
use tracing::{debug, info, instrument};
2021-09-14 01:22:42 +00:00
use tracing_actix_web::TracingLogger;
2021-12-03 20:21:31 +00:00
use tracing_awc::Tracing;
2021-09-16 22:51:20 +00:00
use tracing_futures::Instrument;
2020-06-06 21:41:17 +00:00
mod backgrounded;
2021-10-21 00:28:40 +00:00
mod concurrent_processor;
2020-06-07 00:54:06 +00:00
mod config;
2022-03-24 22:09:15 +00:00
mod details;
2021-10-14 00:06:53 +00:00
mod either;
2020-06-06 21:41:17 +00:00
mod error;
2021-08-31 16:02:30 +00:00
mod exiftool;
mod ffmpeg;
2021-10-23 19:14:12 +00:00
mod file;
mod generate;
mod ingest;
2021-10-21 00:28:40 +00:00
mod init_tracing;
mod magick;
mod middleware;
2021-10-21 00:28:40 +00:00
mod process;
mod processor;
mod queue;
mod range;
mod repo;
mod serde_str;
2021-10-23 04:48:56 +00:00
mod store;
mod stream;
2021-10-23 19:14:12 +00:00
mod tmp_file;
2020-06-11 16:46:00 +00:00
mod validate;
2020-06-06 21:41:17 +00:00
use crate::magick::ValidInputType;
use self::{
backgrounded::Backgrounded,
2022-03-28 04:27:07 +00:00
config::{Configuration, ImageFormat, Operation},
2022-03-24 22:09:15 +00:00
details::Details,
2021-10-14 00:06:53 +00:00
either::Either,
2021-09-14 01:22:42 +00:00
error::{Error, UploadError},
ingest::Session,
2021-10-21 00:28:40 +00:00
init_tracing::init_tracing,
2022-03-25 23:47:50 +00:00
magick::details_hint,
2021-09-14 01:22:42 +00:00
middleware::{Deadline, Internal},
queue::queue_generate,
2022-04-28 01:13:49 +00:00
repo::{
Alias, DeleteToken, FullRepo, HashRepo, IdentifierRepo, Repo, SettingsRepo, UploadId,
UploadResult,
},
2022-03-28 04:27:07 +00:00
serde_str::Serde,
2022-09-24 22:18:53 +00:00
store::{
file_store::FileStore,
object_store::{ObjectStore, ObjectStoreConfig},
Identifier, Store, StoreConfig,
},
stream::{StreamLimit, StreamTimeout},
};
const MEGABYTES: usize = 1024 * 1024;
const MINUTES: u32 = 60;
const HOURS: u32 = 60 * MINUTES;
const DAYS: u32 = 24 * HOURS;
2022-03-28 04:27:07 +00:00
static DO_CONFIG: Lazy<(Configuration, Operation)> =
Lazy::new(|| config::configure().expect("Failed to configure"));
static CONFIG: Lazy<Configuration> = Lazy::new(|| DO_CONFIG.0.clone());
static OPERATION: Lazy<Operation> = Lazy::new(|| DO_CONFIG.1.clone());
2022-04-07 17:56:40 +00:00
static PROCESS_SEMAPHORE: Lazy<Semaphore> = Lazy::new(|| {
tracing::trace_span!(parent: None, "Initialize semaphore")
.in_scope(|| Semaphore::new(num_cpus::get().saturating_sub(1).max(1)))
});
2022-04-08 18:03:00 +00:00
async fn ensure_details<R: FullRepo, S: Store + 'static>(
repo: &R,
store: &S,
alias: &Alias,
) -> Result<Details, Error> {
let identifier = repo.identifier_from_alias::<S::Identifier>(alias).await?;
let details = repo.details(&identifier).await?;
if let Some(details) = details {
debug!("details exist");
Ok(details)
} else {
debug!("generating new details from {:?}", identifier);
let hint = details_hint(alias);
let new_details = Details::from_store(store.clone(), identifier.clone(), hint).await?;
debug!("storing details for {:?}", identifier);
repo.relate_details(&identifier, &new_details).await?;
debug!("stored");
Ok(new_details)
}
}
struct Upload<R: FullRepo + 'static, S: Store + 'static>(Value<Session<R, S>>);
impl<R: FullRepo, S: Store + 'static> FormData for Upload<R, S> {
type Item = Session<R, S>;
type Error = Error;
fn form(req: &HttpRequest) -> Form<Self::Item, Self::Error> {
// Create a new Multipart Form validator
//
// This form is expecting a single array field, 'images' with at most 10 files in it
let repo = req
.app_data::<web::Data<R>>()
.expect("No repo in request")
.clone();
let store = req
.app_data::<web::Data<S>>()
.expect("No store in request")
.clone();
Form::new()
.max_files(10)
.max_file_size(CONFIG.media.max_file_size * MEGABYTES)
.transform_error(transform_error)
.field(
"images",
Field::array(Field::file(move |filename, _, stream| {
let repo = repo.clone();
let store = store.clone();
let span = tracing::info_span!("file-upload", ?filename);
let stream = stream.map_err(Error::from);
Box::pin(
async move {
ingest::ingest(&**repo, &**store, stream, None, true, false).await
}
.instrument(span),
)
})),
)
}
fn extract(value: Value<Session<R, S>>) -> Result<Self, Self::Error> {
Ok(Upload(value))
}
}
struct Import<R: FullRepo + 'static, S: Store + 'static>(Value<Session<R, S>>);
impl<R: FullRepo, S: Store + 'static> FormData for Import<R, S> {
type Item = Session<R, S>;
type Error = Error;
fn form(req: &actix_web::HttpRequest) -> Form<Self::Item, Self::Error> {
let repo = req
.app_data::<web::Data<R>>()
.expect("No repo in request")
.clone();
let store = req
.app_data::<web::Data<S>>()
.expect("No store in request")
.clone();
// 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(10)
.max_file_size(CONFIG.media.max_file_size * MEGABYTES)
.transform_error(transform_error)
.field(
"images",
Field::array(Field::file(move |filename, _, stream| {
let repo = repo.clone();
let store = store.clone();
let span = tracing::info_span!("file-import", ?filename);
let stream = stream.map_err(Error::from);
Box::pin(
async move {
ingest::ingest(
&**repo,
&**store,
stream,
Some(Alias::from_existing(&filename)),
!CONFIG.media.skip_validate_imports,
false,
)
.await
}
.instrument(span),
)
})),
)
}
fn extract(value: Value<Self::Item>) -> Result<Self, Self::Error>
where
Self: Sized,
{
Ok(Import(value))
}
}
2020-06-06 21:41:17 +00:00
/// Handle responding to succesful uploads
#[instrument(name = "Uploaded files", skip(value))]
async fn upload<R: FullRepo, S: Store + 'static>(
Multipart(Upload(value)): Multipart<Upload<R, S>>,
repo: web::Data<R>,
store: web::Data<S>,
) -> Result<HttpResponse, Error> {
handle_upload(value, repo, store).await
}
/// Handle responding to succesful uploads
#[instrument(name = "Imported files", skip(value))]
async fn import<R: FullRepo, S: Store + 'static>(
Multipart(Import(value)): Multipart<Import<R, S>>,
repo: web::Data<R>,
store: web::Data<S>,
) -> Result<HttpResponse, Error> {
handle_upload(value, repo, store).await
}
/// Handle responding to succesful uploads
#[instrument(name = "Uploaded files", skip(value))]
async fn handle_upload<R: FullRepo, S: Store + 'static>(
value: Value<Session<R, S>>,
repo: web::Data<R>,
2021-11-01 02:11:35 +00:00
store: web::Data<S>,
) -> 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();
2021-09-12 00:53:26 +00:00
let images = images
.into_iter()
.filter_map(|i| i.file())
.collect::<Vec<_>>();
2021-09-12 00:53:26 +00:00
for image in &images {
if let Some(alias) = image.result.alias() {
info!("Uploaded {} as {:?}", image.filename, alias);
2021-09-12 00:53:26 +00:00
let delete_token = image.result.delete_token().await?;
let details = ensure_details(&repo, &store, alias).await?;
2020-06-07 00:29:15 +00:00
files.push(serde_json::json!({
2022-03-26 21:49:23 +00:00
"file": alias.to_string(),
"delete_token": delete_token.to_string(),
"details": details,
2020-06-07 00:29:15 +00:00
}));
}
}
for mut image in images {
image.result.disarm();
2021-09-12 00:53:26 +00:00
}
2022-04-02 23:53:03 +00:00
2021-02-10 22:57:42 +00:00
Ok(HttpResponse::Created().json(&serde_json::json!({
"msg": "ok",
"files": files
})))
}
struct BackgroundedUpload<R: FullRepo + 'static, S: Store + 'static>(Value<Backgrounded<R, S>>);
impl<R: FullRepo, S: Store + 'static> FormData for BackgroundedUpload<R, S> {
type Item = Backgrounded<R, S>;
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<R>>()
.expect("No repo in request")
.clone();
let store = req
.app_data::<web::Data<S>>()
.expect("No store in request")
.clone();
Form::new()
.max_files(10)
.max_file_size(CONFIG.media.max_file_size * MEGABYTES)
.transform_error(transform_error)
.field(
"images",
Field::array(Field::file(move |filename, _, stream| {
let repo = (**repo).clone();
let store = (**store).clone();
let span = tracing::info_span!("file-proxy", ?filename);
let stream = stream.map_err(Error::from);
Box::pin(
async move { Backgrounded::proxy(repo, store, stream).await }
.instrument(span),
)
})),
)
}
fn extract(value: Value<Self::Item>) -> Result<Self, Self::Error>
where
Self: Sized,
{
Ok(BackgroundedUpload(value))
}
}
#[instrument(name = "Uploaded files", skip(value))]
async fn upload_backgrounded<R: FullRepo, S: Store>(
Multipart(BackgroundedUpload(value)): Multipart<BackgroundedUpload<R, S>>,
repo: web::Data<R>,
) -> 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")
.to_bytes()?;
queue::queue_ingest(&repo, identifier, upload_id, None, true, false).await?;
files.push(serde_json::json!({
2022-04-03 02:15:39 +00:00
"upload_id": upload_id.to_string(),
}));
}
for image in images {
image.result.disarm();
}
2022-04-03 02:15:39 +00:00
Ok(HttpResponse::Accepted().json(&serde_json::json!({
"msg": "ok",
2022-04-03 02:15:39 +00:00
"uploads": files
})))
}
2022-04-03 02:15:39 +00:00
#[derive(Debug, serde::Deserialize)]
struct ClaimQuery {
upload_id: Serde<UploadId>,
}
/// Claim a backgrounded upload
#[instrument(name = "Waiting on upload", skip(repo))]
2022-04-08 18:03:00 +00:00
async fn claim_upload<R: FullRepo, S: Store + 'static>(
2022-04-03 02:15:39 +00:00
repo: web::Data<R>,
2022-04-08 18:03:00 +00:00
store: web::Data<S>,
2022-04-03 02:15:39 +00:00
query: web::Query<ClaimQuery>,
) -> Result<HttpResponse, Error> {
let upload_id = Serde::into_inner(query.into_inner().upload_id);
match actix_rt::time::timeout(Duration::from_secs(10), repo.wait(upload_id)).await {
Ok(wait_res) => {
let upload_result = wait_res?;
repo.claim(upload_id).await?;
match upload_result {
UploadResult::Success { alias, token } => {
let details = ensure_details(&repo, &store, &alias).await?;
2022-04-08 18:03:00 +00:00
2022-04-03 02:15:39 +00:00
Ok(HttpResponse::Ok().json(&serde_json::json!({
"msg": "ok",
"files": [{
"file": alias.to_string(),
"delete_token": token.to_string(),
2022-04-08 18:03:00 +00:00
"details": details,
2022-04-03 02:15:39 +00:00
}]
})))
}
UploadResult::Failure { message } => Ok(HttpResponse::UnprocessableEntity().json(
&serde_json::json!({
"msg": message,
}),
)),
}
}
Err(_) => Ok(HttpResponse::NoContent().finish()),
}
}
2020-06-24 16:58:46 +00:00
#[derive(Debug, serde::Deserialize)]
struct UrlQuery {
url: String,
#[serde(default)]
backgrounded: bool,
#[serde(default)]
ephemeral: bool,
2020-06-24 16:58:46 +00:00
}
/// download an image from a URL
#[instrument(name = "Downloading file", skip(client, repo))]
async fn download<R: FullRepo + 'static, S: Store + 'static>(
client: web::Data<Client>,
repo: web::Data<R>,
2021-11-01 02:11:35 +00:00
store: web::Data<S>,
query: web::Query<UrlQuery>,
) -> Result<HttpResponse, Error> {
2022-04-08 18:16:22 +00:00
let res = client.get(&query.url).send().await?;
if !res.status().is_success() {
return Err(UploadError::Download(res.status()).into());
}
let stream = res
.map_err(Error::from)
.limit((CONFIG.media.max_file_size * MEGABYTES) as u64);
if query.backgrounded {
2022-04-08 18:16:22 +00:00
do_download_backgrounded(stream, repo, store, query.ephemeral).await
} else {
2022-04-08 18:16:22 +00:00
do_download_inline(stream, repo, store, query.ephemeral).await
}
}
2022-04-08 18:16:22 +00:00
#[instrument(name = "Downloading file inline", skip(stream))]
async fn do_download_inline<R: FullRepo + 'static, S: Store + 'static>(
2022-09-24 22:18:53 +00:00
stream: impl Stream<Item = Result<web::Bytes, Error>> + Unpin + 'static,
repo: web::Data<R>,
store: web::Data<S>,
is_cached: bool,
) -> Result<HttpResponse, Error> {
let mut session = ingest::ingest(&repo, &store, stream, None, true, is_cached).await?;
let alias = session.alias().expect("alias should exist").to_owned();
2021-09-12 00:53:26 +00:00
let delete_token = session.delete_token().await?;
let details = ensure_details(&repo, &store, &alias).await?;
session.disarm();
2021-02-10 22:57:42 +00:00
Ok(HttpResponse::Created().json(&serde_json::json!({
"msg": "ok",
"files": [{
2022-03-26 21:49:23 +00:00
"file": alias.to_string(),
"delete_token": delete_token.to_string(),
"details": details,
}]
})))
}
2022-04-08 18:16:22 +00:00
#[instrument(name = "Downloading file in background", skip(stream))]
async fn do_download_backgrounded<R: FullRepo + 'static, S: Store + 'static>(
2022-09-24 22:18:53 +00:00
stream: impl Stream<Item = Result<web::Bytes, Error>> + Unpin + 'static,
2022-04-08 00:07:30 +00:00
repo: web::Data<R>,
store: web::Data<S>,
is_cached: bool,
2022-04-08 00:07:30 +00:00
) -> Result<HttpResponse, Error> {
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")
.to_bytes()?;
queue::queue_ingest(&repo, identifier, upload_id, None, true, is_cached).await?;
2022-04-08 00:07:30 +00:00
backgrounded.disarm();
2022-04-08 00:07:30 +00:00
Ok(HttpResponse::Accepted().json(&serde_json::json!({
"msg": "ok",
"uploads": [{
"upload_id": upload_id.to_string(),
}]
})))
}
2020-06-15 02:41:45 +00:00
/// Delete aliases and files
#[instrument(name = "Deleting file", skip(repo))]
async fn delete<R: FullRepo>(
repo: web::Data<R>,
2020-06-07 00:29:15 +00:00
path_entries: web::Path<(String, String)>,
) -> Result<HttpResponse, Error> {
let (token, alias) = path_entries.into_inner();
2020-06-07 00:29:15 +00:00
2022-03-26 21:49:23 +00:00
let token = DeleteToken::from_existing(&token);
let alias = Alias::from_existing(&alias);
2022-03-26 21:49:23 +00:00
queue::cleanup_alias(&repo, alias, token).await?;
2020-06-07 00:29:15 +00:00
Ok(HttpResponse::NoContent().finish())
}
2020-06-24 16:58:46 +00:00
type ProcessQuery = Vec<(String, String)>;
2022-03-26 21:49:23 +00:00
fn prepare_process(
2020-06-24 16:58:46 +00:00
query: web::Query<ProcessQuery>,
ext: &str,
2022-03-28 04:27:07 +00:00
) -> Result<(ImageFormat, Alias, PathBuf, Vec<String>), Error> {
2020-06-24 16:58:46 +00:00
let (alias, operations) =
query
.into_inner()
.into_iter()
.fold((String::new(), Vec::new()), |(s, mut acc), (k, v)| {
if k == "src" {
(v, acc)
} else {
acc.push((k, v));
(s, acc)
}
});
2020-06-06 21:41:17 +00:00
2021-09-12 15:42:44 +00:00
if alias.is_empty() {
return Err(UploadError::MissingAlias.into());
2020-06-24 16:58:46 +00:00
}
2022-03-26 21:49:23 +00:00
let alias = Alias::from_existing(&alias);
2020-06-06 21:41:17 +00:00
let operations = operations
.into_iter()
.filter(|(k, _)| CONFIG.media.filters.contains(&k.to_lowercase()))
.collect::<Vec<_>>();
2020-06-24 16:58:46 +00:00
let format = ext
2022-03-28 04:27:07 +00:00
.parse::<ImageFormat>()
.map_err(|_| UploadError::UnsupportedFormat)?;
2021-10-21 01:13:39 +00:00
let ext = format.to_string();
let (thumbnail_path, thumbnail_args) = self::processor::build_chain(&operations, &ext)?;
2022-03-26 21:49:23 +00:00
Ok((format, alias, thumbnail_path, thumbnail_args))
}
#[instrument(name = "Fetching derived details", skip(repo))]
async fn process_details<R: FullRepo, S: Store>(
query: web::Query<ProcessQuery>,
ext: web::Path<String>,
repo: web::Data<R>,
) -> Result<HttpResponse, Error> {
let (_, alias, thumbnail_path, _) = prepare_process(query, ext.as_str())?;
let hash = repo.hash(&alias).await?;
let identifier = repo
.variant_identifier::<S::Identifier>(hash, thumbnail_path.to_string_lossy().to_string())
2021-10-19 04:37:11 +00:00
.await?
.ok_or(UploadError::MissingAlias)?;
let details = repo.details(&identifier).await?;
let details = details.ok_or(UploadError::NoFiles)?;
2021-02-10 22:57:42 +00:00
Ok(HttpResponse::Ok().json(&details))
}
/// Process files
#[instrument(name = "Serving processed image", skip(repo))]
async fn process<R: FullRepo, S: Store + 'static>(
2022-03-01 17:23:15 +00:00
range: Option<web::Header<Range>>,
query: web::Query<ProcessQuery>,
ext: web::Path<String>,
repo: web::Data<R>,
2021-11-01 02:11:35 +00:00
store: web::Data<S>,
) -> Result<HttpResponse, Error> {
let (format, alias, thumbnail_path, thumbnail_args) = prepare_process(query, ext.as_str())?;
repo.check_cached(&alias).await?;
let path_string = thumbnail_path.to_string_lossy().to_string();
let hash = repo.hash(&alias).await?;
let identifier_opt = repo
.variant_identifier::<S::Identifier>(hash.clone(), path_string)
2021-11-01 02:11:35 +00:00
.await?;
2021-10-23 04:48:56 +00:00
if let Some(identifier) = identifier_opt {
let details = repo.details(&identifier).await?;
let details = if let Some(details) = details {
debug!("details exist");
details
} else {
debug!("generating new details from {:?}", identifier);
let new_details = Details::from_store(
(**store).clone(),
identifier.clone(),
Some(ValidInputType::from_format(format)),
)
.await?;
debug!("storing details for {:?}", identifier);
repo.relate_details(&identifier, &new_details).await?;
debug!("stored");
new_details
};
2021-10-19 04:37:11 +00:00
return ranged_file_resp(&store, identifier, range, details).await;
2021-10-19 04:37:11 +00:00
}
let (details, bytes) = generate::generate(
&repo,
&store,
format,
alias,
thumbnail_path,
thumbnail_args,
hash,
)
.await?;
2021-10-19 04:37:11 +00:00
2022-03-01 17:23:15 +00:00
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) {
2021-10-19 04:37:11 +00:00
let mut builder = HttpResponse::PartialContent();
builder.insert_header(content_range);
2022-03-01 17:23:15 +00:00
let stream = range::chop_bytes(range, bytes, len)?;
2021-10-19 04:37:11 +00:00
(builder, Either::left(Either::left(stream)))
2021-10-19 04:37:11 +00:00
} else {
(
HttpResponse::RangeNotSatisfiable(),
Either::left(Either::right(empty())),
)
2021-10-19 04:37:11 +00:00
}
} else {
return Err(UploadError::Range.into());
2021-10-19 04:37:11 +00:00
}
} else {
(HttpResponse::Ok(), Either::right(once(ready(Ok(bytes)))))
};
Ok(srv_response(
builder,
stream,
details.content_type(),
7 * DAYS,
details.system_time(),
))
2020-06-24 16:58:46 +00:00
}
2022-07-17 16:15:43 +00:00
#[instrument(name = "Serving processed image headers", skip(repo))]
async fn process_head<R: FullRepo, S: Store + 'static>(
range: Option<web::Header<Range>>,
query: web::Query<ProcessQuery>,
ext: web::Path<String>,
repo: web::Data<R>,
store: web::Data<S>,
) -> Result<HttpResponse, Error> {
let (format, alias, thumbnail_path, _) = prepare_process(query, ext.as_str())?;
repo.check_cached(&alias).await?;
let path_string = thumbnail_path.to_string_lossy().to_string();
let hash = repo.hash(&alias).await?;
let identifier_opt = repo
.variant_identifier::<S::Identifier>(hash.clone(), path_string)
.await?;
if let Some(identifier) = identifier_opt {
let details = repo.details(&identifier).await?;
let details = if let Some(details) = details {
debug!("details exist");
details
} else {
debug!("generating new details from {:?}", identifier);
let new_details = Details::from_store(
(**store).clone(),
identifier.clone(),
Some(ValidInputType::from_format(format)),
)
.await?;
debug!("storing details for {:?}", identifier);
repo.relate_details(&identifier, &new_details).await?;
debug!("stored");
new_details
};
return ranged_file_head_resp(&store, identifier, range, details).await;
2022-07-17 16:15:43 +00:00
}
Ok(HttpResponse::NotFound().finish())
}
/// Process files
#[instrument(name = "Spawning image process", skip(repo))]
async fn process_backgrounded<R: FullRepo, S: Store>(
query: web::Query<ProcessQuery>,
ext: web::Path<String>,
repo: web::Data<R>,
) -> Result<HttpResponse, Error> {
let (target_format, source, process_path, process_args) = prepare_process(query, ext.as_str())?;
let path_string = process_path.to_string_lossy().to_string();
let hash = repo.hash(&source).await?;
let identifier_opt = repo
.variant_identifier::<S::Identifier>(hash.clone(), path_string)
.await?;
if identifier_opt.is_some() {
return Ok(HttpResponse::Accepted().finish());
}
queue_generate(&repo, target_format, source, process_path, process_args).await?;
Ok(HttpResponse::Accepted().finish())
}
/// Fetch file details
#[instrument(name = "Fetching details", skip(repo))]
async fn details<R: FullRepo, S: Store + 'static>(
alias: web::Path<Serde<Alias>>,
repo: web::Data<R>,
2021-11-01 02:11:35 +00:00
store: web::Data<S>,
) -> Result<HttpResponse, Error> {
2022-03-26 21:49:23 +00:00
let alias = alias.into_inner();
let details = ensure_details(&repo, &store, &alias).await?;
2021-02-10 22:57:42 +00:00
Ok(HttpResponse::Ok().json(&details))
}
2020-06-24 16:58:46 +00:00
/// Serve files
#[instrument(name = "Serving file", skip(repo))]
async fn serve<R: FullRepo, S: Store + 'static>(
2022-03-01 17:23:15 +00:00
range: Option<web::Header<Range>>,
alias: web::Path<Serde<Alias>>,
repo: web::Data<R>,
2021-11-01 02:11:35 +00:00
store: web::Data<S>,
) -> Result<HttpResponse, Error> {
2022-03-26 21:49:23 +00:00
let alias = alias.into_inner();
repo.check_cached(&alias).await?;
let identifier = repo.identifier_from_alias::<S::Identifier>(&alias).await?;
let details = ensure_details(&repo, &store, &alias).await?;
2020-06-24 16:58:46 +00:00
ranged_file_resp(&store, identifier, range, details).await
}
2021-01-14 01:08:04 +00:00
2022-07-17 16:15:43 +00:00
#[instrument(name = "Serving file headers", skip(repo))]
async fn serve_head<R: FullRepo, S: Store + 'static>(
range: Option<web::Header<Range>>,
alias: web::Path<Serde<Alias>>,
repo: web::Data<R>,
store: web::Data<S>,
) -> Result<HttpResponse, Error> {
let alias = alias.into_inner();
repo.check_cached(&alias).await?;
let identifier = repo.identifier_from_alias::<S::Identifier>(&alias).await?;
let details = ensure_details(&repo, &store, &alias).await?;
2022-07-17 16:15:43 +00:00
ranged_file_head_resp(&store, identifier, range, details).await
2022-07-17 16:15:43 +00:00
}
async fn ranged_file_head_resp<S: Store + 'static>(
store: &S,
identifier: S::Identifier,
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.content_type(),
7 * DAYS,
details.system_time(),
)
.finish())
}
async fn ranged_file_resp<S: Store + 'static>(
2021-11-01 02:11:35 +00:00
store: &S,
2021-10-23 04:48:56 +00:00
identifier: S::Identifier,
2022-03-01 17:23:15 +00:00
range: Option<web::Header<Range>>,
2021-01-14 16:26:17 +00:00
details: Details,
) -> Result<HttpResponse, Error> {
2022-03-01 17:23:15 +00:00
let (builder, stream) = if let Some(web::Header(range_header)) = range {
//Range header exists - return as ranged
2022-03-01 17:23:15 +00:00
if let Some(range) = range::single_bytes_range(&range_header) {
let len = store.len(&identifier).await?;
2022-03-01 17:23:15 +00:00
if let Some(content_range) = range::to_content_range(range, len) {
let mut builder = HttpResponse::PartialContent();
builder.insert_header(content_range);
2021-10-23 04:48:56 +00:00
(
builder,
Either::left(Either::left(
range::chop_store(range, store, &identifier, len)
.await?
.map_err(Error::from),
)),
2021-10-23 04:48:56 +00:00
)
} else {
(
HttpResponse::RangeNotSatisfiable(),
Either::left(Either::right(empty())),
)
}
} else {
return Err(UploadError::Range.into());
2021-01-14 01:08:04 +00:00
}
} else {
2021-01-14 01:08:04 +00:00
//No Range header in the request - return the entire document
let stream = store
.to_stream(&identifier, None, None)
.await?
.map_err(Error::from);
(HttpResponse::Ok(), Either::right(stream))
2021-01-14 01:08:04 +00:00
};
Ok(srv_response(
builder,
2021-10-20 23:58:32 +00:00
stream,
details.content_type(),
7 * DAYS,
details.system_time(),
))
}
2020-06-06 21:41:17 +00:00
// A helper method to produce responses with proper cache headers
fn srv_response<S, E>(
2022-07-17 16:15:43 +00:00
builder: HttpResponseBuilder,
stream: S,
ext: mime::Mime,
expires: u32,
modified: SystemTime,
) -> HttpResponse
where
2021-10-20 23:58:32 +00:00
S: Stream<Item = Result<web::Bytes, E>> + 'static,
2021-06-19 19:39:41 +00:00
E: std::error::Error + 'static,
actix_web::Error: From<E>,
2021-01-14 01:08:04 +00:00
{
let stream = stream.timeout(Duration::from_secs(5)).map(|res| match res {
Ok(Ok(item)) => Ok(item),
Ok(Err(e)) => Err(actix_web::Error::from(e)),
Err(e) => Err(Error::from(e).into()),
});
2022-07-17 16:15:43 +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
2021-02-10 22:57:42 +00:00
.insert_header(LastModified(modified.into()))
.insert_header(CacheControl(vec![
2021-01-14 01:08:04 +00:00
CacheDirective::Public,
CacheDirective::MaxAge(expires),
CacheDirective::Extension("immutable".to_owned(), None),
]))
2021-02-10 22:57:42 +00:00
.insert_header((ACCEPT_RANGES, "bytes"))
2022-07-17 16:15:43 +00:00
.content_type(ext.to_string());
builder
}
#[instrument(name = "Spawning variant cleanup", skip(repo))]
async fn clean_variants<R: FullRepo>(repo: web::Data<R>) -> Result<HttpResponse, Error> {
queue::cleanup_all_variants(&repo).await?;
Ok(HttpResponse::NoContent().finish())
}
#[derive(Debug, serde::Deserialize)]
2022-03-26 21:49:23 +00:00
struct AliasQuery {
alias: Serde<Alias>,
}
#[instrument(name = "Purging file", skip(repo))]
async fn purge<R: FullRepo>(
2022-03-26 21:49:23 +00:00
query: web::Query<AliasQuery>,
repo: web::Data<R>,
) -> Result<HttpResponse, Error> {
let alias = query.into_inner().alias;
let aliases = repo.aliases_from_alias(&alias).await?;
let hash = repo.hash(&alias).await?;
queue::cleanup_hash(&repo, hash).await?;
2021-02-10 22:57:42 +00:00
Ok(HttpResponse::Ok().json(&serde_json::json!({
"msg": "ok",
2022-03-26 21:49:23 +00:00
"aliases": aliases.iter().map(|a| a.to_string()).collect::<Vec<_>>()
})))
}
#[instrument(name = "Fetching aliases", skip(repo))]
async fn aliases<R: FullRepo>(
2022-03-26 21:49:23 +00:00
query: web::Query<AliasQuery>,
repo: web::Data<R>,
) -> Result<HttpResponse, Error> {
let alias = query.into_inner().alias;
let aliases = repo.aliases_from_alias(&alias).await?;
2021-02-10 22:57:42 +00:00
Ok(HttpResponse::Ok().json(&serde_json::json!({
"msg": "ok",
2022-03-26 21:49:23 +00:00
"aliases": aliases.iter().map(|a| a.to_string()).collect::<Vec<_>>()
})))
}
2021-10-23 04:48:56 +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
}
2021-10-21 00:28:40 +00:00
fn build_client() -> awc::Client {
Client::builder()
2021-12-03 20:21:31 +00:00
.wrap(Tracing)
.add_default_header(("User-Agent", "pict-rs v0.4.0-main"))
.finish()
}
2022-03-29 18:18:47 +00:00
fn next_worker_id() -> String {
static WORKER_ID: AtomicU64 = AtomicU64::new(0);
let next_id = WORKER_ID.fetch_add(1, Ordering::Relaxed);
format!("{}-{}", CONFIG.server.worker_id, next_id)
}
2022-09-24 22:18:53 +00:00
async fn launch<R: FullRepo + 'static, SC: StoreConfig + 'static>(
repo: R,
2022-09-24 22:18:53 +00:00
store_config: SC,
2022-03-29 01:47:46 +00:00
) -> color_eyre::Result<()> {
2022-04-02 23:53:03 +00:00
repo.requeue_in_progress(CONFIG.server.worker_id.as_bytes().to_vec())
.await?;
HttpServer::new(move || {
2022-09-24 22:18:53 +00:00
let store = store_config.clone().build();
let repo = repo.clone();
2022-03-29 18:18:47 +00:00
2022-04-07 17:56:40 +00:00
tracing::trace_span!(parent: None, "Spawn task").in_scope(|| {
actix_rt::spawn(queue::process_cleanup(
repo.clone(),
store.clone(),
next_worker_id(),
))
});
tracing::trace_span!(parent: None, "Spawn task").in_scope(|| {
actix_rt::spawn(queue::process_images(
repo.clone(),
store.clone(),
next_worker_id(),
))
});
2022-03-29 18:18:47 +00:00
App::new()
2021-09-18 21:34:23 +00:00
.wrap(TracingLogger::default())
2021-09-11 20:31:00 +00:00
.wrap(Deadline)
.app_data(web::Data::new(repo))
2022-03-29 18:18:47 +00:00
.app_data(web::Data::new(store))
.app_data(web::Data::new(build_client()))
.service(
web::scope("/image")
.service(
web::resource("")
.guard(guard::Post())
2022-09-24 22:18:53 +00:00
.route(web::post().to(upload::<R, SC::Store>)),
)
.service(
2022-04-03 02:15:39 +00:00
web::scope("/backgrounded")
.service(
web::resource("")
.guard(guard::Post())
2022-09-24 22:18:53 +00:00
.route(web::post().to(upload_backgrounded::<R, SC::Store>)),
2022-04-03 02:15:39 +00:00
)
.service(
2022-09-24 22:18:53 +00:00
web::resource("/claim")
.route(web::get().to(claim_upload::<R, SC::Store>)),
2022-04-03 02:15:39 +00:00
),
)
2022-09-24 22:18:53 +00:00
.service(
web::resource("/download").route(web::get().to(download::<R, SC::Store>)),
)
2020-06-07 00:29:15 +00:00
.service(
web::resource("/delete/{delete_token}/{filename}")
.route(web::delete().to(delete::<R>))
.route(web::get().to(delete::<R>)),
)
.service(
2022-07-17 16:15:43 +00:00
web::resource("/original/{filename}")
2022-09-24 22:18:53 +00:00
.route(web::get().to(serve::<R, SC::Store>))
.route(web::head().to(serve_head::<R, SC::Store>)),
2022-07-17 16:15:43 +00:00
)
.service(
web::resource("/process.{ext}")
2022-09-24 22:18:53 +00:00
.route(web::get().to(process::<R, SC::Store>))
.route(web::head().to(process_head::<R, SC::Store>)),
2020-06-07 00:29:15 +00:00
)
.service(
web::resource("/process_backgrounded.{ext}")
2022-09-24 22:18:53 +00:00
.route(web::get().to(process_backgrounded::<R, SC::Store>)),
)
.service(
web::scope("/details")
.service(
2021-10-23 04:48:56 +00:00
web::resource("/original/{filename}")
2022-09-24 22:18:53 +00:00
.route(web::get().to(details::<R, SC::Store>)),
)
.service(
web::resource("/process.{ext}")
2022-09-24 22:18:53 +00:00
.route(web::get().to(process_details::<R, SC::Store>)),
),
),
)
.service(
2020-07-11 21:28:49 +00:00
web::scope("/internal")
2022-03-28 04:27:07 +00:00
.wrap(Internal(
CONFIG.server.api_key.as_ref().map(|s| s.to_owned()),
))
2022-09-24 22:18:53 +00:00
.service(web::resource("/import").route(web::post().to(import::<R, SC::Store>)))
.service(
web::resource("/variants").route(web::delete().to(clean_variants::<R>)),
)
.service(web::resource("/purge").route(web::post().to(purge::<R>)))
.service(web::resource("/aliases").route(web::get().to(aliases::<R>))),
)
})
2022-03-28 04:27:07 +00:00
.bind(CONFIG.server.address)?
.run()
.await?;
2022-04-28 01:13:49 +00:00
self::tmp_file::remove_tmp_dir().await?;
2021-10-23 19:14:12 +00:00
Ok(())
}
2021-10-23 04:48:56 +00:00
2022-09-24 22:18:53 +00:00
async fn migrate_inner<S1>(repo: &Repo, from: S1, to: config::Store) -> color_eyre::Result<()>
2021-11-01 02:11:35 +00:00
where
S1: Store,
{
match to {
2022-03-28 04:27:07 +00:00
config::Store::Filesystem(config::Filesystem { path }) => {
2022-09-24 22:18:53 +00:00
let to = FileStore::build(path.clone(), repo.clone()).await?.build();
2022-09-24 19:18:49 +00:00
match repo {
Repo::Sled(repo) => migrate_store(repo, from, to).await?,
}
2021-11-01 02:11:35 +00:00
}
2022-03-28 04:27:07 +00:00
config::Store::ObjectStorage(config::ObjectStorage {
endpoint,
2021-11-01 02:11:35 +00:00
bucket_name,
2022-09-24 19:18:49 +00:00
use_path_style,
2021-11-01 02:11:35 +00:00
region,
access_key,
secret_key,
session_token,
2022-03-25 23:47:50 +00:00
}) => {
2021-11-01 02:11:35 +00:00
let to = ObjectStore::build(
2022-09-24 19:18:49 +00:00
endpoint.clone(),
2021-11-01 02:11:35 +00:00
bucket_name,
2022-09-24 22:18:53 +00:00
if use_path_style {
2022-09-24 19:18:49 +00:00
UrlStyle::Path
} else {
UrlStyle::VirtualHost
},
2022-09-24 22:18:53 +00:00
region,
access_key,
secret_key,
session_token,
2022-03-26 21:49:23 +00:00
repo.clone(),
)
2022-09-24 22:18:53 +00:00
.await?
.build();
2022-09-24 19:18:49 +00:00
match repo {
Repo::Sled(repo) => migrate_store(repo, from, to).await?,
}
2021-11-01 02:11:35 +00:00
}
}
Ok(())
}
2021-10-23 04:48:56 +00:00
#[actix_rt::main]
2022-03-29 01:47:46 +00:00
async fn main() -> color_eyre::Result<()> {
2022-03-28 04:27:07 +00:00
init_tracing(&CONFIG.tracing)?;
2021-10-23 04:48:56 +00:00
2022-03-28 04:27:07 +00:00
let repo = Repo::open(CONFIG.repo.clone())?;
2022-04-03 20:07:31 +00:00
repo.from_db(CONFIG.old_db.path.clone()).await?;
2021-11-01 02:11:35 +00:00
2022-03-28 04:27:07 +00:00
match (*OPERATION).clone() {
Operation::Run => (),
Operation::MigrateStore { from, to } => {
2022-03-25 23:47:50 +00:00
match from {
2022-03-28 04:27:07 +00:00
config::Store::Filesystem(config::Filesystem { path }) => {
2022-09-24 22:18:53 +00:00
let from = FileStore::build(path.clone(), repo.clone()).await?.build();
migrate_inner(&repo, from, to).await?;
2022-03-25 23:47:50 +00:00
}
2022-03-28 04:27:07 +00:00
config::Store::ObjectStorage(config::ObjectStorage {
endpoint,
2021-11-01 02:11:35 +00:00
bucket_name,
2022-09-24 19:18:49 +00:00
use_path_style,
2022-03-25 23:47:50 +00:00
region,
access_key,
secret_key,
session_token,
}) => {
let from = ObjectStore::build(
2022-09-24 22:18:53 +00:00
endpoint,
bucket_name,
if use_path_style {
2022-09-24 19:18:49 +00:00
UrlStyle::Path
} else {
UrlStyle::VirtualHost
},
2022-09-24 22:18:53 +00:00
region,
access_key,
secret_key,
2022-03-25 23:47:50 +00:00
session_token,
2022-03-26 21:49:23 +00:00
repo.clone(),
)
2022-09-24 22:18:53 +00:00
.await?
.build();
2022-03-25 23:47:50 +00:00
2022-09-24 22:18:53 +00:00
migrate_inner(&repo, from, to).await?;
2022-03-25 23:47:50 +00:00
}
2021-11-01 02:11:35 +00:00
}
2022-03-25 23:47:50 +00:00
return Ok(());
}
2021-11-01 02:11:35 +00:00
}
2021-10-23 04:48:56 +00:00
2022-03-28 04:27:07 +00:00
match CONFIG.store.clone() {
config::Store::Filesystem(config::Filesystem { path }) => {
2022-04-24 22:10:15 +00:00
repo.migrate_identifiers().await?;
let store = FileStore::build(path, repo.clone()).await?;
match repo {
2022-09-24 19:18:49 +00:00
Repo::Sled(sled_repo) => launch::<_, FileStore>(sled_repo, store).await,
}
}
2022-03-28 04:27:07 +00:00
config::Store::ObjectStorage(config::ObjectStorage {
endpoint,
bucket_name,
2022-09-24 19:18:49 +00:00
use_path_style,
region,
access_key,
secret_key,
session_token,
2022-03-25 23:47:50 +00:00
}) => {
let store = ObjectStore::build(
2022-09-24 22:18:53 +00:00
endpoint,
bucket_name,
if use_path_style {
2022-09-24 19:18:49 +00:00
UrlStyle::Path
} else {
UrlStyle::VirtualHost
},
2022-09-24 22:18:53 +00:00
region,
access_key,
secret_key,
2022-03-25 23:47:50 +00:00
session_token,
repo.clone(),
2022-03-26 21:49:23 +00:00
)
.await?;
match repo {
2022-09-24 22:18:53 +00:00
Repo::Sled(sled_repo) => launch::<_, ObjectStoreConfig>(sled_repo, store).await,
}
}
}
2021-10-23 04:48:56 +00:00
}
const STORE_MIGRATION_PROGRESS: &str = "store-migration-progress";
async fn migrate_store<R, S1, S2>(repo: &R, from: S1, to: S2) -> Result<(), Error>
where
S1: Store,
S2: Store,
R: IdentifierRepo + HashRepo + SettingsRepo,
{
let stream = repo.hashes().await;
let mut stream = Box::pin(stream);
while let Some(hash) = stream.next().await {
let hash = hash?;
if let Some(identifier) = repo
.motion_identifier(hash.as_ref().to_vec().into())
.await?
{
let new_identifier = migrate_file(&from, &to, &identifier).await?;
migrate_details(repo, identifier, &new_identifier).await?;
repo.relate_motion_identifier(hash.as_ref().to_vec().into(), &new_identifier)
.await?;
}
for (variant, identifier) in repo.variants(hash.as_ref().to_vec().into()).await? {
let new_identifier = migrate_file(&from, &to, &identifier).await?;
migrate_details(repo, identifier, &new_identifier).await?;
repo.relate_variant_identifier(hash.as_ref().to_vec().into(), variant, &new_identifier)
.await?;
}
let identifier = repo.identifier(hash.as_ref().to_vec().into()).await?;
let new_identifier = migrate_file(&from, &to, &identifier).await?;
migrate_details(repo, identifier, &new_identifier).await?;
repo.relate_identifier(hash.as_ref().to_vec().into(), &new_identifier)
.await?;
repo.set(STORE_MIGRATION_PROGRESS, hash.as_ref().to_vec().into())
.await?;
}
// clean up the migration key to avoid interfering with future migrations
repo.remove(STORE_MIGRATION_PROGRESS).await?;
Ok(())
}
async fn migrate_file<S1, S2>(
from: &S1,
to: &S2,
identifier: &S1::Identifier,
) -> Result<S2::Identifier, Error>
where
S1: Store,
S2: Store,
{
let stream = from.to_stream(identifier, None, None).await?;
2022-09-24 22:18:53 +00:00
let new_identifier = to.save_stream(stream).await?;
Ok(new_identifier)
}
async fn migrate_details<R, I1, I2>(repo: &R, from: I1, to: &I2) -> Result<(), Error>
where
R: IdentifierRepo,
I1: Identifier,
I2: Identifier,
{
if let Some(details) = repo.details(&from).await? {
repo.relate_details(to, &details).await?;
repo.cleanup(&from).await?;
}
Ok(())
}