-- The entire future agency migration, sketched:
create table public.organizations (id uuid primary key, name text not null, …);
alter table public.workspaces add column organization_id uuid references public.organizations(id);
-- billing moves from workspace owner to organization; no fact table is touched.

auth.users
  └── workspace_members ──┐
                          │  (v1: exactly one 'owner' row per workspace)
  workspaces ─────────────┘   one tracked brand · one market · one cadence · the tenancy boundary
    │
    ├── brands                 name, kind('self'|'competitor'), tracking_state, discovery_source,
    │     │                    primary_domain + domain_status, colour
    │     └── brand_aliases    alias → exactly one brand per workspace (unique, normalised)
    │
    ├── topics                 name, position
    │     └── prompts          text, country_code, language, intent_class, branding_class, is_active
    │           │
    │           └── runs ◄──── THE ATOM.  one (prompt × engine × run_date)
    │                 │        answer_text, model_version, status, metered, cost_micros, latency_ms
    │                 ├── run_mentions    brand_id, char_start/char_end, sentence span, surface_form,
    │                 │                   match_method, is_recommended
    │                 └── run_citations   url_id, raw_uri, order_index, resolution_status
    │
    ├── usage_periods          period_start/end, plan_id, run_limit, runs_used, cost_micros_used
    ├── brand_daily_metrics    run_count, mention_run_count, mention_count, position_sum/count
    └── domain_daily_metrics   run_count, retrieved_run_count, retrieval_count

  engines                      global reference: 'perplexity-sonar' | 'openai-web-search'
                               | 'gemini-grounded' | 'fixture'
  domains ── urls              global reference corpus (see §6 and trade-off T2)

create unique index brands_unique_domain_per_workspace
  on public.brands (workspace_id, normalized_primary_domain)
  where normalized_primary_domain is not null and domain_status <> 'rejected';

create unique index brand_aliases_unique_per_workspace
  on public.brand_aliases (workspace_id, normalized_alias);

domain_status text not null default 'unverified'
  check (domain_status in ('unverified','corroborated','confirmed','rejected')),
domain_evidence_run_count integer not null default 0,

select b.id, b.name, b.tracking_state, count(*) as mentions
  from public.run_mentions m
  join public.brands b on b.id = m.brand_id
 where m.workspace_id = $1
 group by b.id
having count(*) > (select count(*) from public.run_mentions m2
                     join public.brands s on s.id = m2.brand_id
                    where m2.workspace_id = $1 and s.kind = 'self')
   and b.tracking_state <> 'tracked';
-- Any row returned is a rival out-mentioning the customer while sitting outside the ranking.

raw_uri text not null,                      -- exactly what the provider returned
url_id uuid references public.urls(id),     -- null until resolved
resolution_status text not null default 'resolved'
  check (resolution_status in ('resolved','unresolved','redirect_unresolved','invalid')),
check (resolution_status <> 'resolved' or url_id is not null)

brand_daily_metrics(workspace_id, brand_id, engine_id, bucket_date,
                    run_count,           -- ← the n, always present
                    mention_run_count,   -- visibility numerator
                    mention_count,       -- SoV numerator
                    position_sum, position_count)

/** A measured value that always travels with the sample it was computed from. */
export interface Measure {
  /** Successes — mentions, retrievals, whatever the numerator is. */
  readonly successes: number;
  /** Trials — the sample size this was computed from. Never inferred, never assumed. */
  readonly n: number;
  /** Point estimate, or `null` when `n === 0`. There is no `0` fallback. */
  readonly value: number | null;
  /** Clopper–Pearson 95% interval, or `null` when `n === 0`. */
  readonly interval: { low: number; high: number } | null;
  /** `false` when `n` is below the workspace's minimum-sample threshold. */
  readonly reportable: boolean;
}

export interface Comparison {
  readonly current: Measure;
  /** `null` when the prior window contains no runs at all. */
  readonly previous: Measure | null;
  /** `null` whenever `previous` is `null`. Never `current - 0`. */
  readonly delta: number | null;
  /** Why there is no comparison — rendered as copy, not as a number. */
  readonly absence: "no_prior_window" | "insufficient_prior_n" | null;
}

usage_periods(id, workspace_id, period_start, period_end, plan_id,
              run_limit, runs_used, cost_micros_used, …)
unique (workspace_id, period_start)

claim_run_slot(workspace_id, prompt_id, engine_id, run_date) → (run_id, created, reason)

release_run_slot(run_id) → boolean

select p.workspace_id, p.period_start, p.runs_used,
       (select count(*) from public.runs r
         where r.workspace_id = p.workspace_id
           and r.metered
           and r.run_date >= p.period_start
           and r.run_date <  p.period_end) as actual
  from public.usage_periods p
 where p.runs_used <> (…same subquery…);
-- Must return zero rows. Run it after every backfill.

limits: {
  free: { brands: 1, prompts: 10, engines: ["perplexity-sonar"],            runsPerMonth: 50 },
  pro:  { brands: 1, prompts: 100, engines: ["perplexity-sonar",
                                             "openai-web-search",
                                             "gemini-grounded"],            runsPerMonth: 1500 },
}

create or replace function public.is_workspace_member(p_workspace_id uuid)
returns boolean language sql stable security definer set search_path = ''
as $$ select exists (select 1 from public.workspace_members m
                      where m.workspace_id = p_workspace_id
                        and m.user_id = (select auth.uid())); $$;

supabase/migrations/
  20260725120000_workspaces.sql   workspaces, workspace_members, is_workspace_member(), normalisers
  20260725120100_brands.sql       brands, brand_aliases, entity-resolution constraints
  20260725120200_prompts.sql      topics, prompts, engines (+ seed)
  20260725120300_runs.sql         runs (the atom), run_mentions, domains, urls, run_citations
  20260725120400_metering.sql     usage_periods, claim_run_slot(), release_run_slot(), ensure_usage_period()
  20260725120500_rollups.sql      brand_daily_metrics, domain_daily_metrics, refresh functions

-- Workspaces: the tenancy boundary. One workspace is one tracked brand in one
-- market. v1 has exactly one member (the owner); the agency layer adds an
-- organizations table above this one without touching any fact table.

create extension if not exists unaccent with schema extensions;

-- Immutable text normaliser used by generated columns and unique indexes.
--
-- CAVEAT, deliberate: BOTH unaccent() overloads are declared STABLE, not
-- IMMUTABLE, because the accent dictionary is a file on disk. Declaring this
-- wrapper IMMUTABLE is the standard documented workaround for indexing
-- unaccented text; Postgres trusts the declaration. The two-argument form with
-- an explicit regdictionary is used so the dictionary is pinned at parse time
-- rather than resolved through search_path. The residual risk is that changing
-- the unaccent rules file under a live database would silently invalidate the
-- indexes below and require a REINDEX. We never change that file.
create or replace function public.rekkal_normalize(value text)
returns text
language sql
immutable
parallel safe
set search_path = ''
as $$
  select nullif(
    lower(btrim(regexp_replace(
      extensions.unaccent('extensions.unaccent'::regdictionary, coalesce(value, '')),
      '\s+', ' ', 'g'))),
    '')
$$;

comment on function public.rekkal_normalize(text) is
  'Case-, accent- and whitespace-insensitive normal form. Immutable so it can back generated columns and unique indexes. Accent folding is mandatory: "Telecom Paris" and "Telecom Paris" must collide.';

-- Domain normaliser: strips scheme, www., port, path, query, fragment and any
-- trailing dot, so efrei.fr, https://www.efrei.fr/ and EFREI.FR/ all collide.
create or replace function public.rekkal_normalize_domain(value text)
returns text
language sql
immutable
parallel safe
set search_path = ''
as $$
  select nullif(
    regexp_replace(
      regexp_replace(
        regexp_replace(
          regexp_replace(lower(btrim(coalesce(value, ''))), '^[a-z][a-z0-9+.\-]*://', ''),
          '[/:?#].*$', ''),
        '^www\.', ''),
      '\.$', ''),
    '')
$$;

create table if not exists public.workspaces (
  id uuid primary key default gen_random_uuid(),
  owner_user_id uuid not null references auth.users (id) on delete cascade,
  name text not null,
  -- The customer's own domain. Seeds the self brand at onboarding.
  brand_domain text,
  -- The single market this workspace measures (decision #3: one brand, one market).
  country_code text not null default 'FR',
  language text not null default 'fr',
  timezone text not null default 'Europe/Paris',
  -- Decision #6: weekly on both tiers. Column, not constant, so the free audit
  -- ('once') and any future cadence are configuration.
  run_cadence text not null default 'weekly'
    check (run_cadence in ('once', 'weekly')),
  -- Minimum sample size below which an aggregate is not reportable (section 7.5).
  -- Per workspace so the threshold is auditable and never applied inconsistently.
  min_sample_n integer not null default 30 check (min_sample_n > 0),
  first_run_completed_at timestamptz,
  created_at timestamptz not null default now(),
  updated_at timestamptz not null default now()
);

comment on table public.workspaces is
  'Tenancy boundary. One tracked brand in one market. Every fact table carries workspace_id directly.';

create index if not exists workspaces_owner_idx on public.workspaces (owner_user_id);

create trigger workspaces_set_updated_at
  before update on public.workspaces
  for each row execute function public.set_updated_at();

create table if not exists public.workspace_members (
  workspace_id uuid not null references public.workspaces (id) on delete cascade,
  user_id uuid not null references auth.users (id) on delete cascade,
  role text not null default 'owner' check (role in ('owner', 'admin', 'member')),
  created_at timestamptz not null default now(),
  primary key (workspace_id, user_id)
);

comment on table public.workspace_members is
  'Membership. v1 writes exactly one owner row per workspace, but every RLS policy in the schema routes through is_workspace_member() so adding seats is an INSERT, not a policy rewrite.';

-- Supports the membership lookup in is_workspace_member().
create index if not exists workspace_members_user_idx
  on public.workspace_members (user_id, workspace_id);

-- The single membership predicate for the entire schema.
-- SECURITY DEFINER is mandatory: without it this would read workspace_members
-- under that table's own policy, which calls this function -> infinite recursion.
create or replace function public.is_workspace_member(p_workspace_id uuid)
returns boolean
language sql
stable
security definer
set search_path = ''
as $$
  select exists (
    select 1
      from public.workspace_members m
     where m.workspace_id = p_workspace_id
       and m.user_id = (select auth.uid())
  );
$$;

comment on function public.is_workspace_member(uuid) is
  'The one tenancy predicate. Every workspace-scoped RLS policy calls it, so seat/role changes are a single-function change.';

alter table public.workspaces enable row level security;
alter table public.workspace_members enable row level security;

create policy "Members can view their workspaces"
  on public.workspaces for select to authenticated
  using (public.is_workspace_member(id));

create policy "Owners can update their workspace"
  on public.workspaces for update to authenticated
  using ((select auth.uid()) = owner_user_id)
  with check ((select auth.uid()) = owner_user_id);

create policy "Members can view workspace membership"
  on public.workspace_members for select to authenticated
  using (public.is_workspace_member(workspace_id));

-- No write policy on workspace_members: rows are created server-side at
-- onboarding via the service-role client.

-- Brands: the tracked customer brand (kind='self'), its competitors, and every
-- brand auto-discovered in an answer. Untracked brands are stored too, because
-- Position ranks over every detected brand (report 4.3) and because a rival
-- out-mentioning the customer must be detectable (report P2.6).

create table if not exists public.brands (
  id uuid primary key default gen_random_uuid(),
  workspace_id uuid not null references public.workspaces (id) on delete cascade,
  name text not null,
  normalized_name text generated always as (public.rekkal_normalize(name)) stored,
  kind text not null default 'competitor' check (kind in ('self', 'competitor')),
  -- 'tracked'   counts toward the ranking and is zero-filled in the rollups
  -- 'suggested' auto-discovered, present in mentions, not yet in the ranking
  -- 'ignored'   the user dismissed it
  tracking_state text not null default 'tracked'
    check (tracking_state in ('tracked', 'suggested', 'ignored')),
  discovery_source text not null default 'user'
    check (discovery_source in ('self', 'user', 'auto')),
  primary_domain text,
  normalized_primary_domain text
    generated always as (public.rekkal_normalize_domain(primary_domain)) stored,
  -- report P2.6 failure 2: an auto-discovered domain is a guess. It is never
  -- used for attribution or joined to Sources until it is corroborated by
  -- evidence (the domain actually cited in runs mentioning this brand) or
  -- confirmed by a human.
  domain_status text not null default 'unverified'
    check (domain_status in ('unverified', 'corroborated', 'confirmed', 'rejected')),
  domain_evidence_run_count integer not null default 0
    check (domain_evidence_run_count >= 0),
  colour text,
  first_seen_run_date date,
  created_at timestamptz not null default now(),
  updated_at timestamptz not null default now(),
  constraint brands_id_workspace_key unique (id, workspace_id)
);

comment on table public.brands is
  'Every brand known to a workspace: the customer (kind=self), tracked competitors, and auto-discovered brands. Untracked brands still accrue mentions so Position and rival detection are correct.';

comment on column public.brands.domain_status is
  'unverified -> guessed, never used for attribution. corroborated -> the domain was cited in runs mentioning this brand. confirmed -> a human said yes. rejected -> a human said no (and the domain is freed for the correct brand).';

-- Exactly one self brand per workspace (decision #3: one workspace, one brand).
create unique index if not exists brands_one_self_per_workspace
  on public.brands (workspace_id) where kind = 'self';

-- report P2.6 failure 1: "Efrei" and "Efrei Paris" cannot both exist against
-- efrei.fr. The second candidate becomes an alias of the first.
create unique index if not exists brands_unique_domain_per_workspace
  on public.brands (workspace_id, normalized_primary_domain)
  where normalized_primary_domain is not null and domain_status <> 'rejected';

create unique index if not exists brands_unique_name_per_workspace
  on public.brands (workspace_id, normalized_name);

create trigger brands_set_updated_at
  before update on public.brands
  for each row execute function public.set_updated_at();

create table if not exists public.brand_aliases (
  id uuid primary key default gen_random_uuid(),
  workspace_id uuid not null references public.workspaces (id) on delete cascade,
  brand_id uuid not null,
  alias text not null,
  normalized_alias text generated always as (public.rekkal_normalize(alias)) stored,
  source text not null default 'user' check (source in ('user', 'auto', 'seed')),
  created_at timestamptz not null default now(),
  -- Composite FK: a child physically cannot claim a workspace its parent is not in.
  foreign key (brand_id, workspace_id)
    references public.brands (id, workspace_id) on delete cascade
);

comment on table public.brand_aliases is
  'Alternate surface forms for a brand. The unique index below makes the mention resolver deterministic instead of first-match-wins.';

-- One normalised alias resolves to exactly one brand per workspace.
create unique index if not exists brand_aliases_unique_per_workspace
  on public.brand_aliases (workspace_id, normalized_alias);

create index if not exists brand_aliases_brand_idx on public.brand_aliases (brand_id);

alter table public.brands enable row level security;
alter table public.brand_aliases enable row level security;

create policy "Members can view brands"
  on public.brands for select to authenticated
  using (public.is_workspace_member(workspace_id));
create policy "Members can insert brands"
  on public.brands for insert to authenticated
  with check (public.is_workspace_member(workspace_id));
create policy "Members can update brands"
  on public.brands for update to authenticated
  using (public.is_workspace_member(workspace_id))
  with check (public.is_workspace_member(workspace_id));
create policy "Members can delete brands"
  on public.brands for delete to authenticated
  using (public.is_workspace_member(workspace_id));

create policy "Members can view brand aliases"
  on public.brand_aliases for select to authenticated
  using (public.is_workspace_member(workspace_id));
create policy "Members can insert brand aliases"
  on public.brand_aliases for insert to authenticated
  with check (public.is_workspace_member(workspace_id));
create policy "Members can update brand aliases"
  on public.brand_aliases for update to authenticated
  using (public.is_workspace_member(workspace_id))
  with check (public.is_workspace_member(workspace_id));
create policy "Members can delete brand aliases"
  on public.brand_aliases for delete to authenticated
  using (public.is_workspace_member(workspace_id));

-- Topics, prompts, and the engine reference table.
--
-- A prompt belongs to exactly one topic. This is deliberate: report 4.2 trap #2
-- ("never sum across tags") is a bug class that only exists with a many-to-many
-- tag model. A single-valued FK makes the over-count structurally impossible.

create table if not exists public.topics (
  id uuid primary key default gen_random_uuid(),
  workspace_id uuid not null references public.workspaces (id) on delete cascade,
  name text not null,
  normalized_name text generated always as (public.rekkal_normalize(name)) stored,
  position integer not null default 0,
  created_at timestamptz not null default now(),
  updated_at timestamptz not null default now(),
  constraint topics_id_workspace_key unique (id, workspace_id)
);

create unique index if not exists topics_unique_name_per_workspace
  on public.topics (workspace_id, normalized_name);

create trigger topics_set_updated_at
  before update on public.topics
  for each row execute function public.set_updated_at();

create table if not exists public.prompts (
  id uuid primary key default gen_random_uuid(),
  workspace_id uuid not null references public.workspaces (id) on delete cascade,
  topic_id uuid not null,
  text text not null,
  normalized_text text generated always as (public.rekkal_normalize(text)) stored,
  -- The market this prompt measures. The same question in three countries is
  -- three prompt rows, which is also the correct pricing: three countries
  -- genuinely is 3x the runs (report 4.5 #1).
  country_code text not null,
  language text not null,
  intent_class text,
  branding_class text,
  is_active boolean not null default true,
  source text not null default 'user' check (source in ('user', 'auto', 'seed')),
  archived_at timestamptz,
  created_at timestamptz not null default now(),
  updated_at timestamptz not null default now(),
  foreign key (topic_id, workspace_id)
    references public.topics (id, workspace_id) on delete cascade,
  constraint prompts_id_workspace_key unique (id, workspace_id)
);

comment on table public.prompts is
  'A monitored question. Exactly one topic per prompt (report 4.2 trap #2). Country and language live here, not on the run, so the same question in two markets is two prompts and two quota lines.';

-- The same question cannot be tracked twice in one market: a duplicate would
-- silently double the run meter for no extra information.
create unique index if not exists prompts_unique_text_per_market
  on public.prompts (workspace_id, normalized_text, country_code)
  where archived_at is null;

create index if not exists prompts_workspace_active_idx
  on public.prompts (workspace_id) where is_active and archived_at is null;
create index if not exists prompts_topic_idx on public.prompts (topic_id);

create trigger prompts_set_updated_at
  before update on public.prompts
  for each row execute function public.set_updated_at();

-- Engines: a stable surface id, decoupled from model version (report 4.1).
-- Text primary key so the adapter registry keys off it and migrations can seed it.
create table if not exists public.engines (
  id text primary key,
  display_name text not null,
  provider text not null,
  status text not null default 'active' check (status in ('active', 'beta', 'disabled')),
  supports_citations boolean not null default true,
  -- Indicative list price per run, for budgeting only. Actual spend is
  -- runs.cost_micros, recorded per call.
  list_cost_micros integer not null default 0,
  created_at timestamptz not null default now()
);

comment on table public.engines is
  'Stable answer-engine surface ids, decoupled from model version. Global reference data; readable by any authenticated user, written only by migration.';

insert into public.engines (id, display_name, provider, status, supports_citations, list_cost_micros)
values
  ('perplexity-sonar',  'Perplexity Sonar',            'perplexity', 'active',   true,   6000),
  ('openai-web-search', 'OpenAI (web search)',         'openai',     'active',   true,  12000),
  ('gemini-grounded',   'Gemini (Google Search)',      'google',     'active',   true,  35000),
  -- Test-only. status='disabled' so the runner refuses to dispatch it for a
  -- real workspace; the fixture adapter is selected explicitly by the test suite.
  ('fixture',           'Fixture (recorded answers)',  'rekkal',     'disabled', true,      0)
on conflict (id) do nothing;

alter table public.topics enable row level security;
alter table public.prompts enable row level security;
alter table public.engines enable row level security;

create policy "Members can view topics"
  on public.topics for select to authenticated
  using (public.is_workspace_member(workspace_id));
create policy "Members can insert topics"
  on public.topics for insert to authenticated
  with check (public.is_workspace_member(workspace_id));
create policy "Members can update topics"
  on public.topics for update to authenticated
  using (public.is_workspace_member(workspace_id))
  with check (public.is_workspace_member(workspace_id));
create policy "Members can delete topics"
  on public.topics for delete to authenticated
  using (public.is_workspace_member(workspace_id));

create policy "Members can view prompts"
  on public.prompts for select to authenticated
  using (public.is_workspace_member(workspace_id));
create policy "Members can insert prompts"
  on public.prompts for insert to authenticated
  with check (public.is_workspace_member(workspace_id));
create policy "Members can update prompts"
  on public.prompts for update to authenticated
  using (public.is_workspace_member(workspace_id))
  with check (public.is_workspace_member(workspace_id));
create policy "Members can delete prompts"
  on public.prompts for delete to authenticated
  using (public.is_workspace_member(workspace_id));

-- Reference data: readable by anyone signed in, writable by no one.
create policy "Engines are readable by authenticated users"
  on public.engines for select to authenticated using (true);

-- THE ATOM.
--
-- One row in public.runs is one prompt, sent to one engine, for one scheduled
-- slot date, in one workspace. It carries the answer exactly as that engine
-- returned it, the execution parameters it actually ran under, and it is the
-- single unit the run meter counts.
--
-- run_date is the SCHEDULED SLOT, not the wall clock. For weekly cadence it is
-- the Monday of the ISO week in the workspace timezone. A run that fails on
-- Monday and is retried on Tuesday reuses this row and this slot, which is
-- exactly why retries cannot double-count. Wall-clock execution is
-- started_at / completed_at; attempt_count records how many tries it took.

create table if not exists public.runs (
  id uuid primary key default gen_random_uuid(),
  workspace_id uuid not null references public.workspaces (id) on delete cascade,
  prompt_id uuid not null,
  engine_id text not null references public.engines (id),
  run_date date not null,

  status text not null default 'pending'
    check (status in ('pending', 'running', 'succeeded', 'failed', 'skipped')),
  -- false once release_run_slot() has refunded a failure that never reached the
  -- provider. The row survives for audit; it just stops counting.
  metered boolean not null default true,
  attempt_count integer not null default 0 check (attempt_count >= 0),

  -- Execution parameters, frozen at dispatch. Copied from the prompt so that
  -- editing a prompt never silently rewrites history.
  country_code text not null,
  language text not null,
  model_version text,

  answer_text text,
  answer_char_length integer check (answer_char_length is null or answer_char_length >= 0),
  -- Report 4.5 #3: raw answers stay hot for 30-90 days, then move to object
  -- storage. Nullable from day one so archival is a background job, not a migration.
  answer_archived_at timestamptz,
  answer_storage_key text,

  -- Which extraction stage actually ran. 'alias_only' means the brand set is
  -- knowably incomplete, so Position over this run is not trustworthy and the
  -- read layer must refuse to render it (report 4.3, P2.6).
  extraction_status text not null default 'pending'
    check (extraction_status in ('pending', 'alias_only', 'llm', 'failed')),
  extracted_brand_count integer,

  latency_ms integer,
  cost_micros bigint not null default 0 check (cost_micros >= 0),
  error_code text,
  error_message text,

  started_at timestamptz,
  completed_at timestamptz,
  created_at timestamptz not null default now(),
  updated_at timestamptz not null default now(),

  foreign key (prompt_id, workspace_id)
    references public.prompts (id, workspace_id) on delete cascade,
  constraint runs_id_workspace_key unique (id, workspace_id),
  constraint runs_answer_present_when_succeeded
    check (status <> 'succeeded' or answer_text is not null or answer_archived_at is not null)
);

comment on table public.runs is
  'THE ATOM: one prompt x one engine x one scheduled slot date. Unique on (prompt_id, engine_id, run_date), which is both the idempotency key and the unit the meter counts. Written only by the runner via the service-role client.';

comment on column public.runs.run_date is
  'The scheduled slot this run belongs to (weekly cadence: the Monday of the ISO week in the workspace timezone), NOT the wall-clock execution date. Retries reuse the slot.';

-- The idempotency key. A re-run is an upsert, never an insert.
create unique index if not exists runs_atom_key
  on public.runs (prompt_id, engine_id, run_date);

create index if not exists runs_workspace_date_idx
  on public.runs (workspace_id, run_date desc);
create index if not exists runs_workspace_engine_date_idx
  on public.runs (workspace_id, engine_id, run_date desc);
create index if not exists runs_dispatch_queue_idx
  on public.runs (workspace_id, status) where status in ('pending', 'running');
create index if not exists runs_meter_idx
  on public.runs (workspace_id, run_date) where metered;

create trigger runs_set_updated_at
  before update on public.runs
  for each row execute function public.set_updated_at();

-- Mentions: occurrence grain, one row per textual occurrence.
--
-- char_start/char_end are UTF-16 CODE-UNIT offsets into runs.answer_text, as
-- produced and consumed by TypeScript. Postgres stores and compares them but
-- never derives them: substring() counts characters, JavaScript counts UTF-16
-- code units, and the two diverge on any astral-plane character.
create table if not exists public.run_mentions (
  id uuid primary key default gen_random_uuid(),
  workspace_id uuid not null references public.workspaces (id) on delete cascade,
  run_id uuid not null,
  brand_id uuid not null,

  char_start integer not null check (char_start >= 0),
  char_end integer not null,
  -- The containing sentence, so the product can show the exact quote.
  sentence_start integer not null check (sentence_start >= 0),
  sentence_end integer not null,
  -- The literal text matched, so "Efrei Paris" is visible even though it
  -- resolved to brand "Efrei".
  surface_form text not null,

  matched_alias_id uuid references public.brand_aliases (id) on delete set null,
  match_method text not null check (match_method in ('alias', 'regex', 'llm')),
  match_confidence numeric(3, 2) not null default 1.00
    check (match_confidence >= 0 and match_confidence <= 1),
  -- Report 5.7: mention order conflates "recommended" with "merely listed".
  -- Nullable, populated only by the LLM stage. Kept now rather than added later
  -- because deriving it needs the raw answer, which archival will move away.
  is_recommended boolean,

  created_at timestamptz not null default now(),

  check (char_end > char_start),
  check (sentence_end > sentence_start),
  check (sentence_start <= char_start and sentence_end >= char_end),
  foreign key (run_id, workspace_id)
    references public.runs (id, workspace_id) on delete cascade,
  foreign key (brand_id, workspace_id)
    references public.brands (id, workspace_id) on delete cascade
);

comment on table public.run_mentions is
  'One row per textual occurrence of a brand in an answer. Occurrence grain yields both metrics: count(distinct run_id) = visibility numerator, count(*) = share-of-voice numerator (confirmed arithmetically in report P2.3).';

-- Re-extraction idempotency at mention grain: re-running the extractor over an
-- existing answer converges instead of duplicating.
create unique index if not exists run_mentions_span_key
  on public.run_mentions (run_id, char_start, char_end, brand_id);
create index if not exists run_mentions_brand_idx
  on public.run_mentions (workspace_id, brand_id, run_id);
create index if not exists run_mentions_run_order_idx
  on public.run_mentions (run_id, char_start);

-- Sources: a global reference corpus. A URL and its title are not customer
-- data; WHICH RUN cited it is, and that lives on run_citations.
create table if not exists public.domains (
  id uuid primary key default gen_random_uuid(),
  domain text not null,
  normalized_domain text
    generated always as (public.rekkal_normalize_domain(domain)) stored,
  classification text,
  first_seen_at timestamptz not null default now(),
  created_at timestamptz not null default now()
);

create unique index if not exists domains_normalized_key
  on public.domains (normalized_domain);

create table if not exists public.urls (
  id uuid primary key default gen_random_uuid(),
  domain_id uuid not null references public.domains (id) on delete cascade,
  url text not null,
  -- URLs can exceed the btree row limit, so the uniqueness key is a hash,
  -- scoped by domain so a collision would also have to collide on domain.
  url_hash text generated always as (md5(url)) stored,
  title text,
  classification text,
  first_seen_at timestamptz not null default now(),
  created_at timestamptz not null default now()
);

create unique index if not exists urls_domain_hash_key on public.urls (domain_id, url_hash);
create index if not exists urls_domain_idx on public.urls (domain_id);

-- Citations: which run cited which URL. Carries the tenancy boundary.
create table if not exists public.run_citations (
  id uuid primary key default gen_random_uuid(),
  workspace_id uuid not null references public.workspaces (id) on delete cascade,
  run_id uuid not null,
  url_id uuid references public.urls (id) on delete set null,
  -- Exactly what the provider returned, before resolution. Report 4.4: Gemini
  -- returns vertexaisearch redirect wrappers; storing them unresolved would make
  -- every domain in the database Google's.
  raw_uri text not null,
  order_index integer not null check (order_index >= 0),
  resolution_status text not null default 'resolved'
    check (resolution_status in ('resolved', 'unresolved', 'redirect_unresolved', 'invalid')),
  -- Where in the answer the engine attached this source, when it says.
  char_start integer,
  char_end integer,
  created_at timestamptz not null default now(),

  foreign key (run_id, workspace_id)
    references public.runs (id, workspace_id) on delete cascade,
  -- An unresolved redirect is kept for later re-resolution and is visibly
  -- missing from the Sources rollup, rather than silently misattributed.
  constraint run_citations_resolved_has_url
    check (resolution_status <> 'resolved' or url_id is not null)
);

create unique index if not exists run_citations_order_key
  on public.run_citations (run_id, order_index);
create index if not exists run_citations_url_idx on public.run_citations (url_id);
create index if not exists run_citations_workspace_idx
  on public.run_citations (workspace_id, run_id);

alter table public.runs enable row level security;
alter table public.run_mentions enable row level security;
alter table public.run_citations enable row level security;
alter table public.domains enable row level security;
alter table public.urls enable row level security;

-- SELECT only. There are deliberately no INSERT/UPDATE/DELETE policies for
-- authenticated users on runs, run_mentions or run_citations: Postgres denies by
-- default, so a valid session token calling PostgREST directly cannot fabricate
-- a run, forge a mention, or move the meter. Rows are written exclusively by the
-- runner via the service-role client (lib/supabase/admin.ts), which bypasses RLS.
create policy "Members can view runs"
  on public.runs for select to authenticated
  using (public.is_workspace_member(workspace_id));
create policy "Members can view run mentions"
  on public.run_mentions for select to authenticated
  using (public.is_workspace_member(workspace_id));
create policy "Members can view run citations"
  on public.run_citations for select to authenticated
  using (public.is_workspace_member(workspace_id));

create policy "Domains are readable by authenticated users"
  on public.domains for select to authenticated using (true);
create policy "Urls are readable by authenticated users"
  on public.urls for select to authenticated using (true);

-- The run meter. One row per workspace per billing period, and it is only ever
-- moved by claim_run_slot() / release_run_slot(). Nothing else writes runs_used.

create table if not exists public.usage_periods (
  id uuid primary key default gen_random_uuid(),
  workspace_id uuid not null references public.workspaces (id) on delete cascade,
  period_start date not null,
  period_end date not null,             -- exclusive
  -- Snapshot of the plan and the limit that was actually in force. Raising a
  -- plan's quota next quarter must not retroactively rewrite past enforcement.
  plan_id text not null,
  run_limit integer not null check (run_limit >= 0),
  runs_used integer not null default 0 check (runs_used >= 0),
  cost_micros_used bigint not null default 0 check (cost_micros_used >= 0),
  created_at timestamptz not null default now(),
  updated_at timestamptz not null default now(),
  check (period_end > period_start)
);

comment on table public.usage_periods is
  'The run meter. runs_used is moved only by claim_run_slot()/release_run_slot(). run_limit is snapshotted from config.ts when the period opens, so config.ts stays the single source of truth for pricing while the DB records what was enforced.';

create unique index if not exists usage_periods_workspace_period_key
  on public.usage_periods (workspace_id, period_start);

create trigger usage_periods_set_updated_at
  before update on public.usage_periods
  for each row execute function public.set_updated_at();

-- Open (or refresh) the period covering p_period_start. The limit moves only
-- upward, so a mid-period upgrade grants headroom and a mid-period downgrade
-- never retroactively puts a customer over quota.
create or replace function public.ensure_usage_period(
  p_workspace_id uuid,
  p_period_start date,
  p_period_end date,
  p_plan_id text,
  p_run_limit integer
)
returns uuid
language plpgsql
security definer
set search_path = ''
as $$
declare
  v_id uuid;
begin
  insert into public.usage_periods
    (workspace_id, period_start, period_end, plan_id, run_limit)
  values
    (p_workspace_id, p_period_start, p_period_end, p_plan_id, p_run_limit)
  -- The conflict target is referenced by bare relation name, not schema-
  -- qualified: `public.usage_periods.run_limit` here is a parse error even
  -- though the INSERT target is schema-qualified.
  on conflict (workspace_id, period_start) do update
    set period_end = excluded.period_end,
        plan_id    = excluded.plan_id,
        run_limit  = greatest(usage_periods.run_limit, excluded.run_limit),
        updated_at = now()
  returning id into v_id;

  return v_id;
end;
$$;

-- Atomically claim one metered slot for (prompt, engine, run_date).
--
-- Returns reason:
--   'claimed'          a new run row was created and the meter moved by exactly 1
--   'exists'           the slot was already claimed; the meter did NOT move
--   'quota_exhausted'  runs_used >= run_limit; nothing was inserted
--   'no_open_period'   no usage period covers run_date; call ensure_usage_period first
--   'unknown_prompt'   the prompt does not exist in this workspace
--
-- There is no window in which a run exists unmetered, or is metered twice.
create or replace function public.claim_run_slot(
  p_workspace_id uuid,
  p_prompt_id uuid,
  p_engine_id text,
  p_run_date date
)
returns table (run_id uuid, created boolean, reason text)
language plpgsql
security definer
set search_path = ''
as $$
declare
  v_period_id uuid;
  v_runs_used integer;
  v_run_limit integer;
  v_run_id uuid;
  v_country text;
  v_language text;
begin
  -- 1. Already claimed? Idempotent no-op. This is the whole re-run guarantee.
  select r.id into v_run_id
    from public.runs r
   where r.prompt_id = p_prompt_id
     and r.engine_id = p_engine_id
     and r.run_date  = p_run_date;

  if found then
    return query select v_run_id, false, 'exists'::text;
    return;
  end if;

  -- 2. Serialise concurrent dispatchers on the workspace's period row.
  select p.id, p.runs_used, p.run_limit
    into v_period_id, v_runs_used, v_run_limit
    from public.usage_periods p
   where p.workspace_id = p_workspace_id
     and p_run_date >= p.period_start
     and p_run_date <  p.period_end
     for update;

  if not found then
    return query select null::uuid, false, 'no_open_period'::text;
    return;
  end if;

  -- 3. Quota.
  if v_runs_used >= v_run_limit then
    return query select null::uuid, false, 'quota_exhausted'::text;
    return;
  end if;

  -- Freeze the execution parameters from the prompt.
  select pr.country_code, pr.language
    into v_country, v_language
    from public.prompts pr
   where pr.id = p_prompt_id and pr.workspace_id = p_workspace_id;

  if not found then
    return query select null::uuid, false, 'unknown_prompt'::text;
    return;
  end if;

  -- 4. Insert, and move the meter only if the insert actually created a row.
  insert into public.runs
    (workspace_id, prompt_id, engine_id, run_date, status, country_code, language)
  values
    (p_workspace_id, p_prompt_id, p_engine_id, p_run_date, 'pending', v_country, v_language)
  on conflict (prompt_id, engine_id, run_date) do nothing
  returning id into v_run_id;

  if v_run_id is null then
    -- Lost a race; the winning transaction already metered it.
    select r.id into v_run_id
      from public.runs r
     where r.prompt_id = p_prompt_id
       and r.engine_id = p_engine_id
       and r.run_date  = p_run_date;
    return query select v_run_id, false, 'exists'::text;
    return;
  end if;

  update public.usage_periods
     set runs_used = runs_used + 1,
         updated_at = now()
   where id = v_period_id;

  return query select v_run_id, true, 'claimed'::text;
end;
$$;

-- Refund a slot whose engine call never reached the provider (connection
-- refused, 429 with no completion, our own crash). Idempotent: the `where
-- metered` guard means a second call is a no-op. A provider 500 AFTER
-- generation still counts, because it still cost money.
create or replace function public.release_run_slot(p_run_id uuid)
returns boolean
language plpgsql
security definer
set search_path = ''
as $$
declare
  v_workspace_id uuid;
  v_run_date date;
begin
  update public.runs
     set metered = false, updated_at = now()
   where id = p_run_id and metered
  returning workspace_id, run_date into v_workspace_id, v_run_date;

  if not found then
    return false;
  end if;

  update public.usage_periods
     set runs_used = greatest(runs_used - 1, 0),
         updated_at = now()
   where workspace_id = v_workspace_id
     and v_run_date >= period_start
     and v_run_date <  period_end;

  return true;
end;
$$;

alter table public.usage_periods enable row level security;

-- Read-only for members. No write policy: the meter is unforgeable from a
-- browser because there is no policy to exploit.
create policy "Members can view their usage periods"
  on public.usage_periods for select to authenticated
  using (public.is_workspace_member(workspace_id));

-- Precomputed rollups. Sums and counts only -- no column in this schema holds a
-- ratio (report 4.1). Every ratio is recomputable from its components at any
-- aggregation level, and the n comes along for free at each one.
--
-- The naming convention is the enforcement: every metric column ends in _count
-- or _sum.

create table if not exists public.brand_daily_metrics (
  workspace_id uuid not null references public.workspaces (id) on delete cascade,
  brand_id uuid not null,
  engine_id text not null references public.engines (id),
  bucket_date date not null,

  -- THE n. Denominator for every brand metric, at every grain.
  -- The > 0 check is what makes "no data" row absence rather than a zero-n row,
  -- so an empty comparison window is NULL by construction (report P2.5).
  run_count integer not null check (run_count > 0),
  mention_run_count integer not null default 0 check (mention_run_count >= 0),
  mention_count integer not null default 0 check (mention_count >= 0),
  position_sum integer not null default 0 check (position_sum >= 0),
  position_count integer not null default 0 check (position_count >= 0),
  updated_at timestamptz not null default now(),

  primary key (workspace_id, brand_id, engine_id, bucket_date),
  foreign key (brand_id, workspace_id)
    references public.brands (id, workspace_id) on delete cascade,
  check (mention_run_count <= run_count)
);

comment on table public.brand_daily_metrics is
  'Sums and counts, never ratios. run_count is the sample size for every brand metric. A row exists only where runs actually happened (check run_count > 0), so an absent comparison window is NULL rather than a delta against zero.';

create index if not exists brand_daily_metrics_window_idx
  on public.brand_daily_metrics (workspace_id, bucket_date desc, brand_id);

create table if not exists public.domain_daily_metrics (
  workspace_id uuid not null references public.workspaces (id) on delete cascade,
  domain_id uuid not null references public.domains (id) on delete cascade,
  engine_id text not null references public.engines (id),
  bucket_date date not null,

  run_count integer not null check (run_count > 0),
  -- Distinct runs that retrieved this domain. This is the citation-rate
  -- denominator (report P2.3); dividing by retrieval_count is the deprecated
  -- metric, so both are stored and the correct one is always available.
  retrieved_run_count integer not null default 0 check (retrieved_run_count >= 0),
  retrieval_count integer not null default 0 check (retrieval_count >= 0),
  updated_at timestamptz not null default now(),

  primary key (workspace_id, domain_id, engine_id, bucket_date),
  check (retrieved_run_count <= run_count)
);

create index if not exists domain_daily_metrics_window_idx
  on public.domain_daily_metrics (workspace_id, bucket_date desc, domain_id);

-- Recompute one workspace-day of brand metrics. Idempotent (delete + insert).
--
-- Position ranks over EVERY detected brand, tracked or not (report 4.3), which
-- is why untracked brands still accrue mention rows.
--
-- Tracked brands are zero-filled for every (engine, date) that had runs, so
-- "measured zero with n=12" has a row and is distinguishable from "not
-- measured". Untracked brands get rows only where they were mentioned.
create or replace function public.refresh_brand_daily_metrics(
  p_workspace_id uuid,
  p_bucket_date date
)
returns integer
language plpgsql
security definer
set search_path = ''
as $$
declare
  v_rows integer;
begin
  delete from public.brand_daily_metrics
   where workspace_id = p_workspace_id and bucket_date = p_bucket_date;

  with scoped as (
    select r.id as run_id, r.engine_id
      from public.runs r
     where r.workspace_id = p_workspace_id
       and r.run_date = p_bucket_date
       and r.status = 'succeeded'
  ),
  totals as (
    select s.engine_id, count(*)::integer as run_count
      from scoped s
     group by s.engine_id
  ),
  per_run as (
    select m.run_id, m.brand_id, s.engine_id,
           count(*)::integer as mention_count,
           min(m.char_start) as first_char_start
      from public.run_mentions m
      join scoped s on s.run_id = m.run_id
     where m.workspace_id = p_workspace_id
     group by m.run_id, m.brand_id, s.engine_id
  ),
  positioned as (
    select p.*,
           rank() over (partition by p.run_id order by p.first_char_start)::integer as position
      from per_run p
  ),
  agg as (
    select p.brand_id, p.engine_id,
           count(distinct p.run_id)::integer as mention_run_count,
           sum(p.mention_count)::integer     as mention_count,
           sum(p.position)::integer          as position_sum,
           count(*)::integer                 as position_count
      from positioned p
     group by p.brand_id, p.engine_id
  )
  insert into public.brand_daily_metrics
    (workspace_id, brand_id, engine_id, bucket_date,
     run_count, mention_run_count, mention_count, position_sum, position_count)
  select p_workspace_id, b.id, t.engine_id, p_bucket_date,
         t.run_count,
         coalesce(a.mention_run_count, 0),
         coalesce(a.mention_count, 0),
         coalesce(a.position_sum, 0),
         coalesce(a.position_count, 0)
    from totals t
    cross join public.brands b
    left join agg a on a.brand_id = b.id and a.engine_id = t.engine_id
   where b.workspace_id = p_workspace_id
     and (b.tracking_state = 'tracked' or a.brand_id is not null);

  get diagnostics v_rows = row_count;
  return v_rows;
end;
$$;

-- Recompute one workspace-day of source metrics. Only resolved citations count:
-- an unresolved Gemini redirect wrapper is visibly missing rather than
-- misattributed to google.com.
create or replace function public.refresh_domain_daily_metrics(
  p_workspace_id uuid,
  p_bucket_date date
)
returns integer
language plpgsql
security definer
set search_path = ''
as $$
declare
  v_rows integer;
begin
  delete from public.domain_daily_metrics
   where workspace_id = p_workspace_id and bucket_date = p_bucket_date;

  with scoped as (
    select r.id as run_id, r.engine_id
      from public.runs r
     where r.workspace_id = p_workspace_id
       and r.run_date = p_bucket_date
       and r.status = 'succeeded'
  ),
  totals as (
    select s.engine_id, count(*)::integer as run_count
      from scoped s
     group by s.engine_id
  ),
  cited as (
    select u.domain_id, s.engine_id, c.run_id
      from public.run_citations c
      join scoped s on s.run_id = c.run_id
      join public.urls u on u.id = c.url_id
     where c.workspace_id = p_workspace_id
       and c.resolution_status = 'resolved'
  )
  insert into public.domain_daily_metrics
    (workspace_id, domain_id, engine_id, bucket_date,
     run_count, retrieved_run_count, retrieval_count)
  select p_workspace_id, c.domain_id, c.engine_id, p_bucket_date,
         t.run_count,
         count(distinct c.run_id)::integer,
         count(*)::integer
    from cited c
    join totals t on t.engine_id = c.engine_id
   group by c.domain_id, c.engine_id, t.run_count;

  get diagnostics v_rows = row_count;
  return v_rows;
end;
$$;

alter table public.brand_daily_metrics enable row level security;
alter table public.domain_daily_metrics enable row level security;

create policy "Members can view brand metrics"
  on public.brand_daily_metrics for select to authenticated
  using (public.is_workspace_member(workspace_id));
create policy "Members can view domain metrics"
  on public.domain_daily_metrics for select to authenticated
  using (public.is_workspace_member(workspace_id));

/** A stable answer-engine surface id. Matches public.engines.id. */
export type EngineId =
  | "perplexity-sonar"
  | "openai-web-search"
  | "gemini-grounded"
  | "fixture";

/** A source an engine cited, normalised across providers. */
export interface NormalizedSource {
  /** Fully-resolved absolute URL. Gemini vertexaisearch wrappers resolved. */
  readonly url: string | null;
  /** Exactly what the provider returned, before resolution. Always present. */
  readonly rawUri: string;
  readonly title: string | null;
  /** 0-based order the engine listed it in. */
  readonly orderIndex: number;
  readonly resolutionStatus:
    | "resolved" | "unresolved" | "redirect_unresolved" | "invalid";
  /** UTF-16 offsets into answerText, when the engine attributes the source. */
  readonly charStart: number | null;
  readonly charEnd: number | null;
}

export interface EngineResult {
  /** The answer, verbatim. Mention offsets are UTF-16 indexes into this string. */
  readonly answerText: string;
  readonly sources: readonly NormalizedSource[];
  readonly modelVersion: string;
  readonly costMicros: number | null;
  readonly latencyMs: number;
}

/**
 * One contract, three real implementations plus a fixture. A fourth engine is
 * a new file and one `engines` row -- no change to the runner.
 */
export interface EngineAdapter {
  readonly id: EngineId;
  /** False for the fixture adapter, so the suite runs with no provider keys. */
  readonly requiresApiKey: boolean;
  isConfigured(): boolean;
  run(request: EngineRequest): Promise<EngineResult>;
}

resolveDuePrompts(workspaceId, slotDate)      active prompts x plan engines
  -> ensureUsagePeriod(workspaceId, slotDate) snapshot plan + limit from config.ts
  -> for each (prompt, engine):
       claimRunSlot(...)                      'exists' -> skip, meter untouched
                                              'quota_exhausted' -> stop, report
       adapter.run(...)                       on pre-billing failure: releaseRunSlot()
       persistAnswer(...)                     status, answer_text, cost, latency
       extractMentions(...)                   stage 1 always; stage 2 if configured
       persistCitations(...)                  resolve redirects, upsert domains/urls
  -> refreshBrandDailyMetrics(workspaceId, slotDate)
  -> refreshDomainDailyMetrics(workspaceId, slotDate)