Shuffle/supabase/migrations/20261103235947_catalog_search.sql

-- KAT-01 / PR 3a: Katalogsuche (Spec docs/specs/3-katalog-suche.md, Abschnitt 3.2 und 4.2).
-- Nur additiv. Enthält:
--   pg_trgm, unaccent            Erweiterungen im Schema extensions
--   search_normalize(text)       Basisfunktion für alle normalisierten Titel (auch für #13)
--   title_search, name_search    generierte Spalten an games, publishers und game_alt_titles + Trigramm-Indizes
--   game_alt_titles              alternative Titel mit Sprache (Klärung 3); nur lesbar für API-Rollen
--   games_manual_fields_check    vollständige Schlüsselliste von main plus 'alt_titles'
--   search_catalog_games         RPC für die Suche (Vertrag: Spec 4.2)

-- ---------------------------------------------------------------------------------------------
-- 1. Erweiterungen (Bestandteil des Supabase-Postgres-Images)
-- ---------------------------------------------------------------------------------------------
create extension if not exists pg_trgm with schema extensions;
create extension if not exists unaccent with schema extensions;

-- ---------------------------------------------------------------------------------------------
-- 2. search_normalize: die einzige Basisfunktion für normalisierte Titel (E3)
-- Schritte: Marken- und Schutzzeichen entfernen, Akzente/Umlaute entfernen, Kleinschreibung, alles
-- außer a-z/0-9 zu einem Leerzeichen, Ränder kürzen, leer = null.
-- ---------------------------------------------------------------------------------------------
create function public.search_normalize(p text)
returns text
language sql
immutable
strict
parallel safe
set search_path = ''
as $$
  select nullif(
    btrim(
      regexp_replace(
        lower(extensions.unaccent('extensions.unaccent'::regdictionary, regexp_replace(p, '[®©™℗℠]', '', 'g'))),
        '[^a-z0-9]+', ' ', 'g')),
    '')
$$;

comment on function public.search_normalize(text) is
  'Normalisierung für Suche und Titelvergleich (KAT-01, gemeinsame Basis mit KAT-04). '
  'Rein rechnend, ohne Tabellenzugriff: deshalb für anon, authenticated und service_role ausführbar '
  '(generierte Spalten und security-invoker-Funktionen rechnen mit den Rechten des Aufrufers). '
  'immutable ist eine Zusicherung: unaccent selbst ist nur stable; der Wrapper nennt das Wörterbuch ausdrücklich. '
  'Ändert sich diese Funktion oder die Datei unaccent.rules (Image-Upgrade), müssen die generierten Spalten '
  'games.title_search, game_alt_titles.title_search und publishers.name_search per Migration neu angelegt werden '
  '(Prüfabfrage: docs/katalog-pflege.md, Abschnitt Image-Upgrade). Artikel werden hier nie entfernt.';

-- Rechte-Regel: zuerst allen entziehen, dann nur die vorgesehenen Grants.
revoke execute on function public.search_normalize(text) from public, anon, authenticated, service_role;
grant execute on function public.search_normalize(text) to anon, authenticated, service_role;

-- ---------------------------------------------------------------------------------------------
-- 3. Generierte Spalten und Trigramm-Indizes
-- Die Indizes sind keine Zusage für Geschwindigkeit: Bei 2-Zeichen-Wörtern, der Regel ohne Leerzeichen
-- und den RLS-Prüfungen für anon werden sie weitgehend nicht genutzt (Messung: supabase/perf/).
-- ---------------------------------------------------------------------------------------------
alter table public.games
  add column title_search text generated always as (public.search_normalize(title)) stored;
alter table public.publishers
  add column name_search text generated always as (public.search_normalize(name)) stored;

-- Der Partial-Index passt nur zu Abfragen mit genau den Bedingungen der View catalog_games.
create index games_title_search_trgm_idx on public.games
  using gin (title_search extensions.gin_trgm_ops)
  where visibility = 'catalog' and hidden_at is null and merged_into_id is null;
create index publishers_name_search_trgm_idx on public.publishers
  using gin (name_search extensions.gin_trgm_ops);

-- ---------------------------------------------------------------------------------------------
-- 4. Alternative Titel (Klärung 3, E2). Pflege nur über Studio und das Startbestand-Werkzeug.
-- ---------------------------------------------------------------------------------------------
create table public.game_alt_titles (
  id           uuid primary key default gen_random_uuid(),
  game_id      uuid not null references public.games (id) on delete cascade,
  title        text not null check (char_length(title) between 1 and 200),
  locale       text not null check (locale in ('de', 'en')),
  title_search text generated always as (public.search_normalize(title)) stored
               constraint game_alt_titles_title_search_check check (title_search is not null),
  created_at   timestamptz not null default now(),
  updated_at   timestamptz not null default now(),
  -- Konfliktziel des Startbestand-Werkzeugs; derselbe Titel in derselben Sprache nur einmal.
  constraint game_alt_titles_game_locale_title_key unique (game_id, locale, title_search)
);

create trigger set_updated_at before update on public.game_alt_titles
  for each row execute function public.set_updated_at();
create trigger game_links_mark_manual after insert or update or delete on public.game_alt_titles
  for each row execute function public.game_links_mark_manual('alt_titles');

create index game_alt_titles_title_search_trgm_idx on public.game_alt_titles
  using gin (title_search extensions.gin_trgm_ops);

alter table public.game_alt_titles enable row level security;
revoke all on public.game_alt_titles from anon, authenticated;
grant select on public.game_alt_titles to anon, authenticated;

-- Lesen: wer das Spiel sehen darf (Muster game_publishers_select). Schreiben: keine Policy, kein Recht.
create policy game_alt_titles_select on public.game_alt_titles
  for select to anon, authenticated
  using (exists (select 1 from public.games g where g.id = game_id));

-- ---------------------------------------------------------------------------------------------
-- 5. manual_fields: vollständige Liste von main (Migration catalog_foundation) plus 'alt_titles'
-- ---------------------------------------------------------------------------------------------
alter table public.games
  drop constraint games_manual_fields_check,
  add constraint games_manual_fields_check
    check (manual_fields <@ array['title', 'description', 'description_locale', 'year',
      'min_players', 'max_players', 'min_minutes', 'max_minutes', 'min_age', 'image_path',
      'genres', 'mechanics', 'categories', 'publishers', 'alt_titles']::text[]);

-- ---------------------------------------------------------------------------------------------
-- 6. RPC search_catalog_games (Vertrag zwischen 3a und 3b: Spec 4.2)
-- security invoker: Für den Aufrufer (auch anon) gelten die Policies von games, catalog_sources,
-- game_publishers, publishers, game_alt_titles, game_genres und game_mechanics; die Funktion umgeht
-- nichts. Gelesen wird über die View catalog_games (ausgeblendete und zusammengeführte Spiele fehlen,
-- Bild und Text sind maskiert); title_search kommt per Primärschlüssel aus games.
-- Der Partial-Index games_title_search_trgm_idx ist nur über die Bedingungen der View nutzbar.
-- Ein set statement_timeout an dieser Funktion wirkt nicht (gilt erst für die nächste Anweisung);
-- es gelten die Zeitlimits der Rollen. Suchbegriffe werden nirgends gespeichert oder protokolliert.
-- ---------------------------------------------------------------------------------------------
create function public.search_catalog_games(
  p_query        text     default null,
  p_players      smallint default null,
  p_max_minutes  smallint default null,
  p_age          smallint default null,
  p_genre_ids    uuid[]   default null,
  p_mechanic_ids uuid[]   default null,
  p_limit        integer  default 30,
  p_offset       integer  default 0
)
returns table (
  id               uuid,
  title            text,
  year             smallint,
  min_players      smallint,
  max_players      smallint,
  min_minutes      smallint,
  max_minutes      smallint,
  min_age          smallint,
  image_path       text,
  publisher_names  text[],
  matched_alt_title text,
  match_rank       smallint,
  total_count      bigint
)
language plpgsql
stable
set search_path = ''
as $$
#variable_conflict use_column
declare
  v_limit     integer := coalesce(p_limit, 30);
  v_offset    integer := coalesce(p_offset, 0);
  v_q         text;
  v_qc        text;
  v_words     text[] := '{}'::text[];
  v_patterns  text[] := '{}'::text[];   -- '%wort%' je Suchwort, für like all
  v_has_query boolean := false;
  v_genres    uuid[];
  v_mechanics uuid[];
  v_genre_n   integer;
  v_mech_n    integer;
begin
  -- Eingabegrenzen (Fehler statt stilles Abschneiden)
  if p_query is not null and char_length(p_query) > 100 then
    raise exception 'search_query_too_long' using errcode = 'P0001';
  end if;
  if (p_players is not null and p_players not between 1 and 99)
     or (p_max_minutes is not null and p_max_minutes not between 1 and 1440)
     or (p_age is not null and p_age not between 0 and 99)
     or cardinality(p_genre_ids) > 20
     or cardinality(p_mechanic_ids) > 20 then
    raise exception 'search_filter_invalid' using errcode = 'P0001';
  end if;
  if v_limit not between 1 and 50 or v_offset not between 0 and 10000 then
    raise exception 'search_paging_invalid' using errcode = 'P0001';
  end if;

  -- Suchbegriff: null/nur Leerraum = kein Suchbegriff; normalisiert null (z. B. '!!!') = keine Treffer;
  -- weniger als 2 Zeichen = kein Suchbegriff.
  if p_query is not null and p_query ~ '\S' then
    v_q := public.search_normalize(p_query);
    if v_q is null then
      return;
    end if;
    if cardinality(string_to_array(v_q, ' ')) > 10 then
      raise exception 'search_query_too_long' using errcode = 'P0001';
    end if;
    if char_length(v_q) >= 2 then
      v_has_query := true;
      v_qc := replace(v_q, ' ', '');
      v_words := coalesce(
        array(select distinct w from unnest(string_to_array(v_q, ' ')) as w where char_length(w) >= 2), '{}');
      v_patterns := array(select '%' || w || '%' from unnest(v_words) as w);
    end if;
  end if;

  v_genres    := array(select distinct x from unnest(p_genre_ids) as x where x is not null);
  v_mechanics := array(select distinct x from unnest(p_mechanic_ids) as x where x is not null);
  v_genre_n   := cardinality(v_genres);
  v_mech_n    := cardinality(v_mechanics);

  if not v_has_query then
    -- Ohne Suchbegriff: Filter wirken, Sortierung nach Titel, match_rank ist null.
    return query
    select s.id, s.title, s.year, s.min_players, s.max_players, s.min_minutes, s.max_minutes, s.min_age,
           s.image_path,
           coalesce((select array_agg(p.name order by p.name_search, p.name)
                       from public.game_publishers gp
                       join public.publishers p on p.id = gp.publisher_id
                      where gp.game_id = s.id), '{}'::text[]),
           null::text, null::smallint, s.total
    from (
      select c.id, c.title, c.year, c.min_players, c.max_players, c.min_minutes, c.max_minutes, c.min_age,
             c.image_path, g.title_search, count(*) over () as total
      from public.catalog_games c
      join public.games g on g.id = c.id
      where (p_players is null or (c.min_players <= p_players and p_players <= c.max_players))
        and (p_max_minutes is null or coalesce(c.max_minutes, c.min_minutes) <= p_max_minutes)
        and (p_age is null or c.min_age <= p_age)
        and (v_genre_n = 0 or c.id in (select gg.game_id from public.game_genres gg where gg.genre_id = any (v_genres)
                                    group by gg.game_id having count(*) = v_genre_n))
        and (v_mech_n = 0 or c.id in (select gm.game_id from public.game_mechanics gm where gm.mechanic_id = any (v_mechanics)
                                    group by gm.game_id having count(*) = v_mech_n))
      order by g.title_search collate "C", c.id
      limit v_limit offset v_offset
    ) s
    order by s.title_search collate "C", s.id;
    return;
  end if;

  return query
  with base as (
    select c.id, c.title, c.year, c.min_players, c.max_players, c.min_minutes, c.max_minutes, c.min_age,
           c.image_path, g.title_search
    from public.catalog_games c
    join public.games g on g.id = c.id
    where (p_players is null or (c.min_players <= p_players and p_players <= c.max_players))
      and (p_max_minutes is null or coalesce(c.max_minutes, c.min_minutes) <= p_max_minutes)
      and (p_age is null or c.min_age <= p_age)
      and (v_genre_n = 0 or c.id in (select gg.game_id from public.game_genres gg where gg.genre_id = any (v_genres)
                                    group by gg.game_id having count(*) = v_genre_n))
      and (v_mech_n = 0 or c.id in (select gm.game_id from public.game_mechanics gm where gm.mechanic_id = any (v_mechanics)
                                    group by gm.game_id having count(*) = v_mech_n))
  ),
  -- Kandidaten-Titel: Haupttitel (alt_title null) und alle alternativen Titel
  cand as (
    select b.id, b.title_search as ts, null::text as alt_title from base b
    union all
    select b.id, a.title_search, a.title
    from base b join public.game_alt_titles a on a.game_id = b.id
  ),
  -- Relevanzstufe je Kandidaten-Titel (null = kein Titeltreffer). Nur günstige Vergleiche; die teure
  -- Trigramm-Ähnlichkeit wird erst für die Treffer berechnet (sims, alt_pick).
  ranked as (
    select t.id, t.ts, t.alt_title,
           case
             -- Stufe 0: gleich dem Suchbegriff (auch ohne Leerzeichen gleich; das schließt ts = v_q ein)
             when replace(t.ts, ' ', '') = v_qc then 0
             when t.ts like v_q || '%' then 1
             when t.ts like '% ' || v_q || '%' then 2
             when replace(t.ts, ' ', '') like '%' || v_qc || '%'
                  or (cardinality(v_patterns) > 0 and t.ts like all (v_patterns))
               then 3
           end as rk
    from cand t
  ),
  agg as (
    select r.id,
           min(r.rk) as best_rk,
           min(r.rk) filter (where r.alt_title is null) as main_rk,
           min(r.rk) filter (where r.alt_title is not null) as alt_rk
    from ranked r
    group by r.id
  ),
  -- der alternative Titel mit der besten Stufe (bei Gleichstand: größte Ähnlichkeit, dann Schreibweise)
  alt_pick as (
    select distinct on (r.id) r.id, r.alt_title
    from ranked r
    where r.alt_title is not null and r.rk is not null
    order by r.id, r.rk, extensions.similarity(r.ts, v_q) desc, r.ts, r.alt_title
  ),
  -- Rang 4: welches Suchwort kommt in welchem Spiel in einem Titel oder einem Verlagsnamen vor?
  -- (als Mengenoperation statt korrelierter Unterabfrage je Spiel: sonst wächst die Zeit quadratisch)
  word_hits as (
    select r.id, w.w from ranked r cross join unnest(v_words) as w(w) where r.ts like '%' || w.w || '%'
    union
    select gp.game_id, w.w
    from public.publishers p
    join public.game_publishers gp on gp.publisher_id = p.id
    cross join unnest(v_words) as w(w)
    where p.name_search like '%' || w.w || '%'
  ),
  word_cover as (
    select wh.id, count(*) as n from word_hits wh group by wh.id
  ),
  hit as (
    select b.*, a.best_rk, a.main_rk, a.alt_rk, ap.alt_title
    from base b
    join agg a on a.id = b.id
    left join alt_pick ap on ap.id = b.id
    left join word_cover wc on wc.id = b.id
    where a.best_rk is not null
       -- Rang 4: jedes Wort kommt in einem Titel oder einem Verlagsnamen des Spiels vor
       or (cardinality(v_words) > 0 and wc.n = cardinality(v_words))
  ),
  -- Ähnlichkeit (zweites Sortierkriterium): die höchste über Haupttitel und alle alternativen Titel
  sims as (
    select r.id, max(extensions.similarity(r.ts, v_q)) as sim
    from ranked r
    join hit h on h.id = r.id
    group by r.id
  ),
  paged as (
    select h.*, sm.sim, coalesce(h.best_rk, 4) as rk, count(*) over () as total
    from hit h
    join sims sm on sm.id = h.id
    order by coalesce(h.best_rk, 4), sm.sim desc nulls last, h.title_search collate "C", h.id
    limit v_limit offset v_offset
  )
  select s.id, s.title, s.year, s.min_players, s.max_players, s.min_minutes, s.max_minutes, s.min_age,
         s.image_path,
         coalesce((select array_agg(p.name order by p.name_search, p.name)
                     from public.game_publishers gp
                     join public.publishers p on p.id = gp.publisher_id
                    where gp.game_id = s.id), '{}'::text[]),
         -- nur, wenn ein alternativer Titel besser getroffen hat als der Haupttitel
         case when s.alt_rk is not null and (s.main_rk is null or s.alt_rk < s.main_rk) then s.alt_title end,
         s.rk::smallint, s.total
  from paged s
  order by s.rk, s.sim desc nulls last, s.title_search collate "C", s.id;
end;
$$;

-- Rechte-Regel: erst allen entziehen, dann nur anon und authenticated (erste Abfrage-RPC für API-Rollen).
revoke execute on function public.search_catalog_games(text, smallint, smallint, smallint, uuid[], uuid[], integer, integer)
  from public, anon, authenticated, service_role;
grant execute on function public.search_catalog_games(text, smallint, smallint, smallint, uuid[], uuid[], integer, integer)
  to anon, authenticated;