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

"""Resolve a durable worker scope before entering FORCE RLS outbox rows."""

from alembic import op

revision = "20260817_514"
down_revision = "20260817_513"
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.execute(r"""
      CREATE TABLE public.tenant_control_outbox_scope_lookup (
        outbox_uid uuid PRIMARY KEY,
        tenant_id text NOT NULL REFERENCES public.tenants(tenant_id) ON DELETE RESTRICT,
        created_at timestamptz NOT NULL DEFAULT clock_timestamp()
      );
      ALTER TABLE public.tenant_control_outbox_scope_lookup OWNER TO dataops_tenant_foundation_owner;
      REVOKE ALL ON public.tenant_control_outbox_scope_lookup FROM PUBLIC,dataops_app_runtime,dataops_tenant_control;
      ALTER TABLE public.tenant_control_outbox NO FORCE ROW LEVEL SECURITY;
      INSERT INTO public.tenant_control_outbox_scope_lookup(outbox_uid,tenant_id)
        SELECT outbox_uid,tenant_id FROM public.tenant_control_outbox
        ON CONFLICT(outbox_uid) DO NOTHING;
      ALTER TABLE public.tenant_control_outbox FORCE ROW LEVEL SECURITY;

      CREATE OR REPLACE FUNCTION public.tenant_control_claim_outbox_enqueue()
      RETURNS trigger LANGUAGE plpgsql SECURITY DEFINER SET search_path=pg_catalog,public AS $tenant_outbox_enqueue$
      DECLARE v_outbox uuid;
      BEGIN
        v_outbox:=gen_random_uuid();
        INSERT INTO public.tenant_control_outbox_scope_lookup(outbox_uid,tenant_id) VALUES(v_outbox,NEW.tenant_id);
        PERFORM set_config('dataops.tenant_id',NEW.tenant_id,true);
        INSERT INTO public.tenant_control_outbox(outbox_uid,tenant_id,claim_uid,event_type,payload_digest)
          VALUES(v_outbox,NEW.tenant_id,NEW.claim_uid,
            CASE WHEN NEW.action='lifecycle_transition' THEN 'tenant.lifecycle.claim' ELSE 'tenant.quota.claim' END,
            NEW.request_digest);
        INSERT INTO public.tenant_audit_events(tenant_id,event_type,payload_digest,lease_fence)
          VALUES(NEW.tenant_id,'tenant.outbox.queued',NEW.request_digest,0);
        RETURN NEW;
      END; $tenant_outbox_enqueue$;
      ALTER FUNCTION public.tenant_control_claim_outbox_enqueue() OWNER TO dataops_tenant_foundation_owner;

      CREATE OR REPLACE FUNCTION public.tenant_control_worker_claim(p_payload jsonb)
      RETURNS jsonb LANGUAGE plpgsql SECURITY DEFINER SET search_path=pg_catalog,public AS $tenant_worker_claim$
      DECLARE v_outbox uuid; v_worker text; v_seconds integer; v_row public.tenant_control_outbox%ROWTYPE; v_tenant record; v_scope text;
      BEGIN
        IF NOT pg_has_role(session_user,'dataops_tenant_control','MEMBER') THEN RAISE EXCEPTION 'tenant_control_identity_required'; END IF;
        IF jsonb_typeof(p_payload)<>'object' OR NOT (p_payload ?& ARRAY['outbox_uid','worker_id','lease_seconds']) OR p_payload-ARRAY['outbox_uid','worker_id','lease_seconds']<>'{}'::jsonb THEN RAISE EXCEPTION 'tenant_payload_closed'; END IF;
        IF coalesce(p_payload->>'outbox_uid','') !~* '^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$'
          OR coalesce(p_payload->>'worker_id','') !~ '^[A-Za-z0-9._:-]{1,120}$'
          OR coalesce(p_payload->>'lease_seconds','') !~ '^[1-9][0-9]{0,3}$' THEN RAISE EXCEPTION 'tenant_payload_invalid'; END IF;
        v_outbox:=(p_payload->>'outbox_uid')::uuid; v_worker:=p_payload->>'worker_id'; v_seconds:=(p_payload->>'lease_seconds')::integer;
        SELECT tenant_id INTO v_scope FROM public.tenant_control_outbox_scope_lookup WHERE outbox_uid=v_outbox;
        IF NOT FOUND THEN RAISE EXCEPTION 'tenant_outbox_missing'; END IF;
        PERFORM set_config('dataops.tenant_id',v_scope,true);
        SELECT * INTO v_row FROM public.tenant_control_outbox WHERE outbox_uid=v_outbox FOR UPDATE;
        IF NOT FOUND OR v_row.tenant_id<>v_scope THEN RAISE EXCEPTION 'tenant_outbox_missing'; END IF;
        SELECT state,lease_fence INTO v_tenant FROM public.tenants WHERE tenant_id=v_row.tenant_id FOR UPDATE;
        IF NOT FOUND OR v_tenant.state<>'active' THEN RAISE EXCEPTION 'tenant_worker_scope_denied'; END IF;
        IF v_row.state='completed' THEN RAISE EXCEPTION 'tenant_outbox_replayed'; END IF;
        IF v_row.state='leased' AND v_row.lease_expires_at>clock_timestamp() THEN RAISE EXCEPTION 'tenant_outbox_leased'; END IF;
        UPDATE public.tenant_control_outbox SET state='leased',worker_id=v_worker,leased_at=clock_timestamp(),lease_expires_at=clock_timestamp()+make_interval(secs=>v_seconds),lease_fence=lease_fence+1
          WHERE outbox_uid=v_outbox RETURNING * INTO v_row;
        INSERT INTO public.tenant_audit_events(tenant_id,event_type,payload_digest,lease_fence)
          VALUES(v_row.tenant_id,'tenant.outbox.leased',v_row.payload_digest,v_row.lease_fence);
        RETURN jsonb_build_object('outbox_uid',v_row.outbox_uid,'tenant_id',v_row.tenant_id,'event_type',v_row.event_type,'payload_digest',v_row.payload_digest,'lease_fence',v_row.lease_fence,'lease_expires_at',v_row.lease_expires_at);
      END; $tenant_worker_claim$;
      ALTER FUNCTION public.tenant_control_worker_claim(jsonb) OWNER TO dataops_tenant_foundation_owner;
      REVOKE ALL ON FUNCTION public.tenant_control_worker_claim(jsonb) FROM PUBLIC,dataops_app_runtime;
      GRANT EXECUTE ON FUNCTION public.tenant_control_worker_claim(jsonb) TO dataops_tenant_control;
    """)


def downgrade() -> None:
    op.execute("""
      DO $$ BEGIN
        IF EXISTS (SELECT 1 FROM public.tenant_control_outbox_scope_lookup LIMIT 1) THEN RAISE EXCEPTION 'tenant outbox scope downgrade refused while work exists'; END IF;
      END $$;
      DROP TABLE public.tenant_control_outbox_scope_lookup;
    """)