use std::io::Cursor; use std::path::Path; use anyhow::Context; use chrono::{DateTime, NaiveDateTime, Utc}; use image::codecs::jpeg::JpegEncoder; use image::imageops::FilterType; use image::DynamicImage; use serde::Deserialize; use uuid::Uuid; use crate::models::{Photo, PhotoStatus}; use crate::s3; use crate::state::AppState; const PREVIEW_EDGE: u32 = 2048; const THUMB_EDGE: u32 = 512; pub const RAW_EXTENSIONS: &[&str] = &[ "3fr", "arw", "cr2", "cr3", "dng", "erf", "iiq", "kdc", "mef", "mos", "nef", "nrw", "orf", "pef", "raf", "raw", "rw2", "rwl", "srw", "x3f", ]; pub fn is_raw_filename(filename: &str) -> bool { Path::new(filename) .extension() .and_then(|e| e.to_str()) .map(|e| RAW_EXTENSIONS.contains(&e.to_lowercase().as_str())) .unwrap_or(false) } #[derive(Deserialize)] struct ProcessPayload { photo_id: Uuid, } pub async fn process_photo_job(state: &AppState, payload: &serde_json::Value) -> anyhow::Result<()> { let payload: ProcessPayload = serde_json::from_value(payload.clone())?; process_photo(state, payload.photo_id).await } pub async fn process_photo(state: &AppState, photo_id: Uuid) -> anyhow::Result<()> { let photo: Option = sqlx::query_as("select * from photos where id = $1") .bind(photo_id) .fetch_optional(&state.db) .await?; let Some(photo) = photo else { tracing::warn!("photo {photo_id} no longer exists, skipping"); return Ok(()); }; sqlx::query("update photos set status = $2, error = null where id = $1") .bind(photo_id) .bind(PhotoStatus::Processing.as_str()) .execute(&state.db) .await?; let dir = tempfile::tempdir().context("creating temp dir")?; let extension = Path::new(&photo.filename) .extension() .and_then(|e| e.to_str()) .unwrap_or("bin") .to_lowercase(); let src_path = dir.path().join(format!("original.{extension}")); let (sha256, crc32) = download(state, &s3::original_key(photo_id, &photo.filename), &src_path).await?; if photo.sha256.is_none() || photo.crc32.is_none() { // Best-effort backfill for pre-hash photos; a duplicate in the same // album trips the unique index, which is fine — skip silently. if let Err(e) = sqlx::query( "update photos set sha256 = coalesce(sha256, $2), crc32 = coalesce(crc32, $3) where id = $1", ) .bind(photo_id) .bind(&sha256) .bind(crc32) .execute(&state.db) .await { tracing::debug!("hash backfill skipped for {photo_id}: {e}"); } } let meta = exif_metadata(&src_path).await?; let render_input = if RAW_EXTENSIONS.contains(&extension.as_str()) { extract_embedded_jpeg(&src_path).await? } else { tokio::fs::read(&src_path).await.context("reading original")? }; let orientation = meta.orientation; let (preview, thumb, width, height) = tokio::task::spawn_blocking(move || render(&render_input, orientation)) .await .context("render task panicked")??; s3::put_bytes(state, &s3::preview_key(photo_id), preview, "image/jpeg").await?; s3::put_bytes(state, &s3::thumb_key(photo_id), thumb, "image/jpeg").await?; let updated = sqlx::query( "update photos set status = $5, error = null, width = $2, height = $3, taken_at = coalesce($4, taken_at), processed_at = now() where id = $1", ) .bind(photo_id) .bind(width as i32) .bind(height as i32) .bind(meta.taken_at) .bind(PhotoStatus::Ready.as_str()) .execute(&state.db) .await?; if updated.rows_affected() == 0 { // Photo was deleted while we were processing; its delete_s3_prefix job // may already have run, so remove the derivatives we just re-created. tracing::warn!("photo {photo_id} deleted during processing; cleaning up derivatives"); s3::delete_prefix(state, &s3::photo_prefix(photo_id)).await?; } Ok(()) } /// Download the original, hashing along the way so legacy photos (uploaded /// before hashes existed) can be backfilled. async fn download(state: &AppState, key: &str, path: &Path) -> anyhow::Result<(String, i64)> { use sha2::Digest; use tokio::io::{AsyncReadExt, AsyncWriteExt}; let object = state .s3 .get_object() .bucket(&state.config.s3_bucket) .key(key) .send() .await .with_context(|| format!("fetching s3://{}/{key}", state.config.s3_bucket))?; let mut reader = object.body.into_async_read(); let mut file = tokio::fs::File::create(path).await?; let mut sha = sha2::Sha256::new(); let mut crc = crc32fast::Hasher::new(); let mut buf = vec![0u8; 128 * 1024]; loop { let n = reader.read(&mut buf).await?; if n == 0 { break; } sha.update(&buf[..n]); crc.update(&buf[..n]); file.write_all(&buf[..n]).await?; } file.flush().await?; Ok((hex::encode(sha.finalize()), i64::from(crc.finalize()))) } #[derive(Default)] struct ExifMeta { orientation: u32, taken_at: Option>, } async fn exif_metadata(path: &Path) -> anyhow::Result { let output = tokio::process::Command::new("exiftool") .args([ "-j", "-d", "%Y-%m-%dT%H:%M:%S", "-Orientation#", "-DateTimeOriginal", "-CreateDate", "-OffsetTimeOriginal", "-OffsetTime", ]) .arg(path) .output() .await; let output = match output { Err(e) if e.kind() == std::io::ErrorKind::NotFound => { anyhow::bail!("exiftool is not installed or not on PATH") } other => other.context("running exiftool")?, }; if !output.status.success() { tracing::warn!( "exiftool metadata read failed: {}", String::from_utf8_lossy(&output.stderr) ); return Ok(ExifMeta::default()); } let parsed: Vec = serde_json::from_slice(&output.stdout).context("parsing exiftool json")?; let entry = parsed.first().cloned().unwrap_or_default(); let orientation = entry .get("Orientation") .and_then(|v| v.as_u64()) .map(|v| v as u32) .unwrap_or(1); let offset = ["OffsetTimeOriginal", "OffsetTime"] .iter() .find_map(|field| entry.get(*field).and_then(|v| v.as_str())); let taken_at = ["DateTimeOriginal", "CreateDate"] .iter() .filter_map(|field| entry.get(*field).and_then(|v| v.as_str())) .find_map(|s| parse_exif_datetime(s, offset)); Ok(ExifMeta { orientation, taken_at, }) } /// EXIF datetimes are camera-local wall-clock time; apply the EXIF offset tag /// when the camera recorded one, otherwise fall back to treating it as UTC. fn parse_exif_datetime(s: &str, offset: Option<&str>) -> Option> { if let Some(offset) = offset { if let Ok(dt) = DateTime::parse_from_str(&format!("{s}{offset}"), "%Y-%m-%dT%H:%M:%S%:z") { return Some(dt.with_timezone(&Utc)); } } NaiveDateTime::parse_from_str(s, "%Y-%m-%dT%H:%M:%S") .ok() .map(|naive| naive.and_utc()) } /// Extract the largest embedded JPEG preview from a raw file using exiftool. async fn extract_embedded_jpeg(path: &Path) -> anyhow::Result> { let mut best: Vec = Vec::new(); for tag in ["-JpgFromRaw", "-PreviewImage", "-OtherImage", "-ThumbnailImage"] { let output = tokio::process::Command::new("exiftool") .args(["-b", tag]) .arg(path) .output() .await .context("running exiftool")?; if output.status.success() && output.stdout.len() > best.len() { best = output.stdout; } // A full-size embedded preview is comfortably above this; stop early. if best.len() > 200_000 { break; } } anyhow::ensure!( best.len() > 1_000, "no usable embedded preview found in raw file" ); Ok(best) } fn render(bytes: &[u8], orientation: u32) -> anyhow::Result<(Vec, Vec, u32, u32)> { let img = image::load_from_memory(bytes).context("decoding image")?; let img = apply_orientation(img, orientation); let (width, height) = (img.width(), img.height()); let preview = if width.max(height) > PREVIEW_EDGE { img.resize(PREVIEW_EDGE, PREVIEW_EDGE, FilterType::Triangle) } else { img }; let thumb = preview.resize(THUMB_EDGE, THUMB_EDGE, FilterType::Lanczos3); Ok(( encode_jpeg(&preview, 86)?, encode_jpeg(&thumb, 82)?, width, height, )) } fn encode_jpeg(img: &DynamicImage, quality: u8) -> anyhow::Result> { let rgb = img.to_rgb8(); let mut buf = Cursor::new(Vec::new()); let encoder = JpegEncoder::new_with_quality(&mut buf, quality); rgb.write_with_encoder(encoder).context("encoding jpeg")?; Ok(buf.into_inner()) } fn apply_orientation(img: DynamicImage, orientation: u32) -> DynamicImage { match orientation { 2 => img.fliph(), 3 => img.rotate180(), 4 => img.flipv(), 5 => img.rotate90().fliph(), 6 => img.rotate90(), 7 => img.rotate270().fliph(), 8 => img.rotate270(), _ => img, } }