-- ============================================================================ -- PHASE 1 CRITICAL FIXES - Bulletproof Pipeline -- ============================================================================ -- 1. Add idempotency parameter to RPC -- 2. Remove item-level exception handling (ensure full rollback) -- 3. Add timeout protection -- 4. Add idempotency check at start of transaction -- ============================================================================ -- Drop and recreate the main RPC with fixes DROP FUNCTION IF EXISTS process_approval_transaction(UUID, UUID[], UUID, UUID, TEXT); CREATE OR REPLACE FUNCTION process_approval_transaction( p_submission_id UUID, p_item_ids UUID[], p_moderator_id UUID, p_submitter_id UUID, p_request_id TEXT DEFAULT NULL, p_idempotency_key TEXT DEFAULT NULL ) RETURNS JSONB LANGUAGE plpgsql SECURITY DEFINER SET search_path = public AS $$ DECLARE v_start_time TIMESTAMPTZ; v_result JSONB; v_item RECORD; v_item_data JSONB; v_entity_id UUID; v_approval_results JSONB[] := ARRAY[]::JSONB[]; v_final_status TEXT; v_all_approved BOOLEAN := TRUE; v_some_approved BOOLEAN := FALSE; v_items_processed INTEGER := 0; v_existing_key RECORD; BEGIN v_start_time := clock_timestamp(); -- ======================================================================== -- STEP 0: TIMEOUT PROTECTION -- ======================================================================== SET LOCAL statement_timeout = '60s'; SET LOCAL lock_timeout = '10s'; SET LOCAL idle_in_transaction_session_timeout = '30s'; RAISE NOTICE '[%] Starting atomic approval transaction for submission %', COALESCE(p_request_id, 'NO_REQUEST_ID'), p_submission_id; -- ======================================================================== -- STEP 0.5: IDEMPOTENCY CHECK -- ======================================================================== IF p_idempotency_key IS NOT NULL THEN SELECT * INTO v_existing_key FROM submission_idempotency_keys WHERE idempotency_key = p_idempotency_key; IF FOUND THEN IF v_existing_key.status = 'completed' THEN RAISE NOTICE '[%] Idempotency key already processed, returning cached result', COALESCE(p_request_id, 'NO_REQUEST_ID'); RETURN v_existing_key.result_data; ELSIF v_existing_key.status = 'processing' AND v_existing_key.created_at > NOW() - INTERVAL '5 minutes' THEN RAISE EXCEPTION 'Request already in progress' USING ERRCODE = '40P01'; -- deadlock_detected (will trigger retry) END IF; -- If stale 'processing' key (>5 min old), continue and overwrite END IF; END IF; -- ======================================================================== -- STEP 1: Set session variables (transaction-scoped with is_local=true) -- ======================================================================== PERFORM set_config('app.current_user_id', p_submitter_id::text, true); PERFORM set_config('app.submission_id', p_submission_id::text, true); PERFORM set_config('app.moderator_id', p_moderator_id::text, true); -- ======================================================================== -- STEP 2: Validate submission ownership and lock status -- ======================================================================== IF NOT EXISTS ( SELECT 1 FROM content_submissions WHERE id = p_submission_id AND (assigned_to = p_moderator_id OR assigned_to IS NULL) AND status IN ('pending', 'partially_approved') ) THEN RAISE EXCEPTION 'Submission not found, locked by another moderator, or already processed' USING ERRCODE = '42501'; END IF; -- ======================================================================== -- STEP 3: Process each item sequentially within this transaction -- NO EXCEPTION HANDLER - Let failures trigger full rollback -- ======================================================================== FOR v_item IN SELECT si.*, ps.name as park_name, ps.slug as park_slug, ps.description as park_description, ps.park_type, ps.status as park_status, ps.location_id, ps.operator_id, ps.property_owner_id, ps.opening_date as park_opening_date, ps.closing_date as park_closing_date, ps.opening_date_precision as park_opening_date_precision, ps.closing_date_precision as park_closing_date_precision, ps.website_url as park_website_url, ps.phone as park_phone, ps.email as park_email, ps.banner_image_url as park_banner_image_url, ps.banner_image_id as park_banner_image_id, ps.card_image_url as park_card_image_url, ps.card_image_id as park_card_image_id, rs.name as ride_name, rs.slug as ride_slug, rs.park_id as ride_park_id, rs.ride_type, rs.status as ride_status, rs.manufacturer_id, rs.ride_model_id, rs.opening_date as ride_opening_date, rs.closing_date as ride_closing_date, rs.opening_date_precision as ride_opening_date_precision, rs.closing_date_precision as ride_closing_date_precision, rs.description as ride_description, rs.banner_image_url as ride_banner_image_url, rs.banner_image_id as ride_banner_image_id, rs.card_image_url as ride_card_image_url, rs.card_image_id as ride_card_image_id, cs.name as company_name, cs.slug as company_slug, cs.description as company_description, cs.website_url as company_website_url, cs.founded_year, cs.banner_image_url as company_banner_image_url, cs.banner_image_id as company_banner_image_id, cs.card_image_url as company_card_image_url, cs.card_image_id as company_card_image_id, rms.name as ride_model_name, rms.slug as ride_model_slug, rms.manufacturer_id as ride_model_manufacturer_id, rms.ride_type as ride_model_ride_type, rms.description as ride_model_description, rms.banner_image_url as ride_model_banner_image_url, rms.banner_image_id as ride_model_banner_image_id, rms.card_image_url as ride_model_card_image_url, rms.card_image_id as ride_model_card_image_id FROM submission_items si LEFT JOIN park_submissions ps ON si.park_submission_id = ps.id LEFT JOIN ride_submissions rs ON si.ride_submission_id = rs.id LEFT JOIN company_submissions cs ON si.company_submission_id = cs.id LEFT JOIN ride_model_submissions rms ON si.ride_model_submission_id = rms.id WHERE si.id = ANY(p_item_ids) ORDER BY si.order_index, si.created_at LOOP v_items_processed := v_items_processed + 1; -- Build item data based on entity type IF v_item.item_type = 'park' THEN v_item_data := jsonb_build_object( 'name', v_item.park_name, 'slug', v_item.park_slug, 'description', v_item.park_description, 'park_type', v_item.park_type, 'status', v_item.park_status, 'location_id', v_item.location_id, 'operator_id', v_item.operator_id, 'property_owner_id', v_item.property_owner_id, 'opening_date', v_item.park_opening_date, 'closing_date', v_item.park_closing_date, 'opening_date_precision', v_item.park_opening_date_precision, 'closing_date_precision', v_item.park_closing_date_precision, 'website_url', v_item.park_website_url, 'phone', v_item.park_phone, 'email', v_item.park_email, 'banner_image_url', v_item.park_banner_image_url, 'banner_image_id', v_item.park_banner_image_id, 'card_image_url', v_item.park_card_image_url, 'card_image_id', v_item.park_card_image_id ); ELSIF v_item.item_type = 'ride' THEN v_item_data := jsonb_build_object( 'name', v_item.ride_name, 'slug', v_item.ride_slug, 'park_id', v_item.ride_park_id, 'ride_type', v_item.ride_type, 'status', v_item.ride_status, 'manufacturer_id', v_item.manufacturer_id, 'ride_model_id', v_item.ride_model_id, 'opening_date', v_item.ride_opening_date, 'closing_date', v_item.ride_closing_date, 'opening_date_precision', v_item.ride_opening_date_precision, 'closing_date_precision', v_item.ride_closing_date_precision, 'description', v_item.ride_description, 'banner_image_url', v_item.ride_banner_image_url, 'banner_image_id', v_item.ride_banner_image_id, 'card_image_url', v_item.ride_card_image_url, 'card_image_id', v_item.ride_card_image_id ); ELSIF v_item.item_type IN ('manufacturer', 'operator', 'property_owner', 'designer') THEN v_item_data := jsonb_build_object( 'name', v_item.company_name, 'slug', v_item.company_slug, 'description', v_item.company_description, 'website_url', v_item.company_website_url, 'founded_year', v_item.founded_year, 'banner_image_url', v_item.company_banner_image_url, 'banner_image_id', v_item.company_banner_image_id, 'card_image_url', v_item.company_card_image_url, 'card_image_id', v_item.company_card_image_id ); ELSIF v_item.item_type = 'ride_model' THEN v_item_data := jsonb_build_object( 'name', v_item.ride_model_name, 'slug', v_item.ride_model_slug, 'manufacturer_id', v_item.ride_model_manufacturer_id, 'ride_type', v_item.ride_model_ride_type, 'description', v_item.ride_model_description, 'banner_image_url', v_item.ride_model_banner_image_url, 'banner_image_id', v_item.ride_model_banner_image_id, 'card_image_url', v_item.ride_model_card_image_url, 'card_image_id', v_item.ride_model_card_image_id ); ELSE RAISE EXCEPTION 'Unsupported item_type: %', v_item.item_type; END IF; -- Execute action based on action_type IF v_item.action_type = 'create' THEN v_entity_id := create_entity_from_submission( v_item.item_type, v_item_data, p_submitter_id ); ELSIF v_item.action_type = 'update' THEN v_entity_id := update_entity_from_submission( v_item.item_type, v_item_data, v_item.target_entity_id, p_submitter_id ); ELSIF v_item.action_type = 'delete' THEN PERFORM delete_entity_from_submission( v_item.item_type, v_item.target_entity_id, p_submitter_id ); v_entity_id := v_item.target_entity_id; ELSE RAISE EXCEPTION 'Unknown action_type: %', v_item.action_type; END IF; -- Update submission_item to approved status UPDATE submission_items SET status = 'approved', approved_entity_id = v_entity_id, updated_at = NOW() WHERE id = v_item.id; -- Track success v_approval_results := array_append( v_approval_results, jsonb_build_object( 'itemId', v_item.id, 'entityId', v_entity_id, 'itemType', v_item.item_type, 'actionType', v_item.action_type, 'success', true ) ); v_some_approved := TRUE; RAISE NOTICE '[%] Approved item % (type=%s, action=%s, entityId=%s)', COALESCE(p_request_id, 'NO_REQUEST_ID'), v_item.id, v_item.item_type, v_item.action_type, v_entity_id; END LOOP; -- Clear session variables immediately after use PERFORM set_config('app.current_user_id', '', true); PERFORM set_config('app.submission_id', '', true); PERFORM set_config('app.moderator_id', '', true); -- ======================================================================== -- STEP 4: Determine final submission status -- ======================================================================== v_final_status := 'approved'; -- All items must succeed or transaction rolls back -- ======================================================================== -- STEP 5: Update submission status -- ======================================================================== UPDATE content_submissions SET status = v_final_status, reviewer_id = p_moderator_id, reviewed_at = NOW(), assigned_to = NULL, locked_until = NULL WHERE id = p_submission_id; -- ======================================================================== -- STEP 6: Log metrics (non-critical - wrapped in exception handler) -- ======================================================================== BEGIN INSERT INTO approval_transaction_metrics ( submission_id, moderator_id, submitter_id, items_count, duration_ms, success, request_id ) VALUES ( p_submission_id, p_moderator_id, p_submitter_id, array_length(p_item_ids, 1), EXTRACT(EPOCH FROM (clock_timestamp() - v_start_time)) * 1000, TRUE, p_request_id ); EXCEPTION WHEN OTHERS THEN RAISE WARNING 'Failed to log metrics, but approval succeeded: %', SQLERRM; -- Don't re-raise - metrics are non-critical END; -- ======================================================================== -- STEP 7: Build result -- ======================================================================== v_result := jsonb_build_object( 'success', TRUE, 'results', to_jsonb(v_approval_results), 'submissionStatus', v_final_status, 'itemsProcessed', v_items_processed, 'allApproved', TRUE ); RAISE NOTICE '[%] Transaction completed successfully in %ms', COALESCE(p_request_id, 'NO_REQUEST_ID'), EXTRACT(EPOCH FROM (clock_timestamp() - v_start_time)) * 1000; RETURN v_result; EXCEPTION WHEN OTHERS THEN -- ANY unhandled error triggers automatic ROLLBACK RAISE WARNING '[%] Transaction failed, rolling back: % (SQLSTATE: %)', COALESCE(p_request_id, 'NO_REQUEST_ID'), SQLERRM, SQLSTATE; -- Log failed transaction metrics (best effort) BEGIN INSERT INTO approval_transaction_metrics ( submission_id, moderator_id, submitter_id, items_count, duration_ms, success, rollback_triggered, error_message, request_id ) VALUES ( p_submission_id, p_moderator_id, p_submitter_id, array_length(p_item_ids, 1), EXTRACT(EPOCH FROM (clock_timestamp() - v_start_time)) * 1000, FALSE, TRUE, SQLERRM, p_request_id ); EXCEPTION WHEN OTHERS THEN RAISE WARNING 'Failed to log rollback metrics: %', SQLERRM; END; -- Clear session variables before re-raising PERFORM set_config('app.current_user_id', '', true); PERFORM set_config('app.submission_id', '', true); PERFORM set_config('app.moderator_id', '', true); -- Re-raise the exception to trigger ROLLBACK RAISE; END; $$; -- Grant execute permissions GRANT EXECUTE ON FUNCTION process_approval_transaction TO authenticated;