macltok

use std::{

    borrow::Cow,

    collections::{HashMap, VecDeque},

    io::Read as _,

    panic::AssertUnwindSafe,

    sync::Arc,

    time::Duration,

};

use futures_util::{

    FutureExt, StreamExt,

    future::{Shared, WeakShared},

};

use matrix_sdk::{

    Client, Room, SendOutsideWasm, SyncOutsideWasm,

    media::{MediaFormat, MediaRequestParameters, MediaThumbnailSettings},

    room::reply::{EnforceThread, Reply as SdkReply},

    ruma::{

        api::{

            Metadata, OutgoingRequest,

            auth_scheme::SendAccessToken,

            client::{authenticated_media, media},

        },

        events::room::{

            EncryptedFileHash, EncryptedFileHashAlgorithm, EncryptedFileInfo, MediaSource,

            message::{

                AddMentions, Relation, ReplyWithinThread, RoomMessageEventContentWithoutRelation,

            },

        },

    },

};

use matrix_sdk_crypto::{AttachmentDecryptor, MediaEncryptionInfo};

use ruma::{MxcUri, OwnedEventId};

use tokio::sync::Mutex;

use super::{executor::EngineBoxFuture, url_preview::UrlPreviews};

use crate::{

    AttachmentEncryption, AttachmentUpload, Error, ErrorKind, LinkPreview, Media, MediaFile,

    MediaId, MediaKind, MediaLocation, MediaSize,

    normalise::{MediaRegistration, MediaScope, MediaSourceInfo},

    upload::prepare_upload,

};

/// a stalled download gives up after this long without a byte

const DOWNLOAD_IDLE: Duration = Duration::from_secs(60);

/// the sdk client gives up on any request after thirty seconds, far too soon for a video

const DOWNLOAD_TOTAL: Duration = Duration::from_secs(60 * 60);

type FlightFuture = EngineBoxFuture<'static, Result<Media, Error>>;

type Flight = Shared<FlightFuture>;

/// owns everything needed to turn a media id or an mxc uri into bytes, so the

/// handle can answer media queries without a trip through the engine

pub(crate) struct MediaService {

    client: Client,

    registry: Mutex<MediaRegistry>,

    flights: Mutex<HashMap<String, WeakShared<FlightFuture>>>,

    max_bytes: usize,

    context_limit: usize,

    url_previews: UrlPreviews,

}

impl MediaService {

    pub(crate) fn new(client: Client, max_bytes: usize, context_limit: usize) -> Self {

        Self {

            url_previews: UrlPreviews::new(client.clone()),

            client,

            registry: Mutex::new(MediaRegistry::default()),

            flights: Mutex::new(HashMap::new()),

            max_bytes,

            context_limit,

        }

    }

    /// one event context session's media, bounded by `EngineConfig::max_context_media`

    pub(crate) fn context_scope(&self) -> MediaScope {

        MediaScope::bounded(self.context_limit)

    }

    pub(crate) async fn url_preview(&self, url: &str) -> Result<Option<LinkPreview>, Error> {

        self.url_previews.preview(url).await

    }

    pub(crate) async fn register(&self, scope: &MediaScope, media: Vec<MediaRegistration>) {

        if !media.is_empty() {

            self.registry.lock().await.insert_all(scope, media);

        }

    }

    pub(crate) async fn release(&self, scope: &MediaScope) {

        self.registry.lock().await.release(scope);

    }

    pub(crate) async fn clear(&self) {

        self.registry.lock().await.clear();

    }

    pub(crate) async fn timeline_media(

        &self,

        media_id: &MediaId,

        size: MediaSize,

    ) -> Result<Media, Error> {

        let chosen = choose_source(self.registered(media_id).await?, size)?;

        if chosen

            .declared_size

            .is_some_and(|size| size > self.max_bytes as u64)

        {

            return Err(too_large("media exceeds the configured byte limit"));

        }

        self.fetch_or_file(chosen.request, chosen.mimetype).await

    }

    pub(crate) async fn media_location(

        &self,

        media_id: &MediaId,

        size: MediaSize,

    ) -> Result<MediaLocation, Error> {

        let media = self.registered(media_id).await?;

        let chosen = choose_source(media.clone(), size)?;

        if is_local(&chosen.request.source) {

            return Ok(MediaLocation::Local);

        }

        if !matches!(chosen.request.format, MediaFormat::Thumbnail(_)) {

            return Ok(MediaLocation::File(

                media_file(

                    &self.client,

                    chosen.request,

                    chosen.mimetype,

                    chosen.declared_size,

                )

                .await?,

            ));

        }

        let url = media_http_request(&self.client, &chosen.request)

            .await?

            .uri()

            .to_string();

        let fallback = MediaRequestParameters {

            source: media.full.source,

            format: MediaFormat::File,

        };

        Ok(MediaLocation::ServerThumbnail {

            url,

            fallback: media_file(

                &self.client,

                fallback,

                media.full.mimetype,

                media.full.declared_size,

            )

            .await?,

        })

    }

    async fn registered(&self, media_id: &MediaId) -> Result<MediaRegistration, Error> {

        if let Some(uri) = media_id.as_plain() {

            return plain_registration(media_id, uri);

        }

        self.registry

            .lock()

            .await

            .get(media_id)

            .cloned()

            .ok_or_else(|| {

                Error::new(

                    ErrorKind::MediaUnknown,

                    "media is not registered in a retained timeline",

                )

            })

    }

    /// plenty of servers won't thumbnail everything, and the file itself is the next best

    /// thing. the byte cap still holds either way

    async fn fetch_or_file(

        &self,

        request: MediaRequestParameters,

        mimetype: Option<String>,

    ) -> Result<Media, Error> {

        if !matches!(request.format, MediaFormat::Thumbnail(_)) {

            return self.fetch(request, mimetype).await;

        }

        match self.fetch(request.clone(), mimetype.clone()).await {

            Err(error) if error.kind == ErrorKind::MediaUnavailable => {

                let file = MediaRequestParameters {

                    source: request.source,

                    format: MediaFormat::File,

                };

                self.fetch(file, mimetype).await

            }

            result => result,

        }

    }

    async fn fetch(

        &self,

        request: MediaRequestParameters,

        mimetype: Option<String>,

    ) -> Result<Media, Error> {

        let key = flight_key(&request);

        let flight = {

            let mut flights = self.flights.lock().await;

            flights.retain(|_, flight| flight.upgrade().is_some());

            match flights.get(&key).and_then(WeakShared::upgrade) {

                Some(flight) => flight,

                None => {

                    let prepared = AssertUnwindSafe(fetch(

                        self.client.clone(),

                        request,

                        mimetype,

                        self.max_bytes,

                    ))

                    .catch_unwind()

                    .map(|result| {

                        result.unwrap_or_else(|_| {

                            Err(Error::new(ErrorKind::Other, "media fetch panicked"))

                        })

                    });

                    let boxed: FlightFuture = Box::pin(prepared);

                    let flight: Flight = boxed.shared();

                    if let Some(weak) = flight.downgrade() {

                        flights.insert(key, weak);

                    }

                    flight

                }

            }

        };

        flight.await

    }

}

struct ChosenSource {

    request: MediaRequestParameters,

    mimetype: Option<String>,

    declared_size: Option<u64>,

}

fn plain_registration(media_id: &MediaId, uri: &MxcUri) -> Result<MediaRegistration, Error> {

    if !uri.is_valid() {

        return Err(Error::new(ErrorKind::Invalid, "mxc uri is malformed"));

    }

    Ok(MediaRegistration {

        id: media_id.clone(),

        kind: MediaKind::Image,

        full: MediaSourceInfo {

            source: MediaSource::Plain(uri.to_owned()),

            mimetype: None,

            declared_size: None,

        },

        thumbnail: None,

    })

}

fn choose_source(media: MediaRegistration, size: MediaSize) -> Result<ChosenSource, Error> {

    let (source, format, mimetype, declared_size) = match size {

        MediaSize::Full => (

            media.full.source,

            MediaFormat::File,

            media.full.mimetype,

            media.full.declared_size,

        ),

        MediaSize::Thumbnail { width, height } => match media.thumbnail {

            Some(thumbnail) => (

                thumbnail.source,

                MediaFormat::File,

                thumbnail.mimetype,

                thumbnail.declared_size,

            ),

            // the server can only resize what it can read

            None if matches!(media.kind, MediaKind::Image | MediaKind::Video)

                && matches!(media.full.source, MediaSource::Plain(_)) =>

            {

                // the server picks the thumbnail's type, so the original's would be a lie

                let format = if is_local(&media.full.source) {

                    MediaFormat::File

                } else {

                    thumbnail_format(width, height)

                };

                (media.full.source, format, None, None)

            }

            None => {

                return Err(Error::new(

                    ErrorKind::MediaUnavailable,

                    "the requested media size is unavailable",

                ));

            }

        },

    };

    Ok(ChosenSource {

        request: MediaRequestParameters { source, format },

        mimetype,

        declared_size,

    })

}

async fn media_file(

    client: &Client,

    request: MediaRequestParameters,

    mimetype: Option<String>,

    declared_size: Option<u64>,

) -> Result<MediaFile, Error> {

    let url = media_http_request(client, &request)

        .await?

        .uri()

        .to_string();

    let encryption = match &request.source {

        MediaSource::Plain(_) => None,

        MediaSource::Encrypted(file) => {

            let (EncryptedFileInfo::V2(info), Some(EncryptedFileHash::Sha256(sha256))) = (

                &file.info,

                file.hashes.get(&EncryptedFileHashAlgorithm::Sha256),

            ) else {

                return Err(Error::new(

                    ErrorKind::MediaUnavailable,

                    "encrypted media is not v2 with a sha256 hash",

                ));

            };

            Some(AttachmentEncryption {

                key: info.k.encode(),

                iv: info.iv.encode(),

                sha256: sha256.encode(),

            })

        }

    };

    Ok(MediaFile {

        url,

        mimetype,

        declared_size,

        encryption,

    })

}

/// servers without spec v1.11 animated thumbnails ignore the flag and send a still frame

fn thumbnail_format(width: u32, height: u32) -> MediaFormat {

    let mut settings = MediaThumbnailSettings::new(width.into(), height.into());

    settings.animated = true;

    MediaFormat::Thumbnail(settings)

}

fn is_local(source: &MediaSource) -> bool {

    let uri = match source {

        MediaSource::Plain(uri) => uri,

        MediaSource::Encrypted(file) => &file.url,

    };

    uri.server_name()

        .is_ok_and(|server| server.as_str() == "send-queue.localhost")

}

/// plain and encrypted sources share a cache key in the sdk, but must never share a flight

fn flight_key(request: &MediaRequestParameters) -> String {

    let uri = match &request.source {

        MediaSource::Plain(uri) => format!("plain|{uri}"),

        MediaSource::Encrypted(file) => format!("encrypted|{}", file.url),

    };

    match &request.format {

        MediaFormat::File => uri.to_owned(),

        MediaFormat::Thumbnail(settings) => format!(

            "{uri}|{}|{}x{}|{}",

            settings.method, settings.width, settings.height, settings.animated

        ),

    }

}

fn too_large(message: &'static str) -> Error {

    Error::new(ErrorKind::MediaTooLarge, message)

}

pub(crate) async fn download_file(

    client: &Client,

    source: MediaSource,

    mimetype: Option<String>,

    max_bytes: usize,

) -> Result<Media, Error> {

    let request = MediaRequestParameters {

        source,

        format: MediaFormat::File,

    };

    fetch(client.clone(), request, mimetype, max_bytes).await

}

async fn fetch(

    client: Client,

    request: MediaRequestParameters,

    mimetype: Option<String>,

    max_bytes: usize,

) -> Result<Media, Error> {

    if let Some(cached) = cached_media(&client, &request).await? {

        if cached.len() > max_bytes {

            if let Ok(store) = client.media_store().lock().await {

                let _ = store.remove_media_content(&request).await;

            }

            return Err(too_large("cached media exceeds the configured byte limit"));

        }

        // plain and encrypted share a cache key, so a plain fetch of an encrypted file's uri

        // leaves ciphertext here. its hash gives it away

        if is_ciphertext(&cached, &request.source) {

            let bytes = maybe_decrypt(cached, &request, max_bytes)?;

            cache(&client, &request, &bytes).await;

            return Ok(media(bytes, mimetype));

        }

        return Ok(media(cached, mimetype));

    }

    let (bytes, served_mimetype) = download_bounded(&client, &request, max_bytes).await?;

    // the sdk caches plaintext under the encrypted key, so we do too or nobody can read anything

    let bytes = maybe_decrypt(bytes, &request, max_bytes)?;

    cache(&client, &request, &bytes).await;

    Ok(media(bytes, mimetype.or(served_mimetype)))

}

async fn cache(client: &Client, request: &MediaRequestParameters, bytes: &[u8]) {

    if let Ok(store) = client.media_store().lock().await {

        let _ = store

            .add_media_content(

                request,

                bytes.to_vec(),

                matrix_sdk::media::store::IgnoreMediaRetentionPolicy::No,

            )

            .await;

    }

}

fn is_ciphertext(bytes: &[u8], source: &MediaSource) -> bool {

    use sha2::{Digest, Sha256};

    let MediaSource::Encrypted(file) = source else {

        return false;

    };

    file.hashes

        .get(&ruma::events::room::EncryptedFileHashAlgorithm::Sha256)

        .is_some_and(|hash| hash.as_bytes() == Sha256::digest(bytes).as_slice())

}

fn media(bytes: Vec<u8>, mimetype: Option<String>) -> Media {

    let mimetype = mimetype

        .or_else(|| sniff_image(&bytes).map(str::to_owned))

        .unwrap_or_else(|| "application/octet-stream".to_owned());

    Media {

        bytes: Arc::from(bytes),

        mimetype,

    }

}

/// the cache forgets content types, and avatars never declare one. images are the common case

fn sniff_image(bytes: &[u8]) -> Option<&'static str> {

    match bytes {

        [0x89, b'P', b'N', b'G', ..] => Some("image/png"),

        [0xff, 0xd8, 0xff, ..] => Some("image/jpeg"),

        [b'G', b'I', b'F', b'8', ..] => Some("image/gif"),

        [

            b'R',

            b'I',

            b'F',

            b'F',

            _,

            _,

            _,

            _,

            b'W',

            b'E',

            b'B',

            b'P',

            ..,

        ] => Some("image/webp"),

        _ => None,

    }

}

async fn cached_media(

    client: &Client,

    request: &MediaRequestParameters,

) -> Result<Option<Vec<u8>>, Error> {

    fn cache_failed(

        source: impl std::error::Error + SendOutsideWasm + SyncOutsideWasm + 'static,

    ) -> Error {

        Error::with_source(ErrorKind::Store, "reading media cache failed", source)

    }

    client

        .media_store()

        .lock()

        .await

        .map_err(cache_failed)?

        .get_media_content(request)

        .await

        .map_err(cache_failed)

}

fn maybe_decrypt(

    bytes: Vec<u8>,