agenciapsilmno/database-novo/migrations/20260612000004_f1_clone_drop_functions.sql

-- =============================================================================
-- F1.4 — Schema-per-tenant: clone/drop + registro + roteamento de canais
--
-- * public.tenant_schemas — registro dos schemas provisionados (alimenta o
--   gerador do config.toml na F5)
-- * public.channel_routing — índice global de roteamento: webhooks inbound
--   (Twilio/Evolution) precisam descobrir o tenant do canal ANTES de saber o
--   schema (decisão Q3: notification_channels mora no schema do tenant).
--   Mantido por trigger em cada tenant_<slug>.notification_channels.
-- * clone_tenant_template(tenant_id) — instancia tenant_<slug> a partir do
--   _tenant_template: tabelas + sequences locais + FKs + seeds + views + RLS
--   (policies com tenant_id EMBUTIDO — modelo multi-membership) + realtime +
--   grants + trigger de roteamento.
-- * drop_tenant_schema(tenant_id) — protegido (assert tenant_%).
--
-- NOTA: clones criados na F1/F2 ainda NÃO têm triggers de negócio (F6) e não
-- estão expostos no PostgREST (F5). _meta.triggers_pending registra isso.
-- =============================================================================

BEGIN;

-- ---------------------------------------------------------------------------
-- Registro de schemas provisionados
-- ---------------------------------------------------------------------------
CREATE TABLE IF NOT EXISTS public.tenant_schemas (
    tenant_id        uuid PRIMARY KEY REFERENCES public.tenants(id) ON DELETE CASCADE,
    schema_name      text NOT NULL UNIQUE,
    template_version int  NOT NULL,
    created_at       timestamptz NOT NULL DEFAULT now()
);

ALTER TABLE public.tenant_schemas ENABLE ROW LEVEL SECURITY;
DROP POLICY IF EXISTS tenant_schemas_select ON public.tenant_schemas;
CREATE POLICY tenant_schemas_select ON public.tenant_schemas
    FOR SELECT TO authenticated
    USING (public.is_tenant_member(tenant_id) OR public.is_saas_admin());

-- ---------------------------------------------------------------------------
-- Índice global de roteamento de canais (webhook inbound -> tenant)
-- ---------------------------------------------------------------------------
CREATE TABLE IF NOT EXISTS public.channel_routing (
    channel_id            uuid PRIMARY KEY,
    tenant_id             uuid NOT NULL REFERENCES public.tenants(id) ON DELETE CASCADE,
    channel               text NOT NULL,
    provider              text,
    sender_address        text,
    twilio_subaccount_sid text,
    twilio_phone_number   text,
    metadata              jsonb,
    is_active             boolean NOT NULL DEFAULT true,
    updated_at            timestamptz NOT NULL DEFAULT now()
);

CREATE INDEX IF NOT EXISTS channel_routing_tenant_idx ON public.channel_routing (tenant_id);
CREATE INDEX IF NOT EXISTS channel_routing_sender_idx ON public.channel_routing (sender_address) WHERE sender_address IS NOT NULL;
CREATE INDEX IF NOT EXISTS channel_routing_twilio_phone_idx ON public.channel_routing (twilio_phone_number) WHERE twilio_phone_number IS NOT NULL;
CREATE INDEX IF NOT EXISTS channel_routing_twilio_sid_idx ON public.channel_routing (twilio_subaccount_sid) WHERE twilio_subaccount_sid IS NOT NULL;

-- Tabela de infra: só service_role (edge functions) e saas admin enxergam
ALTER TABLE public.channel_routing ENABLE ROW LEVEL SECURITY;
DROP POLICY IF EXISTS channel_routing_saas_admin ON public.channel_routing;
CREATE POLICY channel_routing_saas_admin ON public.channel_routing
    FOR ALL TO authenticated
    USING (public.is_saas_admin())
    WITH CHECK (public.is_saas_admin());

-- Trigger anexado a cada tenant_<slug>.notification_channels pelo clone
CREATE OR REPLACE FUNCTION public.trg_sync_channel_routing()
RETURNS trigger
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path TO 'public', 'pg_temp'
AS $$
DECLARE
    v_tenant_id uuid;
BEGIN
    v_tenant_id := public.tenant_id_for_schema(TG_TABLE_SCHEMA);
    IF v_tenant_id IS NULL THEN
        RAISE WARNING 'trg_sync_channel_routing: schema % sem tenant correspondente', TG_TABLE_SCHEMA;
        RETURN COALESCE(NEW, OLD);
    END IF;
    IF TG_OP = 'DELETE' THEN
        DELETE FROM public.channel_routing WHERE channel_id = OLD.id;
        RETURN OLD;
    END IF;
    INSERT INTO public.channel_routing AS cr
        (channel_id, tenant_id, channel, provider, sender_address,
         twilio_subaccount_sid, twilio_phone_number, metadata, is_active, updated_at)
    VALUES
        (NEW.id, v_tenant_id, NEW.channel, NEW.provider, NEW.sender_address,
         NEW.twilio_subaccount_sid, NEW.twilio_phone_number, NEW.metadata,
         COALESCE(NEW.is_active, false) AND NEW.deleted_at IS NULL, now())
    ON CONFLICT (channel_id) DO UPDATE SET
        tenant_id = EXCLUDED.tenant_id,
        channel = EXCLUDED.channel,
        provider = EXCLUDED.provider,
        sender_address = EXCLUDED.sender_address,
        twilio_subaccount_sid = EXCLUDED.twilio_subaccount_sid,
        twilio_phone_number = EXCLUDED.twilio_phone_number,
        metadata = EXCLUDED.metadata,
        is_active = EXCLUDED.is_active,
        updated_at = now();
    RETURN NEW;
END;
$$;

-- ---------------------------------------------------------------------------
-- clone_tenant_template
-- ---------------------------------------------------------------------------
CREATE OR REPLACE FUNCTION public.clone_tenant_template(p_tenant_id uuid)
RETURNS text
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path TO 'public', 'pg_temp'
AS $$
DECLARE
    v_slug text;
    v_schema text;
    v_version int;
    t text;
    r record;
    v_def text;
    v_seq text;
    v_n int;
    v_pending text[];
    v_failed text[];
BEGIN
    SELECT slug INTO v_slug FROM public.tenants WHERE id = p_tenant_id;
    IF v_slug IS NULL THEN
        RAISE EXCEPTION 'clone_tenant_template: tenant % não existe ou sem slug', p_tenant_id;
    END IF;
    v_schema := public.tenant_schema_name(v_slug);
    IF v_schema IS NULL THEN
        RAISE EXCEPTION 'clone_tenant_template: slug % inválido', v_slug;
    END IF;
    IF EXISTS (SELECT 1 FROM pg_namespace WHERE nspname = v_schema) THEN
        RAISE EXCEPTION 'clone_tenant_template: schema % já existe', v_schema;
    END IF;
    IF NOT EXISTS (SELECT 1 FROM pg_namespace WHERE nspname = '_tenant_template') THEN
        RAISE EXCEPTION 'clone_tenant_template: _tenant_template não existe (rode a F1.3)';
    END IF;
    SELECT (value)::int INTO v_version FROM _tenant_template._meta WHERE key = 'template_version';

    EXECUTE format('CREATE SCHEMA %I', v_schema);

    -- nomes qualificados nas definições geradas pelo catálogo
    PERFORM pg_catalog.set_config('search_path', 'pg_catalog', true);

    -- 1. tabelas
    FOR r IN
        SELECT table_name AS tab FROM information_schema.tables
        WHERE table_schema = '_tenant_template' AND table_type = 'BASE TABLE'
          AND table_name NOT LIKE '\_%'
        ORDER BY table_name
    LOOP
        EXECUTE format('CREATE TABLE %I.%I (LIKE _tenant_template.%I INCLUDING ALL)',
                       v_schema, r.tab, r.tab);
    END LOOP;

    -- 2. sequences locais (defaults que apontam pro template)
    FOR r IN
        SELECT c.relname AS tab, a.attname AS col
        FROM pg_attrdef d
        JOIN pg_class c ON c.oid = d.adrelid
        JOIN pg_attribute a ON a.attrelid = d.adrelid AND a.attnum = d.adnum
        WHERE c.relnamespace = v_schema::regnamespace
          AND pg_get_expr(d.adbin, d.adrelid) LIKE 'nextval(''_tenant_template.%'
    LOOP
        v_seq := r.tab || '_' || r.col || '_seq';
        EXECUTE format('CREATE SEQUENCE %I.%I', v_schema, v_seq);
        EXECUTE format('ALTER TABLE %I.%I ALTER COLUMN %I SET DEFAULT nextval(%L::regclass)',
                       v_schema, r.tab, r.col, format('%I.%I', v_schema, v_seq));
        EXECUTE format('ALTER SEQUENCE %I.%I OWNED BY %I.%I.%I',
                       v_schema, v_seq, v_schema, r.tab, r.col);
    END LOOP;

    -- 3. seeds (linhas-default do sistema guardadas no template)
    -- Sem session_replication_role (postgres não é superuser no Supabase):
    -- ordem de FK resolvida por tentativa-e-repetição em rounds.
    v_pending := ARRAY[]::text[];
    FOR r IN
        SELECT table_name AS tab FROM information_schema.tables
        WHERE table_schema = '_tenant_template' AND table_type = 'BASE TABLE'
          AND table_name NOT LIKE '\_%'
        ORDER BY table_name
    LOOP
        EXECUTE format('SELECT count(*) FROM _tenant_template.%I', r.tab) INTO v_n;
        IF v_n > 0 THEN v_pending := v_pending || r.tab; END IF;
    END LOOP;

    WHILE coalesce(array_length(v_pending, 1), 0) > 0 LOOP
        v_failed := ARRAY[]::text[];
        FOR r IN SELECT unnest(v_pending) AS tab LOOP
            BEGIN
                EXECUTE format('INSERT INTO %I.%I SELECT * FROM _tenant_template.%I',
                               v_schema, r.tab, r.tab);
            EXCEPTION WHEN foreign_key_violation THEN
                v_failed := v_failed || r.tab;
            END;
        END LOOP;
        IF array_length(v_failed, 1) = array_length(v_pending, 1) THEN
            RAISE EXCEPTION 'clone_tenant_template: dependência circular nos seeds: %', v_failed;
        END IF;
        v_pending := v_failed;
    END LOOP;

    -- 4. FKs (intra-schema e pra public/auth)
    FOR r IN
        SELECT cl.relname AS tab, con.conname, pg_get_constraintdef(con.oid) AS def
        FROM pg_constraint con
        JOIN pg_class cl ON cl.oid = con.conrelid
        WHERE con.contype = 'f'
          AND cl.relnamespace = '_tenant_template'::regnamespace
        ORDER BY cl.relname, con.conname
    LOOP
        v_def := replace(r.def, ' REFERENCES _tenant_template.', format(' REFERENCES %I.', v_schema));
        EXECUTE format('ALTER TABLE %I.%I ADD CONSTRAINT %I %s', v_schema, r.tab, r.conname, v_def);
    END LOOP;

    -- 5. views (placeholders __SCHEMA__ / __TENANT_ID__)
    PERFORM pg_catalog.set_config('search_path', 'public, pg_catalog', true);
    FOR r IN SELECT * FROM _tenant_template._views ORDER BY position LOOP
        EXECUTE replace(replace(r.definition, '__SCHEMA__', quote_ident(v_schema)),
                        '__TENANT_ID__', p_tenant_id::text);
    END LOOP;

    -- 6. RLS: tenant_id embutido (multi-membership: o usuário só enxerga
    --    schemas de tenants onde tenant_members o lista como ativo)
    FOR r IN
        SELECT table_name AS tab FROM information_schema.tables
        WHERE table_schema = '_tenant_template' AND table_type = 'BASE TABLE'
          AND table_name NOT LIKE '\_%'
    LOOP
        EXECUTE format('ALTER TABLE %I.%I ENABLE ROW LEVEL SECURITY', v_schema, r.tab);
        EXECUTE format(
            'CREATE POLICY tenant_member_full ON %I.%I FOR ALL TO authenticated USING (public.is_tenant_member(%L::uuid)) WITH CHECK (public.is_tenant_member(%L::uuid))',
            v_schema, r.tab, p_tenant_id, p_tenant_id);
        EXECUTE format(
            'CREATE POLICY saas_admin_full ON %I.%I FOR ALL TO authenticated USING (public.is_saas_admin()) WITH CHECK (public.is_saas_admin())',
            v_schema, r.tab);
    END LOOP;

    -- 7. trigger de roteamento de canais
    EXECUTE format(
        'CREATE TRIGGER trg_channel_routing AFTER INSERT OR UPDATE OR DELETE ON %I.notification_channels FOR EACH ROW EXECUTE FUNCTION public.trg_sync_channel_routing()',
        v_schema);

    -- 8. realtime
    FOR r IN SELECT table_name FROM _tenant_template._realtime_tables LOOP
        EXECUTE format('ALTER PUBLICATION supabase_realtime ADD TABLE %I.%I', v_schema, r.table_name);
    END LOOP;

    -- 9. grants (espelha o padrão do Supabase pra schemas expostos)
    EXECUTE format('GRANT USAGE ON SCHEMA %I TO anon, authenticated, service_role', v_schema);
    EXECUTE format('GRANT ALL ON ALL TABLES IN SCHEMA %I TO anon, authenticated, service_role', v_schema);
    EXECUTE format('GRANT ALL ON ALL SEQUENCES IN SCHEMA %I TO anon, authenticated, service_role', v_schema);
    EXECUTE format('ALTER DEFAULT PRIVILEGES IN SCHEMA %I GRANT ALL ON TABLES TO anon, authenticated, service_role', v_schema);
    EXECUTE format('ALTER DEFAULT PRIVILEGES IN SCHEMA %I GRANT ALL ON SEQUENCES TO anon, authenticated, service_role', v_schema);

    INSERT INTO public.tenant_schemas (tenant_id, schema_name, template_version)
    VALUES (p_tenant_id, v_schema, COALESCE(v_version, 1));

    RETURN v_schema;
END;
$$;

-- ---------------------------------------------------------------------------
-- drop_tenant_schema
-- ---------------------------------------------------------------------------
CREATE OR REPLACE FUNCTION public.drop_tenant_schema(p_tenant_id uuid)
RETURNS void
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path TO 'public', 'pg_temp'
AS $$
DECLARE
    v_schema text;
BEGIN
    SELECT schema_name INTO v_schema FROM public.tenant_schemas WHERE tenant_id = p_tenant_id;
    IF v_schema IS NULL THEN
        v_schema := public.tenant_schema_for(p_tenant_id);
    END IF;
    IF v_schema IS NULL OR v_schema NOT LIKE 'tenant\_%' THEN
        RAISE EXCEPTION 'drop_tenant_schema: schema inválido pra tenant % (%)', p_tenant_id, v_schema;
    END IF;
    IF NOT EXISTS (SELECT 1 FROM pg_namespace WHERE nspname = v_schema) THEN
        RAISE EXCEPTION 'drop_tenant_schema: schema % não existe', v_schema;
    END IF;
    DELETE FROM public.channel_routing WHERE tenant_id = p_tenant_id;
    DELETE FROM public.tenant_schemas WHERE tenant_id = p_tenant_id;
    EXECUTE format('DROP SCHEMA %I CASCADE', v_schema);
END;
$$;

-- Clone/drop são operações de provisionamento: só service_role (edge) e postgres
REVOKE ALL ON FUNCTION public.clone_tenant_template(uuid) FROM PUBLIC, anon, authenticated;
REVOKE ALL ON FUNCTION public.drop_tenant_schema(uuid) FROM PUBLIC, anon, authenticated;
GRANT EXECUTE ON FUNCTION public.clone_tenant_template(uuid) TO service_role;
GRANT EXECUTE ON FUNCTION public.drop_tenant_schema(uuid) TO service_role;

COMMIT;