Detect & correct SQL database instances with incorrect labels

with original_labels as (
  select
    coalesce(name, title) as title,
    name as id,
    project,
    sp_connection_name as cred,
    '' as zone,
    coalesce(labels, '{}' :: jsonb) as labels,
    l.key,
    l.value
  from
    gcp_sql_database_instance
    left join jsonb_each_text(labels) as l(key, value) on true
),
updated_labels as (
  select
    id,
    key as old_key,
    case
      when false then key
      else key
    end as new_key,
    value
  from
    original_labels
),
required_labels as (
  select
    r.id,
    null as old_key,
    a.key as new_key,
    a.value
  from
    (
      select
        distinct name as id
      from
        gcp_sql_database_instance
    ) r
    cross join (
      values
        (null, null)
    ) as a(key, value)
  where
    not exists (
      select
        1
      from
        updated_labels ul
      where
        ul.id = r.id
        and ul.new_key = a.key
    )
),
all_labels as (
  select
    id,
    old_key,
    new_key,
    value
  from
    updated_labels
  union all
  select
    id,
    old_key,
    new_key,
    value
  from
    required_labels
  where
    new_key is not null
),
allowed_labels as (
  select
    distinct id,
    new_key
  from
    (
      select
        id,
        new_key,
        case
          when new_key like '%' then true
          else false
        end as allowed
      from
        all_labels
    ) a
  where
    allowed = true
),
remove_labels as (
  select
    distinct id,
    key
  from
    (
      select
        id,
        new_key as key,
        case
          when new_key like '%' then false
          else false
        end as remove
      from
        all_labels
    ) r
  where
    remove = true
  union
  select
    id,
    old_key as key
  from
    all_labels
  where
    old_key is not null
    and old_key != new_key
  union
  select
    id,
    new_key as key
  from
    all_labels a
  where
    not exists (
      select
        1
      from
        allowed_labels al
      where
        al.id = a.id
        and al.new_key = a.new_key
    )
),
updated_values as (
  select
    id,
    new_key,
    value as old_value,
    case
      when false then value
      else value
    end as updated_value
  from
    all_labels
)
select
  *
from
  (
    select
      l.title,
      l.id :: text,
      l.project,
      l.zone,
      l.cred,
      coalesce(
        (
          select
            jsonb_agg(key)
          from
            remove_labels rl
          where
            rl.id = l.id
            and key is not null
        ),
        '[]' :: jsonb
      ) as remove,
      coalesce(
        (
          select
            jsonb_object_agg(al.new_key, al.value)
          from
            all_labels al
          where
            al.id = l.id
            and al.new_key != coalesce(al.old_key, '')
            and not exists (
              select
                1
              from
                remove_labels rl
              where
                rl.id = al.id
                and rl.key = al.new_key
            )
        ),
        '{}' :: jsonb
      ) || coalesce(
        (
          select
            jsonb_object_agg(uv.new_key, uv.updated_value)
          from
            updated_values uv
          where
            uv.id = l.id
            and uv.updated_value != uv.old_value
            and not exists (
              select
                1
              from
                remove_labels rl
              where
                rl.id = uv.id
                and rl.key = uv.new_key
            )
        ),
        '{}' :: jsonb
      ) as upsert
    from
      original_labels l
    group by
      l.title,
      l.id,
      l.project,
      l.zone,
      l.cred
  ) result
where
  remove != '[]' :: jsonb
  or upsert != '{}' :: jsonb;