DataOps-platform/migrations/versions/20260802_474_connector_version_guardrails.py

"""Continuously reject ambiguous active connector versions."""

from alembic import op

revision = "20260802_474"
down_revision = "20260802_473"
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.execute("""
    DO $$ BEGIN
      IF EXISTS (
        SELECT 1
          FROM public.connector_principals p
          JOIN (
            SELECT connector_id
              FROM public.connector_manifests
             WHERE status='active'
             GROUP BY connector_id
            HAVING COUNT(DISTINCT connector_version)>1
          ) ambiguous USING(connector_id)
      ) THEN
        RAISE EXCEPTION
          'connector 474 upgrade rejected: principal has multiple active manifest versions; provide an explicit principal-to-version mapping';
      END IF;
    END $$;
    CREATE OR REPLACE FUNCTION public.enforce_connector_active_version()
    RETURNS trigger LANGUAGE plpgsql AS $$
    DECLARE target_connector VARCHAR(64);
    BEGIN
      target_connector := COALESCE(NEW.connector_id, OLD.connector_id);
      IF EXISTS (SELECT 1 FROM public.connector_principals WHERE connector_id=target_connector)
         AND (SELECT COUNT(DISTINCT connector_version)
                FROM public.connector_manifests
               WHERE connector_id=target_connector AND status='active') > 1 THEN
        RAISE EXCEPTION
          'multiple active connector versions require an explicit one-active-version mapping';
      END IF;
      RETURN COALESCE(NEW, OLD);
    END $$;
    CREATE TRIGGER trg_connector_manifest_active_version
      AFTER INSERT OR UPDATE OF connector_id,connector_version,status
      ON public.connector_manifests
      FOR EACH ROW EXECUTE FUNCTION public.enforce_connector_active_version();
    CREATE TRIGGER trg_connector_principal_active_version
      AFTER INSERT OR UPDATE OF connector_id,connector_version
      ON public.connector_principals
      FOR EACH ROW EXECUTE FUNCTION public.enforce_connector_active_version();
    """)


def downgrade() -> None:
    op.execute("""
    DO $$ BEGIN
      IF EXISTS (
        SELECT 1
          FROM public.connector_principals
         GROUP BY connector_id,source_uid,business_domain_uid,environment
        HAVING COUNT(DISTINCT connector_version)>1
      ) THEN
        RAISE EXCEPTION
          'connector downgrade below 474 rejected: multiple principal versions share one legacy binding; consolidate explicitly before downgrade';
      END IF;
    END $$;
    DROP TRIGGER IF EXISTS trg_connector_principal_active_version ON public.connector_principals;
    DROP TRIGGER IF EXISTS trg_connector_manifest_active_version ON public.connector_manifests;
    DROP FUNCTION IF EXISTS public.enforce_connector_active_version();
    """)