BEGIN;

-- A dispatch reservation is a durable lease, not proof that FCM accepted the
-- message. Existing pre-outbox rows are treated as completed so a rollout
-- never replays historical notifications unexpectedly.
ALTER TABLE public.push_dispatch_attempts
  ADD COLUMN status text NOT NULL DEFAULT 'pending',
  ADD COLUMN attempt_count integer NOT NULL DEFAULT 0,
  ADD COLUMN dispatch_lease_token uuid,
  ADD COLUMN lease_expires_at timestamptz,
  ADD COLUMN next_retry_at timestamptz,
  ADD COLUMN completed_at timestamptz,
  ADD COLUMN last_error_code text,
  ADD COLUMN updated_at timestamptz NOT NULL DEFAULT now();

UPDATE public.push_dispatch_attempts
SET status = 'succeeded',
    completed_at = created_at,
    updated_at = now()
WHERE status = 'pending';

ALTER TABLE public.push_dispatch_attempts
  ADD CONSTRAINT push_dispatch_status_check CHECK (
    status IN ('pending', 'processing', 'succeeded', 'partial', 'failed')
  ),
  ADD CONSTRAINT push_dispatch_attempt_count_check CHECK (attempt_count >= 0),
  ADD CONSTRAINT push_dispatch_error_code_check CHECK (
    last_error_code IS NULL OR last_error_code ~ '^[a-z0-9_]{1,64}$'
  ),
  ADD CONSTRAINT push_dispatch_lease_shape_check CHECK (
    (dispatch_lease_token IS NULL AND lease_expires_at IS NULL)
    OR (dispatch_lease_token IS NOT NULL AND lease_expires_at IS NOT NULL)
  );

CREATE INDEX idx_push_dispatch_due
  ON public.push_dispatch_attempts(next_retry_at, created_at)
  WHERE status IN ('pending', 'processing');

CREATE TABLE public.push_deliveries (
  id                   uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  attempt_id           uuid NOT NULL REFERENCES public.push_dispatch_attempts(id) ON DELETE CASCADE,
  push_token_id        uuid NOT NULL,
  target_user_id       uuid NOT NULL REFERENCES auth.users(id) ON DELETE CASCADE,
  provider             text NOT NULL CHECK (provider IN ('fcm', 'apns', 'webpush')),
  token_fingerprint    text NOT NULL CHECK (token_fingerprint ~ '^[0-9a-f]{64}$'),
  status               text NOT NULL DEFAULT 'pending' CHECK (
    status IN ('pending', 'leased', 'delivered', 'stale', 'retryable', 'permanent_failed')
  ),
  attempt_count        integer NOT NULL DEFAULT 0 CHECK (attempt_count >= 0),
  delivery_lease_token uuid,
  lease_expires_at     timestamptz,
  next_retry_at        timestamptz,
  last_error_code      text CHECK (
    last_error_code IS NULL OR last_error_code ~ '^[a-z0-9_]{1,64}$'
  ),
  completed_at         timestamptz,
  created_at           timestamptz NOT NULL DEFAULT now(),
  updated_at           timestamptz NOT NULL DEFAULT now(),
  UNIQUE (attempt_id, push_token_id),
  CHECK (
    (status = 'leased' AND delivery_lease_token IS NOT NULL AND lease_expires_at IS NOT NULL)
    OR (status <> 'leased' AND delivery_lease_token IS NULL AND lease_expires_at IS NULL)
  )
);

CREATE INDEX idx_push_deliveries_attempt_status
  ON public.push_deliveries(attempt_id, status, next_retry_at);

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

CREATE OR REPLACE FUNCTION public.push_dispatch_target_user(
  target_event_type text,
  target_resource_id uuid
)
RETURNS uuid
LANGUAGE plpgsql
STABLE
SECURITY DEFINER
SET search_path = ''
AS $$
DECLARE
  target_user_id uuid;
BEGIN
  IF auth.role() <> 'service_role' THEN
    RAISE EXCEPTION 'service_role_required' USING ERRCODE = '42501';
  END IF;
  CASE target_event_type
    WHEN 'transcription.completed' THEN
      SELECT history.user_id INTO target_user_id
      FROM public.history AS history
      WHERE history.id = target_resource_id AND history.status = 'completed';
    WHEN 'team.invite.created' THEN
      SELECT account.id INTO target_user_id
      FROM public.team_invites AS invite
      JOIN auth.users AS account ON lower(account.email) = lower(invite.email)
      WHERE invite.id = target_resource_id
        AND invite.accepted_at IS NULL
        AND invite.expires_at > now()
      ORDER BY account.created_at, account.id
      LIMIT 1;
    WHEN 'billing.status.changed' THEN
      SELECT subscription.user_id INTO target_user_id
      FROM public.subscriptions AS subscription
      WHERE subscription.id = target_resource_id;
    ELSE
      RAISE EXCEPTION 'invalid_push_event' USING ERRCODE = '22023';
  END CASE;
  RETURN target_user_id;
END;
$$;

CREATE OR REPLACE FUNCTION public.reserve_push_dispatch(
  push_event_type text,
  push_resource_id uuid
)
RETURNS jsonb
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = ''
AS $$
DECLARE
  current_user_id uuid := auth.uid();
  normalized_event_type text := lower(btrim(push_event_type));
  existing_attempt public.push_dispatch_attempts;
  created_attempt public.push_dispatch_attempts;
  new_lease_token uuid := gen_random_uuid();
BEGIN
  IF current_user_id IS NULL THEN
    RAISE EXCEPTION 'authentication_required' USING ERRCODE = '42501';
  END IF;
  IF push_resource_id IS NULL OR normalized_event_type NOT IN (
    'transcription.completed', 'team.invite.created', 'billing.status.changed'
  ) THEN
    RAISE EXCEPTION 'invalid_push_event' USING ERRCODE = '22023';
  END IF;

  -- RPC callers cannot poison another user's immutable event reservation.
  IF normalized_event_type = 'transcription.completed' AND NOT EXISTS (
    SELECT 1 FROM public.history
    WHERE id = push_resource_id AND user_id = current_user_id AND status = 'completed'
  ) THEN
    RAISE EXCEPTION 'push_resource_unavailable' USING ERRCODE = 'P0002';
  ELSIF normalized_event_type = 'team.invite.created' AND NOT EXISTS (
    SELECT 1 FROM public.team_invites
    WHERE id = push_resource_id AND invited_by = current_user_id
      AND accepted_at IS NULL AND expires_at > now()
  ) THEN
    RAISE EXCEPTION 'push_resource_unavailable' USING ERRCODE = 'P0002';
  ELSIF normalized_event_type = 'billing.status.changed' AND NOT EXISTS (
    SELECT 1 FROM public.subscriptions
    WHERE id = push_resource_id AND user_id = current_user_id
  ) THEN
    RAISE EXCEPTION 'push_resource_unavailable' USING ERRCODE = 'P0002';
  END IF;

  PERFORM pg_advisory_xact_lock(hashtextextended(current_user_id::text, 73048));
  PERFORM pg_advisory_xact_lock(
    hashtextextended(normalized_event_type || ':' || push_resource_id::text, 73051)
  );

  SELECT attempt.* INTO existing_attempt
  FROM public.push_dispatch_attempts AS attempt
  WHERE attempt.event_type = normalized_event_type
    AND attempt.resource_id = push_resource_id
    AND (
      normalized_event_type IN ('transcription.completed', 'team.invite.created')
      OR attempt.status IN ('pending', 'processing')
      OR attempt.created_at > now() - interval '30 seconds'
    )
  ORDER BY
    CASE WHEN attempt.status IN ('pending', 'processing') THEN 0 ELSE 1 END,
    attempt.created_at DESC
  LIMIT 1
  FOR UPDATE;

  IF existing_attempt.id IS NOT NULL THEN
    IF existing_attempt.status IN ('succeeded', 'partial', 'failed') THEN
      RETURN jsonb_build_object(
        'reserved', false, 'duplicate', true,
        'attempt_id', existing_attempt.id, 'status', existing_attempt.status
      );
    END IF;
    IF existing_attempt.lease_expires_at IS NOT NULL
       AND existing_attempt.lease_expires_at > now() THEN
      RETURN jsonb_build_object(
        'reserved', false, 'duplicate', true,
        'attempt_id', existing_attempt.id, 'status', 'processing'
      );
    END IF;
    IF existing_attempt.next_retry_at IS NOT NULL
       AND existing_attempt.next_retry_at > now() THEN
      RETURN jsonb_build_object(
        'reserved', false, 'duplicate', true,
        'attempt_id', existing_attempt.id, 'status', 'pending',
        'next_retry_at', existing_attempt.next_retry_at
      );
    END IF;
    UPDATE public.push_dispatch_attempts
    SET status = 'processing',
        attempt_count = attempt_count + 1,
        dispatch_lease_token = new_lease_token,
        lease_expires_at = now() + interval '2 minutes',
        next_retry_at = NULL,
        completed_at = NULL,
        last_error_code = NULL,
        updated_at = now()
    WHERE id = existing_attempt.id
    RETURNING * INTO existing_attempt;
    RETURN jsonb_build_object(
      'reserved', true, 'duplicate', false,
      'attempt_id', existing_attempt.id, 'status', existing_attempt.status,
      'dispatch_lease_token', new_lease_token
    );
  END IF;

  IF (
    SELECT count(*) FROM public.push_dispatch_attempts AS attempt
    WHERE attempt.actor_kind = 'user' AND attempt.caller_id = current_user_id
      AND attempt.created_at > now() - interval '1 minute'
  ) >= 10 OR (
    SELECT count(*) FROM public.push_dispatch_attempts AS attempt
    WHERE attempt.actor_kind = 'user' AND attempt.caller_id = current_user_id
      AND attempt.created_at > now() - interval '24 hours'
  ) >= 100 THEN
    RAISE EXCEPTION 'push_rate_limited' USING ERRCODE = '54000';
  END IF;

  INSERT INTO public.push_dispatch_attempts (
    caller_id, actor_kind, event_type, resource_id, status, attempt_count,
    dispatch_lease_token, lease_expires_at
  ) VALUES (
    current_user_id, 'user', normalized_event_type, push_resource_id,
    'processing', 1, new_lease_token, now() + interval '2 minutes'
  ) RETURNING * INTO created_attempt;
  RETURN jsonb_build_object(
    'reserved', true, 'duplicate', false,
    'attempt_id', created_attempt.id, 'status', created_attempt.status,
    'dispatch_lease_token', new_lease_token
  );
END;
$$;

CREATE OR REPLACE FUNCTION public.reserve_system_push_dispatch(
  push_event_type text,
  push_resource_id uuid
)
RETURNS jsonb
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = ''
AS $$
DECLARE
  normalized_event_type text := lower(btrim(push_event_type));
  existing_attempt public.push_dispatch_attempts;
  created_attempt public.push_dispatch_attempts;
  new_lease_token uuid := gen_random_uuid();
  target_user_id uuid;
BEGIN
  IF auth.role() <> 'service_role' THEN
    RAISE EXCEPTION 'service_role_required' USING ERRCODE = '42501';
  END IF;
  IF push_resource_id IS NULL OR normalized_event_type NOT IN (
    'transcription.completed', 'team.invite.created', 'billing.status.changed'
  ) THEN
    RAISE EXCEPTION 'invalid_push_event' USING ERRCODE = '22023';
  END IF;
  target_user_id := public.push_dispatch_target_user(normalized_event_type, push_resource_id);
  IF target_user_id IS NULL THEN
    RAISE EXCEPTION 'push_resource_unavailable' USING ERRCODE = 'P0002';
  END IF;

  PERFORM pg_advisory_xact_lock(hashtextextended('system-push-global', 73050));
  PERFORM pg_advisory_xact_lock(
    hashtextextended(normalized_event_type || ':' || push_resource_id::text, 73051)
  );

  SELECT attempt.* INTO existing_attempt
  FROM public.push_dispatch_attempts AS attempt
  WHERE attempt.event_type = normalized_event_type
    AND attempt.resource_id = push_resource_id
    AND (
      normalized_event_type IN ('transcription.completed', 'team.invite.created')
      OR attempt.status IN ('pending', 'processing')
      OR attempt.created_at > now() - interval '30 seconds'
    )
  ORDER BY
    CASE WHEN attempt.status IN ('pending', 'processing') THEN 0 ELSE 1 END,
    attempt.created_at DESC
  LIMIT 1
  FOR UPDATE;

  IF existing_attempt.id IS NOT NULL THEN
    IF existing_attempt.status IN ('succeeded', 'partial', 'failed') THEN
      RETURN jsonb_build_object(
        'reserved', false, 'duplicate', true,
        'attempt_id', existing_attempt.id, 'status', existing_attempt.status
      );
    END IF;
    IF existing_attempt.lease_expires_at IS NOT NULL
       AND existing_attempt.lease_expires_at > now() THEN
      RETURN jsonb_build_object(
        'reserved', false, 'duplicate', true,
        'attempt_id', existing_attempt.id, 'status', 'processing'
      );
    END IF;
    IF existing_attempt.next_retry_at IS NOT NULL
       AND existing_attempt.next_retry_at > now() THEN
      RETURN jsonb_build_object(
        'reserved', false, 'duplicate', true,
        'attempt_id', existing_attempt.id, 'status', 'pending',
        'next_retry_at', existing_attempt.next_retry_at
      );
    END IF;
    UPDATE public.push_dispatch_attempts
    SET status = 'processing',
        attempt_count = attempt_count + 1,
        dispatch_lease_token = new_lease_token,
        lease_expires_at = now() + interval '2 minutes',
        next_retry_at = NULL,
        completed_at = NULL,
        last_error_code = NULL,
        updated_at = now()
    WHERE id = existing_attempt.id
    RETURNING * INTO existing_attempt;
    RETURN jsonb_build_object(
      'reserved', true, 'duplicate', false,
      'attempt_id', existing_attempt.id, 'status', existing_attempt.status,
      'dispatch_lease_token', new_lease_token
    );
  END IF;

  IF (
    SELECT count(*) FROM public.push_dispatch_attempts AS attempt
    WHERE attempt.actor_kind = 'system'
      AND attempt.created_at > now() - interval '1 minute'
  ) >= 500 OR (
    SELECT count(*) FROM public.push_dispatch_attempts AS attempt
    WHERE attempt.actor_kind = 'system'
      AND attempt.created_at > now() - interval '24 hours'
  ) >= 10000 THEN
    RAISE EXCEPTION 'push_rate_limited' USING ERRCODE = '54000';
  END IF;

  INSERT INTO public.push_dispatch_attempts (
    caller_id, actor_kind, event_type, resource_id, status, attempt_count,
    dispatch_lease_token, lease_expires_at
  ) VALUES (
    NULL, 'system', normalized_event_type, push_resource_id,
    'processing', 1, new_lease_token, now() + interval '2 minutes'
  ) RETURNING * INTO created_attempt;
  RETURN jsonb_build_object(
    'reserved', true, 'duplicate', false,
    'attempt_id', created_attempt.id, 'status', created_attempt.status,
    'dispatch_lease_token', new_lease_token
  );
END;
$$;

CREATE OR REPLACE FUNCTION public.lease_push_deliveries(
  target_attempt_id uuid,
  target_dispatch_lease_token uuid
)
RETURNS TABLE (
  delivery_id uuid,
  delivery_lease_token uuid,
  push_token_id uuid,
  provider text,
  registration_id text,
  last_registered_at timestamptz
)
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = ''
AS $$
DECLARE
  dispatch public.push_dispatch_attempts;
  resolved_target_user_id uuid;
BEGIN
  IF auth.role() <> 'service_role' THEN
    RAISE EXCEPTION 'service_role_required' USING ERRCODE = '42501';
  END IF;
  SELECT * INTO dispatch FROM public.push_dispatch_attempts
  WHERE id = target_attempt_id FOR UPDATE;
  IF dispatch.id IS NULL THEN
    RAISE EXCEPTION 'push_dispatch_not_found' USING ERRCODE = 'P0002';
  END IF;
  IF dispatch.status IN ('succeeded', 'partial', 'failed') THEN
    RETURN;
  END IF;
  IF target_dispatch_lease_token IS NULL
     OR dispatch.dispatch_lease_token IS DISTINCT FROM target_dispatch_lease_token
     OR dispatch.lease_expires_at IS NULL
     OR dispatch.lease_expires_at <= now() THEN
    RAISE EXCEPTION 'push_dispatch_lease_conflict' USING ERRCODE = '40001';
  END IF;

  resolved_target_user_id := public.push_dispatch_target_user(dispatch.event_type, dispatch.resource_id);
  IF resolved_target_user_id IS NULL THEN
    UPDATE public.push_dispatch_attempts
    SET status = 'succeeded', completed_at = now(), last_error_code = 'no_targets',
        dispatch_lease_token = NULL, lease_expires_at = NULL,
        next_retry_at = NULL, updated_at = now()
    WHERE id = dispatch.id;
    RETURN;
  END IF;

  -- Refresh the delivery snapshot without duplicating raw registration ids.
  INSERT INTO public.push_deliveries (
    attempt_id, push_token_id, target_user_id, provider, token_fingerprint
  )
  SELECT
    dispatch.id,
    token.id,
    resolved_target_user_id,
    token.provider,
    encode(extensions.digest(token.token, 'sha256'), 'hex')
  FROM public.push_tokens AS token
  JOIN public.devices AS device
    ON device.id = token.device_id
   AND device.user_id = token.user_id
   AND device.revoked_at IS NULL
  WHERE token.user_id = resolved_target_user_id
    AND token.last_registered_at >= now() - interval '35 days'
  ON CONFLICT ON CONSTRAINT push_deliveries_attempt_id_push_token_id_key DO NOTHING;

  -- Missing, revoked, or old registrations are terminal neutral outcomes.
  UPDATE public.push_deliveries AS delivery
  SET status = 'stale', completed_at = now(), last_error_code = 'registration_stale',
      delivery_lease_token = NULL, lease_expires_at = NULL,
      next_retry_at = NULL, updated_at = now()
  WHERE delivery.attempt_id = dispatch.id
    AND delivery.status IN ('pending', 'leased', 'retryable')
    AND NOT EXISTS (
      SELECT 1
      FROM public.push_tokens AS token
      JOIN public.devices AS device
        ON device.id = token.device_id
       AND device.user_id = token.user_id
       AND device.revoked_at IS NULL
      WHERE token.id = delivery.push_token_id
        AND token.user_id = resolved_target_user_id
        AND token.last_registered_at >= now() - interval '35 days'
        AND encode(extensions.digest(token.token, 'sha256'), 'hex') = delivery.token_fingerprint
    );

  DELETE FROM public.push_tokens AS token
  WHERE token.user_id = resolved_target_user_id
    AND token.last_registered_at < now() - interval '35 days';

  UPDATE public.push_deliveries
  SET status = CASE WHEN attempt_count >= 5 THEN 'permanent_failed' ELSE 'retryable' END,
      next_retry_at = CASE
        WHEN attempt_count >= 5 THEN NULL
        ELSE now() + make_interval(secs => least(3600, (30 * power(2, greatest(0, attempt_count - 1)))::integer))
      END,
      completed_at = CASE WHEN attempt_count >= 5 THEN now() ELSE NULL END,
      last_error_code = CASE WHEN attempt_count >= 5 THEN 'delivery_attempts_exhausted' ELSE 'delivery_lease_expired' END,
      delivery_lease_token = NULL,
      lease_expires_at = NULL,
      updated_at = now()
  WHERE attempt_id = dispatch.id AND status = 'leased' AND lease_expires_at <= now();

  UPDATE public.push_deliveries
  SET status = 'permanent_failed', completed_at = now(), next_retry_at = NULL,
      last_error_code = 'delivery_attempts_exhausted', updated_at = now()
  WHERE attempt_id = dispatch.id AND status = 'retryable' AND attempt_count >= 5;

  IF NOT EXISTS (
    SELECT 1 FROM public.push_deliveries WHERE attempt_id = dispatch.id
  ) THEN
    UPDATE public.push_dispatch_attempts
    SET status = 'succeeded', completed_at = now(), last_error_code = 'no_targets',
        dispatch_lease_token = NULL, lease_expires_at = NULL,
        next_retry_at = NULL, updated_at = now()
    WHERE id = dispatch.id;
    RETURN;
  END IF;

  RETURN QUERY
  WITH candidates AS (
    SELECT delivery.id
    FROM public.push_deliveries AS delivery
    WHERE delivery.attempt_id = dispatch.id
      AND (
        delivery.status = 'pending'
        OR (delivery.status = 'retryable' AND coalesce(delivery.next_retry_at, now()) <= now())
      )
      AND delivery.attempt_count < 5
    ORDER BY delivery.created_at, delivery.id
    FOR UPDATE SKIP LOCKED
    LIMIT 100
  ), leased AS (
    UPDATE public.push_deliveries AS delivery
    SET status = 'leased',
        attempt_count = delivery.attempt_count + 1,
        delivery_lease_token = gen_random_uuid(),
        lease_expires_at = now() + interval '60 seconds',
        next_retry_at = NULL,
        completed_at = NULL,
        last_error_code = NULL,
        updated_at = now()
    FROM candidates
    WHERE delivery.id = candidates.id
    RETURNING delivery.*
  )
  SELECT
    leased.id,
    leased.delivery_lease_token,
    leased.push_token_id,
    leased.provider,
    token.token,
    token.last_registered_at
  FROM leased
  JOIN public.push_tokens AS token
    ON token.id = leased.push_token_id
   AND token.user_id = leased.target_user_id
   AND encode(extensions.digest(token.token, 'sha256'), 'hex') = leased.token_fingerprint
  ORDER BY leased.created_at, leased.id;
END;
$$;

CREATE OR REPLACE FUNCTION public.finalize_push_delivery(
  target_delivery_id uuid,
  target_delivery_lease_token uuid,
  delivery_outcome text,
  delivery_error_code text DEFAULT NULL
)
RETURNS jsonb
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = ''
AS $$
DECLARE
  delivery public.push_deliveries;
  normalized_outcome text := lower(btrim(delivery_outcome));
  normalized_error text := nullif(lower(btrim(delivery_error_code)), '');
  next_status text;
  retry_at timestamptz;
BEGIN
  IF auth.role() <> 'service_role' THEN
    RAISE EXCEPTION 'service_role_required' USING ERRCODE = '42501';
  END IF;
  IF normalized_outcome NOT IN (
    'delivered', 'stale', 'retryable_failure', 'permanent_failure'
  ) OR (normalized_error IS NOT NULL AND normalized_error !~ '^[a-z0-9_]{1,64}$') THEN
    RAISE EXCEPTION 'invalid_delivery_outcome' USING ERRCODE = '22023';
  END IF;
  IF normalized_outcome IN ('retryable_failure', 'permanent_failure')
     AND normalized_error IS NULL THEN
    RAISE EXCEPTION 'delivery_error_code_required' USING ERRCODE = '22023';
  END IF;

  SELECT * INTO delivery FROM public.push_deliveries
  WHERE id = target_delivery_id FOR UPDATE;
  IF delivery.id IS NULL THEN
    RAISE EXCEPTION 'push_delivery_not_found' USING ERRCODE = 'P0002';
  END IF;
  IF delivery.status <> 'leased'
     OR target_delivery_lease_token IS NULL
     OR delivery.delivery_lease_token IS DISTINCT FROM target_delivery_lease_token
     OR delivery.lease_expires_at IS NULL
     OR delivery.lease_expires_at <= now() THEN
    RAISE EXCEPTION 'push_delivery_lease_conflict' USING ERRCODE = '40001';
  END IF;

  IF normalized_outcome = 'delivered' THEN
    next_status := 'delivered';
  ELSIF normalized_outcome = 'stale' THEN
    next_status := 'stale';
    DELETE FROM public.push_tokens AS token
    WHERE token.id = delivery.push_token_id
      AND encode(extensions.digest(token.token, 'sha256'), 'hex') = delivery.token_fingerprint;
  ELSIF normalized_outcome = 'retryable_failure' AND delivery.attempt_count < 5 THEN
    next_status := 'retryable';
    retry_at := now() + make_interval(
      secs => least(3600, (30 * power(2, greatest(0, delivery.attempt_count - 1)))::integer)
    );
  ELSE
    next_status := 'permanent_failed';
  END IF;

  UPDATE public.push_deliveries
  SET status = next_status,
      next_retry_at = retry_at,
      last_error_code = CASE
        WHEN next_status = 'delivered' THEN NULL
        WHEN next_status = 'stale' THEN coalesce(normalized_error, 'registration_stale')
        ELSE normalized_error
      END,
      completed_at = CASE
        WHEN next_status IN ('delivered', 'stale', 'permanent_failed') THEN now()
        ELSE NULL
      END,
      delivery_lease_token = NULL,
      lease_expires_at = NULL,
      updated_at = now()
  WHERE id = delivery.id;

  RETURN jsonb_build_object(
    'delivery_id', delivery.id,
    'status', next_status,
    'next_retry_at', retry_at
  );
END;
$$;

CREATE OR REPLACE FUNCTION public.finalize_push_dispatch(
  target_attempt_id uuid,
  target_dispatch_lease_token uuid
)
RETURNS jsonb
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = ''
AS $$
DECLARE
  dispatch public.push_dispatch_attempts;
  delivered_count integer;
  stale_count integer;
  retryable_count integer;
  permanent_count integer;
  leased_count integer;
  pending_count integer;
  next_retry timestamptz;
  final_status text;
  is_complete boolean;
BEGIN
  IF auth.role() <> 'service_role' THEN
    RAISE EXCEPTION 'service_role_required' USING ERRCODE = '42501';
  END IF;
  SELECT * INTO dispatch FROM public.push_dispatch_attempts
  WHERE id = target_attempt_id FOR UPDATE;
  IF dispatch.id IS NULL THEN
    RAISE EXCEPTION 'push_dispatch_not_found' USING ERRCODE = 'P0002';
  END IF;
  IF dispatch.status IN ('succeeded', 'partial', 'failed') THEN
    RETURN jsonb_build_object(
      'attempt_id', dispatch.id, 'status', dispatch.status, 'complete', true
    );
  END IF;
  IF target_dispatch_lease_token IS NULL
     OR dispatch.dispatch_lease_token IS DISTINCT FROM target_dispatch_lease_token
     OR dispatch.lease_expires_at IS NULL
     OR dispatch.lease_expires_at <= now() THEN
    RAISE EXCEPTION 'push_dispatch_lease_conflict' USING ERRCODE = '40001';
  END IF;

  UPDATE public.push_deliveries
  SET status = CASE WHEN attempt_count >= 5 THEN 'permanent_failed' ELSE 'retryable' END,
      next_retry_at = CASE
        WHEN attempt_count >= 5 THEN NULL
        ELSE now() + make_interval(secs => least(3600, (30 * power(2, greatest(0, attempt_count - 1)))::integer))
      END,
      completed_at = CASE WHEN attempt_count >= 5 THEN now() ELSE NULL END,
      last_error_code = CASE WHEN attempt_count >= 5 THEN 'delivery_attempts_exhausted' ELSE 'delivery_lease_expired' END,
      delivery_lease_token = NULL, lease_expires_at = NULL, updated_at = now()
  WHERE attempt_id = dispatch.id AND status = 'leased' AND lease_expires_at <= now();

  SELECT
    count(*) FILTER (WHERE status = 'delivered'),
    count(*) FILTER (WHERE status = 'stale'),
    count(*) FILTER (WHERE status = 'retryable'),
    count(*) FILTER (WHERE status = 'permanent_failed'),
    count(*) FILTER (WHERE status = 'leased'),
    count(*) FILTER (WHERE status = 'pending'),
    min(CASE
      WHEN status = 'pending' THEN now()
      WHEN status = 'retryable' THEN coalesce(next_retry_at, now())
      WHEN status = 'leased' THEN lease_expires_at
    END)
  INTO delivered_count, stale_count, retryable_count, permanent_count,
       leased_count, pending_count, next_retry
  FROM public.push_deliveries
  WHERE attempt_id = dispatch.id;

  IF retryable_count + leased_count + pending_count > 0 THEN
    final_status := CASE WHEN leased_count > 0 THEN 'processing' ELSE 'pending' END;
    is_complete := false;
    UPDATE public.push_dispatch_attempts
    SET status = final_status,
        dispatch_lease_token = NULL,
        lease_expires_at = NULL,
        next_retry_at = coalesce(next_retry, now()),
        completed_at = NULL,
        last_error_code = CASE WHEN retryable_count > 0 THEN 'delivery_retry_pending' ELSE NULL END,
        updated_at = now()
    WHERE id = dispatch.id;
  ELSE
    is_complete := true;
    final_status := CASE
      WHEN delivered_count > 0 AND permanent_count = 0 THEN 'succeeded'
      WHEN delivered_count > 0 AND permanent_count > 0 THEN 'partial'
      WHEN delivered_count = 0 AND permanent_count > 0 THEN 'failed'
      ELSE 'succeeded'
    END;
    UPDATE public.push_dispatch_attempts
    SET status = final_status,
        dispatch_lease_token = NULL,
        lease_expires_at = NULL,
        next_retry_at = NULL,
        completed_at = now(),
        last_error_code = CASE
          WHEN final_status = 'partial' THEN 'delivery_partial_failure'
          WHEN final_status = 'failed' THEN 'delivery_failed'
          WHEN delivered_count = 0 THEN 'no_targets'
          ELSE NULL
        END,
        updated_at = now()
    WHERE id = dispatch.id;
  END IF;

  RETURN jsonb_build_object(
    'attempt_id', dispatch.id,
    'status', final_status,
    'complete', is_complete,
    'delivered', delivered_count,
    'stale', stale_count,
    'retryable_failed', retryable_count,
    'permanent_failed', permanent_count,
    'next_retry_at', next_retry
  );
END;
$$;

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 $$
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;

  RETURN QUERY
  WITH due AS (
    SELECT attempt.id
    FROM public.push_dispatch_attempts AS attempt
    WHERE (
      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 coalesce(attempt.next_retry_at, attempt.created_at), attempt.id
    FOR UPDATE SKIP LOCKED
    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.push_dispatch_target_user(text, uuid)
  FROM PUBLIC, anon, authenticated;
REVOKE ALL ON FUNCTION public.lease_push_deliveries(uuid, uuid)
  FROM PUBLIC, anon, authenticated;
REVOKE ALL ON FUNCTION public.finalize_push_delivery(uuid, uuid, text, text)
  FROM PUBLIC, anon, authenticated;
REVOKE ALL ON FUNCTION public.finalize_push_dispatch(uuid, uuid)
  FROM PUBLIC, anon, authenticated;
REVOKE ALL ON FUNCTION public.claim_due_push_dispatches(integer)
  FROM PUBLIC, anon, authenticated;
GRANT EXECUTE ON FUNCTION public.push_dispatch_target_user(text, uuid) TO service_role;
GRANT EXECUTE ON FUNCTION public.lease_push_deliveries(uuid, uuid) TO service_role;
GRANT EXECUTE ON FUNCTION public.finalize_push_delivery(uuid, uuid, text, text) TO service_role;
GRANT EXECUTE ON FUNCTION public.finalize_push_dispatch(uuid, uuid) TO service_role;
GRANT EXECUTE ON FUNCTION public.claim_due_push_dispatches(integer) TO service_role;

COMMIT;