DataOps-platform/deployment/migrations/versions/20260817_508_tenant_controlled_lifecycle.py

"""Claim-bound, fenced tenant lifecycle and legal-hold boundary."""

from alembic import op

revision = "20260817_508"
down_revision = "20260817_507"
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.execute(
        r"""
        CREATE TABLE public.tenant_legal_holds (
          hold_ref text PRIMARY KEY CHECK (hold_ref ~ '^[A-Za-z0-9._:-]{1,120}$'),
          tenant_id text NOT NULL REFERENCES public.tenants(tenant_id) ON DELETE RESTRICT,
          state text NOT NULL CHECK (state IN ('active','released')),
          evidence_digest char(64) NOT NULL,
          created_at timestamptz NOT NULL DEFAULT clock_timestamp(),
          released_at timestamptz
        );
        ALTER TABLE public.tenant_legal_holds OWNER TO dataops_tenant_foundation_owner;
        ALTER TABLE public.tenant_legal_holds ENABLE ROW LEVEL SECURITY;
        ALTER TABLE public.tenant_legal_holds FORCE ROW LEVEL SECURITY;
        CREATE POLICY tenant_legal_hold_scope ON public.tenant_legal_holds
          USING (tenant_id=current_setting('dataops.tenant_id',true))
          WITH CHECK (tenant_id=current_setting('dataops.tenant_id',true));
        REVOKE ALL ON public.tenant_legal_holds FROM PUBLIC,dataops_app_runtime,dataops_tenant_control;

        CREATE FUNCTION public.tenant_control_lifecycle_mutate(p_claim_uid uuid,p_nonce uuid,p_request jsonb)
        RETURNS jsonb LANGUAGE plpgsql SECURITY DEFINER SET search_path=pg_catalog,public AS $tenant_lifecycle_control$
        DECLARE v_claim jsonb; v_tenant text; v_digest text; v_action text; v_target text; v_fence bigint;
        DECLARE v_tenant_row record; v_replay record;
        BEGIN
          IF jsonb_typeof(p_request)<>'object' OR NOT (p_request ?& ARRAY['operation','expected_fence','idempotency_key','approval_ref','backup_digest','retention_seconds'])
             OR p_request-ARRAY['operation','expected_fence','idempotency_key','approval_ref','backup_digest','retention_seconds']<>'{}'::jsonb THEN RAISE EXCEPTION 'tenant_payload_closed'; END IF;
          v_action:=p_request->>'operation';
          IF v_action NOT IN ('freeze','begin_recovery','activate','mark_deletion_candidate','delete','rollback')
             OR p_request->>'expected_fence' !~ '^[0-9]{1,18}$'
             OR p_request->>'idempotency_key' !~ '^[A-Za-z0-9._:-]{1,120}$'
             OR jsonb_typeof(p_request->'approval_ref') NOT IN ('string','null')
             OR jsonb_typeof(p_request->'backup_digest') NOT IN ('string','null')
             OR jsonb_typeof(p_request->'retention_seconds') NOT IN ('string','null') THEN RAISE EXCEPTION 'tenant_payload_invalid'; END IF;
          v_claim:=public.tenant_control_consume_claim(p_claim_uid,p_nonce,'lifecycle_transition',p_request);
          v_tenant:=v_claim->>'tenant_id'; v_digest:=v_claim->>'request_digest';
          PERFORM set_config('dataops.tenant_id',v_tenant,true);
          PERFORM pg_advisory_xact_lock(hashtext(v_tenant || '-lifecycle'));
          SELECT * INTO v_replay FROM public.tenant_lifecycle_events WHERE tenant_id=v_tenant AND idempotency_key=p_request->>'idempotency_key' FOR UPDATE;
          IF FOUND THEN
            IF v_replay.request_digest<>v_digest THEN RAISE EXCEPTION 'tenant_replay_conflict'; END IF;
            RETURN jsonb_build_object('tenant_id',v_tenant,'lease_fence',v_replay.lease_fence,'state',v_replay.next_state);
          END IF;
          SELECT * INTO v_tenant_row FROM public.tenants WHERE tenant_id=v_tenant FOR UPDATE;
          IF NOT FOUND OR v_tenant_row.lease_fence<>(p_request->>'expected_fence')::bigint THEN RAISE EXCEPTION 'tenant_fence_conflict'; END IF;
          IF v_action IN ('begin_recovery','activate') AND coalesce(p_request->>'approval_ref','') !~ '^[A-Za-z0-9._:-]{1,120}$' THEN RAISE EXCEPTION 'tenant_recovery_approval_required'; END IF;
          IF v_action IN ('mark_deletion_candidate','delete') THEN
            IF coalesce(p_request->>'approval_ref','') !~ '^[A-Za-z0-9._:-]{1,120}$' THEN RAISE EXCEPTION 'tenant_deletion_approval_required'; END IF;
            IF coalesce(p_request->>'backup_digest','') !~ '^[0-9a-f]{64}$' THEN RAISE EXCEPTION 'tenant_backup_required'; END IF;
            IF coalesce(p_request->>'retention_seconds','') !~ '^[1-9][0-9]{0,12}$' THEN RAISE EXCEPTION 'tenant_retention_required'; END IF;
            IF EXISTS(SELECT 1 FROM public.tenant_legal_holds WHERE tenant_id=v_tenant AND state='active') THEN RAISE EXCEPTION 'tenant_hold_active'; END IF;
          END IF;
          v_target:=CASE
            WHEN v_tenant_row.state='active' AND v_action='freeze' THEN 'frozen'
            WHEN v_tenant_row.state='frozen' AND v_action='begin_recovery' THEN 'recovering'
            WHEN v_tenant_row.state='recovering' AND v_action='activate' THEN 'active'
            WHEN v_tenant_row.state IN ('active','frozen') AND v_action='mark_deletion_candidate' THEN 'deletion_candidate'
            WHEN v_tenant_row.state='deletion_candidate' AND v_action='delete' THEN 'deleted'
            WHEN v_tenant_row.state='deletion_candidate' AND v_action='rollback' THEN 'frozen'
            ELSE NULL END;
          IF v_target IS NULL THEN RAISE EXCEPTION 'tenant_transition_denied'; END IF;
          UPDATE public.tenants SET state=v_target,lease_fence=lease_fence+1,updated_at=clock_timestamp()
            WHERE tenant_id=v_tenant AND lease_fence=v_tenant_row.lease_fence RETURNING lease_fence INTO v_fence;
          IF NOT FOUND THEN RAISE EXCEPTION 'tenant_fence_conflict'; END IF;
          INSERT INTO public.tenant_lifecycle_events(tenant_id,idempotency_key,request_digest,prior_state,next_state,lease_fence,approval_ref,backup_digest,retention_seconds)
            VALUES(v_tenant,p_request->>'idempotency_key',v_digest,v_tenant_row.state,v_target,v_fence,nullif(p_request->>'approval_ref',''),nullif(p_request->>'backup_digest',''),nullif(p_request->>'retention_seconds','')::bigint);
          INSERT INTO public.tenant_audit_events(tenant_id,event_type,payload_digest,lease_fence) VALUES(v_tenant,'tenant.lifecycle',v_digest,v_fence);
          RETURN jsonb_build_object('tenant_id',v_tenant,'lease_fence',v_fence,'state',v_target);
        END; $tenant_lifecycle_control$;
        ALTER FUNCTION public.tenant_control_lifecycle_mutate(uuid,uuid,jsonb) OWNER TO dataops_tenant_foundation_owner;
        REVOKE ALL ON FUNCTION public.tenant_control_lifecycle_mutate(uuid,uuid,jsonb) FROM PUBLIC,dataops_app_runtime;
        GRANT EXECUTE ON FUNCTION public.tenant_control_lifecycle_mutate(uuid,uuid,jsonb) TO dataops_tenant_control;
        """
    )


def downgrade() -> None:
    op.execute(
        """
        DO $$ BEGIN
          IF EXISTS (SELECT 1 FROM public.tenant_legal_holds LIMIT 1)
             OR EXISTS (SELECT 1 FROM public.tenant_lifecycle_events WHERE next_state IN ('deletion_candidate','deleted')) THEN
            RAISE EXCEPTION 'tenant lifecycle downgrade refused while lifecycle evidence exists';
          END IF;
        END $$;
        REVOKE EXECUTE ON FUNCTION public.tenant_control_lifecycle_mutate(uuid,uuid,jsonb) FROM dataops_tenant_control;
        DROP FUNCTION public.tenant_control_lifecycle_mutate(uuid,uuid,jsonb);
        DROP TABLE public.tenant_legal_holds;
        """
    )