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

"""Enterprise connector SDK, machine identity, execution and graph ledger."""

from alembic import op

revision = "20260802_472"
down_revision = "20260802_471"
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.execute("""
    CREATE TABLE public.connector_manifests (
      uid UUID PRIMARY KEY, connector_id VARCHAR(64) NOT NULL, connector_version VARCHAR(40) NOT NULL,
      sdk_version VARCHAR(20) NOT NULL, display_name VARCHAR(200) NOT NULL,
      capabilities JSONB NOT NULL, config_schema JSONB NOT NULL, status VARCHAR(20) NOT NULL,
      created_by UUID NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
      updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
      CONSTRAINT uq_connector_manifest_version UNIQUE (connector_id, connector_version),
      CONSTRAINT ck_connector_manifest_status CHECK (status IN ('active','retired')),
      CONSTRAINT ck_connector_manifest_capabilities_array CHECK (jsonb_typeof(capabilities)='array'),
      CONSTRAINT ck_connector_manifest_schema_object CHECK (jsonb_typeof(config_schema)='object')
    );
    CREATE TABLE public.connector_principals (
      uid UUID PRIMARY KEY, connector_id VARCHAR(64) NOT NULL, source_uid UUID NOT NULL,
      business_domain_uid UUID NOT NULL, environment VARCHAR(20) NOT NULL,
      allowed_operations TEXT[] NOT NULL, allowed_scopes JSONB NOT NULL,
      status VARCHAR(20) NOT NULL, created_by UUID NOT NULL,
      created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, revoked_at TIMESTAMPTZ,
      CONSTRAINT uq_connector_principal_binding UNIQUE (connector_id, source_uid, business_domain_uid, environment),
      CONSTRAINT ck_connector_principal_environment CHECK (environment IN ('development','staging','production')),
      CONSTRAINT ck_connector_principal_status CHECK (status IN ('active','revoked')),
      CONSTRAINT ck_connector_principal_scope CHECK (jsonb_typeof(allowed_scopes)='object')
    );
    CREATE TABLE public.connector_machine_credentials (
      uid UUID PRIMARY KEY, principal_uid UUID NOT NULL REFERENCES public.connector_principals(uid),
      token_hash CHAR(64) NOT NULL UNIQUE, status VARCHAR(20) NOT NULL,
      issued_by UUID NOT NULL, issued_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
      expires_at TIMESTAMPTZ NOT NULL, first_used_at TIMESTAMPTZ, use_count INTEGER NOT NULL DEFAULT 0,
      revoked_at TIMESTAMPTZ, rotated_from_uid UUID REFERENCES public.connector_machine_credentials(uid),
      CONSTRAINT ck_connector_credential_status CHECK (status IN ('active','revoked','rotated','expired','replayed')),
      CONSTRAINT ck_connector_credential_ttl CHECK (expires_at <= issued_at + INTERVAL '15 minutes'),
      CONSTRAINT ck_connector_credential_use_count CHECK (use_count >= 0)
    );
    CREATE INDEX ix_connector_credential_principal_status ON public.connector_machine_credentials(principal_uid,status,expires_at);
    CREATE TABLE public.connector_runs (
      uid UUID PRIMARY KEY, idempotency_key CHAR(64) NOT NULL UNIQUE,
      connector_id VARCHAR(64) NOT NULL, connector_version VARCHAR(40) NOT NULL,
      source_uid UUID NOT NULL, operation VARCHAR(30) NOT NULL, status VARCHAR(20) NOT NULL,
      attempt_count INTEGER NOT NULL DEFAULT 0, checkpoint JSONB NOT NULL DEFAULT '{}'::jsonb,
      cursor JSONB NOT NULL DEFAULT '{}'::jsonb, error_category VARCHAR(30), error_code VARCHAR(80),
      resumed_from_run_uid UUID REFERENCES public.connector_runs(uid), dry_run BOOLEAN NOT NULL DEFAULT FALSE,
      actor_uid UUID NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
      updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
      CONSTRAINT fk_connector_run_manifest FOREIGN KEY (connector_id,connector_version)
        REFERENCES public.connector_manifests(connector_id,connector_version),
      CONSTRAINT ck_connector_run_operation CHECK (operation IN ('discover','snapshot','incremental','lineage','profile','cancel','resume','evidence')),
      CONSTRAINT ck_connector_run_status CHECK (status IN ('running','succeeded','dry_run','failed','cancelled','resumable')),
      CONSTRAINT ck_connector_run_attempt CHECK (attempt_count BETWEEN 0 AND 5)
    );
    CREATE INDEX ix_connector_runs_source_created ON public.connector_runs(source_uid,created_at DESC);
    CREATE TABLE public.connector_run_attempts (
      uid UUID PRIMARY KEY, run_uid UUID NOT NULL REFERENCES public.connector_runs(uid) ON DELETE CASCADE,
      attempt_number INTEGER NOT NULL, status VARCHAR(20) NOT NULL, error_category VARCHAR(30),
      started_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, finished_at TIMESTAMPTZ,
      CONSTRAINT uq_connector_run_attempt UNIQUE(run_uid,attempt_number)
    );
    CREATE TABLE public.connector_checkpoints (
      uid UUID PRIMARY KEY, run_uid UUID NOT NULL REFERENCES public.connector_runs(uid) ON DELETE CASCADE,
      sequence_number INTEGER NOT NULL, cursor JSONB NOT NULL, checkpoint JSONB NOT NULL,
      content_hash CHAR(64) NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
      CONSTRAINT uq_connector_checkpoint_sequence UNIQUE(run_uid,sequence_number)
    );
    CREATE TABLE public.connector_evidence (
      uid UUID PRIMARY KEY, run_uid UUID NOT NULL REFERENCES public.connector_runs(uid) ON DELETE CASCADE,
      evidence_type VARCHAR(40) NOT NULL, payload JSONB NOT NULL, content_hash CHAR(64) NOT NULL,
      byte_size INTEGER NOT NULL, redacted BOOLEAN NOT NULL,
      created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
      CONSTRAINT ck_connector_evidence_size CHECK (byte_size BETWEEN 0 AND 32768)
    );
    CREATE TABLE public.connector_graph_edges (
      uid UUID PRIMARY KEY, source_uid UUID NOT NULL, from_type VARCHAR(30) NOT NULL,
      from_key VARCHAR(500) NOT NULL, relation_type VARCHAR(40) NOT NULL,
      to_type VARCHAR(30) NOT NULL, to_key VARCHAR(500) NOT NULL,
      business_domain_uid UUID, run_uid UUID REFERENCES public.connector_runs(uid),
      run_status VARCHAR(20), evidence JSONB NOT NULL DEFAULT '{}'::jsonb,
      created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
      CONSTRAINT uq_connector_graph_edge UNIQUE(source_uid,from_type,from_key,relation_type,to_type,to_key),
      CONSTRAINT ck_connector_graph_node_types CHECK (from_type IN ('source','asset','process','business_domain','run') AND to_type IN ('source','asset','process','business_domain','run'))
    );
    CREATE INDEX ix_connector_graph_source ON public.connector_graph_edges(source_uid,relation_type);
    CREATE INDEX ix_connector_graph_to ON public.connector_graph_edges(to_type,to_key);
    CREATE TABLE public.connector_audit_events (
      uid UUID PRIMARY KEY, principal_uid UUID REFERENCES public.connector_principals(uid),
      credential_uid UUID REFERENCES public.connector_machine_credentials(uid), event_type VARCHAR(80) NOT NULL,
      actor_uid UUID, success BOOLEAN NOT NULL, safe_detail VARCHAR(500) NOT NULL,
      created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
    );
    CREATE INDEX ix_connector_audit_principal_created ON public.connector_audit_events(principal_uid,created_at DESC);
    """)


def downgrade() -> None:
    op.execute("""
    DROP TABLE IF EXISTS public.connector_audit_events;
    DROP TABLE IF EXISTS public.connector_graph_edges;
    DROP TABLE IF EXISTS public.connector_evidence;
    DROP TABLE IF EXISTS public.connector_checkpoints;
    DROP TABLE IF EXISTS public.connector_run_attempts;
    DROP TABLE IF EXISTS public.connector_runs;
    DROP TABLE IF EXISTS public.connector_machine_credentials;
    DROP TABLE IF EXISTS public.connector_principals;
    DROP TABLE IF EXISTS public.connector_manifests;
    """)