/* pg_synonym--1.0.sql */

-- Refuse to run as a plain SQL script: this file must be loaded by CREATE EXTENSION.
\echo Use "CREATE EXTENSION pg_synonym" to load this file. \quit

/*
 * How it works, in one paragraph
 * ------------------------------
 * PostgreSQL extensions cannot add new SQL syntax, so there is no
 * CREATE SYNONYM statement. Instead, synonym.create('vendas.cliente',
 * 'erp.clientes') records the synonym in synonym.catalog and creates a view
 *
 *     CREATE VIEW vendas.cliente WITH (security_invoker = true)
 *         AS SELECT * FROM erp.clientes;
 *
 * PostgreSQL makes simple views like this one updatable, so SELECT, INSERT,
 * UPDATE and DELETE all work through it, and security_invoker makes the
 * target's own permissions and row-level security apply to whoever uses
 * the synonym. Two event triggers keep the view in step with its target:
 * one in C (src/pg_synonym.c) runs before DROP TABLE / DROP COLUMN /
 * ALTER COLUMN TYPE, the other (below) runs after DDL and rebuilds what is
 * out of date.
 */

GRANT USAGE ON SCHEMA synonym TO PUBLIC;

------------------------------------------------------------------------
-- The catalog: one row per synonym. This is the source of truth; the
-- views are rebuilt from it whenever needed.
------------------------------------------------------------------------
CREATE TABLE synonym.catalog (
    syn_schema  name        NOT NULL,
    syn_name    name        NOT NULL,
    tgt_schema  name        NOT NULL,
    tgt_name    name        NOT NULL,
    owner       name        NOT NULL DEFAULT current_user,
    created_at  timestamptz NOT NULL DEFAULT now(),
    PRIMARY KEY (syn_schema, syn_name)
);

COMMENT ON TABLE synonym.catalog IS
'Synonyms managed by pg_synonym. Use synonym.create() and synonym.drop() instead of changing this table.';

-- Include the rows in pg_dump output
SELECT pg_catalog.pg_extension_config_dump('synonym.catalog', '');

-- Anyone may read the catalog. Writes go through synonym.create() and
-- synonym.drop(), which run as the caller; these policies make sure a user
-- can only add synonyms in schemas where they may create objects, and only
-- remove synonyms they own.
ALTER TABLE synonym.catalog ENABLE ROW LEVEL SECURITY;

CREATE POLICY catalog_select ON synonym.catalog FOR SELECT USING (true);

CREATE POLICY catalog_insert ON synonym.catalog FOR INSERT
    WITH CHECK (owner = current_user
                AND pg_catalog.has_schema_privilege(syn_schema, 'CREATE'));

CREATE POLICY catalog_delete ON synonym.catalog FOR DELETE
    USING (owner = current_user
           OR EXISTS (SELECT 1 FROM pg_catalog.pg_roles r
                      WHERE r.rolname = owner
                        AND pg_catalog.pg_has_role(current_user, r.oid, 'MEMBER')));

GRANT SELECT, INSERT, DELETE ON synonym.catalog TO PUBLIC;

------------------------------------------------------------------------
-- Small helpers
------------------------------------------------------------------------

-- Column names and types of a relation, in order.
CREATE FUNCTION synonym._columns(rel oid, OUT names name[], OUT types text[])
LANGUAGE sql STABLE
SET search_path = pg_catalog, pg_temp
AS $$
    SELECT coalesce(array_agg(a.attname ORDER BY a.attnum), '{}'),
           coalesce(array_agg(format_type(a.atttypid, a.atttypmod) ORDER BY a.attnum), '{}')
    FROM pg_attribute a
    WHERE a.attrelid = rel AND a.attnum > 0 AND NOT a.attisdropped
$$;

/*
 * The state of every synonym (or of one, when p_schema/p_name are given),
 * computed in a single query so that it stays fast with many synonyms:
 *
 *   ok              the view exists and matches its target
 *   pending         the target does not exist (yet); there is no view
 *   missing_view    the target exists but the view does not: create it
 *   columns_changed the target's columns changed: update the view
 *   target_renamed  the target was renamed or moved: update the catalog
 *   conflict        another object already uses the synonym's name
 *   schema_missing  the synonym's schema no longer exists
 *
 * view_oid is the synonym's view (NULL if there is none, or on conflict).
 * target_oid is the relation the view reads from; when the target is
 * renamed or moved the view keeps pointing at it, so we find it through
 * the view's dependencies rather than by name.
 */
CREATE FUNCTION synonym._states(p_schema name DEFAULT NULL, p_name name DEFAULT NULL)
RETURNS TABLE (syn_schema name, syn_name name, view_oid oid, target_oid oid, state text)
LANGUAGE sql STABLE
SET search_path = pg_catalog, pg_temp
AS $$
    WITH s AS (
        SELECT c.syn_schema, c.syn_name, c.tgt_schema, c.tgt_name, c.owner,
               sn.oid AS syn_nsp,
               v.oid AS view_oid, v.relkind AS view_kind, pg_get_userbyid(v.relowner) AS view_owner,
               t.oid AS tgt_oid
        FROM synonym.catalog c
        LEFT JOIN pg_namespace sn ON sn.nspname = c.syn_schema
        LEFT JOIN pg_class v      ON v.relnamespace = sn.oid AND v.relname = c.syn_name
        LEFT JOIN pg_namespace tn ON tn.nspname = c.tgt_schema
        LEFT JOIN pg_class t      ON t.relnamespace = tn.oid AND t.relname = c.tgt_name
        WHERE (p_schema IS NULL OR c.syn_schema = p_schema)
          AND (p_name IS NULL OR c.syn_name = p_name)
    ),
    r AS (
        SELECT s.*, ref.oid AS ref_oid, refn.nspname AS ref_schema, ref.relname AS ref_name
        FROM s
        LEFT JOIN LATERAL (
            SELECT d.refobjid
            FROM pg_rewrite rw
            JOIN pg_depend d ON d.classid = 'pg_rewrite'::regclass AND d.objid = rw.oid
            WHERE rw.ev_class = s.view_oid
              AND s.view_kind = 'v'
              AND d.refclassid = 'pg_class'::regclass
              AND d.refobjid <> s.view_oid
            LIMIT 1
        ) dep ON true
        LEFT JOIN pg_class ref      ON ref.oid = dep.refobjid
        LEFT JOIN pg_namespace refn ON refn.oid = ref.relnamespace
    )
    SELECT r.syn_schema, r.syn_name,
           CASE WHEN r.view_kind = 'v' AND r.view_owner = r.owner THEN r.view_oid END,
           coalesce(r.ref_oid, r.tgt_oid),
           CASE
               WHEN r.syn_nsp IS NULL THEN 'schema_missing'
               WHEN r.view_oid IS NULL THEN
                   CASE WHEN r.tgt_oid IS NULL THEN 'pending' ELSE 'missing_view' END
               WHEN r.view_kind <> 'v' OR r.view_owner <> r.owner THEN 'conflict'
               WHEN r.ref_oid IS NOT NULL
                    AND (r.ref_schema <> r.tgt_schema OR r.ref_name <> r.tgt_name) THEN 'target_renamed'
               -- a view over a table with no columns has no dependencies to follow
               WHEN coalesce(r.ref_oid, r.tgt_oid) IS NULL THEN 'ok'
               WHEN vc.names = tc.names AND vc.types = tc.types THEN 'ok'
               ELSE 'columns_changed'
           END
    FROM r
    -- the same as synonym._columns(), written inline so the planner can do it in bulk
    LEFT JOIN LATERAL (
        SELECT coalesce(array_agg(a.attname ORDER BY a.attnum), '{}') AS names,
               coalesce(array_agg(format_type(a.atttypid, a.atttypmod) ORDER BY a.attnum), '{}') AS types
        FROM pg_attribute a
        WHERE a.attrelid = r.view_oid AND a.attnum > 0 AND NOT a.attisdropped
    ) vc ON r.view_oid IS NOT NULL
    LEFT JOIN LATERAL (
        SELECT coalesce(array_agg(a.attname ORDER BY a.attnum), '{}') AS names,
               coalesce(array_agg(format_type(a.atttypid, a.atttypmod) ORDER BY a.attnum), '{}') AS types
        FROM pg_attribute a
        WHERE a.attrelid = coalesce(r.ref_oid, r.tgt_oid) AND a.attnum > 0 AND NOT a.attisdropped
    ) tc ON r.view_oid IS NOT NULL
$$;

-- Is this relation a synonym view?
CREATE FUNCTION synonym._is_synonym(rel oid)
RETURNS boolean
LANGUAGE sql STABLE
SET search_path = pg_catalog, pg_temp
AS $$
    SELECT EXISTS (
        SELECT 1 FROM synonym.catalog c
        WHERE to_regclass(format('%I.%I', c.syn_schema, c.syn_name)) = rel)
$$;

-- Create (or extend) the view behind a synonym.
CREATE FUNCTION synonym._create_view(p_schema name, p_name name,
                                     p_tschema name, p_tname name,
                                     p_owner name, p_replace boolean)
RETURNS void
LANGUAGE plpgsql
SET search_path = pg_catalog, pg_temp
AS $$
BEGIN
    EXECUTE format('CREATE %sVIEW %I.%I WITH (security_invoker = true) AS SELECT * FROM %I.%I',
                   CASE WHEN p_replace THEN 'OR REPLACE ' ELSE '' END,
                   p_schema, p_name, p_tschema, p_tname);

    IF NOT p_replace THEN
        EXECUTE format('ALTER VIEW %I.%I OWNER TO %I', p_schema, p_name, p_owner);
        -- Safe because of security_invoker: the target's own permissions still apply
        EXECUTE format('GRANT SELECT, INSERT, UPDATE, DELETE ON %I.%I TO PUBLIC', p_schema, p_name);
    END IF;

    EXECUTE format('COMMENT ON VIEW %I.%I IS %L', p_schema, p_name,
                   format('Synonym for %I.%I (managed by pg_synonym)', p_tschema, p_tname));
END
$$;

REVOKE ALL ON FUNCTION synonym._create_view(name, name, name, name, name, boolean) FROM PUBLIC;

------------------------------------------------------------------------
-- Bringing a synonym in line with its target
------------------------------------------------------------------------

/*
 * Make one synonym match the catalog and its target. Returns the final
 * state ('ok', 'pending' or 'missing'); raises an error when it cannot.
 *
 * SECURITY DEFINER: the person running ALTER TABLE on the target is often
 * not the synonym's owner, but the synonym must still be updated. All
 * privilege checks that matter are done explicitly below.
 */
CREATE FUNCTION synonym._sync_one(p_schema name, p_name name)
RETURNS text
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = pg_catalog, pg_temp
SET pg_synonym.syncing = on
AS $$
DECLARE
    r       synonym.catalog%ROWTYPE;
    i       record;
    v_kind  "char";
    vn name[]; vt text[];
    tn name[]; tt text[];
    k int;
BEGIN
    FOR attempt IN 1..5 LOOP
        SELECT * INTO r FROM synonym.catalog c
        WHERE c.syn_schema = p_schema AND c.syn_name = p_name;
        SELECT * INTO i FROM synonym._states(p_schema, p_name);
        IF NOT FOUND THEN
            RETURN 'missing';
        END IF;

        CASE i.state
        WHEN 'ok', 'pending' THEN
            RETURN i.state;

        WHEN 'schema_missing' THEN
            -- DROP SCHEMA ... CASCADE took the synonym with it, as it would any
            -- other object in that schema
            DELETE FROM synonym.catalog c
            WHERE c.syn_schema = p_schema AND c.syn_name = p_name;
            RETURN 'missing';

        WHEN 'conflict' THEN
            RAISE EXCEPTION 'relation "%.%" already exists and is not managed by pg_synonym',
                            r.syn_schema, r.syn_name
                USING ERRCODE = 'duplicate_table',
                      HINT = 'Rename or drop that relation, then run synonym.refresh().';

        WHEN 'target_renamed' THEN
            UPDATE synonym.catalog c
            SET tgt_schema = n.nspname, tgt_name = cl.relname
            FROM pg_class cl JOIN pg_namespace n ON n.oid = cl.relnamespace
            WHERE cl.oid = i.target_oid
              AND c.syn_schema = p_schema AND c.syn_name = p_name;

        WHEN 'missing_view' THEN
            SELECT c.relkind INTO v_kind FROM pg_class c WHERE c.oid = i.target_oid;
            IF v_kind NOT IN ('r', 'p', 'v', 'm', 'f') THEN
                RAISE EXCEPTION '"%.%" is not a table, view, materialized view or foreign table',
                                r.tgt_schema, r.tgt_name
                    USING ERRCODE = 'wrong_object_type';
            END IF;
            IF synonym._is_synonym(i.target_oid) THEN
                RAISE EXCEPTION '"%.%" is itself a synonym', r.tgt_schema, r.tgt_name
                    USING ERRCODE = 'feature_not_supported',
                          HINT = 'A synonym must point at a real table or view.';
            END IF;
            IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = r.owner) THEN
                -- The owner was dropped while the synonym was pending: it goes too
                DELETE FROM synonym.catalog c
                WHERE c.syn_schema = p_schema AND c.syn_name = p_name;
                RAISE WARNING 'pg_synonym: removed synonym %.% because its owner "%" no longer exists',
                              quote_ident(p_schema), quote_ident(p_name), r.owner;
                RETURN 'missing';
            END IF;
            IF NOT has_schema_privilege(r.owner, r.syn_schema, 'CREATE') THEN
                RAISE EXCEPTION 'owner "%" of synonym "%.%" no longer has CREATE privilege on schema "%"',
                                r.owner, r.syn_schema, r.syn_name, r.syn_schema
                    USING ERRCODE = 'insufficient_privilege';
            END IF;
            PERFORM synonym._create_view(r.syn_schema, r.syn_name, r.tgt_schema, r.tgt_name, r.owner, false);

        WHEN 'columns_changed' THEN
            SELECT c.names, c.types INTO vn, vt FROM synonym._columns(i.view_oid) c;
            SELECT c.names, c.types INTO tn, tt FROM synonym._columns(i.target_oid) c;

            IF cardinality(vn) < cardinality(tn)
               AND vn = tn[1:cardinality(vn)] AND vt = tt[1:cardinality(vt)] THEN
                -- Only new columns at the end: extend the view in place
                PERFORM synonym._create_view(r.syn_schema, r.syn_name, r.tgt_schema, r.tgt_name, r.owner, true);

            ELSIF cardinality(vn) = cardinality(tn) AND vt = tt THEN
                -- Only renamed columns. Two passes, so swapped names never collide.
                FOR k IN 1..cardinality(vn) LOOP
                    IF vn[k] <> tn[k] THEN
                        EXECUTE format('ALTER VIEW %I.%I RENAME COLUMN %I TO %I',
                                       r.syn_schema, r.syn_name, vn[k], '__pg_synonym_' || k);
                    END IF;
                END LOOP;
                FOR k IN 1..cardinality(vn) LOOP
                    IF vn[k] <> tn[k] THEN
                        EXECUTE format('ALTER VIEW %I.%I RENAME COLUMN %I TO %I',
                                       r.syn_schema, r.syn_name, '__pg_synonym_' || k, tn[k]);
                    END IF;
                END LOOP;

            ELSE
                -- Anything else: rebuild the view
                BEGIN
                    EXECUTE format('DROP VIEW %I.%I', r.syn_schema, r.syn_name);
                EXCEPTION WHEN dependent_objects_still_exist THEN
                    RAISE EXCEPTION 'cannot rebuild synonym "%.%" because other objects depend on it',
                                    r.syn_schema, r.syn_name
                        USING ERRCODE = 'dependent_objects_still_exist',
                              HINT = 'Drop the objects that use the synonym, run synonym.refresh(), then recreate them.';
                END;
                PERFORM synonym._create_view(r.syn_schema, r.syn_name, r.tgt_schema, r.tgt_name, r.owner, false);
            END IF;
        END CASE;
    END LOOP;

    RAISE EXCEPTION 'pg_synonym could not bring synonym "%.%" up to date', p_schema, p_name
        USING ERRCODE = 'internal_error';
END
$$;

/*
 * Bring every synonym up to date. Problems become WARNINGs rather than
 * errors: a broken synonym must never make someone's unrelated ALTER TABLE
 * fail.
 */
CREATE FUNCTION synonym._sync_all()
RETURNS void
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = pg_catalog, pg_temp
SET pg_synonym.syncing = on
AS $$
DECLARE
    r record;
BEGIN
    FOR r IN
        SELECT st.syn_schema, st.syn_name
        FROM synonym._states() st
        WHERE st.state NOT IN ('ok', 'pending')
        ORDER BY 1, 2
    LOOP
        BEGIN
            PERFORM synonym._sync_one(r.syn_schema, r.syn_name);
        EXCEPTION WHEN OTHERS THEN
            RAISE WARNING 'pg_synonym: could not update synonym %.%: %',
                          quote_ident(r.syn_schema), quote_ident(r.syn_name), SQLERRM;
        END;
    END LOOP;
END
$$;

/*
 * Called (from C) just before DROP TABLE, ALTER TABLE ... DROP COLUMN and
 * ALTER TABLE ... ALTER COLUMN TYPE. Drops the synonym views that point at
 * the given relations, so the command is not blocked by them. The synonyms
 * stay in the catalog and are rebuilt when the command finishes.
 *
 * Not callable from SQL by ordinary users (EXECUTE is revoked below).
 */
CREATE FUNCTION synonym._detach(p_targets oid[], p_cascade boolean)
RETURNS integer
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = pg_catalog, pg_temp
SET pg_synonym.syncing = on
AS $$
DECLARE
    r record;
    n integer := 0;
BEGIN
    FOR r IN
        SELECT st.syn_schema, st.syn_name
        FROM synonym._states() st
        WHERE st.view_oid IS NOT NULL
          AND st.target_oid = ANY (p_targets)
    LOOP
        BEGIN
            EXECUTE format('DROP VIEW %I.%I %s', r.syn_schema, r.syn_name,
                           CASE WHEN p_cascade THEN 'CASCADE' ELSE 'RESTRICT' END);
        EXCEPTION WHEN dependent_objects_still_exist THEN
            RAISE EXCEPTION 'other objects depend on synonym %.%',
                            quote_ident(r.syn_schema), quote_ident(r.syn_name)
                USING ERRCODE = 'dependent_objects_still_exist',
                      DETAIL = 'The synonym must be rebuilt for this command, and the objects that use it would be lost.',
                      HINT = 'Drop those objects first, or use CASCADE if the command accepts it (DROP TABLE and DROP COLUMN do).';
        END;
        n := n + 1;
    END LOOP;
    RETURN n;
END
$$;

REVOKE ALL ON FUNCTION synonym._detach(oid[], boolean) FROM PUBLIC;

------------------------------------------------------------------------
-- Event triggers
------------------------------------------------------------------------

-- Before: the C function in src/pg_synonym.c
CREATE FUNCTION synonym._on_ddl_start()
RETURNS event_trigger
AS 'MODULE_PATHNAME', 'pg_synonym_ddl_start'
LANGUAGE C;

CREATE EVENT TRIGGER pg_synonym_ddl_start ON ddl_command_start
    WHEN TAG IN ('ALTER TABLE', 'ALTER FOREIGN TABLE',
                 'DROP TABLE', 'DROP VIEW', 'DROP MATERIALIZED VIEW', 'DROP FOREIGN TABLE')
    EXECUTE FUNCTION synonym._on_ddl_start();

-- After: rebuild whatever is out of date
CREATE FUNCTION synonym._on_ddl_end()
RETURNS event_trigger
LANGUAGE plpgsql
SET search_path = pg_catalog, pg_temp
AS $$
BEGIN
    -- Our own CREATE VIEW / DROP VIEW commands must not start another round
    IF current_setting('pg_synonym.syncing', true) = 'on' THEN
        RETURN;
    END IF;
    -- Fast path for databases without synonyms
    IF NOT EXISTS (SELECT 1 FROM synonym.catalog) THEN
        RETURN;
    END IF;
    PERFORM synonym._sync_all();
END
$$;

CREATE EVENT TRIGGER pg_synonym_ddl_end ON ddl_command_end
    WHEN TAG IN ('CREATE TABLE', 'CREATE TABLE AS', 'SELECT INTO',
                 'CREATE VIEW', 'CREATE MATERIALIZED VIEW', 'CREATE FOREIGN TABLE',
                 'IMPORT FOREIGN SCHEMA', 'CREATE SCHEMA',
                 'ALTER TABLE', 'ALTER VIEW', 'ALTER MATERIALIZED VIEW', 'ALTER FOREIGN TABLE',
                 'ALTER SCHEMA',
                 'DROP TABLE', 'DROP VIEW', 'DROP MATERIALIZED VIEW', 'DROP FOREIGN TABLE',
                 'DROP SCHEMA', 'DROP OWNED')
    EXECUTE FUNCTION synonym._on_ddl_end();

/*
 * When someone drops a synonym's view directly (DROP VIEW vendas.cliente,
 * or DROP OWNED BY its owner), they mean to drop the synonym, so we forget
 * it. Views that disappear only as a side effect (CASCADE from the
 * target's schema, for example) are not "original" drops: those synonyms
 * stay and wait for their target to come back.
 *
 * Runs as the person dropping the view. The catalog's DELETE policy lets
 * them remove the row because they own the view (or are a superuser).
 */
CREATE FUNCTION synonym._on_sql_drop()
RETURNS event_trigger
LANGUAGE plpgsql
SET search_path = pg_catalog, pg_temp
AS $$
BEGIN
    IF current_setting('pg_synonym.syncing', true) = 'on' THEN
        RETURN;
    END IF;
    DELETE FROM synonym.catalog c
    USING pg_event_trigger_dropped_objects() d
    WHERE d.original
      AND d.object_type = 'view'
      AND d.schema_name = c.syn_schema
      AND d.object_name = c.syn_name;
END
$$;

CREATE EVENT TRIGGER pg_synonym_sql_drop ON sql_drop
    WHEN TAG IN ('DROP VIEW', 'DROP OWNED')
    EXECUTE FUNCTION synonym._on_sql_drop();

------------------------------------------------------------------------
-- Public API
------------------------------------------------------------------------

/*
 * synonym.create('vendas.cliente', 'erp.clientes')
 *
 * Same as SQL Server's  CREATE SYNONYM vendas.cliente FOR erp.clientes
 *
 * Names without a schema follow the usual PostgreSQL rules: the synonym
 * goes into the first schema of search_path, and the target is looked up
 * through search_path. The target does not need to exist yet.
 */
CREATE FUNCTION synonym."create"(synonym_name text, target_object text)
RETURNS void
LANGUAGE plpgsql
AS $$
DECLARE
    s         text[];
    t         text[];
    v_schema  name;
    v_name    name;
    v_tschema name;
    v_tname   name;
    v_toid    oid;
    v_kind    "char";
    v_state   text;
    part      text;
BEGIN
    IF synonym_name IS NULL OR target_object IS NULL THEN
        RAISE EXCEPTION 'synonym name and target must not be NULL'
            USING ERRCODE = 'null_value_not_allowed';
    END IF;

    s := pg_catalog.parse_ident(synonym_name);
    t := pg_catalog.parse_ident(target_object);

    IF cardinality(s) > 2 THEN
        RAISE EXCEPTION 'improper synonym name (too many dotted names): %', synonym_name
            USING ERRCODE = 'syntax_error', HINT = 'Use schema.name or just name.';
    END IF;
    IF cardinality(t) > 2 THEN
        RAISE EXCEPTION 'cross-database synonyms are not supported: %', target_object
            USING ERRCODE = 'feature_not_supported',
                  HINT = 'Create a foreign table with postgres_fdw and point the synonym at it.';
    END IF;
    FOREACH part IN ARRAY s || t LOOP
        IF octet_length(part) > 63 THEN
            RAISE EXCEPTION 'name "%" is too long (maximum is 63 bytes)', part
                USING ERRCODE = 'name_too_long';
        END IF;
    END LOOP;

    -- Where the synonym goes
    IF cardinality(s) = 1 THEN
        v_schema := pg_catalog.current_schema();
        v_name := s[1];
        IF v_schema IS NULL THEN
            RAISE EXCEPTION 'no schema has been selected to create in'
                USING ERRCODE = 'invalid_schema_name';
        END IF;
    ELSE
        v_schema := s[1];
        v_name := s[2];
    END IF;

    IF pg_catalog.to_regnamespace(pg_catalog.quote_ident(v_schema)) IS NULL THEN
        RAISE EXCEPTION 'schema "%" does not exist', v_schema
            USING ERRCODE = 'invalid_schema_name';
    END IF;
    IF v_schema = 'synonym' OR v_schema LIKE 'pg\_temp%' THEN
        RAISE EXCEPTION 'cannot create a synonym in schema "%"', v_schema
            USING ERRCODE = 'invalid_schema_name';
    END IF;
    IF NOT pg_catalog.has_schema_privilege(v_schema, 'CREATE') THEN
        RAISE EXCEPTION 'permission denied for schema %', v_schema
            USING ERRCODE = 'insufficient_privilege';
    END IF;
    IF EXISTS (SELECT 1 FROM synonym.catalog c WHERE c.syn_schema = v_schema AND c.syn_name = v_name) THEN
        RAISE EXCEPTION 'synonym "%.%" already exists', v_schema, v_name
            USING ERRCODE = 'duplicate_object';
    END IF;
    IF pg_catalog.to_regclass(pg_catalog.format('%I.%I', v_schema, v_name)) IS NOT NULL THEN
        RAISE EXCEPTION 'relation "%.%" already exists', v_schema, v_name
            USING ERRCODE = 'duplicate_table';
    END IF;

    -- What it points at
    IF cardinality(t) = 1 THEN
        v_toid := pg_catalog.to_regclass(pg_catalog.quote_ident(t[1]));
        IF v_toid IS NOT NULL THEN
            SELECT n.nspname, c.relname INTO v_tschema, v_tname
            FROM pg_catalog.pg_class c JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
            WHERE c.oid = v_toid;
        ELSE
            v_tschema := pg_catalog.current_schema();
            v_tname := t[1];
        END IF;
    ELSE
        v_tschema := t[1];
        v_tname := t[2];
        v_toid := pg_catalog.to_regclass(pg_catalog.format('%I.%I', v_tschema, v_tname));
    END IF;

    IF v_tschema LIKE 'pg\_temp%' THEN
        RAISE EXCEPTION 'a synonym cannot point at a temporary table'
            USING ERRCODE = 'feature_not_supported';
    END IF;
    IF v_tschema = v_schema AND v_tname = v_name THEN
        RAISE EXCEPTION 'a synonym cannot point at itself'
            USING ERRCODE = 'invalid_parameter_value';
    END IF;
    IF EXISTS (SELECT 1 FROM synonym.catalog c WHERE c.syn_schema = v_tschema AND c.syn_name = v_tname) THEN
        RAISE EXCEPTION '"%.%" is itself a synonym', v_tschema, v_tname
            USING ERRCODE = 'feature_not_supported',
                  HINT = 'A synonym must point at a real table or view.';
    END IF;
    IF v_toid IS NOT NULL THEN
        SELECT c.relkind INTO v_kind FROM pg_catalog.pg_class c WHERE c.oid = v_toid;
        IF v_kind NOT IN ('r', 'p', 'v', 'm', 'f') THEN
            RAISE EXCEPTION '"%.%" is not a table, view, materialized view or foreign table', v_tschema, v_tname
                USING ERRCODE = 'wrong_object_type';
        END IF;
    END IF;

    INSERT INTO synonym.catalog (syn_schema, syn_name, tgt_schema, tgt_name)
    VALUES (v_schema, v_name, v_tschema, v_tname);

    v_state := synonym._sync_one(v_schema, v_name);

    IF v_state = 'pending' THEN
        RAISE NOTICE '"%.%" does not exist yet; synonym "%.%" will start working as soon as it is created',
                     v_tschema, v_tname, v_schema, v_name;
    END IF;
END
$$;

COMMENT ON FUNCTION synonym."create"(text, text) IS
'Creates a synonym: synonym.create(''schema.synonym'', ''schema.target''). Same as SQL Server CREATE SYNONYM.';

/*
 * synonym.drop('vendas.cliente')
 *
 * Same as SQL Server's  DROP SYNONYM [IF EXISTS] vendas.cliente
 */
CREATE FUNCTION synonym."drop"(synonym_name text,
                               if_exists boolean DEFAULT false,
                               "cascade" boolean DEFAULT false)
RETURNS void
LANGUAGE plpgsql
SET pg_synonym.syncing = on
AS $$
DECLARE
    s     text[];
    r     synonym.catalog%ROWTYPE;
    v_oid oid;
    v_kind "char";
    v_owner name;
BEGIN
    IF synonym_name IS NULL THEN
        RAISE EXCEPTION 'synonym name must not be NULL' USING ERRCODE = 'null_value_not_allowed';
    END IF;

    s := pg_catalog.parse_ident(synonym_name);

    IF cardinality(s) = 1 THEN
        -- Like DROP TABLE: the first schema in search_path that has it
        SELECT c.* INTO r
        FROM synonym.catalog c
        JOIN unnest(pg_catalog.current_schemas(false)) WITH ORDINALITY AS p(nsp, ord) ON p.nsp = c.syn_schema
        WHERE c.syn_name = s[1]
        ORDER BY p.ord
        LIMIT 1;
    ELSIF cardinality(s) = 2 THEN
        SELECT c.* INTO r FROM synonym.catalog c WHERE c.syn_schema = s[1] AND c.syn_name = s[2];
    ELSE
        RAISE EXCEPTION 'improper synonym name (too many dotted names): %', synonym_name
            USING ERRCODE = 'syntax_error';
    END IF;

    IF r.syn_name IS NULL THEN
        IF if_exists THEN
            RAISE NOTICE 'synonym "%" does not exist, skipping', synonym_name;
            RETURN;
        END IF;
        RAISE EXCEPTION 'synonym "%" does not exist', synonym_name
            USING ERRCODE = 'undefined_object';
    END IF;

    IF NOT (r.owner = current_user
            OR (SELECT rolsuper FROM pg_catalog.pg_roles WHERE rolname = current_user)
            OR coalesce(pg_catalog.pg_has_role(current_user,
                        (SELECT oid FROM pg_catalog.pg_roles WHERE rolname = r.owner), 'MEMBER'), false))
    THEN
        RAISE EXCEPTION 'must be owner of synonym %.%', r.syn_schema, r.syn_name
            USING ERRCODE = 'insufficient_privilege';
    END IF;

    -- Drop the view (as the caller, who owns it) and then the catalog row
    v_oid := pg_catalog.to_regclass(pg_catalog.format('%I.%I', r.syn_schema, r.syn_name));
    IF v_oid IS NOT NULL THEN
        SELECT c.relkind, pg_catalog.pg_get_userbyid(c.relowner) INTO v_kind, v_owner
        FROM pg_catalog.pg_class c WHERE c.oid = v_oid;
        IF v_kind = 'v' AND v_owner = r.owner THEN
            EXECUTE pg_catalog.format('DROP VIEW %I.%I %s', r.syn_schema, r.syn_name,
                                      CASE WHEN "cascade" THEN 'CASCADE' ELSE 'RESTRICT' END);
        END IF;
    END IF;

    DELETE FROM synonym.catalog c WHERE c.syn_schema = r.syn_schema AND c.syn_name = r.syn_name;
END
$$;

COMMENT ON FUNCTION synonym."drop"(text, boolean, boolean) IS
'Drops a synonym: synonym.drop(''schema.synonym'' [, if_exists [, cascade]]). Same as SQL Server DROP SYNONYM.';

-- Rebuild every synonym now (normally not needed: it happens automatically)
CREATE FUNCTION synonym.refresh()
RETURNS void
LANGUAGE sql
AS $$ SELECT synonym._sync_all() $$;

COMMENT ON FUNCTION synonym.refresh() IS
'Brings every synonym up to date with its target. Normally automatic.';

-- Like SQL Server's sys.synonyms
CREATE VIEW synonym.list AS
SELECT format('%I.%I', c.syn_schema, c.syn_name) AS synonym,
       format('%I.%I', c.tgt_schema, c.tgt_name) AS target,
       c.owner,
       CASE i.state
           WHEN 'ok' THEN 'ok'
           WHEN 'pending' THEN 'pending'
           WHEN 'conflict' THEN 'conflict'
           WHEN 'schema_missing' THEN 'orphaned'
           ELSE 'out of sync'
       END AS status,
       c.created_at
FROM synonym.catalog c
JOIN synonym._states() i USING (syn_schema, syn_name);

COMMENT ON VIEW synonym.list IS
'All synonyms. status: ok, pending (target does not exist yet), conflict (name taken by another object), orphaned (schema dropped), out of sync (run synonym.refresh()).';

GRANT SELECT ON synonym.list TO PUBLIC;