\set ON_ERROR_STOP on

BEGIN;

CREATE OR REPLACE FUNCTION pg_temp.assert_true(condition boolean, message text)
RETURNS void
LANGUAGE plpgsql
AS $$
BEGIN
  IF condition IS NOT TRUE THEN
    RAISE EXCEPTION 'assertion_failed: %', message;
  END IF;
END;
$$;

INSERT INTO auth.users (
  id, aud, role, email, encrypted_password, email_confirmed_at,
  raw_app_meta_data, raw_user_meta_data, created_at, updated_at
) VALUES
  (
    '61000000-0000-4000-8000-000000000001', 'authenticated', 'authenticated',
    'push-outbox-one@example.invalid', crypt('fixture-password', gen_salt('bf')), now(),
    '{"provider":"email","providers":["email"]}'::jsonb, '{}'::jsonb, now(), now()
  ),
  (
    '61000000-0000-4000-8000-000000000002', 'authenticated', 'authenticated',
    'push-outbox-two@example.invalid', crypt('fixture-password', gen_salt('bf')), now(),
    '{"provider":"email","providers":["email"]}'::jsonb, '{}'::jsonb, now(), now()
  );

INSERT INTO public.history (
  id, user_id, title, original_text, mode, status, duration, word_count, app_version
) VALUES
  (
    '61100000-0000-4000-8000-000000000001',
    '61000000-0000-4000-8000-000000000001',
    'Partial', 'partial delivery fixture', 'file-transcription', 'completed', 1, 3, 'e2e'
  ),
  (
    '61100000-0000-4000-8000-000000000002',
    '61000000-0000-4000-8000-000000000001',
    'Lease', 'lease expiry fixture', 'file-transcription', 'completed', 1, 3, 'e2e'
  ),
  (
    '61100000-0000-4000-8000-000000000003',
    '61000000-0000-4000-8000-000000000001',
    'Stale', 'stale registration fixture', 'file-transcription', 'completed', 1, 3, 'e2e'
  );

INSERT INTO public.devices (
  id, user_id, installation_id, platform, device_name, app_version
) VALUES
  (
    '61200000-0000-4000-8000-000000000001',
    '61000000-0000-4000-8000-000000000001',
    '61300000-0000-4000-8000-000000000001',
    'android', 'FCM fixture', 'e2e'
  ),
  (
    '61200000-0000-4000-8000-000000000002',
    '61000000-0000-4000-8000-000000000001',
    '61300000-0000-4000-8000-000000000002',
    'web', 'Web Push fixture', 'e2e'
  ),
  (
    '61200000-0000-4000-8000-000000000003',
    '61000000-0000-4000-8000-000000000001',
    '61300000-0000-4000-8000-000000000003',
    'android', 'Stale fixture', 'e2e'
  );

INSERT INTO public.push_tokens (
  id, user_id, token, platform, device_name, device_id, provider, last_registered_at
) VALUES
  (
    '61400000-0000-4000-8000-000000000001',
    '61000000-0000-4000-8000-000000000001',
    'fcm-outbox-fixture-token-abcdefghijklmnopqrstuvwxyz-001',
    'android', 'FCM fixture', '61200000-0000-4000-8000-000000000001', 'fcm', now()
  ),
  (
    '61400000-0000-4000-8000-000000000002',
    '61000000-0000-4000-8000-000000000001',
    'webpush-outbox-fixture-token-abcdefghijklmnopqrstuvwxyz-002',
    'web', 'Web Push fixture', '61200000-0000-4000-8000-000000000002', 'webpush', now()
  ),
  (
    '61400000-0000-4000-8000-000000000003',
    '61000000-0000-4000-8000-000000000001',
    'fcm-outbox-stale-token-abcdefghijklmnopqrstuvwxyz-003',
    'android', 'Stale fixture', '61200000-0000-4000-8000-000000000003',
    'fcm', now() - interval '36 days'
  );

SELECT pg_temp.assert_true(
  has_function_privilege('authenticated', 'public.reserve_push_dispatch(text,uuid)', 'EXECUTE'),
  'authenticated caller may reserve only after resource ownership checks'
);
SELECT pg_temp.assert_true(
  NOT has_function_privilege('authenticated', 'public.reserve_system_push_dispatch(text,uuid)', 'EXECUTE'),
  'authenticated caller cannot reserve system dispatches'
);
SELECT pg_temp.assert_true(
  NOT has_function_privilege('authenticated', 'public.lease_push_deliveries(uuid,uuid)', 'EXECUTE'),
  'raw registration lease is service-role only'
);
SELECT pg_temp.assert_true(
  NOT has_table_privilege('authenticated', 'public.push_deliveries', 'SELECT'),
  'authenticated caller cannot read outbox token fingerprints'
);
SELECT pg_temp.assert_true(
  NOT has_table_privilege('authenticated', 'public.push_dispatch_claim_events', 'SELECT'),
  'authenticated caller cannot read the service claim ledger'
);

SET LOCAL ROLE authenticated;
SELECT set_config(
  'request.jwt.claims',
  '{"sub":"61000000-0000-4000-8000-000000000001","role":"authenticated"}',
  true
);

DO $$
DECLARE
  first_reservation jsonb;
  duplicate_reservation jsonb;
BEGIN
  first_reservation := public.reserve_push_dispatch(
    'transcription.completed', '61100000-0000-4000-8000-000000000001'
  );
  PERFORM pg_temp.assert_true(
    (first_reservation->>'reserved')::boolean
      AND (first_reservation->>'dispatch_lease_token')::uuid IS NOT NULL,
    'first owned immutable event receives a durable dispatch lease'
  );

  duplicate_reservation := public.reserve_push_dispatch(
    'transcription.completed', '61100000-0000-4000-8000-000000000001'
  );
  PERFORM pg_temp.assert_true(
    (duplicate_reservation->>'duplicate')::boolean
      AND duplicate_reservation->>'status' = 'processing',
    'active immutable event is deduplicated'
  );
END;
$$;

SELECT set_config(
  'request.jwt.claims',
  '{"sub":"61000000-0000-4000-8000-000000000002","role":"authenticated"}',
  true
);
DO $$
BEGIN
  BEGIN
    PERFORM public.reserve_push_dispatch(
      'transcription.completed', '61100000-0000-4000-8000-000000000001'
    );
    RAISE EXCEPTION 'expected ownership failure';
  EXCEPTION WHEN no_data_found THEN
    NULL;
  END;
END;
$$;

RESET ROLE;
SELECT set_config('request.jwt.claims', '{"role":"service_role"}', true);
SET LOCAL ROLE service_role;

DO $$
DECLARE
  dispatch public.push_dispatch_attempts;
  delivery record;
  fcm_delivery_id uuid;
  fcm_lease uuid;
  web_delivery_id uuid;
  web_lease uuid;
  leased_count integer := 0;
  summary jsonb;
  claim record;
BEGIN
  SELECT * INTO dispatch FROM public.push_dispatch_attempts
  WHERE resource_id = '61100000-0000-4000-8000-000000000001';

  FOR delivery IN
    SELECT * FROM public.lease_push_deliveries(dispatch.id, dispatch.dispatch_lease_token)
  LOOP
    leased_count := leased_count + 1;
    IF delivery.provider = 'fcm' THEN
      fcm_delivery_id := delivery.delivery_id;
      fcm_lease := delivery.delivery_lease_token;
    ELSIF delivery.provider = 'webpush' THEN
      web_delivery_id := delivery.delivery_id;
      web_lease := delivery.delivery_lease_token;
    END IF;
  END LOOP;
  PERFORM pg_temp.assert_true(leased_count = 2, 'fresh registrations are leased independently');
  PERFORM pg_temp.assert_true(
    NOT EXISTS (
      SELECT 1 FROM public.push_tokens
      WHERE id = '61400000-0000-4000-8000-000000000003'
    ),
    '35-day stale registration is deleted before delivery'
  );

  PERFORM public.finalize_push_delivery(fcm_delivery_id, fcm_lease, 'delivered', NULL);
  PERFORM public.finalize_push_delivery(
    web_delivery_id, web_lease, 'retryable_failure', 'provider_timeout'
  );
  summary := public.finalize_push_dispatch(dispatch.id, dispatch.dispatch_lease_token);
  PERFORM pg_temp.assert_true(
    summary->>'status' = 'pending'
      AND (summary->>'delivered')::integer = 1
      AND (summary->>'retryable_failed')::integer = 1,
    'partial transient failure remains durable and retryable'
  );

  UPDATE public.push_deliveries
  SET next_retry_at = now() - interval '1 second'
  WHERE id = web_delivery_id;
  UPDATE public.push_dispatch_attempts
  SET next_retry_at = now() - interval '1 second'
  WHERE id = dispatch.id;

  SELECT * INTO claim
  FROM public.claim_due_push_dispatches(100)
  WHERE attempt_id = dispatch.id;
  PERFORM pg_temp.assert_true(
    claim.attempt_id = dispatch.id AND claim.dispatch_lease_token IS NOT NULL,
    'retry worker claims a due partial dispatch'
  );

  leased_count := 0;
  FOR delivery IN
    SELECT * FROM public.lease_push_deliveries(claim.attempt_id, claim.dispatch_lease_token)
  LOOP
    leased_count := leased_count + 1;
    PERFORM pg_temp.assert_true(
      delivery.delivery_id = web_delivery_id,
      'only the failed device is re-leased'
    );
    PERFORM public.finalize_push_delivery(
      delivery.delivery_id,
      delivery.delivery_lease_token,
      'permanent_failure',
      'provider_not_supported'
    );
  END LOOP;
  PERFORM pg_temp.assert_true(leased_count = 1, 'delivered device is never replayed');
  summary := public.finalize_push_dispatch(claim.attempt_id, claim.dispatch_lease_token);
  PERFORM pg_temp.assert_true(
    summary->>'status' = 'partial'
      AND (summary->>'delivered')::integer = 1
      AND (summary->>'permanent_failed')::integer = 1,
    'terminal mixed outcome is recorded as partial'
  );
END;
$$;

DELETE FROM public.push_tokens
WHERE user_id = '61000000-0000-4000-8000-000000000001';
INSERT INTO public.push_tokens (
  id, user_id, token, platform, device_name, device_id, provider, last_registered_at
) VALUES (
  '61400000-0000-4000-8000-000000000004',
  '61000000-0000-4000-8000-000000000001',
  'fcm-outbox-expiry-token-abcdefghijklmnopqrstuvwxyz-004',
  'android', 'FCM fixture', '61200000-0000-4000-8000-000000000001', 'fcm', now()
);

DO $$
DECLARE
  reservation jsonb;
  v_attempt_id uuid;
  dispatch_lease uuid;
  first_delivery record;
  claim record;
  retry_delivery record;
  summary jsonb;
  conflict_seen boolean := false;
BEGIN
  reservation := public.reserve_system_push_dispatch(
    'transcription.completed', '61100000-0000-4000-8000-000000000002'
  );
  v_attempt_id := (reservation->>'attempt_id')::uuid;
  dispatch_lease := (reservation->>'dispatch_lease_token')::uuid;
  SELECT * INTO first_delivery
  FROM public.lease_push_deliveries(v_attempt_id, dispatch_lease);
  PERFORM pg_temp.assert_true(first_delivery.delivery_id IS NOT NULL, 'delivery is initially leased');

  UPDATE public.push_deliveries
  SET lease_expires_at = now() - interval '1 second'
  WHERE id = first_delivery.delivery_id;
  BEGIN
    PERFORM public.finalize_push_delivery(
      first_delivery.delivery_id,
      first_delivery.delivery_lease_token,
      'delivered',
      NULL
    );
  EXCEPTION WHEN serialization_failure THEN
    conflict_seen := true;
  END;
  PERFORM pg_temp.assert_true(conflict_seen, 'late worker cannot finalize an expired delivery lease');

  UPDATE public.push_dispatch_attempts
  SET lease_expires_at = now() - interval '1 second'
  WHERE id = v_attempt_id;
  SELECT * INTO claim
  FROM public.claim_due_push_dispatches(100) AS due
  WHERE due.attempt_id = v_attempt_id;
  PERFORM pg_temp.assert_true(claim.dispatch_lease_token IS NOT NULL, 'expired dispatch is reclaimed');

  PERFORM count(*)
  FROM public.lease_push_deliveries(claim.attempt_id, claim.dispatch_lease_token);
  UPDATE public.push_deliveries
  SET next_retry_at = now() - interval '1 second'
  WHERE id = first_delivery.delivery_id;
  SELECT * INTO retry_delivery
  FROM public.lease_push_deliveries(claim.attempt_id, claim.dispatch_lease_token);
  PERFORM pg_temp.assert_true(
    retry_delivery.delivery_id = first_delivery.delivery_id
      AND retry_delivery.delivery_lease_token <> first_delivery.delivery_lease_token,
    'expired delivery is re-leased with a new fencing token'
  );
  PERFORM public.finalize_push_delivery(
    retry_delivery.delivery_id, retry_delivery.delivery_lease_token, 'delivered', NULL
  );
  summary := public.finalize_push_dispatch(claim.attempt_id, claim.dispatch_lease_token);
  PERFORM pg_temp.assert_true(summary->>'status' = 'succeeded', 'recovered delivery completes');
END;
$$;

DELETE FROM public.push_tokens
WHERE user_id = '61000000-0000-4000-8000-000000000001';
INSERT INTO public.push_tokens (
  id, user_id, token, platform, device_name, device_id, provider, last_registered_at
) VALUES (
  '61400000-0000-4000-8000-000000000005',
  '61000000-0000-4000-8000-000000000001',
  'fcm-outbox-old-token-abcdefghijklmnopqrstuvwxyz-005',
  'android', 'Stale fixture', '61200000-0000-4000-8000-000000000003',
  'fcm', now() - interval '36 days'
);

DO $$
DECLARE
  reservation jsonb;
  v_attempt_id uuid;
  dispatch_lease uuid;
  leased_count integer;
  summary jsonb;
BEGIN
  reservation := public.reserve_system_push_dispatch(
    'transcription.completed', '61100000-0000-4000-8000-000000000003'
  );
  v_attempt_id := (reservation->>'attempt_id')::uuid;
  dispatch_lease := (reservation->>'dispatch_lease_token')::uuid;
  SELECT count(*) INTO leased_count
  FROM public.lease_push_deliveries(v_attempt_id, dispatch_lease);
  PERFORM pg_temp.assert_true(leased_count = 0, 'old registration is never returned raw');
  PERFORM pg_temp.assert_true(
    NOT EXISTS (
      SELECT 1 FROM public.push_tokens
      WHERE id = '61400000-0000-4000-8000-000000000005'
    ),
    'old registration is removed atomically'
  );
  summary := public.finalize_push_dispatch(v_attempt_id, dispatch_lease);
  PERFORM pg_temp.assert_true(
    summary->>'status' = 'succeeded' AND (summary->>'complete')::boolean,
    'no-target stale dispatch terminates without retry spam'
  );
END;
$$;

-- Transactional enqueue stays durable at provider capacity. The drain does
-- not mutate a capped system row, then claims it as soon as one slot is free.
INSERT INTO public.history (
  id, user_id, title, original_text, mode, status, duration, word_count, app_version
) VALUES (
  '61100000-0000-4000-8000-000000000006',
  '61000000-0000-4000-8000-000000000001',
  'Rate limit', 'rate limit fixture', 'file-transcription', 'completed', 1, 3, 'e2e'
);
INSERT INTO public.push_dispatch_attempts (
  caller_id, actor_kind, event_type, resource_id, status, attempt_count,
  next_retry_at, created_at
) VALUES (
  NULL, 'system', 'transcription.completed',
  '61100000-0000-4000-8000-000000000006', 'pending', 0,
  now() - interval '5 seconds', now() - interval '5 seconds'
);
DELETE FROM public.push_dispatch_claim_events;
INSERT INTO public.push_dispatch_claim_events (
  attempt_id, dispatch_lease_token, claimed_at
)
SELECT gen_random_uuid(), gen_random_uuid(), now()
FROM generate_series(1, 500);

DO $$
DECLARE
  target_attempt public.push_dispatch_attempts;
  capped_count integer;
BEGIN
  SELECT * INTO target_attempt
  FROM public.push_dispatch_attempts
  WHERE event_type = 'transcription.completed'
    AND resource_id = '61100000-0000-4000-8000-000000000006';

  SELECT count(*) INTO capped_count
  FROM public.claim_due_push_dispatches(10);
  PERFORM pg_temp.assert_true(
    capped_count = 0,
    'system dispatch capacity returns no claim instead of dropping the event'
  );
  PERFORM pg_temp.assert_true(
    EXISTS (
      SELECT 1 FROM public.push_dispatch_attempts
      WHERE id = target_attempt.id
        AND status = 'pending'
        AND next_retry_at = target_attempt.next_retry_at
        AND dispatch_lease_token IS NULL
    ),
    'rate-limited pending status and next_retry_at remain unchanged'
  );

  DELETE FROM public.push_dispatch_claim_events
  WHERE id = (SELECT min(id) FROM public.push_dispatch_claim_events);
  SELECT count(*) INTO capped_count
  FROM public.claim_due_push_dispatches(10);
  PERFORM pg_temp.assert_true(
    capped_count = 1,
    'one released system capacity slot atomically claims one due dispatch'
  );
  PERFORM pg_temp.assert_true(
    (SELECT count(*) FROM public.push_dispatch_claim_events) = 500,
    'successful claim consumes exactly one global rate slot'
  );
END;
$$;

DELETE FROM public.push_dispatch_claim_events;
UPDATE public.push_dispatch_attempts
SET status = 'succeeded', dispatch_lease_token = NULL, lease_expires_at = NULL,
    next_retry_at = NULL, completed_at = now(), updated_at = now()
WHERE event_type = 'transcription.completed'
  AND resource_id = '61100000-0000-4000-8000-000000000006';

INSERT INTO public.history (
  id, user_id, title, original_text, mode, status, duration, word_count, app_version
) VALUES (
  '61100000-0000-4000-8000-000000000007',
  '61000000-0000-4000-8000-000000000001',
  'Daily rate limit', 'daily rate limit fixture', 'file-transcription',
  'completed', 1, 4, 'e2e'
);
INSERT INTO public.push_dispatch_attempts (
  caller_id, actor_kind, event_type, resource_id, status, attempt_count,
  next_retry_at, created_at
) VALUES (
  NULL, 'system', 'transcription.completed',
  '61100000-0000-4000-8000-000000000007', 'pending', 0,
  now() - interval '5 seconds', now() - interval '5 seconds'
);
INSERT INTO public.push_dispatch_claim_events (
  attempt_id, dispatch_lease_token, claimed_at
)
SELECT gen_random_uuid(), gen_random_uuid(), now() - interval '2 minutes'
FROM generate_series(1, 10000);

DO $$
DECLARE
  target_attempt public.push_dispatch_attempts;
  capped_count integer;
BEGIN
  SELECT * INTO target_attempt
  FROM public.push_dispatch_attempts
  WHERE event_type = 'transcription.completed'
    AND resource_id = '61100000-0000-4000-8000-000000000007';

  SELECT count(*) INTO capped_count
  FROM public.claim_due_push_dispatches(10);
  PERFORM pg_temp.assert_true(
    capped_count = 0,
    '24-hour system capacity returns no claim instead of dropping the event'
  );
  PERFORM pg_temp.assert_true(
    EXISTS (
      SELECT 1 FROM public.push_dispatch_attempts
      WHERE id = target_attempt.id
        AND status = 'pending'
        AND next_retry_at = target_attempt.next_retry_at
        AND dispatch_lease_token IS NULL
    ),
    'daily rate-limited pending status and next_retry_at remain unchanged'
  );

  DELETE FROM public.push_dispatch_claim_events
  WHERE id = (SELECT min(id) FROM public.push_dispatch_claim_events);
  SELECT count(*) INTO capped_count
  FROM public.claim_due_push_dispatches(10);
  PERFORM pg_temp.assert_true(
    capped_count = 1,
    'one released daily capacity slot atomically claims one due dispatch'
  );
  PERFORM pg_temp.assert_true(
    (SELECT count(*) FROM public.push_dispatch_claim_events) = 10000,
    'successful claim consumes exactly one 24-hour global rate slot'
  );
END;
$$;

DELETE FROM public.push_dispatch_claim_events;
UPDATE public.push_dispatch_attempts
SET status = 'succeeded', dispatch_lease_token = NULL, lease_expires_at = NULL,
    next_retry_at = NULL, completed_at = now(), updated_at = now()
WHERE event_type = 'transcription.completed'
  AND resource_id = '61100000-0000-4000-8000-000000000007';

-- Transactional producer hooks enqueue before the business transaction commits.
INSERT INTO public.history (
  id, user_id, title, original_text, mode, status, duration, word_count, app_version
) VALUES (
  '61100000-0000-4000-8000-000000000004',
  '61000000-0000-4000-8000-000000000001',
  'Worker', 'worker completion fixture', 'file-transcription', 'completed', 1, 3, 'e2e'
);
INSERT INTO public.processing_jobs (
  id, user_id, history_id, kind, status, idempotency_key
) VALUES (
  '61500000-0000-4000-8000-000000000001',
  '61000000-0000-4000-8000-000000000001',
  '61100000-0000-4000-8000-000000000004',
  'transcription', 'queued', 'push-outbox-worker-fixture'
);
UPDATE public.processing_jobs
SET status = 'succeeded', progress = 100, completed_at = now()
WHERE id = '61500000-0000-4000-8000-000000000001';
SELECT pg_temp.assert_true(
  (
    SELECT count(*) FROM public.push_dispatch_attempts
    WHERE event_type = 'transcription.completed'
      AND resource_id = '61100000-0000-4000-8000-000000000004'
      AND status = 'pending'
  ) = 1,
  'processing job completion enqueues transcription in the same transaction'
);

INSERT INTO public.history (
  id, user_id, title, original_text, mode, status, duration, word_count, app_version
) VALUES (
  '61100000-0000-4000-8000-000000000005',
  '61000000-0000-4000-8000-000000000001',
  'Reverse order', 'reverse order fixture', 'file-transcription', 'error', 1, 3, 'e2e'
);
INSERT INTO public.processing_jobs (
  id, user_id, history_id, kind, status, progress, idempotency_key, completed_at
) VALUES (
  '61500000-0000-4000-8000-000000000002',
  '61000000-0000-4000-8000-000000000001',
  '61100000-0000-4000-8000-000000000005',
  'transcription', 'succeeded', 100, 'push-outbox-reverse-fixture', now()
);
UPDATE public.history
SET status = 'completed', error_code = NULL
WHERE id = '61100000-0000-4000-8000-000000000005';
SELECT pg_temp.assert_true(
  (
    SELECT count(*) FROM public.push_dispatch_attempts
    WHERE event_type = 'transcription.completed'
      AND resource_id = '61100000-0000-4000-8000-000000000005'
      AND status = 'pending'
  ) = 1,
  'history completion also enqueues when the worker committed first'
);

-- A recovery worker may persist the completed history before inserting its
-- already-succeeded server job. INSERT must be a first-class completion edge.
INSERT INTO public.history (
  id, user_id, title, original_text, mode, status, duration, word_count, app_version
) VALUES (
  '61100000-0000-4000-8000-000000000008',
  '61000000-0000-4000-8000-000000000001',
  'Inserted success', 'inserted success fixture', 'file-transcription',
  'completed', 1, 3, 'e2e'
);
INSERT INTO public.processing_jobs (
  id, user_id, history_id, kind, status, progress, idempotency_key
) VALUES (
  '61500000-0000-4000-8000-000000000003',
  '61000000-0000-4000-8000-000000000001',
  '61100000-0000-4000-8000-000000000008',
  'transcription', 'succeeded', 100, 'push-outbox-inserted-success'
);
UPDATE public.processing_jobs
SET status = 'succeeded', history_id = '61100000-0000-4000-8000-000000000008'
WHERE id = '61500000-0000-4000-8000-000000000003';
SELECT pg_temp.assert_true(
  (
    SELECT count(*) FROM public.push_dispatch_attempts
    WHERE event_type = 'transcription.completed'
      AND resource_id = '61100000-0000-4000-8000-000000000008'
      AND status = 'pending'
  ) = 1,
  'insert-succeeded and same-value replay enqueue exactly one immutable event'
);

-- Some workers know the result before the durable history id. Attaching the
-- id later must enqueue once, while the succeeded job without a link enqueues 0.
INSERT INTO public.history (
  id, user_id, title, original_text, mode, status, duration, word_count, app_version
) VALUES (
  '61100000-0000-4000-8000-000000000009',
  '61000000-0000-4000-8000-000000000001',
  'Linked success', 'linked success fixture', 'file-transcription',
  'completed', 1, 3, 'e2e'
);
INSERT INTO public.processing_jobs (
  id, user_id, history_id, kind, status, progress, idempotency_key
) VALUES (
  '61500000-0000-4000-8000-000000000004',
  '61000000-0000-4000-8000-000000000001',
  NULL, 'transcription', 'succeeded', 100, 'push-outbox-linked-success'
);
SELECT pg_temp.assert_true(
  NOT EXISTS (
    SELECT 1 FROM public.push_dispatch_attempts
    WHERE event_type = 'transcription.completed'
      AND resource_id = '61100000-0000-4000-8000-000000000009'
  ),
  'succeeded job without history linkage enqueues nothing'
);
UPDATE public.processing_jobs
SET history_id = '61100000-0000-4000-8000-000000000009'
WHERE id = '61500000-0000-4000-8000-000000000004';
UPDATE public.processing_jobs
SET history_id = '61100000-0000-4000-8000-000000000009'
WHERE id = '61500000-0000-4000-8000-000000000004';
SELECT pg_temp.assert_true(
  (
    SELECT count(*) FROM public.push_dispatch_attempts
    WHERE event_type = 'transcription.completed'
      AND resource_id = '61100000-0000-4000-8000-000000000009'
      AND status = 'pending'
  ) = 1,
  'late history linkage and same-value replay enqueue exactly one event'
);

INSERT INTO public.teams (id, name, owner_id) VALUES (
  '61600000-0000-4000-8000-000000000001',
  'Outbox Team', '61000000-0000-4000-8000-000000000001'
);
INSERT INTO public.team_members (team_id, user_id, role) VALUES (
  '61600000-0000-4000-8000-000000000001',
  '61000000-0000-4000-8000-000000000001', 'owner'
);
INSERT INTO public.team_invites (
  id, team_id, invited_by, email, role, token
) VALUES (
  '61700000-0000-4000-8000-000000000001',
  '61600000-0000-4000-8000-000000000001',
  '61000000-0000-4000-8000-000000000001',
  'push-outbox-two@example.invalid', 'member',
  'push_outbox_invite_token_1234567890'
);
SELECT pg_temp.assert_true(
  (
    SELECT count(*) FROM public.push_dispatch_attempts
    WHERE event_type = 'team.invite.created'
      AND resource_id = '61700000-0000-4000-8000-000000000001'
      AND status = 'pending'
  ) = 1,
  'team invitation insert enqueues an immutable notification transactionally'
);

UPDATE public.subscriptions
SET status = 'past_due'
WHERE user_id = '61000000-0000-4000-8000-000000000001';
UPDATE public.subscriptions
SET auto_renewing = false
WHERE user_id = '61000000-0000-4000-8000-000000000001';
SELECT pg_temp.assert_true(
  (
    SELECT count(*)
    FROM public.push_dispatch_attempts AS attempt
    JOIN public.subscriptions AS subscription ON subscription.id = attempt.resource_id
    WHERE attempt.event_type = 'billing.status.changed'
      AND subscription.user_id = '61000000-0000-4000-8000-000000000001'
  ) = 1,
  'rapid billing mutations collapse to one transactional outbox event'
);

RESET ROLE;
ROLLBACK;