//! Object-storage side of preview generation: read the recorded state, lazily
//! build a derivative when it is missing, and clean up the encoding that is no
//! longer in use.
//!
//! Everything here is best-effort by design. A missing derivative, a lost lock,
//! or a failed render all resolve to `None` so the caller keeps serving the
//! original — a preview is an optimisation, never a reason for an upload or a
//! read to fail.

use imaging::layout::{
    PREVIEW_LOCK_TTL_SECS, PreviewState, preview_candidate_keys, preview_from_record, preview_key,
    preview_key_base, preview_lock_key,
};
use imaging::preset::PreviewPreset;
use imaging::{PreviewOutcome, render_preview};
use serde_json::Value;
use storage::S3Client;

use crate::redis::{self, AppRedis};

/// Look up an already-stored derivative. `key_base` comes from
/// [`preview_key_base`]; because the encoder choice (JPEG vs WebP) is not
/// derivable from the original's extension, both candidates are probed.
pub async fn find_stored_preview(client: &S3Client, key_base: &str) -> Option<PreviewState> {
    for candidate in preview_candidate_keys(key_base) {
        match client.head(&candidate).await {
            Ok(Some((bytes, content_type))) => {
                let content_type = if content_type.is_empty() {
                    "image/jpeg".to_string()
                } else {
                    content_type
                };
                return Some(PreviewState::ready(candidate, &content_type, bytes));
            }
            Ok(None) => continue,
            Err(error) => {
                tracing::warn!(%error, key = %candidate, "failed to head image preview object");
                continue;
            }
        }
    }
    None
}

/// Resolve the derivative for one original, generating it on first demand.
///
/// `recorded` is whatever the owner already knows from its own metadata; passing
/// `None` means "nothing recorded", which is exactly the legacy-row case this is
/// meant to heal without a batch job.
pub async fn ensure_preview(
    client: &S3Client,
    redis: &AppRedis,
    key_base: &str,
    original_key: &str,
    content_type: &str,
    preset: PreviewPreset,
    recorded: Option<&PreviewState>,
) -> Option<PreviewState> {
    match recorded {
        Some(PreviewState::Ready(ready)) => {
            // Metadata can be ahead of reality if an object was removed by hand;
            // verify once and fall through to regeneration when it is gone.
            match client.head(&ready.key).await {
                Ok(Some((bytes, stored))) if bytes > 0 => {
                    let content_type = if stored.is_empty() {
                        ready.content_type.clone()
                    } else {
                        stored
                    };
                    return Some(PreviewState::ready(ready.key.clone(), &content_type, bytes));
                }
                Ok(_) => {}
                Err(error) => {
                    tracing::warn!(%error, key = %ready.key, "failed to head recorded image preview");
                    return Some(PreviewState::Ready(ready.clone()));
                }
            }
        }
        Some(PreviewState::Skipped(reason)) => {
            tracing::debug!(%reason, key = %key_base, "image preview previously skipped; serving original");
            return None;
        }
        None => {}
    }

    // Single-flight: one request decodes, the rest wait on the same object
    // appearing (they re-check the recorded state on their next read anyway).
    let lock = preview_lock_key(key_base);
    let acquired = redis::set_nx_ex(redis, &lock, "1", PREVIEW_LOCK_TTL_SECS)
        .await
        .unwrap_or(true);
    if !acquired {
        tracing::debug!(key = %key_base, "image preview generation already in flight");
        return find_stored_preview(client, key_base).await;
    }

    let outcome = generate_preview(client, key_base, original_key, content_type, preset).await;
    if let Err(error) = redis::delete(redis, &lock).await {
        tracing::warn!(%error, key = %lock, "failed to release image preview lock");
    }
    outcome
}

/// Read the original, render a derivative, store it, and drop the other
/// encoding so a JPEG->WebP switch cannot leave two "current" previews behind.
async fn generate_preview(
    client: &S3Client,
    key_base: &str,
    original_key: &str,
    content_type: &str,
    preset: PreviewPreset,
) -> Option<PreviewState> {
    if !imaging::is_previewable_content_type(content_type) {
        return None;
    }
    let source = match client.get(original_key).await {
        Ok(bytes) => bytes,
        Err(error) => {
            tracing::warn!(%error, key = %original_key, "failed to read image for preview");
            return None;
        }
    };
    let rendered = render_preview(source, content_type.to_string(), preset).await;
    let preview = match rendered {
        PreviewOutcome::Preview(preview) => preview,
        PreviewOutcome::UseOriginal => {
            tracing::debug!(key = %original_key, "preview would not shrink the original; skipping");
            return None;
        }
        PreviewOutcome::Skipped(reason) => {
            tracing::debug!(%reason, key = %original_key, "image preview skipped");
            return None;
        }
    };
    let key = preview_key(key_base, preview.extension);
    if let Err(error) = client
        .put_with_content_type(&key, &preview.bytes, preview.content_type)
        .await
    {
        tracing::warn!(%error, key = %key, "failed to store image preview");
        return None;
    }
    for stale in preview_candidate_keys(key_base) {
        if stale == key {
            continue;
        }
        if let Err(error) = client.delete(&stale).await {
            tracing::warn!(%error, key = %stale, "failed to delete stale image preview");
        }
    }
    tracing::info!(
        key = %key,
        bytes = preview.bytes.len(),
        width = preview.width,
        height = preview.height,
        "image preview generated"
    );
    Some(PreviewState::ready(
        key,
        preview.content_type,
        preview.bytes.len() as u64,
    ))
}

/// Merge a preview state into a JSON metadata record (avatar / space sidecars).
#[must_use]
pub fn record_preview(mut record: Value, state: &PreviewState) -> Value {
    imaging::layout::preview_into_record(&mut record, state);
    record
}

/// Read the preview state a JSON record carries, if any.
#[must_use]
pub fn recorded_preview(record: &Value) -> Option<PreviewState> {
    preview_from_record(record)
}

/// Convenience: derive the derivative base for a stored original key.
#[must_use]
pub fn preview_key_base_of(original_key: &str) -> String {
    preview_key_base(original_key)
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    #[test]
    fn record_helpers_round_trip() {
        let state = PreviewState::ready("avatars/a.preview.webp".into(), "image/webp", 9);
        let record = record_preview(json!({ "uid": "a" }), &state);
        assert_eq!(recorded_preview(&record), Some(state));
        assert_eq!(record["uid"], json!("a"));
    }

    #[test]
    fn key_base_helper_matches_layout() {
        assert_eq!(preview_key_base_of("avatars/a.png"), "avatars/a.preview");
    }
}