//! The AWS calls behind a match: manifest to S3 with a presigned GET,
//! RunTask, DescribeTasks. Confined here the way atrium is confined to
//! atproto. Credentials and region come from the instance (IMDS); nothing is
//! configured in code.

use std::time::Duration;

use aws_sdk_ecs::types::{
    AwsVpcConfiguration, ContainerOverride, KeyValuePair, NetworkConfiguration, TaskOverride,
};

use crate::config::MatchConfig;

pub struct Aws {
    s3: aws_sdk_s3::Client,
    ecs: aws_sdk_ecs::Client,
}

/// Never carries AWS error values into logs wholesale - like atrium errors,
/// they can embed request details. Log the category; the operator has
/// CloudTrail for the rest.
#[derive(Debug)]
pub struct AwsFailed(pub &'static str);

impl Aws {
    pub async fn new() -> Aws {
        let config = aws_config::load_defaults(aws_config::BehaviorVersion::latest()).await;
        Aws {
            s3: aws_sdk_s3::Client::new(&config),
            ecs: aws_sdk_ecs::Client::new(&config),
        }
    }

    pub async fn put_manifest(
        &self,
        bucket: &str,
        match_id: &str,
        manifest: &serde_json::Value,
    ) -> Result<String, AwsFailed> {
        let key = format!("matches/{match_id}/manifest.json");
        self.s3
            .put_object()
            .bucket(bucket)
            .key(&key)
            .content_type("application/json")
            .body(manifest.to_string().into_bytes().into())
            .send()
            .await
            .map_err(|_| AwsFailed("manifest upload"))?;
        Ok(key)
    }

    /// A sanitised camo for one match, beside that match's manifest rather
    /// than in a shared place: it is fetched once at launch and has no life
    /// after the match, and a per-match key means one player's camo is never
    /// served under a name another match could guess.
    ///
    /// `name` separates the slots' files under that one prefix - `camo.png`
    /// for the player, `camo-bot.png` for the opponent's rolled one.
    pub async fn put_camo(
        &self,
        bucket: &str,
        match_id: &str,
        name: &str,
        png: Vec<u8>,
    ) -> Result<String, AwsFailed> {
        let key = format!("matches/{match_id}/{name}");
        self.s3
            .put_object()
            .bucket(bucket)
            .key(&key)
            .content_type("image/png")
            .body(png.into())
            .send()
            .await
            .map_err(|_| AwsFailed("camo upload"))?;
        Ok(key)
    }

    /// The presigned URL is a bearer credential; it goes into the container
    /// override and nowhere else - never logged, never stored.
    pub async fn presign_get(&self, bucket: &str, key: &str) -> Result<String, AwsFailed> {
        let presigned = self
            .s3
            .get_object()
            .bucket(bucket)
            .key(key)
            .presigned(
                aws_sdk_s3::presigning::PresigningConfig::expires_in(Duration::from_secs(900))
                    .expect("static expiry is valid"),
            )
            .await
            .map_err(|_| AwsFailed("presign"))?;
        Ok(presigned.uri().to_string())
    }

    /// Presign a PUT for one artifact, so the container sends the bytes to S3
    /// instead of through here.
    ///
    /// Minted per upload rather than at manifest build time, and that is the
    /// whole point: a presigned URL dies with the credentials that signed it,
    /// these are IMDS instance-profile credentials, and IMDS rotates them on
    /// its own schedule and will not issue a fresh set on request. Signing at
    /// launch would tie every upload URL's life to how long a match runs, with
    /// a ceiling that is both low and unpredictable. Signing here means the
    /// signature only has to survive one PUT.
    ///
    /// The signature covers the bucket, the key and the method, so the container
    /// can only write the one artifact it asked for.
    /// Read back a camo this service staged for a match.
    ///
    /// Its own reader because it is not an artifact: a camo is written by us
    /// before the match under `matches/<id>/<name>`, and everything the
    /// container uploads lands under `matches/<id>/artifacts/`. Reading one
    /// with the other's key finds nothing, quietly, which is exactly what the
    /// social card did until somebody noticed the machines were grey.
    pub async fn get_camo(
        &self,
        bucket: &str,
        match_id: &str,
        name: &str,
    ) -> Result<Option<Vec<u8>>, AwsFailed> {
        let sent = self
            .s3
            .get_object()
            .bucket(bucket)
            .key(format!("matches/{match_id}/{name}"))
            .send()
            .await;
        let object = match sent {
            Ok(object) => object,
            // A match with no camo staged for a slot is the ordinary case and
            // not a failure. Told apart here because the caller caches on the
            // difference: a card drawn without colours nobody chose can stand
            // for a day, one drawn without colours S3 would not hand over
            // must not.
            Err(e) => {
                return if e.as_service_error().is_some_and(|e| e.is_no_such_key()) {
                    Ok(None)
                } else {
                    Err(AwsFailed("camo read"))
                };
            }
        };
        let bytes = object
            .body
            .collect()
            .await
            .map_err(|_| AwsFailed("camo body"))?;
        Ok(Some(bytes.into_bytes().to_vec()))
    }

    /// Read an artifact a match uploaded.
    ///
    /// The only read of the artifacts bucket. Everything else here signs a
    /// URL and lets the container do the transfer; a finished match's result
    /// has to be read by us, because it is what a score is computed from.
    /// Whether an artifact is there, without pulling it. The page needs to
    /// know a board exists to offer the card drawn on it, and a board render
    /// is megabytes.
    pub async fn has_artifact(
        &self,
        bucket: &str,
        match_id: &str,
        name: &str,
    ) -> Result<bool, AwsFailed> {
        match self
            .s3
            .head_object()
            .bucket(bucket)
            .key(format!("matches/{match_id}/artifacts/{name}"))
            .send()
            .await
        {
            Ok(_) => Ok(true),
            Err(e) => {
                // A head on a missing key answers 404 with no modelled error
                // variant, unlike get_object's NoSuchKey.
                if e.raw_response().is_some_and(|r| r.status().as_u16() == 404) {
                    Ok(false)
                } else {
                    Err(AwsFailed("artifact head"))
                }
            }
        }
    }

    pub async fn get_artifact(
        &self,
        bucket: &str,
        match_id: &str,
        name: &str,
    ) -> Result<Vec<u8>, AwsFailed> {
        let object = self
            .s3
            .get_object()
            .bucket(bucket)
            .key(format!("matches/{match_id}/artifacts/{name}"))
            .send()
            .await
            .map_err(|_| AwsFailed("artifact read"))?;
        let bytes = object
            .body
            .collect()
            .await
            .map_err(|_| AwsFailed("artifact body"))?;
        Ok(bytes.into_bytes().to_vec())
    }

    pub async fn presign_artifact_put(
        &self,
        bucket: &str,
        match_id: &str,
        name: &str,
        expires_in: Duration,
    ) -> Result<String, AwsFailed> {
        let presigned = self
            .s3
            .put_object()
            .bucket(bucket)
            .key(format!("matches/{match_id}/artifacts/{name}"))
            .presigned(
                aws_sdk_s3::presigning::PresigningConfig::expires_in(expires_in)
                    .map_err(|_| AwsFailed("presign config"))?,
            )
            .await
            .map_err(|_| AwsFailed("presign artifact put"))?;
        Ok(presigned.uri().to_string())
    }

    pub async fn run_task(
        &self,
        config: &MatchConfig,
        manifest_url: &str,
        size: &super::sizing::TaskSize,
    ) -> Result<String, AwsFailed> {
        let network = NetworkConfiguration::builder()
            .awsvpc_configuration(
                AwsVpcConfiguration::builder()
                    .set_subnets(Some(config.subnets.clone()))
                    .security_groups(&config.task_security_group)
                    // No NAT gateway: without a public IP the image pull
                    // hangs.
                    .assign_public_ip(aws_sdk_ecs::types::AssignPublicIp::Enabled)
                    .build()
                    .map_err(|_| AwsFailed("network configuration"))?,
            )
            .build();

        // The match's size rides the same per-task override the manifest URL
        // does: the task definition stays the one-human shape, and a bigger
        // match asks for more only for its own task.
        let overrides = TaskOverride::builder()
            .cpu(size.cpu)
            .memory(&size.memory)
            .container_overrides(
                ContainerOverride::builder()
                    .name("arena")
                    .environment(
                        KeyValuePair::builder()
                            .name("ARENA_MANIFEST_URL")
                            .value(manifest_url)
                            .build(),
                    )
                    .build(),
            )
            .build();

        let output = self
            .ecs
            .run_task()
            .cluster(&config.cluster)
            .task_definition(&config.task_family)
            .launch_type(aws_sdk_ecs::types::LaunchType::Fargate)
            .network_configuration(network)
            .overrides(overrides)
            .send()
            .await
            .map_err(|_| AwsFailed("run task"))?;

        output
            .tasks()
            .first()
            .and_then(|task| task.task_arn())
            .map(str::to_owned)
            .ok_or(AwsFailed("task arn missing"))
    }

    /// Last status plus the private IP once the ENI has one. DescribeTasks
    /// exposes the IP as an attachment detail, so no EC2 call is needed.
    pub async fn task_status(&self, cluster: &str, task_arn: &str) -> Result<TaskState, AwsFailed> {
        let output = self
            .ecs
            .describe_tasks()
            .cluster(cluster)
            .tasks(task_arn)
            .send()
            .await
            .map_err(|_| AwsFailed("describe task"))?;
        let task = output.tasks().first().ok_or(AwsFailed("task not found"))?;

        let status = task.last_status().unwrap_or("UNKNOWN").to_owned();
        let ip = task
            .attachments()
            .iter()
            .flat_map(|attachment| attachment.details())
            .find(|detail| detail.name() == Some("privateIPv4Address"))
            .and_then(|detail| detail.value())
            .map(str::to_owned);
        Ok(TaskState {
            status,
            ip,
            timing: TaskTiming {
                provision_ms: delta_ms(task.created_at(), task.connectivity_at()),
                pull_ms: delta_ms(task.pull_started_at(), task.pull_stopped_at()),
                start_ms: delta_ms(task.pull_stopped_at(), task.started_at()),
                total_ms: delta_ms(task.created_at(), task.started_at()),
            },
        })
    }
}

pub struct TaskState {
    pub status: String,
    pub ip: Option<String>,
    pub timing: TaskTiming,
}

/// What ECS can say about a start that the container cannot. The task is only
/// alive for the last of these: everything before `start_ms` happens before any
/// of arena's own phase marks exist, and the pull in particular is invisible
/// from inside. Every field is None until the task reaches the phase, and stays
/// None if ECS omits the timestamp.
#[derive(Debug)]
pub struct TaskTiming {
    /// RunTask accepted to ENI attached.
    pub provision_ms: Option<i64>,
    /// The image pull. The number `plan/arena-image.md` has only ever guessed at.
    pub pull_ms: Option<i64>,
    /// Pull finished to container running - Fargate's own start overhead.
    pub start_ms: Option<i64>,
    /// RunTask accepted to container running.
    pub total_ms: Option<i64>,
}

fn delta_ms(
    from: Option<&aws_sdk_ecs::primitives::DateTime>,
    to: Option<&aws_sdk_ecs::primitives::DateTime>,
) -> Option<i64> {
    let millis = |t: &aws_sdk_ecs::primitives::DateTime| {
        t.secs() * 1000 + i64::from(t.subsec_nanos()) / 1_000_000
    };
    Some(millis(to?) - millis(from?))
}

#[cfg(test)]
mod tests {
    use super::*;
    use aws_sdk_ecs::primitives::DateTime;

    #[test]
    fn both_present_gives_the_millisecond_gap() {
        let from = DateTime::from_millis(1_000);
        let to = DateTime::from_millis(2_500);
        assert_eq!(delta_ms(Some(&from), Some(&to)), Some(1_500));
    }

    #[test]
    fn a_missing_endpoint_gives_no_timing() {
        let present = DateTime::from_millis(1_000);
        assert_eq!(delta_ms(None, Some(&present)), None);
        assert_eq!(delta_ms(Some(&present), None), None);
    }

    #[test]
    fn both_missing_gives_no_timing() {
        assert_eq!(delta_ms(None, None), None);
    }

    /// ECS timestamps are not guaranteed monotonic against the phase order
    /// this module assumes (a clock can be corrected between two events);
    /// `delta_ms` does not clamp, so a `to` earlier than `from` comes out
    /// negative rather than panicking or getting silently flipped.
    #[test]
    fn to_before_from_is_a_negative_delta() {
        let later = DateTime::from_millis(5_000);
        let earlier = DateTime::from_millis(1_000);
        assert_eq!(delta_ms(Some(&later), Some(&earlier)), Some(-4_000));
    }
}