-- ============================================================================
-- Payment provider serialization and entitlement single source of truth
--
-- All Stripe, Payple, Google Play and App Store entitlement changes pass
-- through service-role-only RPCs.  A per-user transaction advisory lock
-- serializes provider races, provider cursors reject out-of-order events, and
-- payment_provider is maintained only as a compatibility mirror of provider.
-- ============================================================================

BEGIN;

ALTER TABLE public.subscriptions
  ADD COLUMN IF NOT EXISTS provider_resource_id text,
  ADD COLUMN IF NOT EXISTS provider_event_id text,
  ADD COLUMN IF NOT EXISTS provider_event_created_at timestamptz;

CREATE UNIQUE INDEX IF NOT EXISTS idx_subscriptions_provider_resource_owner
  ON public.subscriptions(provider, provider_resource_id)
  WHERE provider <> 'none' AND provider_resource_id IS NOT NULL;

CREATE TABLE public.payment_provider_events (
  id                    uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  provider              text NOT NULL
                        CHECK (provider IN ('stripe', 'payple', 'google_play', 'app_store', 'admin')),
  event_id               text NOT NULL CHECK (length(event_id) BETWEEN 3 AND 255),
  user_id                uuid NOT NULL REFERENCES auth.users(id) ON DELETE CASCADE,
  provider_resource_id   text NOT NULL CHECK (length(provider_resource_id) BETWEEN 1 AND 255),
  event_type             text NOT NULL CHECK (length(event_type) BETWEEN 1 AND 100),
  event_created_at       timestamptz NOT NULL,
  payload_digest         text NOT NULL CHECK (payload_digest ~ '^[0-9a-f]{64}$'),
  disposition            text NOT NULL DEFAULT 'received'
                         CHECK (disposition IN ('received', 'applied', 'ignored', 'rejected', 'failed')),
  result                 jsonb,
  received_at            timestamptz NOT NULL DEFAULT now(),
  processed_at           timestamptz,
  UNIQUE (provider, event_id)
);

CREATE INDEX idx_payment_provider_events_user_created
  ON public.payment_provider_events(user_id, event_created_at DESC);

ALTER TABLE public.payment_provider_events ENABLE ROW LEVEL SECURITY;
-- No authenticated policies: provider payload metadata is service-only.

CREATE TABLE public.payment_provider_cursors (
  user_id                uuid NOT NULL REFERENCES auth.users(id) ON DELETE CASCADE,
  provider               text NOT NULL
                         CHECK (provider IN ('stripe', 'payple', 'google_play', 'app_store', 'admin')),
  last_event_created_at  timestamptz NOT NULL,
  last_event_id          text NOT NULL,
  updated_at             timestamptz NOT NULL DEFAULT now(),
  PRIMARY KEY (user_id, provider)
);

ALTER TABLE public.payment_provider_cursors ENABLE ROW LEVEL SECURITY;
-- No authenticated policies: cursors are service-owned serialization state.

CREATE TABLE public.payment_provider_operations (
  id                    uuid PRIMARY KEY DEFAULT gen_random_uuid(),
  user_id               uuid NOT NULL REFERENCES auth.users(id) ON DELETE CASCADE,
  provider              text NOT NULL
                        CHECK (provider IN ('stripe', 'payple', 'google_play', 'app_store')),
  operation_type        text NOT NULL
                        CHECK (operation_type IN ('checkout', 'renewal', 'cancellation')),
  idempotency_key       text NOT NULL CHECK (length(idempotency_key) BETWEEN 12 AND 160),
  requested_tier        text CHECK (requested_tier IN ('pro', 'pro_plus')),
  provider_order_id     text,
  provider_resource_id  text,
  state                 text NOT NULL DEFAULT 'reserved'
                        CHECK (state IN ('reserved', 'external_created', 'charged', 'applied', 'failed')),
  external_reference    text,
  error_code            text,
  expires_at            timestamptz NOT NULL,
  created_at            timestamptz NOT NULL DEFAULT now(),
  updated_at            timestamptz NOT NULL DEFAULT now(),
  UNIQUE (user_id, provider, idempotency_key)
);

CREATE UNIQUE INDEX idx_payment_provider_operations_order
  ON public.payment_provider_operations(provider, provider_order_id)
  WHERE provider_order_id IS NOT NULL;

CREATE INDEX idx_payment_provider_operations_active
  ON public.payment_provider_operations(user_id, expires_at)
  WHERE state IN ('reserved', 'external_created', 'charged');

ALTER TABLE public.payment_provider_operations ENABLE ROW LEVEL SECURITY;
-- No authenticated policies: operations contain server-side payment state.

CREATE TRIGGER set_updated_at_payment_provider_operations
  BEFORE UPDATE ON public.payment_provider_operations
  FOR EACH ROW EXECUTE FUNCTION public.moddatetime();

CREATE OR REPLACE FUNCTION public.mirror_subscription_payment_provider()
RETURNS trigger
LANGUAGE plpgsql
SET search_path = public, pg_temp
AS $$
BEGIN
  NEW.payment_provider := CASE
    WHEN NEW.provider = 'admin' THEN 'none'
    ELSE NEW.provider
  END;
  RETURN NEW;
END;
$$;

DROP TRIGGER IF EXISTS mirror_subscription_payment_provider ON public.subscriptions;
CREATE TRIGGER mirror_subscription_payment_provider
  BEFORE INSERT OR UPDATE OF provider, payment_provider ON public.subscriptions
  FOR EACH ROW EXECUTE FUNCTION public.mirror_subscription_payment_provider();

UPDATE public.subscriptions
   SET payment_provider = CASE WHEN provider = 'admin' THEN 'none' ELSE provider END
 WHERE payment_provider IS DISTINCT FROM
       CASE WHEN provider = 'admin' THEN 'none' ELSE provider END;

CREATE OR REPLACE FUNCTION public.sync_profile_tier_from_subscription()
RETURNS trigger
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = public, pg_temp
AS $$
BEGIN
  UPDATE public.profiles
     SET tier = NEW.tier,
         updated_at = now()
   WHERE id = NEW.user_id
     AND tier IS DISTINCT FROM NEW.tier;
  RETURN NEW;
END;
$$;

DROP TRIGGER IF EXISTS sync_profile_tier_from_subscription ON public.subscriptions;
CREATE TRIGGER sync_profile_tier_from_subscription
  AFTER INSERT OR UPDATE OF tier ON public.subscriptions
  FOR EACH ROW EXECUTE FUNCTION public.sync_profile_tier_from_subscription();

UPDATE public.profiles AS p
   SET tier = s.tier,
       updated_at = now()
  FROM public.subscriptions AS s
 WHERE s.user_id = p.id
   AND p.tier IS DISTINCT FROM s.tier;

-- Reserve a provider operation before any external API call.  A live operation
-- for the user blocks every different request, including the same provider,
-- while an exact idempotency-key replay returns the existing operation.
CREATE OR REPLACE FUNCTION public.reserve_payment_provider_operation(
  p_user_id uuid,
  p_provider text,
  p_operation_type text,
  p_requested_tier text,
  p_idempotency_key text,
  p_provider_order_id text DEFAULT NULL,
  p_provider_resource_id text DEFAULT NULL
) RETURNS jsonb
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = public, pg_temp
AS $$
DECLARE
  v_existing public.payment_provider_operations%ROWTYPE;
  v_subscription public.subscriptions%ROWTYPE;
  v_operation public.payment_provider_operations%ROWTYPE;
BEGIN
  IF p_user_id IS NULL OR NOT EXISTS (SELECT 1 FROM auth.users WHERE id = p_user_id) THEN
    RAISE EXCEPTION 'unknown_user';
  END IF;
  IF p_provider NOT IN ('stripe', 'payple', 'google_play', 'app_store') THEN
    RAISE EXCEPTION 'invalid_provider';
  END IF;
  IF p_operation_type NOT IN ('checkout', 'renewal', 'cancellation') THEN
    RAISE EXCEPTION 'invalid_operation_type';
  END IF;
  IF p_operation_type IN ('checkout', 'renewal') AND p_requested_tier NOT IN ('pro', 'pro_plus') THEN
    RAISE EXCEPTION 'invalid_tier';
  END IF;
  IF p_operation_type = 'cancellation' AND p_requested_tier IS NOT NULL THEN
    RAISE EXCEPTION 'cancellation_tier_must_be_null';
  END IF;
  IF p_idempotency_key IS NULL
     OR p_idempotency_key !~ '^[A-Za-z0-9._:-]{12,160}$' THEN
    RAISE EXCEPTION 'invalid_idempotency_key';
  END IF;
  IF p_provider_order_id IS NOT NULL
     AND p_provider_order_id !~ '^[A-Za-z0-9._-]{8,64}$' THEN
    RAISE EXCEPTION 'invalid_provider_order_id';
  END IF;
  IF p_provider_resource_id IS NOT NULL
     AND length(trim(p_provider_resource_id)) NOT BETWEEN 1 AND 255 THEN
    RAISE EXCEPTION 'invalid_provider_resource_id';
  END IF;

  PERFORM pg_advisory_xact_lock(hashtextextended(p_user_id::text, 73031));

  SELECT *
    INTO v_existing
    FROM public.payment_provider_operations
   WHERE user_id = p_user_id
     AND provider = p_provider
     AND idempotency_key = p_idempotency_key
   FOR UPDATE;

  IF v_existing.id IS NOT NULL THEN
    IF v_existing.operation_type <> p_operation_type
       OR v_existing.requested_tier IS DISTINCT FROM p_requested_tier
       OR v_existing.provider_resource_id IS DISTINCT FROM p_provider_resource_id THEN
      RAISE EXCEPTION 'idempotency_key_payload_mismatch';
    END IF;
    RETURN jsonb_build_object(
      'created', false,
      'operation_id', v_existing.id,
      'state', v_existing.state,
      'provider_order_id', v_existing.provider_order_id,
      'external_reference', v_existing.external_reference,
      'reason', 'idempotent_replay'
    );
  END IF;

  UPDATE public.payment_provider_operations
     SET state = 'failed',
         error_code = 'operation_lease_expired',
         updated_at = now()
   WHERE user_id = p_user_id
     AND state IN ('reserved', 'external_created', 'charged')
     AND expires_at <= now();

  SELECT *
    INTO v_subscription
    FROM public.subscriptions
   WHERE user_id = p_user_id
   FOR UPDATE;

  IF v_subscription.id IS NULL THEN
    INSERT INTO public.subscriptions (user_id, tier, status, provider, payment_provider)
    VALUES (p_user_id, 'free', 'active', 'none', 'none')
    RETURNING * INTO v_subscription;
  END IF;

  IF p_operation_type = 'checkout' THEN
    IF v_subscription.provider NOT IN ('none', p_provider) THEN
      RETURN jsonb_build_object(
        'created', false,
        'state', 'rejected',
        'reason', 'active_subscription_other_provider',
        'owner_provider', v_subscription.provider
      );
    END IF;
  ELSE
    IF v_subscription.provider <> p_provider THEN
      RETURN jsonb_build_object(
        'created', false,
        'state', 'rejected',
        'reason', 'provider_not_owner',
        'owner_provider', v_subscription.provider
      );
    END IF;
    IF p_provider_resource_id IS NOT NULL
       AND v_subscription.provider_resource_id IS DISTINCT FROM p_provider_resource_id THEN
      RETURN jsonb_build_object(
        'created', false,
        'state', 'rejected',
        'reason', 'provider_resource_not_owner'
      );
    END IF;
  END IF;

  IF EXISTS (
    SELECT 1
      FROM public.payment_provider_operations
     WHERE user_id = p_user_id
       AND state IN ('reserved', 'external_created', 'charged')
       AND expires_at > now()
  ) THEN
    RETURN jsonb_build_object(
      'created', false,
      'state', 'rejected',
      'reason', 'payment_operation_in_progress'
    );
  END IF;

  INSERT INTO public.payment_provider_operations (
    user_id,
    provider,
    operation_type,
    idempotency_key,
    requested_tier,
    provider_order_id,
    provider_resource_id,
    expires_at
  ) VALUES (
    p_user_id,
    p_provider,
    p_operation_type,
    p_idempotency_key,
    p_requested_tier,
    nullif(trim(p_provider_order_id), ''),
    nullif(trim(p_provider_resource_id), ''),
    now() + interval '15 minutes'
  )
  RETURNING * INTO v_operation;

  RETURN jsonb_build_object(
    'created', true,
    'operation_id', v_operation.id,
    'state', v_operation.state,
    'provider_order_id', v_operation.provider_order_id,
    'expires_at', v_operation.expires_at
  );
END;
$$;

CREATE OR REPLACE FUNCTION public.mark_payment_provider_operation(
  p_operation_id uuid,
  p_state text,
  p_external_reference text DEFAULT NULL,
  p_error_code text DEFAULT NULL
) RETURNS jsonb
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = public, pg_temp
AS $$
DECLARE
  v_operation public.payment_provider_operations%ROWTYPE;
BEGIN
  IF p_state NOT IN ('external_created', 'charged', 'failed') THEN
    RAISE EXCEPTION 'invalid_operation_state';
  END IF;
  IF p_external_reference IS NOT NULL AND length(p_external_reference) > 255 THEN
    RAISE EXCEPTION 'invalid_external_reference';
  END IF;
  IF p_error_code IS NOT NULL AND p_error_code !~ '^[a-z0-9_:-]{1,100}$' THEN
    RAISE EXCEPTION 'invalid_error_code';
  END IF;

  SELECT *
    INTO v_operation
    FROM public.payment_provider_operations
   WHERE id = p_operation_id;
  IF v_operation.id IS NULL THEN
    RAISE EXCEPTION 'operation_not_found';
  END IF;

  PERFORM pg_advisory_xact_lock(hashtextextended(v_operation.user_id::text, 73031));
  SELECT *
    INTO v_operation
    FROM public.payment_provider_operations
   WHERE id = p_operation_id
   FOR UPDATE;

  IF v_operation.state IN ('applied', 'failed') THEN
    RETURN jsonb_build_object(
      'operation_id', v_operation.id,
      'state', v_operation.state,
      'duplicate', true
    );
  END IF;

  IF p_state = 'charged' AND v_operation.operation_type = 'cancellation' THEN
    RAISE EXCEPTION 'invalid_cancellation_state';
  END IF;

  UPDATE public.payment_provider_operations
     SET state = p_state,
         external_reference = coalesce(nullif(trim(p_external_reference), ''), external_reference),
         error_code = CASE WHEN p_state = 'failed' THEN p_error_code ELSE NULL END,
         updated_at = now()
   WHERE id = p_operation_id
   RETURNING * INTO v_operation;

  RETURN jsonb_build_object(
    'operation_id', v_operation.id,
    'state', v_operation.state,
    'duplicate', false
  );
END;
$$;

-- Record a provider event that carries correlation metadata but is not itself
-- authoritative for entitlement (for example Stripe checkout.session.completed).
-- It is replay-protected without advancing the entitlement ordering cursor.
CREATE OR REPLACE FUNCTION public.record_payment_provider_observation(
  p_user_id uuid,
  p_provider text,
  p_event_id text,
  p_event_created_at timestamptz,
  p_event_type text,
  p_payload_digest text,
  p_provider_resource_id text
) RETURNS jsonb
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = public, pg_temp
AS $$
DECLARE
  v_event public.payment_provider_events%ROWTYPE;
  v_result jsonb;
BEGIN
  IF p_user_id IS NULL OR NOT EXISTS (SELECT 1 FROM auth.users WHERE id = p_user_id) THEN
    RAISE EXCEPTION 'unknown_user';
  END IF;
  IF p_provider NOT IN ('stripe', 'payple', 'google_play', 'app_store', 'admin') THEN
    RAISE EXCEPTION 'invalid_provider';
  END IF;
  IF p_event_id IS NULL OR length(trim(p_event_id)) NOT BETWEEN 3 AND 255
     OR p_event_type IS NULL OR length(trim(p_event_type)) NOT BETWEEN 1 AND 100
     OR p_provider_resource_id IS NULL
     OR length(trim(p_provider_resource_id)) NOT BETWEEN 1 AND 255 THEN
    RAISE EXCEPTION 'invalid_provider_observation';
  END IF;
  IF p_event_created_at IS NULL OR p_event_created_at > now() + interval '10 minutes' THEN
    RAISE EXCEPTION 'invalid_event_created_at';
  END IF;
  IF p_payload_digest IS NULL OR p_payload_digest !~ '^[0-9a-f]{64}$' THEN
    RAISE EXCEPTION 'invalid_payload_digest';
  END IF;

  PERFORM pg_advisory_xact_lock(hashtextextended(p_user_id::text, 73031));
  v_result := jsonb_build_object('observed', true, 'duplicate', false);
  INSERT INTO public.payment_provider_events (
    provider, event_id, user_id, provider_resource_id, event_type,
    event_created_at, payload_digest, disposition, result, processed_at
  ) VALUES (
    p_provider, trim(p_event_id), p_user_id, trim(p_provider_resource_id),
    trim(p_event_type), p_event_created_at, p_payload_digest,
    'ignored', v_result, now()
  )
  ON CONFLICT (provider, event_id) DO NOTHING
  RETURNING * INTO v_event;

  IF v_event.id IS NULL THEN
    SELECT * INTO v_event
      FROM public.payment_provider_events
     WHERE provider = p_provider AND event_id = trim(p_event_id)
     FOR UPDATE;
    IF v_event.user_id <> p_user_id
       OR v_event.provider_resource_id <> trim(p_provider_resource_id)
       OR v_event.event_type <> trim(p_event_type)
       OR v_event.payload_digest <> p_payload_digest THEN
      RAISE EXCEPTION 'provider_event_payload_mismatch';
    END IF;
    RETURN coalesce(v_event.result, v_result) || jsonb_build_object('duplicate', true);
  END IF;
  RETURN v_result;
END;
$$;

-- Apply one authoritative provider event.  Provider ownership is strict: an
-- event can never overwrite another provider.  A cancellation can affect only
-- the exact provider resource currently owning the entitlement.
CREATE OR REPLACE FUNCTION public.apply_payment_provider_event(
  p_user_id uuid,
  p_provider text,
  p_event_id text,
  p_event_created_at timestamptz,
  p_event_type text,
  p_payload_digest text,
  p_provider_resource_id text,
  p_tier text,
  p_status text,
  p_entitled boolean,
  p_current_period_start timestamptz DEFAULT NULL,
  p_current_period_end timestamptz DEFAULT NULL,
  p_cancel_at timestamptz DEFAULT NULL,
  p_auto_renewing boolean DEFAULT NULL,
  p_provider_customer_id text DEFAULT NULL,
  p_provider_order_id text DEFAULT NULL,
  p_store_product_id text DEFAULT NULL,
  p_store_purchase_id uuid DEFAULT NULL,
  p_operation_id uuid DEFAULT NULL
) RETURNS jsonb
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = public, pg_temp
AS $$
DECLARE
  v_event public.payment_provider_events%ROWTYPE;
  v_cursor public.payment_provider_cursors%ROWTYPE;
  v_subscription public.subscriptions%ROWTYPE;
  v_result jsonb;
BEGIN
  IF p_user_id IS NULL OR NOT EXISTS (SELECT 1 FROM auth.users WHERE id = p_user_id) THEN
    RAISE EXCEPTION 'unknown_user';
  END IF;
  IF p_provider NOT IN ('stripe', 'payple', 'google_play', 'app_store', 'admin') THEN
    RAISE EXCEPTION 'invalid_provider';
  END IF;
  IF p_event_id IS NULL OR length(trim(p_event_id)) NOT BETWEEN 3 AND 255 THEN
    RAISE EXCEPTION 'invalid_event_id';
  END IF;
  IF p_event_created_at IS NULL OR p_event_created_at > now() + interval '10 minutes' THEN
    RAISE EXCEPTION 'invalid_event_created_at';
  END IF;
  IF p_event_type IS NULL OR length(trim(p_event_type)) NOT BETWEEN 1 AND 100 THEN
    RAISE EXCEPTION 'invalid_event_type';
  END IF;
  IF p_payload_digest IS NULL OR p_payload_digest !~ '^[0-9a-f]{64}$' THEN
    RAISE EXCEPTION 'invalid_payload_digest';
  END IF;
  IF p_provider_resource_id IS NULL
     OR length(trim(p_provider_resource_id)) NOT BETWEEN 1 AND 255 THEN
    RAISE EXCEPTION 'invalid_provider_resource_id';
  END IF;
  IF p_tier NOT IN ('free', 'pro', 'pro_plus') THEN
    RAISE EXCEPTION 'invalid_tier';
  END IF;
  IF p_entitled AND p_tier = 'free' THEN
    RAISE EXCEPTION 'entitled_tier_must_be_paid';
  END IF;
  IF p_status IS NULL OR p_status NOT IN (
    'active', 'trialing', 'past_due', 'canceled', 'unpaid', 'incomplete',
    'incomplete_expired', 'paused', 'on_hold', 'expired', 'refunded', 'pending'
  ) THEN
    RAISE EXCEPTION 'invalid_status';
  END IF;
  IF p_current_period_start IS NOT NULL
     AND p_current_period_end IS NOT NULL
     AND p_current_period_end < p_current_period_start THEN
    RAISE EXCEPTION 'invalid_subscription_period';
  END IF;
  IF p_provider_customer_id IS NOT NULL AND length(p_provider_customer_id) > 255 THEN
    RAISE EXCEPTION 'invalid_provider_customer_id';
  END IF;
  IF p_provider_order_id IS NOT NULL AND length(p_provider_order_id) > 255 THEN
    RAISE EXCEPTION 'invalid_provider_order_id';
  END IF;

  PERFORM pg_advisory_xact_lock(hashtextextended(p_user_id::text, 73031));

  INSERT INTO public.payment_provider_events (
    provider,
    event_id,
    user_id,
    provider_resource_id,
    event_type,
    event_created_at,
    payload_digest
  ) VALUES (
    p_provider,
    trim(p_event_id),
    p_user_id,
    trim(p_provider_resource_id),
    trim(p_event_type),
    p_event_created_at,
    p_payload_digest
  )
  ON CONFLICT (provider, event_id) DO NOTHING
  RETURNING * INTO v_event;

  IF v_event.id IS NULL THEN
    SELECT *
      INTO v_event
      FROM public.payment_provider_events
     WHERE provider = p_provider
       AND event_id = trim(p_event_id)
     FOR UPDATE;
    IF v_event.user_id <> p_user_id
       OR v_event.provider_resource_id <> trim(p_provider_resource_id)
       OR v_event.event_type <> trim(p_event_type)
       OR v_event.payload_digest <> p_payload_digest THEN
      RAISE EXCEPTION 'provider_event_payload_mismatch';
    END IF;
    RETURN coalesce(
      v_event.result,
      jsonb_build_object(
        'applied', false,
        'duplicate', true,
        'reason', 'event_processing_in_progress'
      )
    ) || jsonb_build_object('duplicate', true);
  END IF;

  SELECT *
    INTO v_cursor
    FROM public.payment_provider_cursors
   WHERE user_id = p_user_id
     AND provider = p_provider
   FOR UPDATE;

  IF v_cursor.user_id IS NOT NULL AND (
    v_cursor.last_event_created_at > p_event_created_at
    OR (
      v_cursor.last_event_created_at = p_event_created_at
      AND v_cursor.last_event_id >= trim(p_event_id)
    )
  ) THEN
    v_result := jsonb_build_object(
      'applied', false,
      'duplicate', false,
      'reason', 'stale_provider_event'
    );
    UPDATE public.payment_provider_events
       SET disposition = 'ignored', result = v_result, processed_at = now()
     WHERE id = v_event.id;
    RETURN v_result;
  END IF;

  INSERT INTO public.payment_provider_cursors (
    user_id, provider, last_event_created_at, last_event_id
  ) VALUES (
    p_user_id, p_provider, p_event_created_at, trim(p_event_id)
  )
  ON CONFLICT (user_id, provider) DO UPDATE
    SET last_event_created_at = EXCLUDED.last_event_created_at,
        last_event_id = EXCLUDED.last_event_id,
        updated_at = now();

  SELECT *
    INTO v_subscription
    FROM public.subscriptions
   WHERE user_id = p_user_id
   FOR UPDATE;

  IF v_subscription.id IS NULL THEN
    INSERT INTO public.subscriptions (user_id, tier, status, provider, payment_provider)
    VALUES (p_user_id, 'free', 'active', 'none', 'none')
    RETURNING * INTO v_subscription;
  END IF;

  IF p_operation_id IS NOT NULL AND NOT EXISTS (
    SELECT 1
      FROM public.payment_provider_operations
     WHERE id = p_operation_id
       AND user_id = p_user_id
       AND provider = p_provider
  ) THEN
    RAISE EXCEPTION 'invalid_payment_operation';
  END IF;

  IF p_entitled AND EXISTS (
    SELECT 1
      FROM public.payment_provider_operations
     WHERE user_id = p_user_id
       AND provider <> p_provider
       AND state IN ('reserved', 'external_created', 'charged')
       AND expires_at > now()
  ) THEN
    v_result := jsonb_build_object(
      'applied', false,
      'duplicate', false,
      'reason', 'other_provider_operation_in_progress'
    );
    UPDATE public.payment_provider_events
       SET disposition = 'rejected', result = v_result, processed_at = now()
     WHERE id = v_event.id;
    RETURN v_result;
  END IF;

  IF p_entitled AND v_subscription.provider NOT IN ('none', p_provider) THEN
    v_result := jsonb_build_object(
      'applied', false,
      'duplicate', false,
      'reason', 'active_subscription_other_provider',
      'owner_provider', v_subscription.provider
    );
    UPDATE public.payment_provider_events
       SET disposition = 'rejected', result = v_result, processed_at = now()
     WHERE id = v_event.id;
    RETURN v_result;
  END IF;

  IF NOT p_entitled AND v_subscription.provider <> p_provider THEN
    v_result := jsonb_build_object(
      'applied', false,
      'duplicate', false,
      'reason', 'provider_not_owner',
      'owner_provider', v_subscription.provider
    );
    UPDATE public.payment_provider_events
       SET disposition = 'ignored', result = v_result, processed_at = now()
     WHERE id = v_event.id;
    RETURN v_result;
  END IF;

  IF NOT p_entitled
     AND v_subscription.provider_resource_id IS DISTINCT FROM trim(p_provider_resource_id) THEN
    v_result := jsonb_build_object(
      'applied', false,
      'duplicate', false,
      'reason', 'provider_resource_not_owner'
    );
    UPDATE public.payment_provider_events
       SET disposition = 'ignored', result = v_result, processed_at = now()
     WHERE id = v_event.id;
    RETURN v_result;
  END IF;

  IF p_entitled THEN
    UPDATE public.subscriptions
       SET tier = p_tier,
           status = p_status,
           current_period_start = p_current_period_start,
           current_period_end = p_current_period_end,
           cancel_at = p_cancel_at,
           provider = p_provider,
           provider_resource_id = trim(p_provider_resource_id),
           provider_event_id = trim(p_event_id),
           provider_event_created_at = p_event_created_at,
           auto_renewing = p_auto_renewing,
           stripe_customer_id = CASE
             WHEN p_provider = 'stripe' THEN coalesce(nullif(trim(p_provider_customer_id), ''), stripe_customer_id)
             ELSE stripe_customer_id
           END,
           stripe_subscription_id = CASE
             WHEN p_provider = 'stripe' THEN trim(p_provider_resource_id)
             ELSE stripe_subscription_id
           END,
           payple_payer_id = CASE
             WHEN p_provider = 'payple' AND p_provider_customer_id = '' THEN NULL
             WHEN p_provider = 'payple' AND p_provider_customer_id IS NOT NULL
               THEN trim(p_provider_customer_id)
             ELSE payple_payer_id
           END,
           payple_pay_oid = CASE
             WHEN p_provider = 'payple' THEN coalesce(nullif(trim(p_provider_order_id), ''), payple_pay_oid)
             ELSE payple_pay_oid
           END,
           store_product_id = CASE
             WHEN p_provider IN ('google_play', 'app_store') THEN p_store_product_id
             ELSE NULL
           END,
           store_purchase_id = CASE
             WHEN p_provider IN ('google_play', 'app_store') THEN p_store_purchase_id
             ELSE NULL
           END,
           renewal_failures = CASE WHEN p_provider = 'payple' THEN 0 ELSE renewal_failures END,
           updated_at = now()
     WHERE user_id = p_user_id;
  ELSE
    UPDATE public.subscriptions
       SET tier = 'free',
           status = p_status,
           current_period_start = coalesce(p_current_period_start, current_period_start),
           current_period_end = coalesce(p_current_period_end, current_period_end),
           cancel_at = coalesce(p_cancel_at, p_current_period_end, now()),
           provider = 'none',
           provider_resource_id = NULL,
           provider_event_id = trim(p_event_id),
           provider_event_created_at = p_event_created_at,
           auto_renewing = false,
           store_product_id = NULL,
           store_purchase_id = NULL,
           updated_at = now()
     WHERE user_id = p_user_id;
  END IF;

  IF p_operation_id IS NOT NULL THEN
    UPDATE public.payment_provider_operations
       SET state = 'applied',
           external_reference = coalesce(
             nullif(trim(p_provider_order_id), ''),
             nullif(trim(p_provider_resource_id), ''),
             external_reference
           ),
           error_code = NULL,
           updated_at = now()
     WHERE id = p_operation_id;
  END IF;

  v_result := jsonb_build_object(
    'applied', true,
    'duplicate', false,
    'provider', CASE WHEN p_entitled THEN p_provider ELSE 'none' END,
    'tier', CASE WHEN p_entitled THEN p_tier ELSE 'free' END,
    'status', p_status,
    'entitled', p_entitled
  );
  UPDATE public.payment_provider_events
     SET disposition = 'applied', result = v_result, processed_at = now()
   WHERE id = v_event.id;
  RETURN v_result;
END;
$$;

CREATE OR REPLACE FUNCTION public.record_payment_provider_renewal_failure(
  p_user_id uuid,
  p_provider text,
  p_event_id text,
  p_event_created_at timestamptz,
  p_payload_digest text,
  p_provider_resource_id text,
  p_error_code text,
  p_failure_threshold integer,
  p_operation_id uuid DEFAULT NULL
) RETURNS jsonb
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = public, pg_temp
AS $$
DECLARE
  v_event public.payment_provider_events%ROWTYPE;
  v_cursor public.payment_provider_cursors%ROWTYPE;
  v_subscription public.subscriptions%ROWTYPE;
  v_failures integer;
  v_revoked boolean := false;
  v_result jsonb;
BEGIN
  IF p_provider NOT IN ('stripe', 'payple', 'google_play', 'app_store') THEN
    RAISE EXCEPTION 'invalid_provider';
  END IF;
  IF p_event_id IS NULL OR length(trim(p_event_id)) NOT BETWEEN 3 AND 255 THEN
    RAISE EXCEPTION 'invalid_event_id';
  END IF;
  IF p_event_created_at IS NULL OR p_event_created_at > now() + interval '10 minutes' THEN
    RAISE EXCEPTION 'invalid_event_created_at';
  END IF;
  IF p_payload_digest IS NULL OR p_payload_digest !~ '^[0-9a-f]{64}$' THEN
    RAISE EXCEPTION 'invalid_payload_digest';
  END IF;
  IF p_provider_resource_id IS NULL OR length(trim(p_provider_resource_id)) NOT BETWEEN 1 AND 255 THEN
    RAISE EXCEPTION 'invalid_provider_resource_id';
  END IF;
  IF p_error_code IS NULL OR p_error_code !~ '^[a-z0-9_:-]{1,100}$' THEN
    RAISE EXCEPTION 'invalid_error_code';
  END IF;
  IF p_failure_threshold NOT BETWEEN 1 AND 10 THEN
    RAISE EXCEPTION 'invalid_failure_threshold';
  END IF;

  PERFORM pg_advisory_xact_lock(hashtextextended(p_user_id::text, 73031));

  INSERT INTO public.payment_provider_events (
    provider, event_id, user_id, provider_resource_id, event_type,
    event_created_at, payload_digest
  ) VALUES (
    p_provider, trim(p_event_id), p_user_id, trim(p_provider_resource_id),
    'renewal.failed', p_event_created_at, p_payload_digest
  )
  ON CONFLICT (provider, event_id) DO NOTHING
  RETURNING * INTO v_event;

  IF v_event.id IS NULL THEN
    SELECT * INTO v_event
      FROM public.payment_provider_events
     WHERE provider = p_provider AND event_id = trim(p_event_id)
     FOR UPDATE;
    IF v_event.user_id <> p_user_id OR v_event.payload_digest <> p_payload_digest THEN
      RAISE EXCEPTION 'provider_event_payload_mismatch';
    END IF;
    RETURN coalesce(v_event.result, jsonb_build_object(
      'applied', false, 'reason', 'event_processing_in_progress'
    )) || jsonb_build_object('duplicate', true);
  END IF;

  SELECT * INTO v_cursor
    FROM public.payment_provider_cursors
   WHERE user_id = p_user_id AND provider = p_provider
   FOR UPDATE;
  IF v_cursor.user_id IS NOT NULL AND (
    v_cursor.last_event_created_at > p_event_created_at
    OR (v_cursor.last_event_created_at = p_event_created_at
        AND v_cursor.last_event_id >= trim(p_event_id))
  ) THEN
    v_result := jsonb_build_object('applied', false, 'duplicate', false, 'reason', 'stale_provider_event');
    UPDATE public.payment_provider_events
       SET disposition = 'ignored', result = v_result, processed_at = now()
     WHERE id = v_event.id;
    RETURN v_result;
  END IF;

  INSERT INTO public.payment_provider_cursors (user_id, provider, last_event_created_at, last_event_id)
  VALUES (p_user_id, p_provider, p_event_created_at, trim(p_event_id))
  ON CONFLICT (user_id, provider) DO UPDATE
    SET last_event_created_at = EXCLUDED.last_event_created_at,
        last_event_id = EXCLUDED.last_event_id,
        updated_at = now();

  SELECT * INTO v_subscription
    FROM public.subscriptions
   WHERE user_id = p_user_id
   FOR UPDATE;
  IF v_subscription.id IS NULL
     OR v_subscription.provider <> p_provider
     OR v_subscription.provider_resource_id IS DISTINCT FROM trim(p_provider_resource_id) THEN
    v_result := jsonb_build_object('applied', false, 'duplicate', false, 'reason', 'provider_resource_not_owner');
    UPDATE public.payment_provider_events
       SET disposition = 'ignored', result = v_result, processed_at = now()
     WHERE id = v_event.id;
    RETURN v_result;
  END IF;

  v_failures := v_subscription.renewal_failures + 1;
  v_revoked := v_failures >= p_failure_threshold
               AND coalesce(v_subscription.current_period_end, now()) <= now();

  UPDATE public.subscriptions
     SET renewal_failures = v_failures,
         status = CASE WHEN v_revoked THEN 'expired' ELSE 'past_due' END,
         tier = CASE WHEN v_revoked THEN 'free' ELSE tier END,
         provider = CASE WHEN v_revoked THEN 'none' ELSE provider END,
         provider_resource_id = CASE WHEN v_revoked THEN NULL ELSE provider_resource_id END,
         auto_renewing = CASE WHEN v_revoked THEN false ELSE auto_renewing END,
         provider_event_id = trim(p_event_id),
         provider_event_created_at = p_event_created_at,
         updated_at = now()
   WHERE user_id = p_user_id;

  IF p_operation_id IS NOT NULL THEN
    UPDATE public.payment_provider_operations
       SET state = 'failed', error_code = p_error_code, updated_at = now()
     WHERE id = p_operation_id
       AND user_id = p_user_id
       AND provider = p_provider;
  END IF;

  v_result := jsonb_build_object(
    'applied', true,
    'duplicate', false,
    'renewal_failures', v_failures,
    'revoked', v_revoked,
    'status', CASE WHEN v_revoked THEN 'expired' ELSE 'past_due' END
  );
  UPDATE public.payment_provider_events
     SET disposition = 'applied', result = v_result, processed_at = now()
   WHERE id = v_event.id;
  RETURN v_result;
END;
$$;

-- Preserve the public signature used by iap-verify and RTDN.  Store receipt
-- persistence happens under the same user lock, then the normalized store
-- entitlement is routed through apply_payment_provider_event.
CREATE OR REPLACE FUNCTION public.apply_verified_store_purchase(
  p_user_id uuid,
  p_platform text,
  p_product_id text,
  p_store_transaction_id text,
  p_token_hash text,
  p_purchase_token text,
  p_purchase_state text,
  p_purchase_at timestamptz,
  p_expires_at timestamptz,
  p_auto_renewing boolean,
  p_acknowledged boolean,
  p_tier text,
  p_entitled boolean,
  p_verification jsonb
) RETURNS jsonb
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = public, pg_temp
AS $$
DECLARE
  v_purchase_id uuid;
  v_existing_user_id uuid;
  v_status text;
  v_event_id text;
  v_result jsonb;
BEGIN
  IF p_user_id IS NULL OR NOT EXISTS (SELECT 1 FROM auth.users WHERE id = p_user_id) THEN
    RAISE EXCEPTION 'unknown_user';
  END IF;
  IF p_platform NOT IN ('google_play', 'app_store') THEN
    RAISE EXCEPTION 'invalid_platform';
  END IF;
  IF p_tier NOT IN ('pro', 'pro_plus') THEN
    RAISE EXCEPTION 'invalid_tier';
  END IF;
  IF p_purchase_state NOT IN ('pending', 'purchased', 'cancelled', 'expired', 'refunded', 'on_hold', 'paused') THEN
    RAISE EXCEPTION 'invalid_purchase_state';
  END IF;
  IF p_token_hash IS NULL OR p_token_hash !~ '^[0-9a-f]{64}$'
     OR p_purchase_token IS NULL OR length(trim(p_purchase_token)) < 8 THEN
    RAISE EXCEPTION 'invalid_purchase_token';
  END IF;

  PERFORM pg_advisory_xact_lock(hashtextextended(p_user_id::text, 73031));

  SELECT user_id INTO v_existing_user_id
    FROM public.iap_purchases
   WHERE platform = p_platform AND token_hash = p_token_hash
   FOR UPDATE;
  IF v_existing_user_id IS NOT NULL AND v_existing_user_id <> p_user_id THEN
    RAISE EXCEPTION 'purchase_owned_by_other_user';
  END IF;

  INSERT INTO public.iap_purchases (
    user_id, platform, product_id, store_transaction_id, token_hash,
    purchase_state, purchase_at, expires_at, auto_renewing,
    acknowledged_at, verified_at
  ) VALUES (
    p_user_id, p_platform, trim(p_product_id), nullif(trim(p_store_transaction_id), ''),
    p_token_hash, p_purchase_state, p_purchase_at, p_expires_at, p_auto_renewing,
    CASE WHEN p_acknowledged THEN now() ELSE NULL END, now()
  )
  ON CONFLICT (platform, token_hash) DO UPDATE
    SET product_id = EXCLUDED.product_id,
        store_transaction_id = EXCLUDED.store_transaction_id,
        purchase_state = EXCLUDED.purchase_state,
        purchase_at = EXCLUDED.purchase_at,
        expires_at = EXCLUDED.expires_at,
        auto_renewing = EXCLUDED.auto_renewing,
        acknowledged_at = CASE
          WHEN EXCLUDED.acknowledged_at IS NOT NULL THEN EXCLUDED.acknowledged_at
          ELSE public.iap_purchases.acknowledged_at
        END,
        verified_at = now(),
        updated_at = now()
  RETURNING id INTO v_purchase_id;

  INSERT INTO public.iap_purchase_receipts (purchase_id, purchase_token, verification)
  VALUES (v_purchase_id, p_purchase_token, coalesce(p_verification, '{}'::jsonb))
  ON CONFLICT (purchase_id) DO UPDATE
    SET purchase_token = EXCLUDED.purchase_token,
        verification = EXCLUDED.verification,
        updated_at = now();

  v_status := CASE p_purchase_state
    WHEN 'purchased' THEN 'active'
    WHEN 'cancelled' THEN 'canceled'
    WHEN 'on_hold' THEN 'on_hold'
    ELSE p_purchase_state
  END;
  v_event_id := concat_ws(
    ':', p_platform, p_token_hash, p_purchase_state,
    coalesce(extract(epoch FROM p_expires_at)::bigint::text, 'none'),
    CASE WHEN p_entitled THEN 'entitled' ELSE 'not-entitled' END
  );

  v_result := public.apply_payment_provider_event(
    p_user_id => p_user_id,
    p_provider => p_platform,
    p_event_id => v_event_id,
    p_event_created_at => clock_timestamp(),
    p_event_type => 'store.purchase.' || p_purchase_state,
    p_payload_digest => encode(extensions.digest(coalesce(p_verification, '{}'::jsonb)::text, 'sha256'), 'hex'),
    p_provider_resource_id => v_purchase_id::text,
    p_tier => CASE WHEN p_entitled THEN p_tier ELSE 'free' END,
    p_status => v_status,
    p_entitled => p_entitled,
    p_current_period_start => p_purchase_at,
    p_current_period_end => p_expires_at,
    p_cancel_at => CASE WHEN p_auto_renewing THEN NULL ELSE p_expires_at END,
    p_auto_renewing => p_auto_renewing,
    p_provider_customer_id => NULL,
    p_provider_order_id => p_store_transaction_id,
    p_store_product_id => trim(p_product_id),
    p_store_purchase_id => v_purchase_id,
    p_operation_id => NULL
  );

  IF p_entitled AND v_result->>'reason' IN (
    'active_subscription_other_provider',
    'other_provider_operation_in_progress'
  ) THEN
    RAISE EXCEPTION 'active_subscription_other_provider';
  END IF;

  RETURN v_result || jsonb_build_object(
    'purchase_id', v_purchase_id,
    'acknowledged', p_acknowledged
  );
END;
$$;

-- Preserve the Google Play wrapper signature and linked-purchase semantics.
CREATE OR REPLACE FUNCTION public.apply_verified_google_play_purchase(
  p_user_id uuid,
  p_platform text,
  p_product_id text,
  p_store_transaction_id text,
  p_token_hash text,
  p_linked_token_hash text,
  p_purchase_token text,
  p_purchase_state text,
  p_purchase_at timestamptz,
  p_expires_at timestamptz,
  p_auto_renewing boolean,
  p_acknowledged boolean,
  p_tier text,
  p_entitled boolean,
  p_verification jsonb
) RETURNS jsonb
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = public, pg_temp
AS $$
DECLARE
  v_linked_purchase_id uuid;
  v_linked_user_id uuid;
  v_linked_revoked boolean := false;
  v_result jsonb;
BEGIN
  IF p_platform <> 'google_play' THEN
    RAISE EXCEPTION 'invalid_platform';
  END IF;

  PERFORM pg_advisory_xact_lock(hashtextextended(p_user_id::text, 73031));

  IF p_linked_token_hash IS NOT NULL THEN
    IF p_linked_token_hash !~ '^[0-9a-f]{64}$' OR p_linked_token_hash = p_token_hash THEN
      RAISE EXCEPTION 'invalid_linked_purchase_token';
    END IF;

    SELECT id, user_id INTO v_linked_purchase_id, v_linked_user_id
      FROM public.iap_purchases
     WHERE platform = 'google_play' AND token_hash = p_linked_token_hash
     FOR UPDATE;
    IF v_linked_user_id IS NOT NULL AND v_linked_user_id <> p_user_id THEN
      RAISE EXCEPTION 'linked_purchase_owned_by_other_user';
    END IF;

    IF v_linked_purchase_id IS NOT NULL THEN
      UPDATE public.iap_purchases
         SET purchase_state = 'expired',
             expires_at = least(coalesce(expires_at, now()), now()),
             auto_renewing = false,
             verified_at = now(),
             updated_at = now()
       WHERE id = v_linked_purchase_id;
      v_linked_revoked := true;

      IF NOT p_entitled AND EXISTS (
        SELECT 1 FROM public.subscriptions
         WHERE user_id = p_user_id
           AND provider = 'google_play'
           AND store_purchase_id = v_linked_purchase_id
      ) THEN
        PERFORM public.apply_payment_provider_event(
          p_user_id => p_user_id,
          p_provider => 'google_play',
          p_event_id => 'google_play:linked-revoked:' || p_linked_token_hash || ':' || p_token_hash,
          p_event_created_at => clock_timestamp(),
          p_event_type => 'store.purchase.linked_revoked',
          p_payload_digest => encode(extensions.digest(p_linked_token_hash || ':' || p_token_hash, 'sha256'), 'hex'),
          p_provider_resource_id => v_linked_purchase_id::text,
          p_tier => 'free',
          p_status => 'expired',
          p_entitled => false,
          p_current_period_start => NULL,
          p_current_period_end => now(),
          p_cancel_at => now(),
          p_auto_renewing => false,
          p_provider_customer_id => NULL,
          p_provider_order_id => NULL,
          p_store_product_id => NULL,
          p_store_purchase_id => v_linked_purchase_id,
          p_operation_id => NULL
        );
      END IF;
    END IF;
  END IF;

  v_result := public.apply_verified_store_purchase(
    p_user_id,
    p_platform,
    p_product_id,
    p_store_transaction_id,
    p_token_hash,
    p_purchase_token,
    p_purchase_state,
    p_purchase_at,
    p_expires_at,
    p_auto_renewing,
    p_acknowledged,
    p_tier,
    p_entitled,
    p_verification
  );

  RETURN v_result || jsonb_build_object('linked_purchase_revoked', v_linked_revoked);
END;
$$;

REVOKE ALL ON TABLE public.payment_provider_events FROM PUBLIC, anon, authenticated;
REVOKE ALL ON TABLE public.payment_provider_cursors FROM PUBLIC, anon, authenticated;
REVOKE ALL ON TABLE public.payment_provider_operations FROM PUBLIC, anon, authenticated;
GRANT ALL ON TABLE public.payment_provider_events TO service_role;
GRANT ALL ON TABLE public.payment_provider_cursors TO service_role;
GRANT ALL ON TABLE public.payment_provider_operations TO service_role;

REVOKE ALL ON FUNCTION public.reserve_payment_provider_operation(
  uuid, text, text, text, text, text, text
) FROM PUBLIC, anon, authenticated;
REVOKE ALL ON FUNCTION public.mark_payment_provider_operation(
  uuid, text, text, text
) FROM PUBLIC, anon, authenticated;
REVOKE ALL ON FUNCTION public.record_payment_provider_observation(
  uuid, text, text, timestamptz, text, text, text
) FROM PUBLIC, anon, authenticated;
REVOKE ALL ON FUNCTION public.apply_payment_provider_event(
  uuid, text, text, timestamptz, text, text, text, text, text, boolean,
  timestamptz, timestamptz, timestamptz, boolean, text, text, text, uuid, uuid
) FROM PUBLIC, anon, authenticated;
REVOKE ALL ON FUNCTION public.record_payment_provider_renewal_failure(
  uuid, text, text, timestamptz, text, text, text, integer, uuid
) FROM PUBLIC, anon, authenticated;
REVOKE ALL ON FUNCTION public.apply_verified_store_purchase(
  uuid, text, text, text, text, text, text, timestamptz, timestamptz,
  boolean, boolean, text, boolean, jsonb
) FROM PUBLIC, anon, authenticated;
REVOKE ALL ON FUNCTION public.apply_verified_google_play_purchase(
  uuid, text, text, text, text, text, text, text, timestamptz, timestamptz,
  boolean, boolean, text, boolean, jsonb
) FROM PUBLIC, anon, authenticated;

GRANT EXECUTE ON FUNCTION public.reserve_payment_provider_operation(
  uuid, text, text, text, text, text, text
) TO service_role;
GRANT EXECUTE ON FUNCTION public.mark_payment_provider_operation(
  uuid, text, text, text
) TO service_role;
GRANT EXECUTE ON FUNCTION public.record_payment_provider_observation(
  uuid, text, text, timestamptz, text, text, text
) TO service_role;
GRANT EXECUTE ON FUNCTION public.apply_payment_provider_event(
  uuid, text, text, timestamptz, text, text, text, text, text, boolean,
  timestamptz, timestamptz, timestamptz, boolean, text, text, text, uuid, uuid
) TO service_role;
GRANT EXECUTE ON FUNCTION public.record_payment_provider_renewal_failure(
  uuid, text, text, timestamptz, text, text, text, integer, uuid
) TO service_role;
GRANT EXECUTE ON FUNCTION public.apply_verified_store_purchase(
  uuid, text, text, text, text, text, text, timestamptz, timestamptz,
  boolean, boolean, text, boolean, jsonb
) TO service_role;
GRANT EXECUTE ON FUNCTION public.apply_verified_google_play_purchase(
  uuid, text, text, text, text, text, text, text, timestamptz, timestamptz,
  boolean, boolean, text, boolean, jsonb
) TO service_role;

COMMIT;