use std::collections::HashSet;

use derive_builder::Builder;
use interim_pgtypes::{
    client::WorkspaceClient, pg_acl::PgPrivilegeType, pg_class::PgClass, pg_role::RoleTree,
    rolnames::ROLE_PREFIX_USER,
};
use sqlx::postgres::types::Oid;
use tracing::{Instrument, info, info_span};
use uuid::Uuid;

use crate::{client::AppDbClient, portal::Portal};

use super::{AccessError, Accessor, Actor};

/// Utility for fetching a [`Portal`], with authorization.
#[derive(Builder, Debug)]
#[builder(
    // Build fn should only be called internally, via `fetch_optional()`.
    build_fn(private, error = "super::AccessError"),
    // Callers interact primarily with the generated builder struct.
    name = "PortalAccessor",
    vis = "pub",
    // "Owned" pattern circumvents the `Clone` trait bound on fields.
    pattern = "owned",
)]
struct GeneratedPortalAccessor<'a> {
    /// Required. ID of the portal to be accessed.
    id: Uuid,

    /// Required. Identity against which to evaluate authorization checks.
    as_actor: Actor,

    /// Required. Client for the backing database.
    using_workspace_client: &'a mut WorkspaceClient,

    /// Required. Client for the application database.
    using_app_db: &'a mut AppDbClient,

    /// Optionally verify that the portal is associated with a specific
    /// workspace ID.
    #[builder(default, setter(strip_option))]
    verify_workspace_id: Option<Uuid>,

    /// Optionally verify that the portal is associated with a specific relation
    /// OID.
    #[builder(default, setter(strip_option))]
    verify_rel_oid: Option<Oid>,

    /// Optionally verify that the actor has or inherits specific Postgres
    /// permissions on the relation in the backing database.
    #[builder(default, setter(into))]
    // Using [`HashSet<PgPrivilegeType>`] with `#[builder(setter(into))]` is
    // more straightforward for callers than asking them to provide a
    // `&'a [PgPrivilege]`. Forcing a conversion to a heap-based data structure
    // feels wrong when callers will often be starting with a statically
    // allocated array of `PgPrivilegeType`s, but that optimization can be made
    // later if the trade-off with ergonomics proves warranted.
    verify_rel_permissions: HashSet<PgPrivilegeType>,

    // Find doc comment on setter method.
    #[builder(default, setter(custom))]
    verify_rel_ownership: bool,

    /// Optionally for performance, relation metadata may be provided by the
    /// caller so that the accessor doesn't need to refetch it behind the
    /// scenes.
    #[builder(default, setter(strip_option))]
    using_rel: Option<&'a PgClass>,
}

impl PortalAccessor<'_> {
    /// Optionally verify that the actor has or belongs to the role that owns
    /// the relation in the backing database.
    pub fn verify_rel_ownership(self) -> Self {
        Self {
            verify_rel_ownership: Some(true),
            ..self
        }
    }
}

impl<'a> Accessor<Portal> for PortalAccessor<'a> {
    async fn fetch_one(self) -> Result<Portal, AccessError> {
        let spec = self.build()?;
        async {
            // FIXME: Do we need to explicitly verify that the actor has
            // database `CONNECT` privileges, or is that implicitly given by the
            // fact that a workspace client has already been acquired?

            let portal = Portal::with_id(spec.id)
                .fetch_optional(spec.using_app_db)
                .await?
                .ok_or(AccessError::NotFound)?;

            spec.verify_workspace_id
                .is_none_or(|value| portal.workspace_id == value)
                .ok_or_else(|| {
                    info!("workspace_id check failed for portal");
                    AccessError::NotFound
                })?;
            spec.verify_rel_oid
                .is_none_or(|value| portal.class_oid == value)
                .ok_or_else(|| {
                    info!("rel_oid check failed for portal");
                    AccessError::NotFound
                })?;

            let rel = if let Some(value) = spec.using_rel {
                value
            } else {
                &PgClass::with_oid(portal.class_oid)
                    .fetch_optional(spec.using_workspace_client)
                    .await?
                    .ok_or_else(|| {
                        info!("unable to fetch PgClass for portal");
                        AccessError::NotFound
                    })?
            };

            let actor_rolname = match spec.as_actor {
                Actor::Bypass => None,
                Actor::User(user_id) => Some(format!(
                    "{ROLE_PREFIX_USER}{user_id}",
                    user_id = user_id.simple()
                )),
            };

            if let Some(actor_rolname) = actor_rolname {
                // Verify ACL permissions.
                //
                // No need to explicitly check whether
                // `verify_rel_permissions` is empty: database queries are
                // only performed based on roles which intersect with the
                // permissions check.
                //
                // This could alternatively be implemented using streams,
                // but this naive for-loop version is much more readable and
                // still bottlenecks on the same serial database access.
                let mut actor_permissions: HashSet<PgPrivilegeType> = HashSet::new();
                for acl_item in rel
                    .relacl
                    .as_ref()
                    .unwrap_or(&vec![])
                    .iter()
                    .filter(|acl_item| {
                        acl_item
                            .privileges
                            .iter()
                            .any(|value| spec.verify_rel_permissions.contains(&value.privilege))
                    })
                {
                    // The naive equality check is technically redundant,
                    // but it allows us to eliminate a database round trip
                    // if it evaluates to true.
                    if acl_item.grantee == actor_rolname
                        || RoleTree::members_of_rolname(&acl_item.grantee)
                            .fetch_tree(spec.using_workspace_client)
                            .await?
                            .map(|tree| tree.flatten_inherited())
                            .unwrap_or_default()
                            .iter()
                            .any(|value| value.rolname == actor_rolname)
                    {
                        for permission in acl_item.privileges.iter() {
                            actor_permissions.insert(permission.privilege);
                        }
                    }
                }
                if !actor_permissions.is_superset(&spec.verify_rel_permissions) {
                    info!("actor lacks postgres privileges");
                    return Err(AccessError::NotFound);
                }

                // Verify relation ownership.
                if spec.verify_rel_ownership
                    // The naive equality check is technically redundant,
                    // but it allows us to eliminate a database round trip
                    // if it evaluates to true.
                    && (rel.regowner == actor_rolname
                        || !RoleTree::members_of_oid(rel.relowner)
                            .fetch_tree(spec.using_workspace_client)
                            .await?
                            .map(|tree| tree.flatten_inherited())
                            .unwrap_or_default()
                            .iter()
                            .any(|value| value.rolname == actor_rolname))
                {
                    info!("actor is not relation owner");
                    return Err(AccessError::NotFound);
                }
            }

            Ok(portal)
        }
        .instrument(info_span!(
            "PortalAccessor::fetch_optional()",
            portal_id = spec.id.to_string(),
        ))
        .await
    }
}