use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use sqlx::{FromRow, PgPool, Postgres, QueryBuilder};
use uuid::Uuid;

use crate::schema::ensure_storage_runtime_schema;
use crate::store::{S3CatalogRuntimeBinding, S3CredentialStoreError, load_s3_catalog_runtime};

#[derive(Debug, Clone, Serialize, FromRow)]
pub struct ManagedFileRecord {
    pub id: String,
    pub name: String,
    pub original_name: Option<String>,
    pub url: Option<String>,
    pub bucket: String,
    pub s3_id: Option<String>,
    pub prefix_path: Option<String>,
    pub size_bytes: Option<i64>,
    pub mime_type: Option<String>,
    pub resource_id: Option<String>,
    pub organization_id: Option<String>,
    pub metadata: Value,
    pub created_at: DateTime<Utc>,
    pub updated_at: DateTime<Utc>,
    pub storage_key: String,
    pub uploaded_by_user_id: Option<String>,
    pub extension: Option<String>,
    pub is_public: bool,
    pub visibility: String,
    pub status: String,
    pub deleted_at: Option<DateTime<Utc>>,
    pub created_by_user_id: String,
    pub file_name: Option<String>,
    pub content_type: Option<String>,
    pub checksum_sha256: Option<String>,
}

#[derive(Debug, Clone, Serialize)]
pub struct ManagedStorageTarget {
    pub s3_id: String,
    pub endpoint: String,
    pub region: String,
    pub bucket: String,
    pub provider: String,
    pub force_path_style: bool,
    pub default_prefix: Option<String>,
    pub public_base_url: Option<String>,
    pub access_key_id: String,
    pub secret_key: String,
    pub session_token: Option<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize, FromRow)]
pub struct PresignedUrlCacheRecord {
    pub file_id: String,
    pub purpose: String,
    pub bucket: String,
    pub storage_key: String,
    pub url: String,
    pub expires_at: DateTime<Utc>,
    pub expires_at_epoch_seconds: i64,
    pub created_at: DateTime<Utc>,
    pub metadata: Value,
}

#[derive(Debug, Clone, Default)]
pub struct FileAccessContext {
    pub user_id: Option<String>,
    pub organization_id: Option<String>,
    pub organization_ids: Vec<String>,
    pub role_ids: Vec<String>,
    pub team_ids: Vec<String>,
    pub is_admin: bool,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum FilePermissionKind {
    Read,
    Write,
    Delete,
    Share,
    Owner,
}

impl FilePermissionKind {
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Read => "read",
            Self::Write => "write",
            Self::Delete => "delete",
            Self::Share => "share",
            Self::Owner => "owner",
        }
    }

    fn implied_permissions(self) -> &'static [&'static str] {
        match self {
            Self::Read => &["read", "owner"],
            Self::Write => &["write", "owner"],
            Self::Delete => &["delete", "owner"],
            Self::Share => &["share", "owner"],
            Self::Owner => &["owner"],
        }
    }

    pub fn parse(raw: &str) -> Option<Self> {
        match raw.trim().to_ascii_lowercase().as_str() {
            "read" => Some(Self::Read),
            "write" => Some(Self::Write),
            "delete" => Some(Self::Delete),
            "share" => Some(Self::Share),
            "owner" => Some(Self::Owner),
            _ => None,
        }
    }
}

#[derive(Debug, Clone, Serialize, FromRow)]
pub struct FilePermissionRecord {
    pub id: Uuid,
    pub file_id: String,
    pub principal_type: String,
    pub principal_id: String,
    pub permission: String,
    pub granted_by_user_id: Option<String>,
    pub created_at: DateTime<Utc>,
    pub expires_at: Option<DateTime<Utc>>,
    pub revoked_at: Option<DateTime<Utc>>,
    pub metadata: Value,
}

#[derive(Debug, Clone)]
pub struct FilePermissionGrantInput {
    pub principal_type: String,
    pub principal_id: String,
    pub permission: FilePermissionKind,
    pub granted_by_user_id: Option<String>,
    pub expires_at: Option<DateTime<Utc>>,
    pub metadata: Value,
}

#[derive(Debug, Clone)]
pub struct ManagedFileUpsertInput {
    pub file_id: Option<String>,
    pub s3_id: Uuid,
    pub bucket: String,
    pub storage_key: String,
    pub organization_id: Option<String>,
    pub uploaded_by_user_id: Option<String>,
    pub name: Option<String>,
    pub original_name: Option<String>,
    pub mime_type: Option<String>,
    pub size_bytes: Option<i64>,
    pub resource_id: Option<String>,
    pub is_public: bool,
    pub visibility: Option<String>,
    pub metadata: Value,
}

#[derive(Debug, Clone, Default)]
pub struct ManagedFilePatchInput {
    pub file_name: Option<String>,
    pub original_name: Option<String>,
    pub content_type: Option<String>,
    pub size_bytes: Option<i64>,
    pub checksum_sha256: Option<String>,
    pub resource_id: Option<String>,
    pub visibility: Option<String>,
    pub status: Option<String>,
    pub metadata: Option<Value>,
}

#[derive(Debug, Clone)]
pub struct ManagedFileActionRecord {
    pub file_id: String,
    pub organization_id: Option<String>,
    pub user_id: Option<String>,
    pub route: String,
    pub action: String,
    pub status: ManagedFileActionStatus,
    pub message: String,
    pub request_payload: Value,
    pub result_payload: Value,
    pub metadata: Value,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
pub enum ManagedFileActionStatus {
    Success,
    Failed,
}

impl ManagedFileActionStatus {
    fn as_audit_status(self) -> &'static str {
        match self {
            Self::Success => "success",
            Self::Failed => "error",
        }
    }
}

#[derive(Debug, Clone, Serialize, FromRow)]
pub struct StorageAuditEventRecord {
    pub id: Uuid,
    pub operation: String,
    pub file_id: Option<String>,
    pub s3_id: Option<Uuid>,
    pub bucket: Option<String>,
    pub storage_key: Option<String>,
    pub actor_user_id: Option<String>,
    pub organization_id: Option<String>,
    pub status: String,
    pub error: Option<String>,
    pub metadata: Value,
    pub created_at: DateTime<Utc>,
}

#[derive(Debug, Clone, Default)]
pub struct StorageAuditListInput {
    pub file_id: Option<String>,
    pub actor_user_id: Option<String>,
    pub limit: usize,
    pub offset: usize,
}

pub async fn ensure_managed_storage_tables(pool: &PgPool) -> Result<(), S3CredentialStoreError> {
    ensure_storage_runtime_schema(pool).await?;
    Ok(())
}

pub async fn load_managed_file(
    pool: &PgPool,
    file_id: &str,
) -> Result<ManagedFileRecord, S3CredentialStoreError> {
    ensure_managed_storage_tables(pool).await?;
    query_file_by_id(pool, file_id).await
}

pub async fn load_managed_storage_target(
    pool: &PgPool,
    s3_id: Uuid,
) -> Result<ManagedStorageTarget, S3CredentialStoreError> {
    ensure_managed_storage_tables(pool).await?;
    let runtime = load_s3_catalog_runtime(pool, s3_id).await?;
    Ok(runtime_binding_to_target(runtime))
}

pub async fn find_existing_file_by_path(
    pool: &PgPool,
    s3_id: Uuid,
    bucket: &str,
    storage_key: &str,
) -> Result<Option<ManagedFileRecord>, S3CredentialStoreError> {
    ensure_managed_storage_tables(pool).await?;

    let sql = file_select_sql(
        r#"
        WHERE s3_id = $1
          AND bucket = $2
          AND storage_key = $3
        ORDER BY updated_at DESC
        LIMIT 1
        "#,
    );
    let row = sqlx::query_as::<_, ManagedFileRecord>(sql.as_str())
        .bind(s3_id)
        .bind(bucket)
        .bind(storage_key)
        .fetch_optional(pool)
        .await?;

    Ok(row)
}

pub async fn upsert_managed_file_for_upload(
    pool: &PgPool,
    input: &ManagedFileUpsertInput,
) -> Result<ManagedFileRecord, S3CredentialStoreError> {
    ensure_managed_storage_tables(pool).await?;

    let normalized_storage_key = normalize_storage_key(&input.storage_key)?;
    let normalized_bucket = normalize_required(&input.bucket, "bucket")?;
    let normalized_name = normalize_optional(input.name.as_deref())
        .unwrap_or_else(|| object_name(&normalized_storage_key));
    let original_name = normalize_optional(input.original_name.as_deref())
        .or_else(|| Some(normalized_name.clone()));
    let prefix_path = object_prefix_path(&normalized_storage_key);
    let extension = file_extension(&normalized_name);
    let visibility = normalize_visibility(
        input.visibility.as_deref(),
        input.is_public,
        input.organization_id.as_deref(),
    )?;

    let existing = find_existing_file_by_path(
        pool,
        input.s3_id,
        &normalized_bucket,
        &normalized_storage_key,
    )
    .await?;
    let file_id = match (&input.file_id, existing.as_ref()) {
        (Some(file_id), Some(existing_file)) if file_id != &existing_file.id => {
            return Err(S3CredentialStoreError::InvalidInput(format!(
                "storage key '{}' is already registered to file '{}'",
                normalized_storage_key, existing_file.id
            )));
        }
        (Some(file_id), _) => normalize_required(file_id, "file_id")?,
        (None, Some(existing_file)) => existing_file.id.clone(),
        (None, None) => Uuid::new_v4().to_string(),
    };

    let created_by = normalize_optional(input.uploaded_by_user_id.as_deref())
        .unwrap_or_else(|| "system".to_string());

    sqlx::query(
        r#"
        INSERT INTO athena.files (
            id,
            organization_id,
            created_by_user_id,
            s3_id,
            bucket,
            storage_key,
            file_name,
            original_name,
            content_type,
            size_bytes,
            visibility,
            status,
            resource_id,
            metadata,
            prefix_path,
            extension,
            deleted_at,
            created_at,
            updated_at
        )
        VALUES (
            $1,
            $2,
            $3,
            $4,
            $5,
            $6,
            $7,
            $8,
            $9,
            $10,
            $11,
            'pending',
            $12,
            $13,
            $14,
            $15,
            NULL,
            now(),
            now()
        )
        ON CONFLICT (id) DO UPDATE
        SET
            organization_id = EXCLUDED.organization_id,
            created_by_user_id = EXCLUDED.created_by_user_id,
            s3_id = EXCLUDED.s3_id,
            bucket = EXCLUDED.bucket,
            storage_key = EXCLUDED.storage_key,
            file_name = EXCLUDED.file_name,
            original_name = EXCLUDED.original_name,
            content_type = COALESCE(EXCLUDED.content_type, athena.files.content_type),
            size_bytes = COALESCE(EXCLUDED.size_bytes, athena.files.size_bytes),
            visibility = EXCLUDED.visibility,
            status = 'pending',
            resource_id = COALESCE(EXCLUDED.resource_id, athena.files.resource_id),
            metadata = COALESCE(athena.files.metadata, '{}'::jsonb) || EXCLUDED.metadata,
            prefix_path = EXCLUDED.prefix_path,
            extension = EXCLUDED.extension,
            deleted_at = NULL,
            updated_at = now()
        "#,
    )
    .bind(&file_id)
    .bind(normalize_optional(input.organization_id.as_deref()))
    .bind(&created_by)
    .bind(input.s3_id)
    .bind(&normalized_bucket)
    .bind(&normalized_storage_key)
    .bind(&normalized_name)
    .bind(original_name)
    .bind(normalize_optional(input.mime_type.as_deref()))
    .bind(input.size_bytes)
    .bind(&visibility)
    .bind(normalize_optional(input.resource_id.as_deref()))
    .bind(normalize_metadata_object(input.metadata.clone()))
    .bind(prefix_path)
    .bind(extension)
    .execute(pool)
    .await?;

    grant_file_permission(
        pool,
        &file_id,
        &FilePermissionGrantInput {
            principal_type: "user".to_string(),
            principal_id: created_by.clone(),
            permission: FilePermissionKind::Owner,
            granted_by_user_id: Some(created_by),
            expires_at: None,
            metadata: json!({
                "source": "storage.files.upload-url",
            }),
        },
    )
    .await?;

    query_file_by_id_any_status(pool, &file_id).await
}

pub async fn grant_file_permission(
    pool: &PgPool,
    file_id: &str,
    input: &FilePermissionGrantInput,
) -> Result<(), S3CredentialStoreError> {
    ensure_managed_storage_tables(pool).await?;

    sqlx::query(
        r#"
        INSERT INTO athena.file_permissions (
            file_id,
            principal_type,
            principal_id,
            permission,
            granted_by_user_id,
            expires_at,
            metadata
        )
        VALUES ($1, $2, $3, $4, $5, $6, $7)
        ON CONFLICT (file_id, principal_type, principal_id, permission)
        DO UPDATE
        SET
            revoked_at = NULL,
            expires_at = EXCLUDED.expires_at,
            granted_by_user_id = EXCLUDED.granted_by_user_id,
            created_at = now(),
            metadata = COALESCE(athena.file_permissions.metadata, '{}'::jsonb) || EXCLUDED.metadata
        "#,
    )
    .bind(file_id)
    .bind(normalize_required(&input.principal_type, "principal_type")?)
    .bind(normalize_required(&input.principal_id, "principal_id")?)
    .bind(input.permission.as_str())
    .bind(normalize_optional(input.granted_by_user_id.as_deref()))
    .bind(input.expires_at)
    .bind(normalize_metadata_object(input.metadata.clone()))
    .execute(pool)
    .await?;

    Ok(())
}

pub async fn revoke_file_permission(
    pool: &PgPool,
    file_id: &str,
    principal_type: &str,
    principal_id: &str,
    permission: FilePermissionKind,
) -> Result<bool, S3CredentialStoreError> {
    ensure_managed_storage_tables(pool).await?;

    let result = sqlx::query(
        r#"
        UPDATE athena.file_permissions
        SET revoked_at = now()
        WHERE file_id = $1
          AND principal_type = $2
          AND principal_id = $3
          AND permission = $4
          AND revoked_at IS NULL
        "#,
    )
    .bind(file_id)
    .bind(normalize_required(principal_type, "principal_type")?)
    .bind(normalize_required(principal_id, "principal_id")?)
    .bind(permission.as_str())
    .execute(pool)
    .await?;

    Ok(result.rows_affected() > 0)
}

pub async fn list_file_permissions(
    pool: &PgPool,
    file_id: &str,
) -> Result<Vec<FilePermissionRecord>, S3CredentialStoreError> {
    ensure_managed_storage_tables(pool).await?;

    let rows = sqlx::query_as::<_, FilePermissionRecord>(
        r#"
        SELECT
            id,
            file_id,
            principal_type,
            principal_id,
            permission,
            granted_by_user_id,
            created_at,
            expires_at,
            revoked_at,
            metadata
        FROM athena.file_permissions
        WHERE file_id = $1
        ORDER BY created_at DESC, id DESC
        "#,
    )
    .bind(file_id)
    .fetch_all(pool)
    .await?;

    Ok(rows)
}

pub async fn authorize_file_access(
    pool: &PgPool,
    file_id: &str,
    access: &FileAccessContext,
    permission: FilePermissionKind,
) -> Result<ManagedFileRecord, S3CredentialStoreError> {
    ensure_managed_storage_tables(pool).await?;
    let file = query_file_by_id(pool, file_id).await?;

    if access.is_admin {
        return Ok(file);
    }

    if permission == FilePermissionKind::Read {
        if file.is_public {
            return Ok(file);
        }

        let organization_ids = merged_organization_ids(access);
        if file.visibility == "organization"
            && file
                .organization_id
                .as_ref()
                .map(|organization_id| {
                    organization_ids
                        .iter()
                        .any(|candidate| candidate == organization_id)
                })
                .unwrap_or(false)
        {
            return Ok(file);
        }
    }

    if access
        .user_id
        .as_deref()
        .map(|user_id| user_id == file.created_by_user_id)
        .unwrap_or(false)
    {
        return Ok(file);
    }

    let Some(permission_found) = permission_exists(pool, file_id, access, permission).await? else {
        return Err(S3CredentialStoreError::AccessDenied);
    };

    if permission_found {
        Ok(file)
    } else {
        Err(S3CredentialStoreError::AccessDenied)
    }
}

pub async fn check_file_permission(
    pool: &PgPool,
    file_id: &str,
    access: &FileAccessContext,
    permission: FilePermissionKind,
) -> Result<bool, S3CredentialStoreError> {
    match authorize_file_access(pool, file_id, access, permission).await {
        Ok(_) => Ok(true),
        Err(S3CredentialStoreError::AccessDenied) => Ok(false),
        Err(err) => Err(err),
    }
}

pub async fn list_files_for_prefix(
    pool: &PgPool,
    organization_id: &str,
    s3_id: Uuid,
    prefix: &str,
) -> Result<Vec<ManagedFileRecord>, S3CredentialStoreError> {
    list_files_for_storage(pool, organization_id, s3_id, Some(prefix)).await
}

pub async fn list_files_for_storage(
    pool: &PgPool,
    organization_id: &str,
    s3_id: Uuid,
    prefix: Option<&str>,
) -> Result<Vec<ManagedFileRecord>, S3CredentialStoreError> {
    ensure_managed_storage_tables(pool).await?;

    let like_pattern = match normalize_optional(prefix) {
        Some(prefix) => Some(format!("{}%", normalize_prefix_for_match(&prefix)?)),
        None => None,
    };
    let organization_filter = normalize_optional(Some(organization_id));

    let rows = if let Some(organization_id) = organization_filter {
        let sql = file_select_sql(
            r#"
            WHERE s3_id = $1
              AND ($2::text IS NULL OR storage_key LIKE $2)
              AND (organization_id = $3 OR organization_id IS NULL)
            ORDER BY storage_key ASC, created_at ASC
            "#,
        );
        sqlx::query_as::<_, ManagedFileRecord>(sql.as_str())
            .bind(s3_id)
            .bind(like_pattern.as_deref())
            .bind(organization_id)
            .fetch_all(pool)
            .await?
    } else {
        let sql = file_select_sql(
            r#"
            WHERE s3_id = $1
              AND ($2::text IS NULL OR storage_key LIKE $2)
            ORDER BY storage_key ASC, created_at ASC
            "#,
        );
        sqlx::query_as::<_, ManagedFileRecord>(sql.as_str())
            .bind(s3_id)
            .bind(like_pattern.as_deref())
            .fetch_all(pool)
            .await?
    };

    Ok(rows)
}

pub async fn search_files(
    pool: &PgPool,
    organization_ids: &[String],
    query: Option<&str>,
    limit: usize,
) -> Result<Vec<ManagedFileRecord>, S3CredentialStoreError> {
    ensure_managed_storage_tables(pool).await?;

    let normalized_query = normalize_optional(query);
    let row_limit = limit.max(1).min(1000) as i64;

    let rows = if organization_ids.is_empty() {
        let sql = file_select_sql(
            r#"
            WHERE (
                $1::text IS NULL
                OR storage_key ILIKE $2
                OR COALESCE(file_name, '') ILIKE $2
            )
            ORDER BY updated_at DESC
            LIMIT $3
            "#,
        );
        sqlx::query_as::<_, ManagedFileRecord>(sql.as_str())
            .bind(normalized_query.as_deref())
            .bind(
                normalized_query
                    .as_ref()
                    .map(|value| format!("%{value}%"))
                    .unwrap_or_else(|| "%".to_string()),
            )
            .bind(row_limit)
            .fetch_all(pool)
            .await?
    } else {
        let sql = file_select_sql(
            r#"
            WHERE (
                organization_id IS NULL
                OR organization_id = ANY($1)
            )
              AND (
                $2::text IS NULL
                OR storage_key ILIKE $3
                OR COALESCE(file_name, '') ILIKE $3
              )
            ORDER BY updated_at DESC
            LIMIT $4
            "#,
        );
        sqlx::query_as::<_, ManagedFileRecord>(sql.as_str())
            .bind(organization_ids)
            .bind(normalized_query.as_deref())
            .bind(
                normalized_query
                    .as_ref()
                    .map(|value| format!("%{value}%"))
                    .unwrap_or_else(|| "%".to_string()),
            )
            .bind(row_limit)
            .fetch_all(pool)
            .await?
    };

    Ok(rows)
}

pub async fn update_file_location(
    pool: &PgPool,
    file_id: &str,
    new_storage_key: &str,
    new_bucket: Option<&str>,
) -> Result<ManagedFileRecord, S3CredentialStoreError> {
    ensure_managed_storage_tables(pool).await?;

    let normalized_storage_key = normalize_storage_key(new_storage_key)?;
    let new_name = object_name(&normalized_storage_key);
    let new_prefix_path = object_prefix_path(&normalized_storage_key);
    let new_extension = file_extension(&new_name);

    let updated = sqlx::query(
        r#"
        UPDATE athena.files
        SET
            file_name = $2,
            original_name = COALESCE(original_name, $2),
            bucket = COALESCE($3, bucket),
            prefix_path = $4,
            storage_key = $5,
            extension = $6,
            updated_at = now()
        WHERE id = $1
          AND deleted_at IS NULL
        "#,
    )
    .bind(file_id)
    .bind(&new_name)
    .bind(normalize_optional(new_bucket))
    .bind(new_prefix_path)
    .bind(&normalized_storage_key)
    .bind(new_extension)
    .execute(pool)
    .await?;

    if updated.rows_affected() == 0 {
        return Err(S3CredentialStoreError::NotFound);
    }

    query_file_by_id(pool, file_id).await
}

pub async fn patch_managed_file(
    pool: &PgPool,
    file_id: &str,
    input: &ManagedFilePatchInput,
) -> Result<ManagedFileRecord, S3CredentialStoreError> {
    ensure_managed_storage_tables(pool).await?;

    let normalized_visibility = match input.visibility.as_deref() {
        Some(value) => Some(normalize_visibility(Some(value), false, None)?),
        None => None,
    };
    let normalized_status = match input.status.as_deref() {
        Some(value) => Some(normalize_status(value)?),
        None => None,
    };

    let mut qb = QueryBuilder::<Postgres>::new("UPDATE athena.files SET ");
    let mut wrote = false;

    macro_rules! push_assignment {
        ($column:literal, $value:expr) => {{
            if wrote {
                qb.push(", ");
            }
            qb.push($column);
            qb.push(" = ");
            qb.push_bind($value);
            wrote = true;
        }};
    }

    if let Some(value) = normalize_optional(input.file_name.as_deref()) {
        push_assignment!("file_name", value.clone());
        if input.original_name.is_none() {
            if wrote {
                qb.push(", ");
            }
            qb.push("original_name = COALESCE(original_name, ");
            qb.push_bind(value);
            qb.push(")");
        }
    }
    if let Some(value) = normalize_optional(input.original_name.as_deref()) {
        push_assignment!("original_name", value);
    }
    if let Some(value) = normalize_optional(input.content_type.as_deref()) {
        push_assignment!("content_type", value);
    }
    if let Some(value) = input.size_bytes {
        push_assignment!("size_bytes", value);
    }
    if let Some(value) = normalize_optional(input.checksum_sha256.as_deref()) {
        push_assignment!("checksum_sha256", value);
    }
    if let Some(value) = normalize_optional(input.resource_id.as_deref()) {
        push_assignment!("resource_id", value);
    }
    if let Some(value) = normalized_visibility {
        push_assignment!("visibility", value);
    }
    if let Some(value) = normalized_status {
        push_assignment!("status", value);
    }
    if let Some(value) = input.metadata.clone() {
        if wrote {
            qb.push(", ");
        }
        qb.push("metadata = COALESCE(metadata, '{}'::jsonb) || ");
        qb.push_bind(normalize_metadata_object(value));
        wrote = true;
    }

    if !wrote {
        return query_file_by_id(pool, file_id).await;
    }

    qb.push(", updated_at = now() WHERE id = ");
    qb.push_bind(file_id);
    qb.push(" AND deleted_at IS NULL");

    let updated = qb.build().execute(pool).await?;
    if updated.rows_affected() == 0 {
        return Err(S3CredentialStoreError::NotFound);
    }

    query_file_by_id(pool, file_id).await
}

pub async fn update_file_status(
    pool: &PgPool,
    file_id: &str,
    status: &str,
    size_bytes: Option<i64>,
    content_type: Option<&str>,
    checksum_sha256: Option<&str>,
    metadata: Option<Value>,
) -> Result<ManagedFileRecord, S3CredentialStoreError> {
    ensure_managed_storage_tables(pool).await?;

    let normalized_status = normalize_status(status)?;

    let updated = sqlx::query(
        r#"
        UPDATE athena.files
        SET
            status = $2,
            size_bytes = COALESCE($3, size_bytes),
            content_type = COALESCE($4, content_type),
            checksum_sha256 = COALESCE($5, checksum_sha256),
            metadata = CASE
                WHEN $6::jsonb IS NULL THEN metadata
                ELSE COALESCE(metadata, '{}'::jsonb) || $6::jsonb
            END,
            deleted_at = CASE WHEN $2 = 'deleted' THEN COALESCE(deleted_at, now()) ELSE NULL END,
            updated_at = now()
        WHERE id = $1
        "#,
    )
    .bind(file_id)
    .bind(normalized_status)
    .bind(size_bytes)
    .bind(normalize_optional(content_type))
    .bind(normalize_optional(checksum_sha256))
    .bind(metadata.map(normalize_metadata_object))
    .execute(pool)
    .await?;

    if updated.rows_affected() == 0 {
        return Err(S3CredentialStoreError::NotFound);
    }

    query_file_by_id_any_status(pool, file_id).await
}

pub async fn set_file_visibility(
    pool: &PgPool,
    file_id: &str,
    visibility: &str,
) -> Result<ManagedFileRecord, S3CredentialStoreError> {
    ensure_managed_storage_tables(pool).await?;

    let normalized_visibility = normalize_visibility(Some(visibility), false, None)?;
    let updated = sqlx::query(
        r#"
        UPDATE athena.files
        SET visibility = $2, updated_at = now()
        WHERE id = $1
          AND deleted_at IS NULL
        "#,
    )
    .bind(file_id)
    .bind(normalized_visibility)
    .execute(pool)
    .await?;

    if updated.rows_affected() == 0 {
        return Err(S3CredentialStoreError::NotFound);
    }

    query_file_by_id(pool, file_id).await
}

pub async fn set_file_public_state(
    pool: &PgPool,
    file_id: &str,
    is_public: bool,
) -> Result<ManagedFileRecord, S3CredentialStoreError> {
    set_file_visibility(pool, file_id, if is_public { "public" } else { "private" }).await
}

pub async fn mark_file_deleted(
    pool: &PgPool,
    file_id: &str,
) -> Result<ManagedFileRecord, S3CredentialStoreError> {
    update_file_status(pool, file_id, "deleted", None, None, None, None).await
}

pub async fn restore_file(
    pool: &PgPool,
    file_id: &str,
) -> Result<ManagedFileRecord, S3CredentialStoreError> {
    ensure_managed_storage_tables(pool).await?;

    let updated = sqlx::query(
        r#"
        UPDATE athena.files
        SET
            status = CASE
                WHEN COALESCE(size_bytes, 0) > 0 THEN 'uploaded'
                ELSE 'pending'
            END,
            deleted_at = NULL,
            updated_at = now()
        WHERE id = $1
        "#,
    )
    .bind(file_id)
    .execute(pool)
    .await?;

    if updated.rows_affected() == 0 {
        return Err(S3CredentialStoreError::NotFound);
    }

    query_file_by_id_any_status(pool, file_id).await
}

pub async fn purge_file(
    pool: &PgPool,
    file_id: &str,
) -> Result<ManagedFileRecord, S3CredentialStoreError> {
    ensure_managed_storage_tables(pool).await?;
    let file = query_file_by_id_any_status(pool, file_id).await?;

    let deleted = sqlx::query("DELETE FROM athena.files WHERE id = $1")
        .bind(file_id)
        .execute(pool)
        .await?;

    if deleted.rows_affected() == 0 {
        return Err(S3CredentialStoreError::NotFound);
    }

    Ok(file)
}

pub async fn load_cached_presigned_url(
    pool: &PgPool,
    file_id: &str,
    purpose: &str,
) -> Result<Option<PresignedUrlCacheRecord>, S3CredentialStoreError> {
    ensure_managed_storage_tables(pool).await?;

    let row = sqlx::query_as::<_, PresignedUrlCacheRecord>(
        r#"
        SELECT
            file_id,
            purpose,
            bucket,
            storage_key,
            url,
            expires_at,
            expires_at_epoch_seconds,
            created_at,
            metadata
        FROM athena.s3_url_cache
        WHERE file_id = $1
          AND purpose = $2
        LIMIT 1
        "#,
    )
    .bind(file_id)
    .bind(purpose)
    .fetch_optional(pool)
    .await?;

    Ok(row)
}

pub async fn upsert_cached_presigned_url(
    pool: &PgPool,
    entry: &PresignedUrlCacheRecord,
) -> Result<(), S3CredentialStoreError> {
    ensure_managed_storage_tables(pool).await?;

    sqlx::query(
        r#"
        INSERT INTO athena.s3_url_cache (
            file_id,
            purpose,
            bucket,
            storage_key,
            url,
            expires_at,
            expires_at_epoch_seconds,
            metadata
        )
        VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
        ON CONFLICT (file_id, purpose)
        DO UPDATE
        SET
            bucket = EXCLUDED.bucket,
            storage_key = EXCLUDED.storage_key,
            url = EXCLUDED.url,
            expires_at = EXCLUDED.expires_at,
            expires_at_epoch_seconds = EXCLUDED.expires_at_epoch_seconds,
            created_at = now(),
            metadata = EXCLUDED.metadata
        "#,
    )
    .bind(&entry.file_id)
    .bind(&entry.purpose)
    .bind(&entry.bucket)
    .bind(&entry.storage_key)
    .bind(&entry.url)
    .bind(entry.expires_at)
    .bind(entry.expires_at_epoch_seconds)
    .bind(normalize_metadata_object(entry.metadata.clone()))
    .execute(pool)
    .await?;

    Ok(())
}

pub async fn delete_cached_presigned_url(
    pool: &PgPool,
    file_id: &str,
    purpose: Option<&str>,
) -> Result<(), S3CredentialStoreError> {
    ensure_managed_storage_tables(pool).await?;

    if let Some(purpose) = purpose {
        sqlx::query("DELETE FROM athena.s3_url_cache WHERE file_id = $1 AND purpose = $2")
            .bind(file_id)
            .bind(purpose)
            .execute(pool)
            .await?;
    } else {
        sqlx::query("DELETE FROM athena.s3_url_cache WHERE file_id = $1")
            .bind(file_id)
            .execute(pool)
            .await?;
    }

    Ok(())
}

pub async fn record_managed_file_action(
    pool: &PgPool,
    record: &ManagedFileActionRecord,
) -> Result<(), S3CredentialStoreError> {
    ensure_managed_storage_tables(pool).await?;

    let file = query_file_by_id_any_status(pool, &record.file_id).await?;
    sqlx::query(
        r#"
        INSERT INTO athena.storage_audit_events (
            operation,
            file_id,
            s3_id,
            bucket,
            storage_key,
            actor_user_id,
            organization_id,
            status,
            error,
            metadata
        )
        VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
        "#,
    )
    .bind(&record.action)
    .bind(&record.file_id)
    .bind(
        file.s3_id
            .as_deref()
            .and_then(|value| Uuid::parse_str(value).ok()),
    )
    .bind(&file.bucket)
    .bind(&file.storage_key)
    .bind(normalize_optional(record.user_id.as_deref()))
    .bind(
        normalize_optional(record.organization_id.as_deref())
            .or_else(|| file.organization_id.clone()),
    )
    .bind(record.status.as_audit_status())
    .bind(if record.status == ManagedFileActionStatus::Failed {
        Some(record.message.clone())
    } else {
        None
    })
    .bind(json!({
        "route": record.route,
        "message": record.message,
        "request": record.request_payload,
        "result": record.result_payload,
        "metadata": record.metadata,
    }))
    .execute(pool)
    .await?;

    Ok(())
}

pub async fn list_storage_audit_events(
    pool: &PgPool,
    input: &StorageAuditListInput,
) -> Result<Vec<StorageAuditEventRecord>, S3CredentialStoreError> {
    ensure_managed_storage_tables(pool).await?;

    let mut qb = QueryBuilder::<Postgres>::new(
        r#"
        SELECT
            id,
            operation,
            file_id,
            s3_id,
            bucket,
            storage_key,
            actor_user_id,
            organization_id,
            status,
            error,
            metadata,
            created_at
        FROM athena.storage_audit_events
        WHERE 1 = 1
        "#,
    );

    if let Some(file_id) = normalize_optional(input.file_id.as_deref()) {
        qb.push(" AND file_id = ");
        qb.push_bind(file_id);
    }
    if let Some(actor_user_id) = normalize_optional(input.actor_user_id.as_deref()) {
        qb.push(" AND actor_user_id = ");
        qb.push_bind(actor_user_id);
    }
    qb.push(" ORDER BY created_at DESC LIMIT ");
    qb.push_bind(input.limit.max(1).min(500) as i64);
    qb.push(" OFFSET ");
    qb.push_bind(input.offset as i64);

    let rows = qb
        .build_query_as::<StorageAuditEventRecord>()
        .fetch_all(pool)
        .await?;
    Ok(rows)
}

async fn permission_exists(
    pool: &PgPool,
    file_id: &str,
    access: &FileAccessContext,
    permission: FilePermissionKind,
) -> Result<Option<bool>, S3CredentialStoreError> {
    let organization_ids = merged_organization_ids(access);
    let has_principals = access.user_id.is_some()
        || !organization_ids.is_empty()
        || !access.role_ids.is_empty()
        || !access.team_ids.is_empty();

    if !has_principals {
        return Ok(None);
    }

    let mut qb =
        QueryBuilder::<Postgres>::new("SELECT 1 FROM athena.file_permissions WHERE file_id = ");
    qb.push_bind(file_id);
    qb.push(" AND revoked_at IS NULL AND (expires_at IS NULL OR expires_at > now()) AND permission IN (");
    let mut permissions = qb.separated(", ");
    for value in permission.implied_permissions() {
        permissions.push_bind(*value);
    }
    qb.push(") AND (");

    let mut wrote_principal = false;
    if let Some(user_id) = access.user_id.as_deref() {
        qb.push("(principal_type = 'user' AND principal_id = ");
        qb.push_bind(user_id);
        qb.push(")");
        wrote_principal = true;
    }
    if !organization_ids.is_empty() {
        if wrote_principal {
            qb.push(" OR ");
        }
        qb.push("(principal_type = 'organization' AND principal_id IN (");
        let mut organization_values = qb.separated(", ");
        for organization_id in &organization_ids {
            organization_values.push_bind(organization_id);
        }
        qb.push("))");
        wrote_principal = true;
    }
    if !access.role_ids.is_empty() {
        if wrote_principal {
            qb.push(" OR ");
        }
        qb.push("(principal_type = 'role' AND principal_id IN (");
        let mut role_ids = qb.separated(", ");
        for role_id in &access.role_ids {
            role_ids.push_bind(role_id);
        }
        qb.push("))");
        wrote_principal = true;
    }
    if !access.team_ids.is_empty() {
        if wrote_principal {
            qb.push(" OR ");
        }
        qb.push("(principal_type = 'team' AND principal_id IN (");
        let mut team_ids = qb.separated(", ");
        for team_id in &access.team_ids {
            team_ids.push_bind(team_id);
        }
        qb.push("))");
    }
    qb.push(") LIMIT 1");

    let found = qb.build().fetch_optional(pool).await?.is_some();
    Ok(Some(found))
}

async fn query_file_by_id(
    pool: &PgPool,
    file_id: &str,
) -> Result<ManagedFileRecord, S3CredentialStoreError> {
    let sql = file_select_sql(
        r#"
        WHERE id = $1
          AND deleted_at IS NULL
        LIMIT 1
        "#,
    );
    let row = sqlx::query_as::<_, ManagedFileRecord>(sql.as_str())
        .bind(file_id)
        .fetch_optional(pool)
        .await?;

    row.ok_or(S3CredentialStoreError::NotFound)
}

async fn query_file_by_id_any_status(
    pool: &PgPool,
    file_id: &str,
) -> Result<ManagedFileRecord, S3CredentialStoreError> {
    let sql = file_select_sql(
        r#"
        WHERE id = $1
        LIMIT 1
        "#,
    );
    let row = sqlx::query_as::<_, ManagedFileRecord>(sql.as_str())
        .bind(file_id)
        .fetch_optional(pool)
        .await?;

    row.ok_or(S3CredentialStoreError::NotFound)
}

fn file_select_sql(suffix: &str) -> String {
    format!(
        r#"
        SELECT
            id,
            COALESCE(file_name, storage_key) AS name,
            COALESCE(original_name, file_name) AS original_name,
            NULL::text AS url,
            bucket,
            s3_id::text AS s3_id,
            prefix_path,
            size_bytes,
            content_type AS mime_type,
            resource_id,
            organization_id,
            metadata,
            created_at,
            updated_at,
            storage_key,
            created_by_user_id AS uploaded_by_user_id,
            extension,
            (visibility = 'public') AS is_public,
            visibility,
            status,
            deleted_at,
            created_by_user_id,
            file_name,
            content_type,
            checksum_sha256
        FROM athena.files
        {suffix}
        "#
    )
}

fn runtime_binding_to_target(binding: S3CatalogRuntimeBinding) -> ManagedStorageTarget {
    ManagedStorageTarget {
        s3_id: binding.catalog.id,
        endpoint: binding.catalog.endpoint.unwrap_or_default(),
        region: binding.catalog.region.unwrap_or_default(),
        bucket: binding.catalog.bucket,
        provider: binding.catalog.provider,
        force_path_style: binding.catalog.force_path_style,
        default_prefix: binding.catalog.default_prefix,
        public_base_url: binding.catalog.public_base_url,
        access_key_id: binding.access_key_id,
        secret_key: binding.secret_key,
        session_token: binding.session_token,
    }
}

fn normalize_storage_key(value: &str) -> Result<String, S3CredentialStoreError> {
    let trimmed = value.trim().trim_matches('/').to_string();
    if trimmed.is_empty() {
        return Err(S3CredentialStoreError::InvalidInput(
            "storage_key must not be empty".to_string(),
        ));
    }
    if trimmed.contains("..") {
        return Err(S3CredentialStoreError::InvalidInput(
            "storage_key must not contain path traversal segments".to_string(),
        ));
    }
    Ok(trimmed)
}

fn normalize_prefix_for_match(prefix: &str) -> Result<String, S3CredentialStoreError> {
    let normalized = normalize_storage_key(prefix)?;
    Ok(format!("{normalized}/"))
}

fn normalize_required(value: &str, field: &str) -> Result<String, S3CredentialStoreError> {
    normalize_optional(Some(value))
        .ok_or_else(|| S3CredentialStoreError::InvalidInput(format!("{field} is required")))
}

fn normalize_optional(value: Option<&str>) -> Option<String> {
    value
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .map(str::to_string)
}

fn normalize_metadata_object(value: Value) -> Value {
    match value {
        Value::Object(map) => Value::Object(map),
        _ => Value::Object(Default::default()),
    }
}

fn normalize_visibility(
    explicit_visibility: Option<&str>,
    is_public: bool,
    organization_id: Option<&str>,
) -> Result<String, S3CredentialStoreError> {
    if let Some(visibility) = normalize_optional(explicit_visibility) {
        match visibility.as_str() {
            "private" | "organization" | "public" => return Ok(visibility),
            _ => {
                return Err(S3CredentialStoreError::InvalidInput(format!(
                    "unsupported visibility '{}'",
                    visibility
                )));
            }
        }
    }
    if is_public {
        return Ok("public".to_string());
    }
    if normalize_optional(organization_id).is_some() {
        return Ok("organization".to_string());
    }
    Ok("private".to_string())
}

fn normalize_status(status: &str) -> Result<String, S3CredentialStoreError> {
    let value = normalize_required(status, "status")?;
    match value.as_str() {
        "pending" | "uploaded" | "failed" | "deleted" => Ok(value),
        _ => Err(S3CredentialStoreError::InvalidInput(format!(
            "unsupported status '{}'",
            value
        ))),
    }
}

fn object_name(key: &str) -> String {
    key.rsplit('/')
        .next()
        .filter(|value| !value.is_empty())
        .unwrap_or(key)
        .to_string()
}

fn object_prefix_path(key: &str) -> Option<String> {
    let trimmed = key.trim_matches('/');
    let (prefix, _) = trimmed.rsplit_once('/')?;
    if prefix.is_empty() {
        return None;
    }
    Some(prefix.to_string())
}

fn file_extension(name: &str) -> Option<String> {
    let (_, ext) = name.rsplit_once('.')?;
    let trimmed = ext.trim();
    if trimmed.is_empty() {
        return None;
    }
    Some(trimmed.to_ascii_lowercase())
}

fn merged_organization_ids(access: &FileAccessContext) -> Vec<String> {
    let mut merged = access.organization_ids.clone();
    if let Some(active) = access.organization_id.as_ref() {
        if !merged.iter().any(|value| value == active) {
            merged.push(active.clone());
        }
    }
    merged
}

#[cfg(test)]
mod tests {
    use super::{
        FilePermissionKind, file_extension, normalize_prefix_for_match, normalize_storage_key,
        normalize_visibility, object_name, object_prefix_path,
    };

    #[test]
    fn file_permission_kind_implies_owner() {
        assert_eq!(
            FilePermissionKind::Read.implied_permissions(),
            &["read", "owner"]
        );
        assert_eq!(
            FilePermissionKind::Delete.implied_permissions(),
            &["delete", "owner"]
        );
    }

    #[test]
    fn normalize_storage_key_trims_slashes() {
        assert_eq!(
            normalize_storage_key("/reports/q1/file.csv/").unwrap(),
            "reports/q1/file.csv"
        );
    }

    #[test]
    fn normalize_prefix_adds_trailing_slash() {
        assert_eq!(
            normalize_prefix_for_match("reports/2026").unwrap(),
            "reports/2026/"
        );
    }

    #[test]
    fn visibility_defaults_to_org_when_org_present() {
        assert_eq!(
            normalize_visibility(None, false, Some("org_1")).unwrap(),
            "organization"
        );
    }

    #[test]
    fn object_name_uses_last_path_segment() {
        assert_eq!(object_name("reports/file.csv"), "file.csv");
        assert_eq!(object_name("file.csv"), "file.csv");
    }

    #[test]
    fn object_prefix_path_uses_parent_prefix() {
        assert_eq!(
            object_prefix_path("reports/2026/file.csv").as_deref(),
            Some("reports/2026")
        );
        assert_eq!(object_prefix_path("file.csv"), None);
    }

    #[test]
    fn file_extension_is_lowercased() {
        assert_eq!(file_extension("Quarterly.PDF").as_deref(), Some("pdf"));
        assert_eq!(file_extension("README").as_deref(), None);
    }
}
