create or replace function public.merge_raw_email_events(
  p_import_id bigint,
  p_rows jsonb
)
returns table(inserted_rows integer, updated_rows integer, failed_rows integer)
language plpgsql
security definer
set search_path = public
as $$
declare
  v_row jsonb;
  v_user_id uuid := auth.uid();
  v_is_admin boolean := public.is_admin(v_user_id);
  v_sent_email_name text;
  v_existing_country text;
  v_target_country text;
begin
  inserted_rows := 0;
  updated_rows := 0;
  failed_rows := 0;

  for v_row in select * from jsonb_array_elements(p_rows)
  loop
    begin
      v_sent_email_name := trim(coalesce(v_row->>'sent_email_name', ''));

      if v_sent_email_name = '' then
        failed_rows := failed_rows + 1;
        continue;
      end if;

      select ree.primary_country_code
      into v_existing_country
      from public.raw_email_events ree
      where ree.sent_email_name = v_sent_email_name;

      v_target_country := upper(coalesce(nullif(v_existing_country, ''), nullif(v_row->>'primary_country_code', '')));

      if not v_is_admin and (v_target_country is null or not public.has_country_access(v_target_country, v_user_id)) then
        failed_rows := failed_rows + 1;
        continue;
      end if;

      if exists (select 1 from public.raw_email_events ree where ree.sent_email_name = v_sent_email_name) then
        updated_rows := updated_rows + 1;
      else
        inserted_rows := inserted_rows + 1;
      end if;

      insert into public.raw_email_events (
        sent_email_name,
        primary_country_code,
        approved_document_name,
        last_open,
        opened,
        total_opens,
        last_click,
        clicked,
        total_clicks,
        sent_date,
        latest_import_id
      )
      values (
        v_sent_email_name,
        upper(coalesce(v_row->>'primary_country_code', '')),
        coalesce(v_row->>'approved_document_name', ''),
        nullif(v_row->>'last_open', '')::timestamptz,
        greatest(coalesce((v_row->>'opened')::integer, 0), 0),
        greatest(coalesce((v_row->>'total_opens')::integer, 0), 0),
        nullif(v_row->>'last_click', '')::timestamptz,
        greatest(coalesce((v_row->>'clicked')::integer, 0), 0),
        greatest(coalesce((v_row->>'total_clicks')::integer, 0), 0),
        nullif(v_row->>'sent_date', '')::timestamptz,
        p_import_id
      )
      on conflict (sent_email_name)
      do update set
        primary_country_code = coalesce(nullif(excluded.primary_country_code, ''), raw_email_events.primary_country_code),
        approved_document_name = coalesce(nullif(excluded.approved_document_name, ''), raw_email_events.approved_document_name),
        last_open = coalesce(excluded.last_open, raw_email_events.last_open),
        opened = coalesce(excluded.opened, raw_email_events.opened, 0),
        total_opens = coalesce(excluded.total_opens, raw_email_events.total_opens, 0),
        last_click = coalesce(excluded.last_click, raw_email_events.last_click),
        clicked = coalesce(excluded.clicked, raw_email_events.clicked, 0),
        total_clicks = coalesce(excluded.total_clicks, raw_email_events.total_clicks, 0),
        sent_date = coalesce(excluded.sent_date, raw_email_events.sent_date),
        latest_import_id = p_import_id,
        updated_at = timezone('utc'::text, now());
    exception
      when others then
        failed_rows := failed_rows + 1;
    end;
  end loop;

  return query select inserted_rows, updated_rows, failed_rows;
end;
$$;

grant execute on function public.merge_raw_email_events(bigint, jsonb) to authenticated;