use axum::{
    Json, Router,
    body::Bytes,
    extract::{DefaultBodyLimit, Path, Query, State},
    http::{HeaderMap, StatusCode},
    response::{IntoResponse, Response},
    routing::{get, post, put},
};
use base64::{Engine, engine::general_purpose::URL_SAFE_NO_PAD};
use futures::StreamExt;
use hmac::{Hmac, KeyInit, Mac};
use models::ci::workflow_artifact::ArtifactState;
use models::ci::{actions_cache, actions_identity, workflow_artifact, workflow_job, workflow_run};
use sea_orm::{
    ActiveModelTrait, ColumnTrait, ConnectionTrait, EntityTrait, QueryFilter, QueryOrder, Set,
    TransactionTrait,
};
use serde::Deserialize;
use serde_json::{Value, json};
use sha2::{Digest, Sha256};
use time::OffsetDateTime;
use tokio::io::{AsyncSeekExt, AsyncWriteExt};
use uuid::Uuid;

use crate::{
    CiService,
    errors::{CiError, CiResult},
};

pub fn router() -> Router<CiService> {
    Router::new()
        .route(
            "/twirp/github.actions.results.api.v1.ArtifactService/{method}",
            post(artifact_rpc),
        )
        .route(
            "/twirp/github.actions.results.api.v1.CacheService/{method}",
            post(cache_rpc),
        )
        .route("/actions-runtime/uploads/{kind}/{id}", put(upload_blob))
        .route("/actions-runtime/oidc", get(oidc_token))
        .route("/.well-known/openid-configuration", get(oidc_configuration))
        .route("/.well-known/jwks.json", get(oidc_jwks))
        .route("/_apis/artifactcache/cache", get(legacy_cache_lookup))
        .route("/_apis/artifactcache/caches", post(legacy_cache_reserve))
        .route(
            "/_apis/artifactcache/caches/{id}",
            post(legacy_cache_finalize).patch(legacy_cache_upload),
        )
        .layer(DefaultBodyLimit::max(64 * 1024 * 1024))
}

pub struct RuntimeError(CiError);
impl From<CiError> for RuntimeError {
    fn from(error: CiError) -> Self {
        Self(error)
    }
}
impl From<storage::StorageError> for RuntimeError {
    fn from(error: storage::StorageError) -> Self {
        Self(error.into())
    }
}
impl From<sea_orm::DbErr> for RuntimeError {
    fn from(error: sea_orm::DbErr) -> Self {
        Self(error.into())
    }
}
impl IntoResponse for RuntimeError {
    fn into_response(self) -> Response {
        let (status, code) = match &self.0 {
            CiError::Unauthorized => (StatusCode::UNAUTHORIZED, "unauthenticated"),
            CiError::Forbidden => (StatusCode::FORBIDDEN, "permission_denied"),
            CiError::NotFound => (StatusCode::NOT_FOUND, "not_found"),
            CiError::Conflict(_) => (StatusCode::CONFLICT, "already_exists"),
            CiError::InvalidArgument(_) => (StatusCode::BAD_REQUEST, "invalid_argument"),
            _ => (StatusCode::INTERNAL_SERVER_ERROR, "internal"),
        };
        (status, Json(json!({"code":code,"msg":self.0.to_string()}))).into_response()
    }
}

#[derive(Clone)]
struct RuntimeContext {
    job: workflow_job::Model,
    run: workflow_run::Model,
}

impl CiService {
    pub fn sign_actions_token(&self, payload: &Value) -> CiResult<String> {
        let header = URL_SAFE_NO_PAD.encode(br#"{"alg":"HS256","typ":"JWT"}"#);
        let data = URL_SAFE_NO_PAD.encode(serde_json::to_vec(payload)?);
        let message = format!("{header}.{data}");
        let mut mac = Hmac::<Sha256>::new_from_slice(self.encryption_key().as_bytes())
            .map_err(|error| CiError::Internal(error.to_string()))?;
        mac.update(message.as_bytes());
        Ok(format!(
            "{message}.{}",
            URL_SAFE_NO_PAD.encode(mac.finalize().into_bytes())
        ))
    }

    fn verify_actions_token(&self, token: &str) -> CiResult<Value> {
        let parts: Vec<_> = token.split('.').collect();
        if parts.len() != 3 {
            return Err(CiError::Unauthorized);
        }
        let mut mac = Hmac::<Sha256>::new_from_slice(self.encryption_key().as_bytes())
            .map_err(|error| CiError::Internal(error.to_string()))?;
        mac.update(format!("{}.{}", parts[0], parts[1]).as_bytes());
        let signature = URL_SAFE_NO_PAD
            .decode(parts[2])
            .map_err(|_| CiError::Unauthorized)?;
        mac.verify_slice(&signature)
            .map_err(|_| CiError::Unauthorized)?;
        let data = URL_SAFE_NO_PAD
            .decode(parts[1])
            .map_err(|_| CiError::Unauthorized)?;
        let value: Value = serde_json::from_slice(&data).map_err(|_| CiError::Unauthorized)?;
        if value["exp"].as_i64().unwrap_or(0) <= OffsetDateTime::now_utc().unix_timestamp() {
            return Err(CiError::Unauthorized);
        }
        Ok(value)
    }

    async fn runtime_context(&self, headers: &HeaderMap) -> CiResult<RuntimeContext> {
        self.ensure_enabled()?;
        let token = headers
            .get("authorization")
            .and_then(|value| value.to_str().ok())
            .and_then(|value| value.strip_prefix("Bearer "))
            .ok_or(CiError::Unauthorized)?;
        let claims = self.verify_actions_token(token)?;
        if claims["aud"] != "gitdata-actions-runtime" {
            return Err(CiError::Unauthorized);
        }
        self.runtime_context_from_claims(&claims).await
    }

    async fn runtime_context_from_claims(&self, claims: &Value) -> CiResult<RuntimeContext> {
        let id = claims["job_id"]
            .as_str()
            .and_then(|value| Uuid::parse_str(value).ok())
            .ok_or(CiError::Unauthorized)?;
        let job = workflow_job::Entity::find_by_id(id)
            .one(self.db_write())
            .await?
            .ok_or(CiError::Unauthorized)?;
        if !job.state.is_active() || job.lease_token.as_deref() != claims["lease_token"].as_str() {
            return Err(CiError::Unauthorized);
        }
        let run = workflow_run::Entity::find_by_id(job.run_id)
            .one(self.db_write())
            .await?
            .ok_or(CiError::Unauthorized)?;
        Ok(RuntimeContext { job, run })
    }

    pub fn actions_runtime_url(&self) -> CiResult<String> {
        let configured = &self.config().ci.runtime_base_url;
        let base = if !configured.is_empty() {
            configured.as_str()
        } else if !self.config().webhook.public_base_url.is_empty() {
            self.config().webhook.public_base_url.as_str()
        } else {
            self.config().git.base_url.as_str()
        };
        let url = reqwest::Url::parse(base).map_err(|error| {
            CiError::InvalidArgument(format!("invalid CI runtime origin: {error}"))
        })?;
        if url.path() != "/" && !url.path().is_empty() {
            return Err(CiError::InvalidArgument(
                "ci.runtime_base_url must be a URL origin without a path prefix".into(),
            ));
        }
        Ok(base.trim_end_matches('/').to_owned())
    }

    pub fn actions_runtime_token(
        &self,
        job: &workflow_job::Model,
        run: &workflow_run::Model,
    ) -> CiResult<String> {
        self.sign_actions_token(&json!({"aud":"gitdata-actions-runtime","job_id":job.id,"run_id":run.id,"repo_id":job.repo_id,"lease_token":job.lease_token,
            "scp":format!("Actions.Results:{}:{}",run.actions_id.unwrap_or(0),job.actions_id.unwrap_or(0)),"iat":OffsetDateTime::now_utc().unix_timestamp(),
            "exp":OffsetDateTime::now_utc().unix_timestamp()+self.config().ci.job_timeout_max_secs as i64+600}))
    }

    fn signed_upload_url(
        &self,
        context: &RuntimeContext,
        kind: &str,
        id: &str,
    ) -> CiResult<String> {
        let token=self.sign_actions_token(&json!({"aud":"gitdata-actions-upload","job_id":context.job.id,"lease_token":context.job.lease_token,"kind":kind,"id":id,"exp":OffsetDateTime::now_utc().unix_timestamp()+3600}))?;
        Ok(format!(
            "{}/actions-runtime/uploads/{kind}/{id}?token={token}",
            self.actions_runtime_url()?
        ))
    }

    async fn numeric_actions_id(&self, entity_id: Uuid) -> CiResult<i64> {
        actions_identity::Entity::find()
            .filter(actions_identity::Column::EntityId.eq(entity_id))
            .one(self.db_write())
            .await?
            .map(|identity| identity.id)
            .ok_or(CiError::NotFound)
    }

    fn check_artifact_scope(&self, context: &RuntimeContext, body: &Value) -> CiResult<()> {
        let run = field(body, "workflowRunBackendId", "workflow_run_backend_id");
        if !run.is_empty() && run != context.run.actions_id.unwrap_or(0).to_string() {
            return Err(CiError::Forbidden);
        }
        Ok(())
    }

    async fn artifact_call(
        &self,
        context: &RuntimeContext,
        method: &str,
        body: &Value,
    ) -> CiResult<Value> {
        self.check_artifact_scope(context, body)?;
        let name = field(body, "name", "name");
        match method {
            "CreateArtifact" => {
                if name.trim().is_empty() || name.len() > 255 {
                    return Err(CiError::InvalidArgument(
                        "artifact name must be 1-255 characters".into(),
                    ));
                }
                let txn = self.db_write().begin().await?;
                txn.execute_unprepared(&format!(
                    "SELECT pg_advisory_xact_lock({})",
                    context.run.id.as_u128() as i64
                ))
                .await?;
                let existing = workflow_artifact::Entity::find()
                    .filter(workflow_artifact::Column::RunId.eq(context.run.id))
                    .filter(workflow_artifact::Column::Name.eq(&name))
                    .one(&txn)
                    .await?;
                if let Some(existing) = existing {
                    if existing.job_id == context.job.id
                        && existing.state == ArtifactState::Uploading
                    {
                        let url =
                            self.signed_upload_url(context, "artifact", &existing.id.to_string())?;
                        return Ok(json!({"ok":true,"signedUploadUrl":url}));
                    }
                    return Err(CiError::Conflict(
                        "artifact name already exists in this run".into(),
                    ));
                }
                let id = Uuid::now_v7();
                let now = OffsetDateTime::now_utc();
                let configured_expiry =
                    now + time::Duration::days(self.config().ci.artifact_retention_days.max(1));
                let requested_expiry = body
                    .get("expiresAt")
                    .or_else(|| body.get("expires_at"))
                    .and_then(Value::as_str)
                    .and_then(|value| {
                        OffsetDateTime::parse(value, &time::format_description::well_known::Rfc3339)
                            .ok()
                    });
                let expires = requested_expiry
                    .map(|value| value.min(configured_expiry))
                    .unwrap_or(configured_expiry);
                workflow_artifact::ActiveModel {
                    id: Set(id),
                    run_id: Set(context.run.id),
                    job_id: Set(context.job.id),
                    name: Set(name),
                    object_key: Set(CiService::artifact_object_key(
                        context.run.id,
                        context.job.id,
                        id,
                    )),
                    size: Set(0),
                    sha256: Set(None),
                    mime_type: Set(body
                        .get("mimeType")
                        .or_else(|| body.get("mime_type"))
                        .and_then(|value| value.as_str().or_else(|| value["value"].as_str()))
                        .map(str::to_owned)),
                    state: Set(ArtifactState::Uploading),
                    expires_at: Set(Some(expires)),
                    created_at: Set(now),
                    updated_at: Set(now),
                }
                .insert(&txn)
                .await?;
                crate::actions::identity(&txn, id, "artifact").await?;
                txn.commit().await?;
                Ok(
                    json!({"ok":true,"signedUploadUrl":self.signed_upload_url(context,"artifact",&id.to_string())?}),
                )
            }
            "FinalizeArtifact" => {
                let row = workflow_artifact::Entity::find()
                    .filter(workflow_artifact::Column::RunId.eq(context.run.id))
                    .filter(workflow_artifact::Column::JobId.eq(context.job.id))
                    .filter(workflow_artifact::Column::Name.eq(name))
                    .one(self.db_write())
                    .await?
                    .ok_or(CiError::NotFound)?;
                let declared = number(body.get("size"))?;
                let (size, hash) = self
                    .uploaded_size_and_hash(&row.object_key, self.config().ci.artifact_max_bytes)
                    .await?;
                if size != declared {
                    return Err(CiError::InvalidArgument(
                        "artifact size does not match upload".into(),
                    ));
                }
                let expected = body
                    .get("hash")
                    .and_then(|v| v.as_str().or_else(|| v["value"].as_str()))
                    .map(|v| v.trim_start_matches("sha256:"));
                if expected.is_some_and(|value| value != hash) {
                    return Err(CiError::InvalidArgument(
                        "artifact digest does not match upload".into(),
                    ));
                }
                let id = self.numeric_actions_id(row.id).await?;
                let mut active: workflow_artifact::ActiveModel = row.into();
                active.size = Set(size as i64);
                active.sha256 = Set(Some(hash));
                active.state = Set(ArtifactState::Ready);
                active.updated_at = Set(OffsetDateTime::now_utc());
                active.update(self.db_write()).await?;
                Ok(json!({"ok":true,"artifactId":id.to_string()}))
            }
            "ListArtifacts" => {
                let rows = workflow_artifact::Entity::find()
                    .filter(workflow_artifact::Column::RunId.eq(context.run.id))
                    .filter(workflow_artifact::Column::State.eq(ArtifactState::Ready))
                    .all(self.db_write())
                    .await?;
                let name_filter = field(body, "nameFilter", "name_filter");
                let id_filter = body
                    .get("idFilter")
                    .or_else(|| body.get("id_filter"))
                    .and_then(|v| v.as_str().or_else(|| v["value"].as_str()));
                let mut artifacts = Vec::new();
                for row in rows {
                    if row
                        .expires_at
                        .is_some_and(|expiry| expiry <= OffsetDateTime::now_utc())
                        || (!name_filter.is_empty() && row.name != name_filter)
                    {
                        continue;
                    }
                    let id = self.numeric_actions_id(row.id).await?;
                    if id_filter.is_some_and(|filter| filter != id.to_string()) {
                        continue;
                    }
                    let job = workflow_job::Entity::find_by_id(row.job_id)
                        .one(self.db_write())
                        .await?
                        .ok_or(CiError::NotFound)?;
                    artifacts.push(json!({"workflowRunBackendId":context.run.actions_id.unwrap_or(0).to_string(),"workflowJobRunBackendId":job.actions_id.unwrap_or(0).to_string(),"databaseId":id.to_string(),"name":row.name,"size":row.size.to_string(),"createdAt":row.created_at.format(&time::format_description::well_known::Rfc3339).unwrap_or_default(),"hash":row.sha256.map(|hash|format!("sha256:{hash}")),"mimeType":row.mime_type}));
                }
                Ok(json!({"artifacts":artifacts}))
            }
            "GetSignedArtifactURL" | "DeleteArtifact" => {
                let row = workflow_artifact::Entity::find()
                    .filter(workflow_artifact::Column::RunId.eq(context.run.id))
                    .filter(workflow_artifact::Column::Name.eq(&name))
                    .one(self.db_write())
                    .await?
                    .ok_or(CiError::NotFound)?;
                if method == "DeleteArtifact" {
                    let id = self.numeric_actions_id(row.id).await?;
                    self.storage().client().delete(&row.object_key).await?;
                    workflow_artifact::Entity::delete_by_id(row.id)
                        .exec(self.db_write())
                        .await?;
                    Ok(json!({"ok":true,"artifactId":id.to_string()}))
                } else {
                    if row.state != ArtifactState::Ready
                        || row
                            .expires_at
                            .is_some_and(|at| at <= OffsetDateTime::now_utc())
                    {
                        return Err(CiError::NotFound);
                    }
                    Ok(
                        json!({"signedUrl":self.storage().client().presign_get(&row.object_key,900)?}),
                    )
                }
            }
            _ => Err(CiError::InvalidArgument(format!(
                "unknown artifact method `{method}`"
            ))),
        }
    }

    async fn uploaded_size_and_hash(&self, key: &str, max: u64) -> CiResult<(u64, String)> {
        let mut stream = self.storage().client().get_stream(key).await?;
        let mut hash = Sha256::new();
        let mut size = 0;
        while let Some(chunk) = stream.next().await {
            let chunk = chunk?;
            size += chunk.len() as u64;
            if size > max {
                return Err(CiError::InvalidArgument(
                    "upload exceeds configured size limit".into(),
                ));
            }
            hash.update(&chunk);
        }
        Ok((
            size,
            hash.finalize()
                .iter()
                .map(|byte| format!("{byte:02x}"))
                .collect(),
        ))
    }

    async fn cache_reserve(
        &self,
        context: &RuntimeContext,
        body: &Value,
    ) -> CiResult<actions_cache::Model> {
        if context.run.event_payload["pull_request"]["head"]["repo"]["fork"]
            .as_bool()
            .unwrap_or(false)
        {
            return Err(CiError::Forbidden);
        }
        let key = field(body, "key", "key");
        let version = field(body, "version", "version");
        if key.is_empty() || key.len() > 512 || version.is_empty() || version.len() > 512 {
            return Err(CiError::InvalidArgument(
                "cache key and version must be 1-512 characters".into(),
            ));
        }
        let txn = self.db_write().begin().await?;
        txn.execute_unprepared(&format!(
            "SELECT pg_advisory_xact_lock({})",
            context.job.repo_id.as_u128() as i64
        ))
        .await?;
        if let Some(row) = actions_cache::Entity::find()
            .filter(actions_cache::Column::RepoId.eq(context.job.repo_id))
            .filter(actions_cache::Column::GitRef.eq(&context.run.git_ref))
            .filter(actions_cache::Column::Key.eq(&key))
            .filter(actions_cache::Column::Version.eq(&version))
            .one(&txn)
            .await?
        {
            if row.job_id == context.job.id && !row.ready {
                return Ok(row);
            }
            return Err(CiError::Conflict("cache entry already exists".into()));
        }
        let now = OffsetDateTime::now_utc();
        let object_key = format!("ci/caches/{}/{}", context.job.repo_id, Uuid::now_v7());
        let row = actions_cache::ActiveModel {
            repo_id: Set(context.job.repo_id),
            job_id: Set(context.job.id),
            git_ref: Set(context.run.git_ref.clone()),
            key: Set(key),
            version: Set(version),
            object_key: Set(object_key),
            size: Set(0),
            ready: Set(false),
            created_at: Set(now),
            last_accessed_at: Set(now),
            expires_at: Set(
                now + time::Duration::days(self.config().ci.cache_retention_days.max(1))
            ),
            ..Default::default()
        }
        .insert(&txn)
        .await?;
        txn.commit().await?;
        Ok(row)
    }

    async fn cache_lookup(
        &self,
        context: &RuntimeContext,
        key: &str,
        restore_keys: &[String],
        version: &str,
    ) -> CiResult<Option<actions_cache::Model>> {
        let repo = models::repo::repos::Entity::find_by_id(context.job.repo_id)
            .one(self.db_write())
            .await?
            .ok_or(CiError::NotFound)?;
        let mut refs = vec![context.run.git_ref.clone()];
        if let Some(base) = &context.run.base_ref {
            refs.push(format!("refs/heads/{base}"))
        }
        refs.push(format!("refs/heads/{}", repo.default_branch));
        refs.dedup();
        let rows = actions_cache::Entity::find()
            .filter(actions_cache::Column::RepoId.eq(repo.id))
            .filter(actions_cache::Column::Version.eq(version))
            .filter(actions_cache::Column::Ready.eq(true))
            .filter(actions_cache::Column::ExpiresAt.gt(OffsetDateTime::now_utc()))
            .order_by_desc(actions_cache::Column::CreatedAt)
            .all(self.db_write())
            .await?;
        for reference in refs {
            let entries: Vec<_> = rows.iter().filter(|row| row.git_ref == reference).collect();
            let matched = entries
                .iter()
                .find(|row| row.key == key)
                .copied()
                .or_else(|| entries.iter().find(|row| row.key.starts_with(key)).copied())
                .or_else(|| {
                    restore_keys.iter().find_map(|prefix| {
                        entries
                            .iter()
                            .find(|row| row.key.starts_with(prefix))
                            .copied()
                    })
                });
            if let Some(row) = matched {
                let mut active: actions_cache::ActiveModel = row.clone().into();
                active.last_accessed_at = Set(OffsetDateTime::now_utc());
                active.expires_at = Set(OffsetDateTime::now_utc()
                    + time::Duration::days(self.config().ci.cache_retention_days.max(1)));
                return Ok(Some(active.update(self.db_write()).await?));
            }
        }
        Ok(None)
    }

    async fn cache_finalize(
        &self,
        context: &RuntimeContext,
        id: Option<i64>,
        body: &Value,
    ) -> CiResult<actions_cache::Model> {
        let mut query = actions_cache::Entity::find()
            .filter(actions_cache::Column::RepoId.eq(context.job.repo_id))
            .filter(actions_cache::Column::JobId.eq(context.job.id));
        if let Some(id) = id {
            query = query.filter(actions_cache::Column::Id.eq(id))
        } else {
            query = query
                .filter(actions_cache::Column::Key.eq(field(body, "key", "key")))
                .filter(actions_cache::Column::Version.eq(field(body, "version", "version")))
        }
        let row = query.one(self.db_write()).await?.ok_or(CiError::NotFound)?;
        let declared = number(
            body.get("sizeBytes")
                .or_else(|| body.get("size_bytes"))
                .or_else(|| body.get("size")),
        )?;
        let (size, _) = self
            .uploaded_size_and_hash(&row.object_key, self.config().ci.cache_max_bytes)
            .await?;
        if size != declared {
            return Err(CiError::InvalidArgument(
                "cache size does not match upload".into(),
            ));
        }
        let mut active: actions_cache::ActiveModel = row.into();
        active.size = Set(size as i64);
        active.ready = Set(true);
        active.last_accessed_at = Set(OffsetDateTime::now_utc());
        Ok(active.update(self.db_write()).await?)
    }

    async fn assemble_blocks(&self, key: &str, blocks: &[String], max: u64) -> CiResult<()> {
        let temporary = std::env::temp_dir().join(format!("gitdata-ci-upload-{}", Uuid::now_v7()));
        let result: CiResult<()> = async {
            let mut file = tokio::fs::OpenOptions::new()
                .create_new(true)
                .write(true)
                .read(true)
                .open(&temporary)
                .await
                .map_err(|error| CiError::Internal(error.to_string()))?;
            let mut size = 0u64;
            for block in blocks {
                let data = self
                    .storage()
                    .client()
                    .get(&format!(
                        "{key}.blocks/{}",
                        crate::secrets::sha256_hex(block.as_bytes())
                    ))
                    .await?;
                size += data.len() as u64;
                if size > max {
                    return Err(CiError::InvalidArgument(
                        "upload exceeds configured size limit".into(),
                    ));
                }
                file.write_all(&data)
                    .await
                    .map_err(|error| CiError::Internal(error.to_string()))?;
            }
            file.seek(std::io::SeekFrom::Start(0))
                .await
                .map_err(|error| CiError::Internal(error.to_string()))?;
            self.storage()
                .client()
                .put_stream(key, &mut file, "application/octet-stream")
                .await?;
            for block in blocks {
                self.storage()
                    .client()
                    .delete(&format!(
                        "{key}.blocks/{}",
                        crate::secrets::sha256_hex(block.as_bytes())
                    ))
                    .await?;
            }
            Ok(())
        }
        .await;
        let _ = tokio::fs::remove_file(temporary).await;
        result
    }
}

fn field(value: &Value, camel: &str, snake: &str) -> String {
    value
        .get(camel)
        .or_else(|| value.get(snake))
        .and_then(|v| v.as_str().or_else(|| v["value"].as_str()))
        .unwrap_or_default()
        .to_owned()
}
fn number(value: Option<&Value>) -> CiResult<u64> {
    value
        .and_then(|value| {
            value
                .as_u64()
                .or_else(|| value.as_str().and_then(|text| text.parse().ok()))
        })
        .ok_or_else(|| CiError::InvalidArgument("expected a nonnegative size".into()))
}

async fn artifact_rpc(
    State(service): State<CiService>,
    Path(method): Path<String>,
    headers: HeaderMap,
    Json(body): Json<Value>,
) -> Result<Json<Value>, RuntimeError> {
    let context = service.runtime_context(&headers).await?;
    Ok(Json(service.artifact_call(&context, &method, &body).await?))
}

#[derive(Deserialize)]
struct OidcQuery {
    audience: Option<String>,
}

async fn oidc_token(
    State(service): State<CiService>,
    headers: HeaderMap,
    Query(query): Query<OidcQuery>,
) -> Result<Json<Value>, RuntimeError> {
    let context = service.runtime_context(&headers).await?;
    if context
        .job
        .actions_context
        .as_ref()
        .and_then(|context| context["permissions"]["id-token"].as_str())
        != Some("write")
    {
        return Err(CiError::Forbidden.into());
    }
    if service.config().ci.oidc_private_key.is_empty() {
        return Err(CiError::InvalidArgument(
            "ci.oidc_private_key is required to issue OIDC tokens".into(),
        )
        .into());
    }
    let github = service.actions_github_context(&context.run).await?;
    let repository = github["repository"].as_str().unwrap_or_default();
    let sub = if let Some(environment) = &context.job.environment {
        format!(
            "repo:{repository}:environment:{}",
            environment.replace(':', "%3A")
        )
    } else if context.run.pr_number.is_some() {
        format!("repo:{repository}:pull_request")
    } else {
        format!("repo:{repository}:ref:{}", context.run.git_ref)
    };
    let now = OffsetDateTime::now_utc().unix_timestamp();
    let issuer = service.actions_runtime_url()?;
    let mut claims = github;
    claims["iss"] = json!(issuer);
    claims["sub"] = json!(sub);
    claims["aud"] = json!(query.audience.unwrap_or_else(|| format!(
        "{}/{}",
        issuer,
        claims["repository_owner"].as_str().unwrap_or_default()
    )));
    claims["iat"] = json!(now);
    claims["nbf"] = json!(now - 5);
    claims["exp"] = json!(now + 300);
    claims["jti"] = json!(Uuid::now_v7());
    claims.as_object_mut().map(|value| value.remove("token"));
    let value = service
        .actions_engine(
            json!({"op":"oidc_sign","token":service.config().ci.oidc_private_key,"value":claims}),
        )
        .await?;
    Ok(Json(json!({"value":value})))
}

async fn oidc_configuration(State(service): State<CiService>) -> Result<Json<Value>, RuntimeError> {
    let issuer = service.actions_runtime_url()?;
    Ok(Json(
        json!({"issuer":issuer,"jwks_uri":format!("{issuer}/.well-known/jwks.json"),"id_token_signing_alg_values_supported":["RS256"],"subject_types_supported":["public"],"response_types_supported":["id_token"]}),
    ))
}

async fn oidc_jwks(State(service): State<CiService>) -> Result<Json<Value>, RuntimeError> {
    if service.config().ci.oidc_private_key.is_empty() {
        return Ok(Json(json!({"keys":[]})));
    }
    Ok(Json(
        service
            .actions_engine(json!({"op":"oidc_jwks","token":service.config().ci.oidc_private_key}))
            .await?,
    ))
}

async fn cache_rpc(
    State(service): State<CiService>,
    Path(method): Path<String>,
    headers: HeaderMap,
    Json(body): Json<Value>,
) -> Result<Json<Value>, RuntimeError> {
    let context = service.runtime_context(&headers).await?;
    let result = match method.as_str() {
        "CreateCacheEntry" => match service.cache_reserve(&context, &body).await {
            Ok(row) => {
                json!({"ok":true,"signedUploadUrl":service.signed_upload_url(&context,"cache",&row.id.to_string())?,"message":""})
            }
            Err(CiError::Conflict(message)) => {
                json!({"ok":false,"signedUploadUrl":"","message":message})
            }
            Err(error) => return Err(error.into()),
        },
        "FinalizeCacheEntryUpload" => {
            let row = service.cache_finalize(&context, None, &body).await?;
            json!({"ok":true,"entryId":row.id.to_string(),"message":""})
        }
        "GetCacheEntryDownloadURL" => {
            let restore = body
                .get("restoreKeys")
                .or_else(|| body.get("restore_keys"))
                .map(|value| ci_protocol::actions::strings(Some(value)))
                .unwrap_or_default();
            match service
                .cache_lookup(
                    &context,
                    &field(&body, "key", "key"),
                    &restore,
                    &field(&body, "version", "version"),
                )
                .await?
            {
                Some(row) => {
                    json!({"ok":true,"signedDownloadUrl":service.storage().client().presign_get(&row.object_key,900)? ,"matchedKey":row.key})
                }
                None => json!({"ok":false,"signedDownloadUrl":"","matchedKey":""}),
            }
        }
        _ => {
            return Err(
                CiError::InvalidArgument(format!("unknown cache method `{method}`")).into(),
            );
        }
    };
    Ok(Json(result))
}

#[derive(Deserialize)]
struct UploadQuery {
    token: String,
    comp: Option<String>,
    blockid: Option<String>,
}

async fn upload_blob(
    State(service): State<CiService>,
    Path((kind, id)): Path<(String, String)>,
    Query(query): Query<UploadQuery>,
    body: Bytes,
) -> Result<Response, RuntimeError> {
    let claims = service.verify_actions_token(&query.token)?;
    if claims["aud"] != "gitdata-actions-upload" || claims["kind"] != kind || claims["id"] != id {
        return Err(CiError::Forbidden.into());
    }
    let context = service.runtime_context_from_claims(&claims).await?;
    let (key, max) = if kind == "artifact" {
        let id = Uuid::parse_str(&id).map_err(|_| CiError::NotFound)?;
        let row = workflow_artifact::Entity::find_by_id(id)
            .one(service.db_write())
            .await?
            .ok_or(CiError::NotFound)?;
        if row.job_id != context.job.id || row.state != ArtifactState::Uploading {
            return Err(CiError::Forbidden.into());
        }
        (row.object_key, service.config().ci.artifact_max_bytes)
    } else if kind == "cache" {
        let id = id.parse::<i64>().map_err(|_| CiError::NotFound)?;
        let row = actions_cache::Entity::find_by_id(id)
            .one(service.db_write())
            .await?
            .ok_or(CiError::NotFound)?;
        if row.job_id != context.job.id || row.ready {
            return Err(CiError::Forbidden.into());
        }
        (row.object_key, service.config().ci.cache_max_bytes)
    } else {
        return Err(CiError::NotFound.into());
    };
    match query.comp.as_deref() {
        Some("block") => {
            let block = query
                .blockid
                .ok_or_else(|| CiError::InvalidArgument("blockid is required".into()))?;
            if block.len() > 256 {
                return Err(CiError::InvalidArgument("blockid is too long".into()).into());
            }
            service
                .storage()
                .client()
                .put(
                    &format!(
                        "{key}.blocks/{}",
                        crate::secrets::sha256_hex(block.as_bytes())
                    ),
                    &body,
                )
                .await
                .map_err(CiError::from)?;
        }
        Some("blocklist") => {
            let text = std::str::from_utf8(&body)
                .map_err(|_| CiError::InvalidArgument("blocklist is not UTF-8".into()))?;
            let mut blocks = Vec::new();
            for part in text.split('<') {
                if let Some(value) = part
                    .strip_prefix("Latest>")
                    .or_else(|| part.strip_prefix("Uncommitted>"))
                    .or_else(|| part.strip_prefix("Committed>"))
                {
                    blocks.push(value.split('<').next().unwrap_or_default().to_owned());
                }
            }
            if blocks.len() > 50000 {
                return Err(CiError::InvalidArgument("too many upload blocks".into()).into());
            }
            service.assemble_blocks(&key, &blocks, max).await?;
        }
        None => {
            if body.len() as u64 > max {
                return Err(CiError::InvalidArgument(
                    "upload exceeds configured size limit".into(),
                )
                .into());
            }
            service
                .storage()
                .client()
                .put(&key, &body)
                .await
                .map_err(CiError::from)?;
        }
        _ => return Err(CiError::InvalidArgument("unsupported upload operation".into()).into()),
    }
    Ok((
        StatusCode::CREATED,
        [
            ("x-ms-request-id", Uuid::now_v7().to_string()),
            ("x-ms-version", "2023-11-03".into()),
            ("etag", format!("\"{}\"", Uuid::now_v7())),
        ],
    )
        .into_response())
}

#[derive(Deserialize)]
struct LegacyLookup {
    keys: String,
    version: String,
}

async fn legacy_cache_lookup(
    State(service): State<CiService>,
    headers: HeaderMap,
    Query(query): Query<LegacyLookup>,
) -> Result<Response, RuntimeError> {
    let context = service.runtime_context(&headers).await?;
    let keys: Vec<String> = query.keys.split(',').map(str::to_owned).collect();
    match service.cache_lookup(&context,keys.first().map(String::as_str).unwrap_or_default(),keys.get(1..).unwrap_or_default(),&query.version).await?{
        Some(row)=>Ok(Json(json!({"cacheKey":row.key,"scope":row.git_ref,"creationTime":row.created_at.format(&time::format_description::well_known::Rfc3339).unwrap_or_default(),"archiveLocation":service.storage().client().presign_get(&row.object_key,900)?})).into_response()),
        None=>Ok(StatusCode::NO_CONTENT.into_response()),
    }
}

async fn legacy_cache_reserve(
    State(service): State<CiService>,
    headers: HeaderMap,
    Json(body): Json<Value>,
) -> Result<Json<Value>, RuntimeError> {
    let context = service.runtime_context(&headers).await?;
    let row = service.cache_reserve(&context, &body).await?;
    Ok(Json(json!({"cacheId":row.id})))
}

async fn legacy_cache_upload(
    State(service): State<CiService>,
    Path(id): Path<i64>,
    headers: HeaderMap,
    body: Bytes,
) -> Result<StatusCode, RuntimeError> {
    let context = service.runtime_context(&headers).await?;
    let row = actions_cache::Entity::find_by_id(id)
        .one(service.db_write())
        .await
        .map_err(CiError::from)?
        .ok_or(CiError::NotFound)?;
    if row.job_id != context.job.id || row.ready {
        return Err(CiError::Forbidden.into());
    }
    let range = headers
        .get("content-range")
        .and_then(|value| value.to_str().ok())
        .ok_or_else(|| CiError::InvalidArgument("content-range is required".into()))?;
    let span = range
        .strip_prefix("bytes ")
        .and_then(|value| value.split('/').next())
        .and_then(|value| value.split_once('-'))
        .ok_or_else(|| CiError::InvalidArgument("invalid content-range".into()))?;
    let start = span
        .0
        .parse::<u64>()
        .map_err(|_| CiError::InvalidArgument("invalid content-range".into()))?;
    let end = span
        .1
        .parse::<u64>()
        .map_err(|_| CiError::InvalidArgument("invalid content-range".into()))?;
    if end < start
        || end - start + 1 != body.len() as u64
        || end >= service.config().ci.cache_max_bytes
    {
        return Err(CiError::InvalidArgument("cache upload range is invalid".into()).into());
    }
    service
        .storage()
        .client()
        .put(&format!("{}.ranges/{start:020}", row.object_key), &body)
        .await
        .map_err(CiError::from)?;
    Ok(StatusCode::NO_CONTENT)
}

async fn legacy_cache_finalize(
    State(service): State<CiService>,
    Path(id): Path<i64>,
    headers: HeaderMap,
    Json(body): Json<Value>,
) -> Result<StatusCode, RuntimeError> {
    let context = service.runtime_context(&headers).await?;
    let row = actions_cache::Entity::find_by_id(id)
        .one(service.db_write())
        .await
        .map_err(CiError::from)?
        .ok_or(CiError::NotFound)?;
    if row.job_id != context.job.id {
        return Err(CiError::Forbidden.into());
    }
    let mut ranges = service
        .storage()
        .client()
        .list(&format!("{}.ranges/", row.object_key))
        .await
        .map_err(CiError::from)?;
    ranges.sort_by(|a, b| a.key.cmp(&b.key));
    let temporary = std::env::temp_dir().join(format!("gitdata-cache-upload-{}", Uuid::now_v7()));
    let assembled: CiResult<()> = async {
        let mut file = tokio::fs::OpenOptions::new()
            .create_new(true)
            .write(true)
            .read(true)
            .open(&temporary)
            .await
            .map_err(|error| CiError::Internal(error.to_string()))?;
        let mut expected = 0;
        for range in &ranges {
            let offset = range
                .key
                .rsplit('/')
                .next()
                .and_then(|value| value.parse::<u64>().ok())
                .ok_or_else(|| CiError::InvalidArgument("invalid cache segment".into()))?;
            if offset != expected {
                return Err(CiError::InvalidArgument(
                    "cache segments overlap or have gaps".into(),
                ));
            }
            let data = service.storage().client().get(&range.key).await?;
            expected += data.len() as u64;
            file.write_all(&data)
                .await
                .map_err(|error| CiError::Internal(error.to_string()))?;
        }
        file.seek(std::io::SeekFrom::Start(0))
            .await
            .map_err(|error| CiError::Internal(error.to_string()))?;
        service
            .storage()
            .client()
            .put_stream(&row.object_key, &mut file, "application/octet-stream")
            .await?;
        Ok(())
    }
    .await;
    let _ = tokio::fs::remove_file(&temporary).await;
    assembled?;
    service.cache_finalize(&context, Some(id), &body).await?;
    for range in ranges {
        service
            .storage()
            .client()
            .delete(&range.key)
            .await
            .map_err(CiError::from)?;
    }
    Ok(StatusCode::NO_CONTENT)
}