/*
 * Author: Ruben Fiszel
 * Copyright: Windmill Labs, Inc 2022
 * This file and its contents are licensed under the AGPLv3 License.
 * Please see the included NOTICE for copyright information and
 * LICENSE-AGPL for a copy of the license.
 */

//! Ownership and grants on the objects of a data table on a cluster Windmill manages.
//!
//! [`datatable_permissions`](crate::datatable_permissions) decides who may connect as which role;
//! this decides what each role may then touch. Every change is a real `GRANT`, `REVOKE`,
//! `ALTER ... OWNER TO` or `ALTER DEFAULT PRIVILEGES`, so Postgres is what enforces it.
//!
//! Reading is open to anyone who reaches the data table. Planning and applying are for those who
//! administer it — admins of the workspace that governs it, and superadmins. All of it is
//! Enterprise Edition ([`crate::datatable_acl_oss`]).

use std::collections::BTreeMap;

use axum::{
    extract::{Extension, Path, Query},
    routing::{get, post},
    Json, Router,
};
use serde::{Deserialize, Serialize};
use tokio::sync::mpsc;
use tokio_postgres::error::{DbError, SqlState};
use tokio_postgres::AsyncMessage;

use windmill_api_auth::ApiAuthed;
use windmill_audit::audit_oss::audit_log;
use windmill_audit::ActionKind;
use windmill_common::datatable_roles::{
    lock_role_catalog, quote_ident, read_role_catalog, read_role_catalog_tx, DatatableRoleCatalog,
    DatatableRoleCluster, ADMIN_DATATABLE_ROLE, CUSTOM_INSTANCE_USER,
};
use windmill_common::error::{pg_error_message, Error, JsonResult, Result};
use windmill_common::workspaces::{resolve_governing_datatable, DataTable, GoverningDatatable};
use windmill_common::{PgDatabase, DB};

use crate::datatable_permissions::{ensure_governs_datatable, ensure_reaches_governing_datatable};

pub(crate) fn routes() -> Router {
    Router::new()
        .route("/datatable_acl/{datatable_name}", get(get_datatable_acl))
        .route(
            "/datatable_acl/{datatable_name}/plan",
            post(plan_datatable_acl),
        )
        .route(
            "/datatable_acl/{datatable_name}/apply",
            post(apply_datatable_acl),
        )
}

/// What a read or a change is about.
#[derive(Deserialize, Serialize, Debug, Clone, PartialEq)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum AclTarget {
    /// The data table's own database — where the privilege to create schemas lives.
    Database,
    Schema {
        schema: String,
    },
    Table {
        schema: String,
        table: String,
    },
}

impl AclTarget {
    /// The schema the target is in, absent for the database itself.
    pub(crate) fn schema(&self) -> Option<&str> {
        match self {
            AclTarget::Database => None,
            AclTarget::Schema { schema } => Some(schema),
            AclTarget::Table { schema, .. } => Some(schema),
        }
    }

    /// What it is called in a message.
    pub(crate) fn label(&self, dbname: &str) -> String {
        match self {
            AclTarget::Database => dbname.to_string(),
            AclTarget::Schema { schema } => schema.clone(),
            AclTarget::Table { schema, table } => format!("{schema}.{table}"),
        }
    }
}

#[derive(Deserialize, Debug)]
pub struct AclTargetQuery {
    kind: String,
    schema: Option<String>,
    table: Option<String>,
}

impl TryFrom<AclTargetQuery> for AclTarget {
    type Error = Error;
    fn try_from(q: AclTargetQuery) -> Result<Self> {
        match (q.kind.as_str(), q.schema, q.table) {
            ("database", _, _) => Ok(AclTarget::Database),
            ("schema", Some(schema), _) => Ok(AclTarget::Schema { schema }),
            ("table", Some(schema), Some(table)) => Ok(AclTarget::Table { schema, table }),
            ("schema" | "table", None, _) => {
                Err(Error::BadRequest("This target needs a schema".to_string()))
            }
            ("table", _, None) => Err(Error::BadRequest(
                "A table target needs a table".to_string(),
            )),
            (kind, _, _) => Err(Error::BadRequest(format!("Unknown ACL target '{kind}'"))),
        }
    }
}

/// Where a set of privileges applies, relative to the target.
///
/// `Future*` covers what does not exist yet: those become `ALTER DEFAULT PRIVILEGES`, which only
/// binds objects created by the roles it names.
#[derive(Deserialize, Serialize, Debug, Clone, Copy, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum GrantScope {
    /// The target itself — the database, the schema, or the table.
    Target,
    AllTables,
    AllSequences,
    AllFunctions,
    FutureTables,
    FutureSequences,
    FutureFunctions,
}

impl GrantScope {
    pub(crate) fn is_future(&self) -> bool {
        matches!(
            self,
            GrantScope::FutureTables | GrantScope::FutureSequences | GrantScope::FutureFunctions
        )
    }
}

/// A change to plan. One at a time: each is confirmed against its own SQL.
#[derive(Deserialize, Serialize, Debug, Clone)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum AclChange {
    /// Hand the target — and, for a schema, everything already in it but an extension's members,
    /// which stay with the extension — to another role.
    SetOwner {
        role: String,
    },
    Grant {
        role: String,
        privileges: Vec<String>,
        scope: GrantScope,
    },
    Revoke {
        role: String,
        privileges: Vec<String>,
        scope: GrantScope,
        /// Objects inside the target, empty for the target itself. `ON ALL TABLES` grants read
        /// back per object, so they are revoked per object — and the same privileges on several
        /// of them are revoked together.
        #[serde(default)]
        objects: Vec<AclObject>,
    },
}

impl AclChange {
    /// The role the change is about, as the editor names it.
    fn role(&self) -> &str {
        match self {
            AclChange::SetOwner { role }
            | AclChange::Grant { role, .. }
            | AclChange::Revoke { role, .. } => role,
        }
    }
}

#[derive(Deserialize, Debug)]
pub struct AclChangeRequest {
    pub target: AclTarget,
    pub change: AclChange,
    /// The statements the plan showed. An apply runs only those: it plans again and refuses if the
    /// result differs.
    #[serde(default)]
    pub statements: Option<Vec<String>>,
}

/// An object inside a schema, named the way `REVOKE ... ON <keyword>` needs it.
#[derive(Deserialize, Serialize, Debug, Clone, PartialEq)]
pub struct AclObject {
    pub name: String,
    /// `TABLE`, `SEQUENCE`, `FUNCTION`, `PROCEDURE` or `TYPE`: what the object is.
    /// [`object_keyword`] turns it into the keyword a revoke takes; a type has none.
    pub kind: String,
    /// A routine is identified by its argument types, not by its name: two `f` in one schema are
    /// two objects. Absent for everything else.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub args: Option<String>,
}

/// A grant as the database has it, under the role names the editor uses.
#[derive(Serialize, Debug, PartialEq)]
pub struct AclGrant {
    /// A data table role's name, `admin` for `custom_instance_user`, else the raw Postgres role
    /// (`PUBLIC` included).
    pub grantee: String,
    pub privileges: Vec<String>,
    /// `None` for the target itself, else the object inside it.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub object: Option<AclObject>,
    /// `TABLES` / `SEQUENCES` / `FUNCTIONS` / `TYPES` (or, on the database, `SCHEMAS`) when this
    /// is a default privilege, which applies to objects that do not exist yet.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub future: Option<String>,
    /// Where the grant comes from, each role once: who granted it, or for a default privilege the
    /// role whose future objects it covers. A revoke of some of the privileges takes them back from
    /// every source that gave them.
    pub sources: Vec<AclSource>,
}

/// One role a grant comes from.
#[derive(Serialize, Debug, PartialEq)]
pub struct AclSource {
    /// Under the same names as [`AclGrant::grantee`].
    pub role: String,
    /// What `role` gave of the grant's privileges. A revoke is held back only by a source out of
    /// reach that gave some of what it takes back.
    pub privileges: Vec<String>,
    /// Whether this data table's connection can take back what `role` gave: on an object only the
    /// owner's grants when it acts for the owner, or else its own (`grant_source!`); for a default
    /// privilege, a creating role it acts for. What a source out of reach gave is not revocable
    /// from here; privileges only other sources gave still are.
    pub reachable: bool,
}

#[derive(Serialize, Debug)]
pub struct DatatableAclInfo {
    /// Under the same names as [`AclGrant::grantee`].
    pub owner: String,
    /// The roles a change may name: `admin`, then every role of the instance catalog. Only for a
    /// caller who may change anything, as the catalog is in the permissions drawer.
    pub roles: Vec<String>,
    /// Whether this caller may plan and apply changes: they administer the data table, on an
    /// edition that has the planner.
    pub editable: bool,
    /// Whether the server is Postgres 17 or later, which added the `MAINTAIN` table privilege.
    pub supports_maintain: bool,
    /// The database the target lives in, which no target carries itself.
    pub dbname: String,
    pub grants: Vec<AclGrant>,
    /// What the target holds that is a target of its own: a database's schemas, a schema's tables.
    pub children: Vec<String>,
}

#[derive(Serialize, Debug)]
pub struct AclPlan {
    pub statements: Vec<String>,
    pub warnings: Vec<String>,
}

/// The Postgres role a role name stands for. A data table role is a login named exactly like the
/// role, so this is the identity — except `admin`, which is `custom_instance_user`.
///
/// Anything else is refused, never resolved to some default: every statement a plan writes names
/// the role it is about.
pub(crate) fn pg_role_of(name: &str, catalog: &DatatableRoleCatalog) -> Result<String> {
    if name == ADMIN_DATATABLE_ROLE {
        return Ok(CUSTOM_INSTANCE_USER.to_string());
    }
    if catalog.values().any(|r| r.name == name) {
        return Ok(name.to_string());
    }
    Err(Error::BadRequest(format!(
        "'{name}' is not a data table role of this instance"
    )))
}

/// The reverse of [`pg_role_of`], for display. A role that is not a data table role reads back as
/// itself.
pub(crate) fn role_name_of(pg_role: &str) -> String {
    if pg_role == CUSTOM_INSTANCE_USER {
        ADMIN_DATATABLE_ROLE.to_string()
    } else {
        pg_role.to_string()
    }
}

/// Every role a change may name: `admin` first, then the catalog.
fn role_names(catalog: &DatatableRoleCatalog) -> Vec<String> {
    let mut names: Vec<String> = catalog.values().map(|r| r.name.clone()).collect();
    names.sort();
    names.insert(0, ADMIN_DATATABLE_ROLE.to_string());
    names
}

/// The cluster whose roles the data table's grants name.
fn ensure_managed(governing: &GoverningDatatable) -> Result<DatatableRoleCluster> {
    governing.role_cluster().ok_or_else(|| {
        Error::BadRequest(format!(
            "Data table '{}' is backed by a Postgres resource, so its access is managed on that \
             server directly. Only a data table on a database Windmill manages has data table \
             roles to grant to.",
            governing.name
        ))
    })
}

/// The data table's `admin` connection, and the notices Postgres sends on it.
///
/// Authorization: connects as `custom_instance_user` with the cluster's stored credentials and
/// checks nothing. Callers MUST have authorized the request first — a request about to be refused must
/// not get as far as this connection.
async fn connect_as_admin_unchecked(
    db: &DB,
    governing: &GoverningDatatable,
) -> Result<(
    tokio_postgres::Client,
    mpsc::UnboundedReceiver<DbError>,
    String,
)> {
    let cluster = ensure_managed(governing)?;
    // Built from the authorized entry, never by resolving the settings again: a save in between
    // could point the entry at a resource on another server and back, and this connection would
    // then alter a database the later checks of the entry never see.
    let dbname = governing
        .datatable
        .database
        .as_ref()
        .expect("a governing entry owns a database")
        .resource_path
        .clone();
    let pg = match cluster {
        DatatableRoleCluster::Instance => {
            let mut pg =
                PgDatabase::parse_uri(&windmill_common::get_database_url().await?.as_str().await)?;
            pg.dbname = dbname.clone();
            pg.user = Some(CUSTOM_INSTANCE_USER.to_string());
            pg.password = Some(windmill_common::utils::get_custom_pg_instance_password(db).await?);
            pg
        }
        DatatableRoleCluster::ExternalInstance => {
            windmill_common::external_instance_pg::external_instance_connection_unchecked(
                db, &dbname, false,
            )
            .await?
        }
    };
    let (client, mut connection) = pg.connect(Some(db)).await?;
    // Unbounded: the driver must never wait on the receiver, which only drains once the statement
    // the driver is carrying has completed.
    let (notices_tx, notices) = mpsc::unbounded_channel();
    tokio::spawn(async move {
        loop {
            match std::future::poll_fn(|cx| connection.poll_message(cx)).await {
                Some(Ok(AsyncMessage::Notice(notice))) => {
                    let _ = notices_tx.send(notice);
                }
                Some(Ok(_)) => {}
                Some(Err(e)) => {
                    tracing::error!("Datatable ACL connection error: {e}");
                    break;
                }
                None => break,
            }
        }
    });
    Ok((client, notices, dbname))
}

/// An object whose ownership follows the schema's.
#[derive(Debug, PartialEq)]
pub(crate) struct OwnedObject {
    /// The keyword `ALTER ... OWNER TO` takes for this kind of object.
    pub(crate) keyword: &'static str,
    /// How Postgres names the object (`pg_identify_object`): schema-qualified, quoted where
    /// needed, with a routine's arguments or an operator class's access method. It goes into the
    /// statement as it is.
    pub(crate) identity: String,
}

/// Everything a schema's change of owner takes along, in schema `$1`, as (kind, identity, owner
/// oid), the first two as `pg_identify_object` gives them. The plan, the check of what this
/// connection may move and the check after the move all read this one list, so what is moved and
/// what is checked cannot differ.
///
/// Read from what depends on the schema rather than catalog by catalog, so no kind of object is
/// left out by omission; array types, row types and indexes depend on another object instead.
/// Left out: an object's internal parts (a range type's constructors and multirange, an identity
/// column's sequence), a sequence tied to a column (it follows its table, and refuses an owner of
/// its own), an extension's members, and what has no `ALTER ... OWNER` at all (an extension, a text
/// search parser or template). Postgres records no owner for the bootstrap superuser, oid 10.
macro_rules! schema_owned_objects {
    () => {
        "SELECT o.type AS kind, o.identity, COALESCE(s.refobjid, 10::oid) AS owner
         FROM pg_depend d
         CROSS JOIN LATERAL pg_identify_object(d.classid, d.objid, 0) o
         LEFT JOIN pg_shdepend s
             ON s.dbid = (SELECT oid FROM pg_database WHERE datname = current_database())
            AND s.classid = d.classid AND s.objid = d.objid AND s.deptype = 'o'
         WHERE d.refclassid = 'pg_namespace'::regclass AND d.deptype = 'n'
           AND d.refobjid = (SELECT oid FROM pg_namespace WHERE nspname = $1)
           AND d.classid <> ALL(ARRAY['pg_extension'::regclass, 'pg_ts_parser'::regclass,
                                      'pg_ts_template'::regclass]::oid[])
           AND NOT EXISTS (
               SELECT 1 FROM pg_depend x
               WHERE x.classid = d.classid AND x.objid = d.objid AND x.objsubid = 0
                 AND (x.deptype IN ('i', 'e')
                      OR (x.deptype = 'a' AND d.classid = 'pg_class'::regclass
                          AND x.refobjsubid <> 0)))"
    };
}

/// The keyword `ALTER ... OWNER TO` takes for a kind of object, as `pg_identify_object` names the
/// kind. A kind missing here is refused rather than skipped, which would leave it behind.
fn owned_keyword(kind: &str) -> Option<&'static str> {
    Some(match kind {
        "table" => "TABLE",
        "view" => "VIEW",
        "materialized view" => "MATERIALIZED VIEW",
        "sequence" => "SEQUENCE",
        "foreign table" => "FOREIGN TABLE",
        "type" => "TYPE",
        "function" | "procedure" | "aggregate" => "ROUTINE",
        "collation" => "COLLATION",
        "conversion" => "CONVERSION",
        "operator" => "OPERATOR",
        "operator class" => "OPERATOR CLASS",
        "operator family" => "OPERATOR FAMILY",
        "statistics object" => "STATISTICS",
        "text search dictionary" => "TEXT SEARCH DICTIONARY",
        "text search configuration" => "TEXT SEARCH CONFIGURATION",
        _ => return None,
    })
}

async fn read_owned_objects(
    client: &tokio_postgres::Client,
    schema: &str,
) -> Result<Vec<OwnedObject>> {
    let rows = client
        .query(
            concat!(
                "SELECT kind, identity FROM (",
                schema_owned_objects!(),
                ") o ORDER BY kind, identity"
            ),
            &[&schema],
        )
        .await
        .map_err(|e| {
            Error::internal_err(format!(
                "Failed to list the objects of schema '{schema}': {}",
                pg_error_message(&e)
            ))
        })?;
    rows.into_iter()
        .map(|row| {
            let kind: &str = row.get(0);
            let identity: String = row.get(1);
            match owned_keyword(kind) {
                Some(keyword) => Ok(OwnedObject { keyword, identity }),
                None => Err(Error::BadRequest(format!(
                    "{identity} is a {kind}, whose owner cannot be changed from here, and schema \
                     {schema} would change hands without it. Move it to another schema first."
                ))),
            }
        })
        .collect()
}

/// Default privileges set database-wide (no `IN SCHEMA`), as `(defaclrole, defaclobjtype, grantee,
/// privilege_type)`. Postgres stores such an entry as a whole acl, the creator's own privileges
/// included, so only what goes beyond the built-in default is a grant. It applies in every schema
/// on top of the schema's own defaults, which cannot take it back.
macro_rules! database_wide_defaults {
    () => {
        "SELECT d.defaclrole, d.defaclobjtype, a.grantee, a.privilege_type
         FROM pg_default_acl d, aclexplode(d.defaclacl) a
         WHERE d.defaclnamespace = 0
           AND NOT EXISTS (
               SELECT 1 FROM aclexplode(acldefault(
                   CASE d.defaclobjtype WHEN 'S' THEN 's' ELSE d.defaclobjtype END, d.defaclrole)) x
               WHERE x.grantee = a.grantee AND x.privilege_type = a.privilege_type)"
    };
}

/// `TABLES`, `SEQUENCES`, `FUNCTIONS` or `TYPES`, as a `defaclobjtype` names them.
fn default_objects_keyword(objtype: &str) -> &'static str {
    match objtype {
        "r" => "TABLES",
        "S" => "SEQUENCES",
        "T" => "TYPES",
        _ => "FUNCTIONS",
    }
}

/// What a schema's owner holds on what gets created there later. A change of owner hands the new
/// owner the same and takes these back: otherwise every former owner keeps reaching whatever the
/// other roles create there.
#[derive(Debug, PartialEq)]
pub(crate) struct FormerOwnerDefaults {
    pub(crate) pg_role: String,
    /// (creating role, `TABLES` / `SEQUENCES` / `FUNCTIONS` / `TYPES`), one per default privilege
    /// it holds.
    pub(crate) defaults: Vec<(String, &'static str)>,
}

/// The default privileges schema `schema`'s owner holds there — `None` when it holds none, or is
/// `new_owner` already. Refused when one was set by a role this connection cannot act for: only
/// a member of the creating role may change its defaults, so the revoke would fail at apply. Also
/// refused when the owner holds defaults set database-wide, which no change to this schema takes
/// back.
async fn read_former_owner_defaults(
    client: &tokio_postgres::Client,
    schema: &str,
    new_owner: &str,
) -> Result<Option<FormerOwnerDefaults>> {
    let database_wide = client
        .query_opt(
            concat!(
                "SELECT pg_get_userbyid(n.nspowner), pg_get_userbyid(g.defaclrole),
                        g.defaclobjtype::text
                 FROM pg_namespace n, (",
                database_wide_defaults!(),
                ") g
                 WHERE n.nspname = $1 AND g.grantee = n.nspowner
                   AND g.defaclobjtype IN ('r', 'S', 'f', 'T')
                   AND n.nspowner <> (SELECT oid FROM pg_roles WHERE rolname = $2)
                 ORDER BY 2, 3
                 LIMIT 1"
            ),
            &[&schema, &new_owner],
        )
        .await
        .map_err(|e| {
            Error::internal_err(format!(
                "Failed to read the database-wide default privileges: {}",
                pg_error_message(&e)
            ))
        })?;
    if let Some(row) = database_wide {
        let pg_role: String = row.get(0);
        let creator: String = row.get(1);
        return Err(Error::BadRequest(format!(
            "{} holds default privileges set database-wide by {}, which no default of schema \
             {schema} takes back, so it would keep reaching what is created here after the change \
             of owner. Revoke them database-wide first: ALTER DEFAULT PRIVILEGES FOR ROLE {} \
             REVOKE ALL PRIVILEGES ON {} FROM {}",
            role_name_of(&pg_role),
            role_name_of(&creator),
            quote_ident(&creator),
            default_objects_keyword(row.get(2)),
            quote_ident(&pg_role)
        )));
    }

    let rows = client
        .query(
            "SELECT DISTINCT pg_get_userbyid(n.nspowner), pg_get_userbyid(d.defaclrole),
                    d.defaclobjtype::text, pg_has_role(d.defaclrole, 'USAGE')
             FROM pg_namespace n
             JOIN pg_default_acl d ON d.defaclnamespace = n.oid
             CROSS JOIN LATERAL aclexplode(d.defaclacl) a
             WHERE n.nspname = $1 AND a.grantee = n.nspowner
               -- What the owner gives itself is about the objects it creates, not about owning
               -- the schema, so a move leaves it alone.
               AND a.grantee <> d.defaclrole
               AND d.defaclobjtype IN ('r', 'S', 'f', 'T')
               AND n.nspowner <> (SELECT oid FROM pg_roles WHERE rolname = $2)
             ORDER BY 2, 3",
            &[&schema, &new_owner],
        )
        .await
        .map_err(|e| {
            Error::internal_err(format!(
                "Failed to read the default privileges of schema '{schema}': {}",
                pg_error_message(&e)
            ))
        })?;
    let Some(first) = rows.first() else {
        return Ok(None);
    };
    let pg_role: String = first.get(0);
    let plural = |row: &tokio_postgres::Row| default_objects_keyword(row.get(2));
    if let Some(row) = rows.iter().find(|row| !row.get::<_, bool>(3)) {
        let creator: String = row.get(1);
        return Err(Error::BadRequest(format!(
            "{} holds default privileges in schema {schema} from {creator}, which this data \
             table's connection cannot act for, so they would outlive the change of owner. \
             Revoke them as {creator} first: ALTER DEFAULT PRIVILEGES FOR ROLE {} IN SCHEMA {} \
             REVOKE ALL PRIVILEGES ON {} FROM {}",
            role_name_of(&pg_role),
            quote_ident(&creator),
            quote_ident(schema),
            plural(row),
            quote_ident(&pg_role)
        )));
    }
    Ok(Some(FormerOwnerDefaults {
        defaults: rows
            .iter()
            .map(|row| (row.get::<_, String>(1), plural(row)))
            .collect(),
        pg_role,
    }))
}

/// What the catalog holds that a plan depends on, read before planning so the planner stays pure.
#[derive(Debug, Default, PartialEq)]
pub(crate) struct CatalogFacts {
    /// Every role that may create objects here, but the one the change is about.
    pub(crate) other_pg_roles: Vec<String>,
    /// For a schema's change of owner: what moves along with it.
    pub(crate) existing_objects: Vec<OwnedObject>,
    /// For a schema's change of owner: the defaults it takes back from the owner it replaces.
    pub(crate) former_owner: Option<FormerOwnerDefaults>,
    /// For a revoke: each grant it takes back.
    pub(crate) revoked_grants: Vec<RevokedGrant>,
}

/// A grant a revoke takes back, as the catalog records it: what `source` gave on `object` (the
/// target itself when `None`), or for a default privilege on what `source` creates later.
#[derive(Debug, PartialEq)]
pub(crate) struct RevokedGrant {
    pub(crate) object: Option<AclObject>,
    /// The role that made the grant: its grantor, or the creating role of a default privilege.
    pub(crate) source: String,
    pub(crate) privileges: Vec<String>,
}

/// An `aclexplode` row's source and whether this connection can take back what it gave, on an
/// object owned by `$owner`. A REVOKE speaks for the grantor Postgres picks itself — the owner,
/// when the connection acts for the owner, and otherwise the connection — and `GRANTED BY` names
/// nobody else, so a grant any other role made stays whatever the connection runs.
macro_rules! grant_source {
    ($owner:literal) => {
        concat!(
            "pg_get_userbyid(a.grantor), a.grantor = CASE WHEN pg_has_role(",
            $owner,
            ", 'USAGE') THEN ",
            $owner,
            " ELSE (SELECT oid FROM pg_roles WHERE rolname = current_user) END"
        )
    };
}

/// What relation `$2` of schema `$1` grants role `$3`, as (source, whether this connection can
/// take it back, privilege).
const RELATION_GRANTS: &str = concat!(
    "SELECT ",
    grant_source!("c.relowner"),
    ", a.privilege_type
     FROM pg_class c
     JOIN pg_namespace n ON n.oid = c.relnamespace,
          aclexplode(COALESCE(c.relacl, acldefault(
              CASE c.relkind WHEN 'S' THEN 's' ELSE 'r' END::\"char\", c.relowner))) a
     WHERE n.nspname = $1 AND c.relname = $2
       AND c.relkind = ANY(ARRAY['r','p','v','m','S','f']::\"char\"[])
       AND a.grantee = (SELECT oid FROM pg_roles WHERE rolname = $3)"
);

/// The grants a revoke takes back, read from the catalog rather than from the request: one per
/// object and source that gave `pg_role` any of `privileges`. Refused when a source's grant is out
/// of this connection's reach (`grant_source!`, or for a default privilege a creating role it does
/// not act for): the revoke would leave that grant in place.
async fn read_revoked_grants(
    client: &tokio_postgres::Client,
    dbname: &str,
    target: &AclTarget,
    scope: GrantScope,
    objects: &[AclObject],
    privileges: &[String],
    pg_role: &str,
) -> Result<Vec<RevokedGrant>> {
    let read_error = |e: tokio_postgres::Error| {
        Error::internal_err(format!(
            "Failed to read what the revoke takes back: {}",
            pg_error_message(&e)
        ))
    };
    let wanted: Vec<String> = privileges.iter().map(|p| p.to_uppercase()).collect();
    // (object, how it reads in a refusal, one row per source and privilege)
    let mut read = Vec::new();
    match (scope, target) {
        (scope, AclTarget::Schema { schema }) if scope.is_future() => {
            let (objtype, plural) = match scope {
                GrantScope::FutureTables => ("r", "tables"),
                GrantScope::FutureSequences => ("S", "sequences"),
                _ => ("f", "functions"),
            };
            let database_wide = client
                .query(
                    concat!(
                        "SELECT pg_get_userbyid(g.defaclrole), g.privilege_type FROM (",
                        database_wide_defaults!(),
                        ") g
                         WHERE g.defaclobjtype::text = $1
                           AND g.grantee = (SELECT oid FROM pg_roles WHERE rolname = $2)
                         ORDER BY 1, 2"
                    ),
                    &[&objtype, &pg_role],
                )
                .await
                .map_err(read_error)?;
            let mut still_granted: BTreeMap<String, Vec<String>> = BTreeMap::new();
            for row in database_wide {
                let privilege: String = row.get(1);
                if wanted.contains(&privilege) {
                    still_granted.entry(row.get(0)).or_default().push(privilege);
                }
            }
            if let Some((creator, taken)) = still_granted.into_iter().next() {
                return Err(Error::BadRequest(format!(
                    "{} also receives {} on {plural} {} creates through a default privilege set \
                     database-wide, which no default of schema {schema} takes back. Revoke it \
                     database-wide first: ALTER DEFAULT PRIVILEGES FOR ROLE {} REVOKE {} ON {} \
                     FROM {}",
                    role_name_of(pg_role),
                    taken.join(", "),
                    role_name_of(&creator),
                    quote_ident(&creator),
                    taken.join(", "),
                    default_objects_keyword(objtype),
                    quote_ident(pg_role)
                )));
            }
            let rows = client
                .query(
                    "SELECT pg_get_userbyid(d.defaclrole), pg_has_role(d.defaclrole, 'USAGE'),
                            a.privilege_type
                     FROM pg_default_acl d
                     JOIN pg_namespace n ON n.oid = d.defaclnamespace,
                          aclexplode(d.defaclacl) a
                     WHERE n.nspname = $1 AND d.defaclobjtype::text = $2
                       AND a.grantee = (SELECT oid FROM pg_roles WHERE rolname = $3)",
                    &[schema, &objtype, &pg_role],
                )
                .await
                .map_err(read_error)?;
            read.push((
                None,
                format!("{plural} created later in schema {schema}"),
                rows,
            ));
        }
        (GrantScope::Target, _) if objects.is_empty() => {
            let rows = match target {
                AclTarget::Database => {
                    client
                        .query(
                            concat!(
                                "SELECT ",
                                grant_source!("d.datdba"),
                                ", a.privilege_type
                                 FROM pg_database d,
                                      aclexplode(COALESCE(d.datacl, acldefault('d', d.datdba))) a
                                 WHERE d.datname = current_database()
                                   AND a.grantee = (SELECT oid FROM pg_roles WHERE rolname = $1)"
                            ),
                            &[&pg_role],
                        )
                        .await
                }
                AclTarget::Schema { schema } => {
                    client
                        .query(
                            concat!(
                                "SELECT ",
                                grant_source!("n.nspowner"),
                                ", a.privilege_type
                                 FROM pg_namespace n,
                                      aclexplode(COALESCE(n.nspacl, acldefault('n', n.nspowner))) a
                                 WHERE n.nspname = $1
                                   AND a.grantee = (SELECT oid FROM pg_roles WHERE rolname = $2)"
                            ),
                            &[schema, &pg_role],
                        )
                        .await
                }
                AclTarget::Table { schema, table } => {
                    client
                        .query(RELATION_GRANTS, &[schema, table, &pg_role])
                        .await
                }
            }
            .map_err(read_error)?;
            read.push((None, target.label(dbname), rows));
        }
        (GrantScope::Target, AclTarget::Schema { schema }) => {
            for object in objects {
                let rows = match object_keyword(&object.kind)? {
                    "ROUTINE" => {
                        client
                            .query(
                                concat!(
                                    "SELECT ",
                                    grant_source!("p.proowner"),
                                    ", a.privilege_type
                                 FROM pg_proc p
                                 JOIN pg_namespace n ON n.oid = p.pronamespace,
                                      aclexplode(COALESCE(p.proacl, acldefault('f', p.proowner))) a
                                 WHERE n.nspname = $1 AND p.proname = $2
                                   AND pg_get_function_identity_arguments(p.oid) = $3
                                   AND a.grantee = (SELECT oid FROM pg_roles WHERE rolname = $4)"
                                ),
                                &[
                                    schema,
                                    &object.name,
                                    &object.args.as_deref().unwrap_or(""),
                                    &pg_role,
                                ],
                            )
                            .await
                    }
                    _ => {
                        client
                            .query(RELATION_GRANTS, &[schema, &object.name, &pg_role])
                            .await
                    }
                }
                .map_err(read_error)?;
                read.push((
                    Some(object.clone()),
                    format!("{} {schema}.{}", object.kind.to_lowercase(), object.name),
                    rows,
                ));
            }
        }
        // Every other scope and target is the planner's to refuse.
        _ => {}
    }

    let mut revoked = Vec::new();
    for (object, label, rows) in read {
        let mut by_source: BTreeMap<String, (bool, Vec<String>)> = BTreeMap::new();
        for row in rows {
            let privilege: String = row.get(2);
            if wanted.contains(&privilege) {
                by_source
                    .entry(row.get(0))
                    .or_insert_with(|| (row.get(1), vec![]))
                    .1
                    .push(privilege);
            }
        }
        for (source, (reachable, mut privileges)) in by_source {
            if !reachable {
                return Err(Error::BadRequest(format!(
                    "{} on {label} was granted to {} by {source}, and Postgres takes a grant \
                     back only through the role that made it, which this data table's \
                     connection cannot speak for here. Revoke it as {source}.",
                    privileges.join(", "),
                    role_name_of(pg_role),
                )));
            }
            privileges.sort();
            privileges.dedup();
            revoked.push(RevokedGrant { object: object.clone(), source, privileges });
        }
    }
    Ok(revoked)
}

/// The keyword a `REVOKE ... ON` takes for one object, checked rather than interpolated: it lands
/// in SQL unquoted.
pub(crate) fn object_keyword(kind: &str) -> Result<&'static str> {
    match kind.to_uppercase().as_str() {
        "TABLE" | "VIEW" | "MATERIALIZED VIEW" | "FOREIGN TABLE" => Ok("TABLE"),
        "SEQUENCE" => Ok("SEQUENCE"),
        // `FUNCTION` names no procedure; `ROUTINE` names either.
        "FUNCTION" | "PROCEDURE" | "ROUTINE" => Ok("ROUTINE"),
        other => Err(Error::BadRequest(format!("Unknown object kind '{other}'"))),
    }
}

/// Every object of a schema, named the way the catalog names it.
async fn read_schema_objects(
    client: &tokio_postgres::Client,
    schema: &str,
) -> Result<Vec<AclObject>> {
    let rows = client
        .query(
            "SELECT CASE c.relkind WHEN 'S' THEN 'SEQUENCE' ELSE 'TABLE' END, c.relname, NULL::text
             FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace
             WHERE n.nspname = $1 AND c.relkind = ANY(ARRAY['r','p','v','m','S','f']::\"char\"[])
             UNION ALL
             SELECT CASE p.prokind WHEN 'p' THEN 'PROCEDURE' ELSE 'FUNCTION' END, p.proname,
                    pg_get_function_identity_arguments(p.oid)
             FROM pg_proc p JOIN pg_namespace n ON n.oid = p.pronamespace
             WHERE n.nspname = $1",
            &[&schema],
        )
        .await
        .map_err(|e| {
            Error::internal_err(format!(
                "Failed to list the objects of schema '{schema}': {}",
                pg_error_message(&e)
            ))
        })?;
    Ok(rows
        .into_iter()
        .map(|row| AclObject { kind: row.get(0), name: row.get(1), args: row.get(2) })
        .collect())
}

/// Replace the objects a revoke names with the catalog's own entry for each.
///
/// A routine is identified by its argument types, and those go into the statement as written —
/// there is no quoting for them — so the request may name an object but never spell one: what
/// reaches the SQL is read back from Postgres. An object that resolves to nothing is refused rather
/// than dropped, since a revoke that silently covers less than it says is worse than an error.
async fn resolve_acl_objects(
    client: &tokio_postgres::Client,
    target: &AclTarget,
    objects: &[AclObject],
) -> Result<Vec<AclObject>> {
    if objects.is_empty() {
        return Ok(vec![]);
    }
    let Some(schema) = target.schema() else {
        return Err(Error::BadRequest(
            "A database has no objects of its own to revoke on".to_string(),
        ));
    };
    let known = read_schema_objects(client, schema).await?;
    objects
        .iter()
        .map(|requested| {
            let keyword = object_keyword(&requested.kind)?;
            known
                .iter()
                .find(|k| {
                    k.name == requested.name
                        && k.args == requested.args
                        && object_keyword(&k.kind).is_ok_and(|k| k == keyword)
                })
                .cloned()
                .ok_or_else(|| {
                    Error::NotFound(format!(
                        "'{}' is not an object of schema '{schema}'",
                        requested.name
                    ))
                })
        })
        .collect()
}

async fn read_owner(client: &tokio_postgres::Client, target: &AclTarget) -> Result<Option<String>> {
    let row = match target {
        AclTarget::Database => {
            client
                .query_opt(
                    "SELECT pg_get_userbyid(datdba) FROM pg_database WHERE datname = current_database()",
                    &[],
                )
                .await
        }
        AclTarget::Schema { schema } => {
            client
                .query_opt(
                    // `public` is owned by `pg_database_owner`, a placeholder role whose membership
                    // is whoever owns the database — naming it back would say nothing, so resolve
                    // it to that owner.
                    "SELECT pg_get_userbyid(owner) FROM (
                         SELECT CASE WHEN n.nspowner = (SELECT oid FROM pg_roles WHERE rolname = 'pg_database_owner')
                                     THEN (SELECT d.datdba FROM pg_database d WHERE d.datname = current_database())
                                     ELSE n.nspowner END AS owner
                         FROM pg_namespace n WHERE n.nspname = $1
                     ) o",
                    &[schema],
                )
                .await
        }
        AclTarget::Table { schema, table } => {
            client
                .query_opt(
                    "SELECT pg_get_userbyid(c.relowner)
                     FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace
                     WHERE n.nspname = $1 AND c.relname = $2",
                    &[schema, table],
                )
                .await
        }
    }
    .map_err(|e| Error::internal_err(format!("Failed to read the owner: {}", pg_error_message(&e))))?;
    Ok(row.map(|row| row.get(0)))
}

async fn read_children(client: &tokio_postgres::Client, target: &AclTarget) -> Result<Vec<String>> {
    let rows = match target {
        AclTarget::Database => {
            client
                .query(
                    "SELECT nspname::text FROM pg_namespace
                     WHERE nspname <> 'information_schema' AND nspname NOT LIKE 'pg\\_%'
                     ORDER BY nspname",
                    &[],
                )
                .await
        }
        AclTarget::Schema { schema } => {
            client
                .query(
                    "SELECT c.relname::text
                     FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace
                     WHERE n.nspname = $1 AND c.relkind = ANY(ARRAY['r','p']::\"char\"[])
                     ORDER BY c.relname",
                    &[schema],
                )
                .await
        }
        AclTarget::Table { .. } => return Ok(vec![]),
    }
    .map_err(|e| {
        Error::internal_err(format!(
            "Failed to list what the target holds: {}",
            pg_error_message(&e)
        ))
    })?;
    Ok(rows.into_iter().map(|row| row.get(0)).collect())
}

async fn get_datatable_acl(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,
    Path((w_id, datatable_name)): Path<(String, String)>,
    Query(query): Query<AclTargetQuery>,
) -> JsonResult<DatatableAclInfo> {
    crate::datatable_acl_oss::ensure_datatable_acl_available()?;
    let target: AclTarget = query.try_into()?;
    let governing = resolve_governing_datatable(&db, &w_id, &datatable_name).await?;
    ensure_reaches_governing_datatable(&db, &w_id, &datatable_name, &governing, &authed).await?;
    let cluster = ensure_managed(&governing)?;
    let editable = ensure_governs_datatable(&db, &authed, &w_id, &governing)
        .await
        .is_ok();
    let roles = if editable {
        role_names(&read_role_catalog(&db, cluster).await?)
    } else {
        vec![]
    };

    let (client, _notices, dbname) = connect_as_admin_unchecked(&db, &governing).await?;
    let owner = read_owner(&client, &target)
        .await?
        .ok_or_else(|| Error::NotFound(format!("{} not found", target.label(&dbname))))?;
    let grants = read_grants(&client, &target).await?;
    let supports_maintain: bool = client
        .query_one(
            "SELECT current_setting('server_version_num')::int >= 170000",
            &[],
        )
        .await
        .map_err(|e| {
            Error::internal_err(format!(
                "Failed to read the server version: {}",
                pg_error_message(&e)
            ))
        })?
        .get(0);
    let children = read_children(&client, &target).await?;

    Ok(Json(DatatableAclInfo {
        owner: role_name_of(&owner),
        roles,
        editable,
        supports_maintain,
        dbname,
        grants,
        children,
    }))
}

async fn read_grants(client: &tokio_postgres::Client, target: &AclTarget) -> Result<Vec<AclGrant>> {
    // `aclexplode` turns an acl array into one row per (grantee, privilege); grantee 0 is PUBLIC,
    // which has no name to resolve. A NULL acl is not "no access" but Postgres's built-in default —
    // the owner holds everything and, on a routine, PUBLIC may EXECUTE — hence `acldefault`. The
    // owner's own entries are left out: what it holds comes with ownership, which the owner shows,
    // not with a grant a revoke here could take back. Each row ends with its source — the grantor,
    // or a default privilege's creating role — and whether this connection can take back what it
    // gave.
    // Column-level grants (`pg_attribute.attacl`) are not supported yet: they are neither read here
    // nor revocable from the editor.
    let mut rows = match target {
        AclTarget::Database => {
            let mut out = client
                .query(
                    concat!(
                        "SELECT CASE WHEN a.grantee = 0 THEN 'PUBLIC' ELSE pg_get_userbyid(a.grantee) END,
                                a.privilege_type, NULL::text, NULL::text, NULL::text, NULL::text, ",
                        grant_source!("d.datdba"),
                        " FROM pg_database d, aclexplode(COALESCE(d.datacl, acldefault('d', d.datdba))) a
                         WHERE d.datname = current_database() AND a.grantee <> d.datdba"
                    ),
                    &[],
                )
                .await
                .map_err(grant_read_error)?;
            out.extend(
                client
                    .query(
                        // Default privileges set database-wide reach what is created in every
                        // schema, so they are the database's to show rather than any schema's.
                        concat!(
                            "SELECT CASE WHEN g.grantee = 0 THEN 'PUBLIC' ELSE pg_get_userbyid(g.grantee) END,
                                    g.privilege_type, NULL::text,
                                    CASE g.defaclobjtype
                                        WHEN 'r' THEN 'TABLES' WHEN 'S' THEN 'SEQUENCES'
                                        WHEN 'f' THEN 'FUNCTIONS' WHEN 'n' THEN 'SCHEMAS'
                                        ELSE 'TYPES' END, NULL::text, NULL::text,
                                    pg_get_userbyid(g.defaclrole), pg_has_role(g.defaclrole, 'USAGE')
                             FROM (",
                            database_wide_defaults!(),
                            ") g"
                        ),
                        &[],
                    )
                    .await
                    .map_err(grant_read_error)?,
            );
            out
        }
        AclTarget::Schema { schema } => {
            let mut out = client
                .query(
                    concat!(
                        "SELECT CASE WHEN a.grantee = 0 THEN 'PUBLIC' ELSE pg_get_userbyid(a.grantee) END,
                                a.privilege_type, NULL::text, NULL::text, NULL::text, NULL::text, ",
                        grant_source!("n.nspowner"),
                        " FROM pg_namespace n, aclexplode(COALESCE(n.nspacl, acldefault('n', n.nspowner))) a
                         WHERE n.nspname = $1 AND a.grantee <> n.nspowner"
                    ),
                    &[schema],
                )
                .await
                .map_err(grant_read_error)?;
            out.extend(
                client
                    .query(
                        concat!(
                            "SELECT CASE WHEN a.grantee = 0 THEN 'PUBLIC' ELSE pg_get_userbyid(a.grantee) END,
                                    a.privilege_type, c.relname, NULL::text,
                                    CASE c.relkind WHEN 'S' THEN 'SEQUENCE' ELSE 'TABLE' END,
                                    NULL::text, ",
                            grant_source!("c.relowner"),
                            " FROM pg_class c
                             JOIN pg_namespace n ON n.oid = c.relnamespace,
                                  aclexplode(COALESCE(c.relacl, acldefault(
                                      CASE c.relkind WHEN 'S' THEN 's' ELSE 'r' END::\"char\", c.relowner))) a
                             WHERE n.nspname = $1
                               AND c.relkind = ANY(ARRAY['r','p','v','m','S','f']::\"char\"[])
                               AND a.grantee <> c.relowner"
                        ),
                        &[schema],
                    )
                    .await
                    .map_err(grant_read_error)?,
            );
            out.extend(
                client
                    .query(
                        // Routines carry their own acl in `pg_proc`; without this a grant made here
                        // would vanish on the next read and could never be revoked back.
                        concat!(
                            "SELECT CASE WHEN a.grantee = 0 THEN 'PUBLIC' ELSE pg_get_userbyid(a.grantee) END,
                                    a.privilege_type, p.proname, NULL::text,
                                    CASE p.prokind WHEN 'p' THEN 'PROCEDURE' ELSE 'FUNCTION' END,
                                    pg_get_function_identity_arguments(p.oid), ",
                            grant_source!("p.proowner"),
                            " FROM pg_proc p
                             JOIN pg_namespace n ON n.oid = p.pronamespace,
                                  aclexplode(COALESCE(p.proacl, acldefault('f', p.proowner))) a
                             WHERE n.nspname = $1 AND a.grantee <> p.proowner"
                        ),
                        &[schema],
                    )
                    .await
                    .map_err(grant_read_error)?,
            );
            out.extend(
                client
                    .query(
                        // `USAGE` on a type is what lets a role use it in a column. Only a type the
                        // schema holds in its own right has an acl: an array, a row type and a
                        // multirange answer to their element, table or range.
                        concat!(
                            "SELECT CASE WHEN a.grantee = 0 THEN 'PUBLIC' ELSE pg_get_userbyid(a.grantee) END,
                                    a.privilege_type, t.typname, NULL::text, 'TYPE', NULL::text, ",
                            grant_source!("t.typowner"),
                            " FROM pg_type t
                         JOIN pg_depend d ON d.classid = 'pg_type'::regclass AND d.objid = t.oid
                          AND d.refclassid = 'pg_namespace'::regclass AND d.deptype = 'n',
                              aclexplode(COALESCE(t.typacl, acldefault('T', t.typowner))) a
                         WHERE d.refobjid = (SELECT oid FROM pg_namespace WHERE nspname = $1)
                           AND a.grantee <> t.typowner
                           AND NOT EXISTS (
                               SELECT 1 FROM pg_depend x
                               WHERE x.classid = 'pg_type'::regclass AND x.objid = t.oid
                                 AND x.objsubid = 0 AND x.deptype = 'i')"
                        ),
                        &[schema],
                    )
                    .await
                    .map_err(grant_read_error)?,
            );
            out.extend(
                client
                    .query(
                        // What a creating role set is taken back `FOR ROLE` that role, which only
                        // a role acting for it may do.
                        "SELECT CASE WHEN a.grantee = 0 THEN 'PUBLIC' ELSE pg_get_userbyid(a.grantee) END,
                                a.privilege_type, NULL::text,
                                CASE d.defaclobjtype
                                    WHEN 'r' THEN 'TABLES' WHEN 'S' THEN 'SEQUENCES'
                                    WHEN 'f' THEN 'FUNCTIONS' ELSE 'TYPES' END, NULL::text, NULL::text,
                                pg_get_userbyid(d.defaclrole), pg_has_role(d.defaclrole, 'USAGE')
                         FROM pg_default_acl d
                         JOIN pg_namespace n ON n.oid = d.defaclnamespace,
                              aclexplode(d.defaclacl) a
                         WHERE n.nspname = $1",
                        &[schema],
                    )
                    .await
                    .map_err(grant_read_error)?,
            );
            out
        }
        AclTarget::Table { schema, table } => client
            .query(
                concat!(
                    "SELECT CASE WHEN a.grantee = 0 THEN 'PUBLIC' ELSE pg_get_userbyid(a.grantee) END,
                            a.privilege_type, NULL::text, NULL::text, NULL::text, NULL::text, ",
                    grant_source!("c.relowner"),
                    " FROM pg_class c
                     JOIN pg_namespace n ON n.oid = c.relnamespace,
                          aclexplode(COALESCE(c.relacl, acldefault(
                              CASE c.relkind WHEN 'S' THEN 's' ELSE 'r' END::\"char\", c.relowner))) a
                     WHERE n.nspname = $1 AND c.relname = $2 AND a.grantee <> c.relowner"
                ),
                &[schema, table],
            )
            .await
            .map_err(grant_read_error)?,
    };

    // One row per privilege and source — and, for default privileges, per creating role. Fold them
    // back into one entry per grantee and object that keeps every source and what each gave: a
    // revoke takes a privilege back from every source that gave it.
    let mut folded: BTreeMap<
        (
            String,
            Option<(String, String, Option<String>)>,
            Option<String>,
        ),
        (Vec<String>, BTreeMap<String, (bool, Vec<String>)>),
    > = BTreeMap::new();
    for row in rows.drain(..) {
        let grantee: String = row.get(0);
        let privilege: String = row.get(1);
        let object: Option<String> = row.get(2);
        let future: Option<String> = row.get(3);
        let object_kind: Option<String> = row.get(4);
        let object_args: Option<String> = row.get(5);
        let source: String = row.get(6);
        let reachable: bool = row.get(7);
        let (privileges, sources) = folded
            .entry((
                role_name_of(&grantee),
                object.map(|name| {
                    (
                        name,
                        object_kind.unwrap_or_else(|| "TABLE".to_string()),
                        object_args,
                    )
                }),
                future,
            ))
            .or_default();
        sources
            .entry(role_name_of(&source))
            .or_insert_with(|| (reachable, vec![]))
            .1
            .push(privilege.clone());
        privileges.push(privilege);
    }
    Ok(folded
        .into_iter()
        .map(|((grantee, object, future), (mut privileges, sources))| {
            privileges.sort();
            privileges.dedup();
            AclGrant {
                grantee,
                privileges,
                object: object.map(|(name, kind, args)| AclObject { name, kind, args }),
                future,
                sources: sources
                    .into_iter()
                    .map(|(role, (reachable, mut privileges))| {
                        privileges.sort();
                        privileges.dedup();
                        AclSource { role, privileges, reachable }
                    })
                    .collect(),
            }
        })
        .collect())
}

fn grant_read_error(e: tokio_postgres::Error) -> Error {
    Error::internal_err(format!("Failed to read grants: {}", pg_error_message(&e)))
}

/// Changing a data table's access is administering it. Checked in full before anything connects
/// with the instance's credentials.
async fn authorize_acl_change(
    db: &DB,
    authed: &ApiAuthed,
    w_id: &str,
    datatable_name: &str,
) -> Result<GoverningDatatable> {
    let governing = resolve_governing_datatable(db, w_id, datatable_name).await?;
    ensure_governs_datatable(db, authed, w_id, &governing).await?;
    ensure_managed(&governing)?;
    Ok(governing)
}

static APPLY_SLOT: tokio::sync::Semaphore = tokio::sync::Semaphore::const_new(1);

/// A role passes on only privileges it holds with grant option, and an instance database
/// provisioned before data table roles gave `custom_instance_user` none. Adds that option to its
/// database and `public` privileges, and nothing else: default privileges are left alone, since a
/// schema's change of owner is planned against them. Best-effort, as a grant it fails to enable is
/// refused when it runs. An external instance database was created with the options, so one
/// missing there is someone's deliberate revoke and is left alone.
async fn ensure_grant_options(
    client: &tokio_postgres::Client,
    db: &DB,
    cluster: DatatableRoleCluster,
    dbname: &str,
) {
    let held = client
        .query_one(
            "SELECT has_database_privilege(current_database(), 'CONNECT WITH GRANT OPTION')
                AND has_database_privilege(current_database(), 'CREATE WITH GRANT OPTION')
                AND (to_regnamespace('public') IS NULL
                     OR (has_schema_privilege('public', 'USAGE WITH GRANT OPTION')
                         AND has_schema_privilege('public', 'CREATE WITH GRANT OPTION')))",
            &[],
        )
        .await
        .is_ok_and(|row| row.get::<_, bool>(0));
    if held || cluster != DatatableRoleCluster::Instance {
        return;
    }
    if let Err(e) = grant_options_as_server(db, dbname).await {
        tracing::warn!("Could not enable grant options on '{dbname}': {e}");
    }
}

/// Only the database's owner, the server's own Postgres user, can hand out an option it holds.
async fn grant_options_as_server(db: &DB, dbname: &str) -> Result<()> {
    let server = PgDatabase::parse_uri(&windmill_common::get_database_url().await?.as_str().await)?;
    let creds = PgDatabase { dbname: dbname.to_string(), ..server };
    let (client, connection) = creds.connect(Some(db)).await?;
    let join_handle = tokio::spawn(async move { connection.await });
    let role = quote_ident(CUSTOM_INSTANCE_USER);
    let result = client
        .batch_execute(&format!(
            "GRANT CONNECT, CREATE ON DATABASE {} TO {role} WITH GRANT OPTION;
             DO $$ BEGIN
               IF to_regnamespace('public') IS NOT NULL THEN
                 GRANT USAGE, CREATE ON SCHEMA public TO {role} WITH GRANT OPTION;
               END IF;
             END $$;",
            quote_ident(dbname)
        ))
        .await;
    drop(client);
    windmill_common::shutdown_pg_connection(join_handle).await?;
    result.map_err(|e| {
        Error::internal_err(format!(
            "Failed to grant options on '{dbname}': {}",
            pg_error_message(&e)
        ))
    })
}

/// Whether the governing entry an apply was authorized on is still the one in the settings, read
/// under the lock: a save in between could have pointed it at another database or changed its roles.
fn entry_unchanged(governing: &GoverningDatatable, entry_now: Option<serde_json::Value>) -> bool {
    let Some(Ok(now)) = entry_now.map(serde_json::from_value::<DataTable>) else {
        return false;
    };
    match (
        serde_json::to_value(&now),
        serde_json::to_value(&governing.datatable),
    ) {
        (Ok(now), Ok(authorized)) => now == authorized,
        _ => false,
    }
}

/// Plan one change against the catalog and the database as they are now.
async fn build_plan(
    client: &tokio_postgres::Client,
    dbname: &str,
    catalog: &DatatableRoleCatalog,
    target: &AclTarget,
    change: &AclChange,
) -> Result<AclPlan> {
    let role = change.role();
    // `admin` is the login the data table itself reaches Postgres through, and the one every
    // change here runs as: a revoke that lands leaves nothing able to grant it back.
    if matches!(change, AclChange::Revoke { .. }) && role == ADMIN_DATATABLE_ROLE {
        return Err(Error::BadRequest(format!(
            "'{ADMIN_DATATABLE_ROLE}' is how this data table reaches its database; \
             its own access is not revocable from here"
        )));
    }
    let pg_role = pg_role_of(role, catalog)?;
    let change = match change {
        AclChange::Revoke { role, privileges, scope, objects } => AclChange::Revoke {
            role: role.clone(),
            privileges: privileges.clone(),
            scope: *scope,
            objects: resolve_acl_objects(client, target, objects).await?,
        },
        change => change.clone(),
    };
    // Default privileges are recorded per creating role, and a schema's new owner is kept in reach
    // of what the others create there, so both are written for every role there is.
    let other_pg_roles = role_names(catalog)
        .iter()
        .map(|name| pg_role_of(name, catalog))
        .filter(|r| r.as_ref().map_or(true, |r| *r != pg_role))
        .collect::<Result<Vec<_>>>()?;
    let (existing_objects, former_owner) = match (&change, target) {
        (AclChange::SetOwner { .. }, AclTarget::Schema { schema }) => (
            read_owned_objects(client, schema).await?,
            read_former_owner_defaults(client, schema, &pg_role).await?,
        ),
        _ => (vec![], None),
    };
    let revoked_grants = match &change {
        AclChange::Revoke { privileges, scope, objects, .. } => {
            read_revoked_grants(
                client, dbname, target, *scope, objects, privileges, &pg_role,
            )
            .await?
        }
        _ => vec![],
    };
    if matches!(change, AclChange::SetOwner { .. }) {
        if let Some((object, owner)) = unmanaged_owner(client, target).await? {
            return Err(Error::BadRequest(format!(
                "{object} is owned by {owner}, which this data table's connection cannot act \
                 for, so its owner cannot be changed from here"
            )));
        }
    }
    let facts = CatalogFacts { other_pg_roles, existing_objects, former_owner, revoked_grants };
    let mut plan =
        crate::datatable_acl_oss::plan_statements(target, &change, dbname, &pg_role, &facts)?;
    if matches!(change, AclChange::SetOwner { .. }) {
        if let Some(missing) = missing_owner_privilege(client, target, &pg_role).await? {
            plan.warnings.push(format!(
                "{role} does not have {missing}, which Postgres requires of a new owner, so this \
                 will be refused. Grant it first."
            ));
        }
    }
    Ok(plan)
}

/// Postgres only hands an object to a role that could have created it: a table to one with
/// `CREATE` on its schema, a schema to one with `CREATE` on the database.
async fn missing_owner_privilege(
    client: &tokio_postgres::Client,
    target: &AclTarget,
    pg_role: &str,
) -> Result<Option<String>> {
    let (row, missing) = match target {
        AclTarget::Table { schema, .. } => (
            client
                .query_one(
                    "SELECT has_schema_privilege($1::name, $2::text, 'CREATE')",
                    &[&pg_role, schema],
                )
                .await,
            format!("CREATE on schema {schema}"),
        ),
        AclTarget::Schema { .. } => (
            client
                .query_one(
                    "SELECT has_database_privilege($1::name, current_database(), 'CREATE')",
                    &[&pg_role],
                )
                .await,
            "CREATE on the database".to_string(),
        ),
        AclTarget::Database => return Ok(None),
    };
    let has: bool = row
        .map_err(|e| {
            Error::internal_err(format!(
                "Failed to read the new owner's privileges: {}",
                pg_error_message(&e)
            ))
        })?
        .get(0);
    Ok((!has).then_some(missing))
}

/// The first thing a change of owner would move that this connection cannot act for, with its
/// owner. Postgres lets only a member of the current owner move an object, and every change runs as
/// `custom_instance_user`, so an object it does not hold the owner of — `public`, owned by the
/// database's owner, above all — is refused here rather than at apply. Running as the instance's
/// own user instead would reach objects Windmill never created.
async fn unmanaged_owner(
    client: &tokio_postgres::Client,
    target: &AclTarget,
) -> Result<Option<(String, String)>> {
    let read_error = |e: tokio_postgres::Error| {
        Error::internal_err(format!(
            "Failed to read who owns what the change moves: {}",
            pg_error_message(&e)
        ))
    };
    match target {
        AclTarget::Database => Ok(None),
        AclTarget::Table { schema, table } => {
            let row = client
                .query_opt(
                    "SELECT pg_get_userbyid(c.relowner)
                     FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace
                     WHERE n.nspname = $1 AND c.relname = $2 AND NOT pg_has_role(c.relowner, 'USAGE')",
                    &[schema, table],
                )
                .await
                .map_err(read_error)?;
            Ok(row.map(|row| (format!("{schema}.{table}"), row.get(0))))
        }
        AclTarget::Schema { schema } => {
            let row = client
                .query_opt(
                    concat!(
                        "SELECT ord, identity, pg_get_userbyid(owner) FROM (
                             SELECT 0 AS ord, NULL::text AS identity, n.nspowner AS owner
                             FROM pg_namespace n WHERE n.nspname = $1
                             UNION ALL
                             SELECT 1, identity, owner FROM (",
                        schema_owned_objects!(),
                        ") m
                         ) o WHERE NOT pg_has_role(owner, 'USAGE') ORDER BY ord, identity LIMIT 1"
                    ),
                    &[schema],
                )
                .await
                .map_err(read_error)?;
            Ok(row.map(|row| {
                let label = row
                    .get::<_, Option<String>>(1)
                    .unwrap_or_else(|| format!("schema {schema}"));
                (label, row.get(2))
            }))
        }
    }
}

async fn plan_datatable_acl(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,
    Path((w_id, datatable_name)): Path<(String, String)>,
    Json(req): Json<AclChangeRequest>,
) -> JsonResult<AclPlan> {
    crate::datatable_acl_oss::ensure_datatable_acl_available()?;
    let governing = authorize_acl_change(&db, &authed, &w_id, &datatable_name).await?;
    let cluster = ensure_managed(&governing)?;
    let catalog = read_role_catalog(&db, cluster).await?;
    let (client, _notices, dbname) = connect_as_admin_unchecked(&db, &governing).await?;
    Ok(Json(
        build_plan(&client, &dbname, &catalog, &req.target, &req.change).await?,
    ))
}

async fn apply_datatable_acl(
    authed: ApiAuthed,
    Extension(db): Extension<DB>,
    Path((w_id, datatable_name)): Path<(String, String)>,
    Json(req): Json<AclChangeRequest>,
) -> Result<String> {
    crate::datatable_acl_oss::ensure_datatable_acl_available()?;
    let confirmed = req.statements.as_ref().ok_or_else(|| {
        Error::BadRequest(
            "An apply runs exactly the statements its plan showed; plan the change first"
                .to_string(),
        )
    })?;
    // Everything that needs the pool happens before the locks: once `tx` holds them, a second pool
    // connection could wait forever on a pool that concurrent applies, queued on the same locks,
    // have exhausted.
    let governing = authorize_acl_change(&db, &authed, &w_id, &datatable_name).await?;
    let cluster = ensure_managed(&governing)?;
    // Applies queue on an instance-wide lock while each holds a direct connection to the instance's
    // Postgres; unbounded, the queue alone could exhaust its connection limit. One at a time per
    // server, and the ones waiting hold no connection at all.
    let _slot = APPLY_SLOT
        .acquire()
        .await
        .map_err(|e| Error::internal_err(format!("ACL apply slot closed: {e}")))?;
    let (mut client, mut notices, dbname) = connect_as_admin_unchecked(&db, &governing).await?;
    ensure_grant_options(&client, &db, cluster, &dbname).await;

    // Held until the change is committed: a role renamed or dropped meanwhile would change what
    // the plan names, and a settings save could move the entry onto another database. Taken in the
    // same order as the permissions save, so the two cannot deadlock.
    let mut tx = db.begin().await?;
    lock_role_catalog(&mut tx).await?;
    let entry_now = sqlx::query_scalar::<_, Option<serde_json::Value>>(
        "SELECT datatable->'datatables'->$2 FROM workspace_settings WHERE workspace_id = $1 FOR UPDATE",
    )
    .bind(&governing.workspace_id)
    .bind(&governing.name)
    .fetch_optional(&mut *tx)
    .await?
    .flatten();
    let catalog = read_role_catalog_tx(&mut tx, cluster).await?;

    let plan = build_plan(&client, &dbname, &catalog, &req.target, &req.change).await?;
    if !entry_unchanged(&governing, entry_now) || &plan.statements != confirmed {
        return Err(Error::BadRequest(
            "The data table or its roles changed since this was planned, so it would no longer \
             run what was confirmed. Plan it again."
                .to_string(),
        ));
    }

    // One transaction: a half-applied ownership transfer leaves one schema's objects owned by two
    // different roles.
    let pg_tx = client.transaction().await.map_err(|e| {
        Error::internal_err(format!(
            "Failed to open a transaction on the data table: {}",
            pg_error_message(&e)
        ))
    })?;
    for statement in &plan.statements {
        pg_tx.batch_execute(statement).await.map_err(|e| {
            Error::ExecutionErr(format!(
                "Failed to run `{statement}`: {}",
                pg_error_message(&e)
            ))
        })?;
        // A privilege the connection cannot pass on is only a warning to Postgres, which then
        // carries on having changed nothing. Returning drops the transaction, rolling back
        // everything before it.
        while let Ok(notice) = notices.try_recv() {
            if *notice.code() == SqlState::WARNING_PRIVILEGE_NOT_GRANTED
                || *notice.code() == SqlState::WARNING_PRIVILEGE_NOT_REVOKED
            {
                return Err(Error::ExecutionErr(format!(
                    "`{statement}` did not take effect ({}), so nothing was applied",
                    notice.message()
                )));
            }
        }
    }

    // A schema's objects were listed before the transaction opened; one committed since would stay
    // with its old owner. One created while this transaction is still open can still slip past, as
    // Postgres has no lock that holds creation in a schema back. That is benign: it stays with its
    // creator, like anything created there later, and moving that table fixes it.
    if let (AclChange::SetOwner { role }, AclTarget::Schema { schema }) = (&req.change, &req.target)
    {
        let new_owner = pg_role_of(role, &catalog)?;
        let straggler = pg_tx
            .query_opt(
                concat!(
                    "SELECT identity FROM (",
                    schema_owned_objects!(),
                    ") m WHERE owner <> (SELECT oid FROM pg_roles WHERE rolname = $2)
                     ORDER BY 1 LIMIT 1"
                ),
                &[schema, &new_owner],
            )
            .await
            .map_err(|e| {
                Error::internal_err(format!(
                    "Failed to check what the schema holds: {}",
                    pg_error_message(&e)
                ))
            })?;
        if let Some(row) = straggler {
            return Err(Error::BadRequest(format!(
                "{} appeared while this ran and would keep its old owner, so nothing was \
                 applied. Plan it again.",
                row.get::<_, String>(0)
            )));
        }
    }

    let target_label = req.target.label(&dbname);
    audit_log(
        &mut *tx,
        &authed,
        "workspaces.datatable_acl",
        ActionKind::Update,
        &governing.workspace_id,
        Some(&governing.name),
        Some(
            [
                ("target", target_label.as_str()),
                ("change", change_kind(&req.change)),
                ("role", req.change.role()),
            ]
            .into(),
        ),
    )
    .await?;
    pg_tx.commit().await.map_err(|e| {
        Error::internal_err(format!(
            "Failed to commit the changes: {}",
            pg_error_message(&e)
        ))
    })?;
    tx.commit().await?;

    windmill_common::feature_usage::log_feature_usage(
        "datatable",
        "acl_applied",
        change_kind(&req.change),
    );

    Ok(format!("Updated access on {target_label}"))
}

/// What kind of change, never what it named: the telemetry key and the audit's summary.
fn change_kind(change: &AclChange) -> &'static str {
    match change {
        AclChange::SetOwner { .. } => "owner",
        AclChange::Grant { scope, .. } | AclChange::Revoke { scope, .. } if scope.is_future() => {
            "default_privileges"
        }
        AclChange::Grant { .. } => "grant",
        AclChange::Revoke { .. } => "revoke",
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use windmill_common::datatable_roles::InstanceDatatableRole;

    #[test]
    fn a_role_is_its_own_postgres_role_except_admin() {
        let catalog: DatatableRoleCatalog = BTreeMap::from([(
            "role1".to_string(),
            InstanceDatatableRole { name: "analytics".to_string(), enabled: true, pwd: None },
        )]);
        assert_eq!(
            pg_role_of("admin", &catalog).unwrap(),
            "custom_instance_user"
        );
        assert_eq!(pg_role_of("analytics", &catalog).unwrap(), "analytics");
        assert_eq!(role_name_of("custom_instance_user"), "admin");
        assert_eq!(role_name_of("analytics"), "analytics");
        // A catalog id, a name the catalog lacks, or the admin login spelled out never stands for
        // some other role.
        for unknown in ["role1", "operator", "custom_instance_user", "PUBLIC", ""] {
            assert!(
                matches!(pg_role_of(unknown, &catalog), Err(Error::BadRequest(_))),
                "{unknown}"
            );
        }
    }

    /// A connection to the test's own database, the way the handlers reach a data table's.
    async fn catalog_client(pool: &sqlx::PgPool) -> tokio_postgres::Client {
        let mut config: tokio_postgres::Config =
            std::env::var("DATABASE_URL").unwrap().parse().unwrap();
        config.dbname(pool.connect_options().get_database().unwrap());
        let (client, connection) = config.connect(tokio_postgres::NoTls).await.unwrap();
        tokio::spawn(connection);
        client
    }

    /// What a revoke takes back is read from the catalog, per object and source, and only the
    /// privileges it asks for: the planner renders exactly this, and refuses an empty read.
    #[sqlx::test(migrations = false)]
    async fn a_revoke_reads_back_only_what_it_asks_for(pool: sqlx::PgPool) {
        let client = catalog_client(&pool).await;
        // A predefined role, so that nothing is granted outside the test's own database.
        client
            .batch_execute(
                "CREATE SCHEMA granted;
                 CREATE TABLE granted.g (id int);
                 GRANT SELECT, INSERT ON granted.g TO pg_read_all_data;",
            )
            .await
            .unwrap();
        let target = AclTarget::Table { schema: "granted".to_string(), table: "g".to_string() };
        let revoked = read_revoked_grants(
            &client,
            "db",
            &target,
            GrantScope::Target,
            &[],
            &["select".to_string()],
            "pg_read_all_data",
        )
        .await
        .unwrap();
        assert_eq!(revoked.len(), 1, "{revoked:?}");
        assert_eq!(revoked[0].object, None);
        assert_eq!(revoked[0].privileges, ["SELECT"]);
        let held_none = read_revoked_grants(
            &client,
            "db",
            &target,
            GrantScope::Target,
            &[],
            &["update".to_string()],
            "pg_read_all_data",
        )
        .await
        .unwrap();
        assert!(held_none.is_empty(), "{held_none:?}");
        // Each source says what it gave: that, and not the whole row, is what a revoke of some of
        // its privileges is held back by.
        let grants = read_grants(
            &client,
            &AclTarget::Schema { schema: "granted".to_string() },
        )
        .await
        .unwrap();
        let on_g = grants
            .iter()
            .find(|g| {
                g.grantee == "pg_read_all_data" && g.object.as_ref().is_some_and(|o| o.name == "g")
            })
            .unwrap();
        assert_eq!(on_g.sources.len(), 1, "{:?}", on_g.sources);
        assert_eq!(on_g.sources[0].privileges, ["INSERT", "SELECT"]);
        assert!(on_g.sources[0].reachable);
    }

    /// A default privilege set database-wide applies in every schema on top of the schema's own,
    /// which cannot take it back: the database shows it, and neither a schema's revoke of it nor a
    /// change of owner away from its grantee may go ahead as if it were gone.
    #[sqlx::test(migrations = false)]
    async fn a_schema_cannot_take_back_a_database_wide_default(pool: sqlx::PgPool) {
        let client = catalog_client(&pool).await;
        client
            .batch_execute(
                "CREATE SCHEMA owned AUTHORIZATION pg_read_all_data;
                 ALTER DEFAULT PRIVILEGES GRANT SELECT ON TABLES TO pg_read_all_data;
                 ALTER DEFAULT PRIVILEGES IN SCHEMA owned
                     GRANT SELECT, INSERT ON TABLES TO pg_read_all_data;
                 CREATE SCHEMA creators AUTHORIZATION pg_monitor;
                 ALTER DEFAULT PRIVILEGES FOR ROLE pg_monitor
                     GRANT SELECT ON TABLES TO pg_read_all_stats;",
            )
            .await
            .unwrap();
        let creator: String = client
            .query_one("SELECT current_user::text", &[])
            .await
            .unwrap()
            .get(0);

        let grants = read_grants(&client, &AclTarget::Database).await.unwrap();
        let database_wide = grants
            .iter()
            .find(|g| g.grantee == "pg_read_all_data" && g.future.as_deref() == Some("TABLES"))
            .unwrap_or_else(|| panic!("{grants:?}"));
        assert_eq!(database_wide.privileges, ["SELECT"]);
        // A database-wide entry also holds its creator's own privileges, which come with creating
        // and are no grant: neither a row nor a refusal may stem from them.
        assert!(
            !grants
                .iter()
                .any(|g| g.future.is_some() && (g.grantee == creator || g.grantee == "pg_monitor")),
            "{grants:?}"
        );
        assert_eq!(
            database_wide.sources.len(),
            1,
            "{:?}",
            database_wide.sources
        );
        assert_eq!(database_wide.sources[0].role, creator);

        let schema = AclTarget::Schema { schema: "owned".to_string() };
        let still_granted = read_revoked_grants(
            &client,
            "db",
            &schema,
            GrantScope::FutureTables,
            &[],
            &["select".to_string()],
            "pg_read_all_data",
        )
        .await;
        assert!(
            matches!(&still_granted, Err(Error::BadRequest(m)) if m.contains("database-wide")),
            "{still_granted:?}"
        );
        let schema_only = read_revoked_grants(
            &client,
            "db",
            &schema,
            GrantScope::FutureTables,
            &[],
            &["insert".to_string()],
            "pg_read_all_data",
        )
        .await
        .unwrap();
        assert_eq!(schema_only.len(), 1, "{schema_only:?}");
        assert_eq!(schema_only[0].privileges, ["INSERT"]);

        let moved = read_former_owner_defaults(&client, "owned", "pg_write_all_data").await;
        assert!(
            matches!(&moved, Err(Error::BadRequest(m)) if m.contains("database-wide")),
            "{moved:?}"
        );
        let moved_from_creator =
            read_former_owner_defaults(&client, "creators", "pg_write_all_data").await;
        assert!(
            matches!(moved_from_creator, Ok(None)),
            "{moved_from_creator:?}"
        );
    }

    /// Defaults on types go with a schema's owner like the other kinds: once a database-wide
    /// default takes PUBLIC's USAGE on types away, they are all that reaches a new type.
    #[sqlx::test(migrations = false)]
    async fn a_schemas_former_owner_defaults_include_types(pool: sqlx::PgPool) {
        let client = catalog_client(&pool).await;
        client
            .batch_execute(
                "CREATE SCHEMA typed AUTHORIZATION pg_read_all_data;
                 ALTER DEFAULT PRIVILEGES IN SCHEMA typed GRANT USAGE ON TYPES TO pg_read_all_data;
                 ALTER DEFAULT PRIVILEGES FOR ROLE pg_read_all_data IN SCHEMA typed
                     GRANT SELECT ON TABLES TO pg_read_all_data;",
            )
            .await
            .unwrap();
        let creator: String = client
            .query_one("SELECT current_user::text", &[])
            .await
            .unwrap()
            .get(0);
        let former = read_former_owner_defaults(&client, "typed", "pg_write_all_data")
            .await
            .unwrap();
        assert_eq!(
            former,
            Some(FormerOwnerDefaults {
                pg_role: "pg_read_all_data".to_string(),
                defaults: vec![(creator, "TYPES")],
            })
        );
    }

    /// A kind of object the list misses stays with its old owner while the schema changes hands,
    /// which only a real catalog shows.
    #[sqlx::test(migrations = false)]
    async fn a_schemas_owner_change_takes_every_object_in_it(pool: sqlx::PgPool) {
        let client = catalog_client(&pool).await;
        client
            .batch_execute(
                "CREATE SCHEMA moved;
                 CREATE TABLE moved.t (id serial PRIMARY KEY, a int, b int);
                 CREATE STATISTICS moved.st ON a, b FROM moved.t;
                 CREATE TABLE moved.pt (id int) PARTITION BY RANGE (id);
                 CREATE TABLE moved.pt1 PARTITION OF moved.pt FOR VALUES FROM (0) TO (10);
                 CREATE TYPE moved.r AS RANGE (subtype = float8);
                 CREATE TYPE moved.pair AS (a int, b int);
                 CREATE TYPE moved.mood AS ENUM ('ok');
                 CREATE DOMAIN moved.tags AS text[];
                 CREATE COLLATION moved.coll (provider = libc, locale = 'C');",
            )
            .await
            .unwrap();
        let owned = read_owned_objects(&client, "moved").await.unwrap();
        // Not the serial's sequence, the range's constructors and multirange, or any array or row
        // type: each follows the object it belongs to.
        assert_eq!(
            owned
                .iter()
                .map(|o| (o.keyword, o.identity.as_str()))
                .collect::<Vec<_>>(),
            [
                ("COLLATION", "moved.coll"),
                ("STATISTICS", "moved.st"),
                ("TABLE", "moved.pt"),
                ("TABLE", "moved.pt1"),
                ("TABLE", "moved.t"),
                ("TYPE", "moved.mood"),
                ("TYPE", "moved.pair"),
                ("TYPE", "moved.r"),
                ("TYPE", "moved.tags"),
            ]
        );
    }
}