DataOps-platform/migrations/versions/20260817_502_tenant_lifecycle_gateway.py

"""Add a DB-clock, fenced lifecycle gateway without rewriting migration 501."""

from alembic import op

revision = "20260817_502"
down_revision = "20260817_501"
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.execute(
        r"""
        CREATE FUNCTION public.tenant_lifecycle_transition(p_payload jsonb)
        RETURNS jsonb LANGUAGE plpgsql SECURITY DEFINER SET search_path=pg_catalog,public AS $tenant_lifecycle$
        DECLARE v_tenant text; v_operation text; v_digest text; v_target text; v_fence bigint;
        DECLARE v_tenant_row record; v_replay record;
        BEGIN
          IF jsonb_typeof(p_payload)<>'object'
             OR NOT (p_payload ?& ARRAY['tenant_id','operation','expected_fence','idempotency_key','approval_ref'])
             OR p_payload-ARRAY['tenant_id','operation','expected_fence','idempotency_key','approval_ref']<>'{}'::jsonb THEN
            RAISE EXCEPTION 'tenant_payload_closed';
          END IF;
          v_tenant:=p_payload->>'tenant_id'; v_operation:=p_payload->>'operation';
          IF v_tenant !~ '^[a-z][a-z0-9-]{0,62}$' OR v_operation NOT IN ('freeze','begin_recovery','activate')
             OR p_payload->>'idempotency_key' !~ '^[A-Za-z0-9._:-]{1,120}$'
             OR p_payload->>'expected_fence' !~ '^[0-9]+$' THEN RAISE EXCEPTION 'tenant_payload_invalid'; END IF;
          IF v_operation IN ('begin_recovery','activate') AND coalesce(p_payload->>'approval_ref','') !~ '^[A-Za-z0-9._:-]{1,120}$' THEN
            RAISE EXCEPTION 'tenant_recovery_approval_required';
          END IF;
          v_digest:=encode(sha256(convert_to(p_payload::text,'utf8')),'hex');
          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_payload->>'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_payload->>'expected_fence')::bigint THEN RAISE EXCEPTION 'tenant_fence_conflict'; END IF;
          v_target:=CASE
            WHEN v_tenant_row.state='active' AND v_operation='freeze' THEN 'frozen'
            WHEN v_tenant_row.state='frozen' AND v_operation='begin_recovery' THEN 'recovering'
            WHEN v_tenant_row.state='recovering' AND v_operation='activate' THEN 'active'
            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)
            VALUES(v_tenant,p_payload->>'idempotency_key',v_digest,v_tenant_row.state,v_target,v_fence,nullif(p_payload->>'approval_ref',''));
          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$;
        ALTER FUNCTION public.tenant_lifecycle_transition(jsonb) OWNER TO dataops_tenant_foundation_owner;
        REVOKE ALL ON FUNCTION public.tenant_lifecycle_transition(jsonb) FROM PUBLIC;
        GRANT EXECUTE ON FUNCTION public.tenant_lifecycle_transition(jsonb) TO dataops_app_runtime;
        """
    )


def downgrade() -> None:
    op.execute(
        """
        REVOKE EXECUTE ON FUNCTION public.tenant_lifecycle_transition(jsonb) FROM dataops_app_runtime;
        DROP FUNCTION IF EXISTS public.tenant_lifecycle_transition(jsonb);
        """
    )