BEGIN;

-- Transactional producers enqueue pending rows even during provider bursts so
-- business state never rolls back merely because notification capacity is
-- exhausted. Rate limiting therefore belongs to the system dispatch lease,
-- immediately before a worker can read registrations or contact FCM.
CREATE TABLE public.push_dispatch_claim_events (
  id                   bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  attempt_id           uuid NOT NULL,
  dispatch_lease_token uuid NOT NULL UNIQUE,
  claimed_at           timestamptz NOT NULL DEFAULT now()
);

CREATE INDEX idx_push_dispatch_claim_events_claimed_at
  ON public.push_dispatch_claim_events(claimed_at);

ALTER TABLE public.push_dispatch_claim_events ENABLE ROW LEVEL SECURITY;
REVOKE ALL ON TABLE public.push_dispatch_claim_events
  FROM PUBLIC, anon, authenticated;
GRANT ALL ON TABLE public.push_dispatch_claim_events TO service_role;

CREATE OR REPLACE FUNCTION public.enforce_system_push_claim_rate()
RETURNS trigger
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = ''
AS $$
BEGIN
  IF NEW.actor_kind <> 'system'
     OR NEW.status <> 'processing'
     OR NEW.dispatch_lease_token IS NULL THEN
    RETURN NEW;
  END IF;
  IF TG_OP = 'UPDATE'
     AND OLD.dispatch_lease_token IS NOT DISTINCT FROM NEW.dispatch_lease_token THEN
    RETURN NEW;
  END IF;

  -- The same lock is acquired by reserve_system_push_dispatch and the due
  -- worker. Reusing it keeps direct dispatch and drain claims in one budget.
  PERFORM pg_advisory_xact_lock(hashtextextended('system-push-global', 73050));

  DELETE FROM public.push_dispatch_claim_events
  WHERE claimed_at < now() - interval '2 days';

  IF (
    SELECT count(*)
    FROM public.push_dispatch_claim_events
    WHERE claimed_at > now() - interval '1 minute'
  ) >= 500 OR (
    SELECT count(*)
    FROM public.push_dispatch_claim_events
    WHERE claimed_at > now() - interval '24 hours'
  ) >= 10000 THEN
    RAISE EXCEPTION 'push_rate_limited' USING ERRCODE = '54000';
  END IF;

  INSERT INTO public.push_dispatch_claim_events (
    attempt_id, dispatch_lease_token
  ) VALUES (
    NEW.id, NEW.dispatch_lease_token
  );
  RETURN NEW;
END;
$$;

CREATE TRIGGER enforce_system_push_claim_rate_on_insert
  BEFORE INSERT ON public.push_dispatch_attempts
  FOR EACH ROW EXECUTE FUNCTION public.enforce_system_push_claim_rate();

CREATE TRIGGER enforce_system_push_claim_rate_on_update
  BEFORE UPDATE OF status, dispatch_lease_token ON public.push_dispatch_attempts
  FOR EACH ROW EXECUTE FUNCTION public.enforce_system_push_claim_rate();

CREATE OR REPLACE FUNCTION public.claim_due_push_dispatches(batch_limit integer DEFAULT 50)
RETURNS TABLE (
  attempt_id uuid,
  event_type text,
  resource_id uuid,
  dispatch_lease_token uuid
)
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = ''
AS $$
DECLARE
  minute_slots integer;
  daily_slots integer;
  system_slots integer;
BEGIN
  IF auth.role() <> 'service_role' THEN
    RAISE EXCEPTION 'service_role_required' USING ERRCODE = '42501';
  END IF;
  IF batch_limit IS NULL OR batch_limit NOT BETWEEN 1 AND 100 THEN
    RAISE EXCEPTION 'invalid_batch_limit' USING ERRCODE = '22023';
  END IF;

  -- Keep capacity calculation, row claims and claim-ledger inserts in the same
  -- transaction. At capacity, system rows are not selected or mutated, so
  -- their pending status and next_retry_at remain intact for a later drain.
  PERFORM pg_advisory_xact_lock(hashtextextended('system-push-global', 73050));
  DELETE FROM public.push_dispatch_claim_events
  WHERE claimed_at < now() - interval '2 days';

  SELECT greatest(0, 500 - count(*))::integer
  INTO minute_slots
  FROM public.push_dispatch_claim_events
  WHERE claimed_at > now() - interval '1 minute';

  SELECT greatest(0, 10000 - count(*))::integer
  INTO daily_slots
  FROM public.push_dispatch_claim_events
  WHERE claimed_at > now() - interval '24 hours';

  system_slots := least(batch_limit, minute_slots, daily_slots);

  RETURN QUERY
  WITH user_due AS (
    SELECT
      attempt.id,
      coalesce(attempt.next_retry_at, attempt.created_at) AS due_at
    FROM public.push_dispatch_attempts AS attempt
    WHERE attempt.actor_kind = 'user'
      AND (
        (attempt.status = 'pending'
          AND coalesce(attempt.next_retry_at, attempt.created_at) <= now())
        OR (attempt.status = 'processing' AND attempt.lease_expires_at <= now())
      )
    ORDER BY due_at, attempt.id
    FOR UPDATE SKIP LOCKED
    LIMIT batch_limit
  ), system_due AS (
    SELECT
      attempt.id,
      coalesce(attempt.next_retry_at, attempt.created_at) AS due_at
    FROM public.push_dispatch_attempts AS attempt
    WHERE attempt.actor_kind = 'system'
      AND (
        (attempt.status = 'pending'
          AND coalesce(attempt.next_retry_at, attempt.created_at) <= now())
        OR (attempt.status = 'processing' AND attempt.lease_expires_at <= now())
      )
    ORDER BY due_at, attempt.id
    FOR UPDATE SKIP LOCKED
    LIMIT system_slots
  ), due AS (
    SELECT candidate.id
    FROM (
      SELECT * FROM user_due
      UNION ALL
      SELECT * FROM system_due
    ) AS candidate
    ORDER BY candidate.due_at, candidate.id
    LIMIT batch_limit
  ), claimed AS (
    UPDATE public.push_dispatch_attempts AS attempt
    SET status = 'processing',
        attempt_count = attempt.attempt_count + 1,
        dispatch_lease_token = gen_random_uuid(),
        lease_expires_at = now() + interval '2 minutes',
        next_retry_at = NULL,
        updated_at = now()
    FROM due
    WHERE attempt.id = due.id
    RETURNING attempt.*
  )
  SELECT claimed.id, claimed.event_type, claimed.resource_id,
         claimed.dispatch_lease_token
  FROM claimed
  ORDER BY claimed.created_at, claimed.id;
END;
$$;

REVOKE ALL ON FUNCTION public.enforce_system_push_claim_rate()
  FROM PUBLIC, anon, authenticated;
REVOKE ALL ON FUNCTION public.claim_due_push_dispatches(integer)
  FROM PUBLIC, anon, authenticated;
GRANT EXECUTE ON FUNCTION public.claim_due_push_dispatches(integer)
  TO service_role;

COMMIT;