Lead scoring

-- better-supabase module: lead-scoring
-- @bs-module lead-scoring@1 managed
-- Scores people, and optionally companies, under a tenant-editable score model. Each record's columns, enrichment status, segments and mail opens, clicks and replies feed the model, and the record is scored again when any of them change.
-- Managed by `better-supabase add`; re-running it overwrites this file.
-- Change it through `blocks` in better-supabase.config.ts and the module's SQL hooks.

create or replace function better_supabase.actor_may_write()
returns boolean
language sql
stable
security invoker
set search_path = ''
as $$
  select case
    when auth.jwt() is null then true
    when auth.jwt() -> 'act' ->> 'kind' = 'support'
      then coalesce(auth.jwt() -> 'act' -> 'read_only' = 'false'::jsonb, false)
    when auth.jwt() -> 'act' ->> 'kind' = 'agent'
      then coalesce(auth.jwt() -> 'act' -> 'read_only', 'false'::jsonb) = 'false'::jsonb
    when auth.jwt() -> 'act' ->> 'kind' is not null then false
    when auth.jwt() -> 'act' ->> 'session_id' is not null
      and auth.jwt() -> 'act' ->> 'kind' is null then false
    else true
  end
$$;
revoke execute on function better_supabase.actor_may_write() from public;
grant execute on function better_supabase.actor_may_write() to anon, authenticated, service_role;

create or replace function better_supabase.raise_actor_readonly()
returns void
language plpgsql
volatile
set search_path = ''
as $$
begin
  raise exception 'This session is read-only' using errcode = '42501', hint = 'ACTOR_READ_ONLY';
end;
$$;
revoke execute on function better_supabase.raise_actor_readonly() from public;
grant execute on function better_supabase.raise_actor_readonly() to anon, authenticated, service_role;
create schema if not exists better_supabase;
grant usage on schema better_supabase to anon, authenticated, service_role;
create table if not exists "better_supabase"."lead_score_checks" (
  "tenant_id" uuid not null,
  "subject_type" text not null,
  "subject_id" text not null,
  -- Set when the record, its segments or its mail changed since the last score.
  "stale" boolean not null default true,
  "checked_at" timestamptz,
  constraint "lead_score_checks_pkey" primary key ("subject_type", "subject_id")
);
create index if not exists "lead_score_checks_tenant_idx" on "better_supabase"."lead_score_checks" ("tenant_id");
create index if not exists "lead_score_checks_due_idx" on "better_supabase"."lead_score_checks" ("checked_at" nulls first) where "stale";

alter table "better_supabase"."lead_score_checks" enable row level security;
revoke all on "better_supabase"."lead_score_checks" from anon, authenticated;
grant select on "better_supabase"."lead_score_checks" to authenticated;
grant all on "better_supabase"."lead_score_checks" to service_role;
drop policy if exists "lead_score_checks_read" on "better_supabase"."lead_score_checks";
create policy "lead_score_checks_read" on "better_supabase"."lead_score_checks" for select to authenticated
  using ("tenant_id" in (select better_supabase.tenant_ids_with('scores.compute')));

-- The values a lead model's rules and metrics read for one record: its
-- columns by database name, segments (the ids of the segments it is in),
-- mail { sent, opened, clicked, replied, lastSentAt, lastRepliedAt } and
-- enrichment { status, provider, fetchedAt }. Null when there is no such
-- record in the organization.
create or replace function "better_supabase"."lead_fields"(tenant uuid, subject_type text, subject_id text)
returns jsonb
language plpgsql
stable
security definer
set search_path = ''
as $$
#variable_conflict use_variable
declare
  v_id uuid;
  v_row jsonb;
  v_tenant uuid;
begin
  if not (coalesce(nullif((select auth.jwt()) ->> 'role', ''), session_user::text) in ('service_role', 'postgres', 'supabase_admin') or (lead_fields.tenant is not null and coalesce(better_supabase.can('tenant', lead_fields.tenant, 'scores.compute'), false))) then
    raise exception 'not allowed to score leads' using errcode = '42501', hint = 'LEAD_SCORING_FORBIDDEN';
  end if;
  begin
    v_id := lead_fields.subject_id::uuid;
  exception when invalid_text_representation then
    return null;
  end;
  if lead_fields.subject_type = 'person' then
    select to_jsonb(x), x."tenant_id" into v_row, v_tenant from "better_supabase"."people" x where x."id" = v_id;
    if v_row is null or v_tenant is distinct from lead_fields.tenant then
      return null;
    end if;
    return v_row || jsonb_build_object(
      'segments', '[]'::jsonb,
      'mail', '{}'::jsonb,
      'enrichment', null
    );
  end if;
  return null;
end;
$$;
revoke execute on function "better_supabase"."lead_fields"(uuid, text, text) from public, anon;
grant execute on function "better_supabase"."lead_fields"(uuid, text, text) to authenticated, service_role;

-- Scores one record under its subject type's lead model and marks it fresh.
create or replace function "better_supabase"."score_lead"(tenant uuid, subject_type text, subject_id text)
returns jsonb
language plpgsql
security definer
set search_path = ''
as $$
#variable_conflict use_variable
declare
  v_fields jsonb;
  v_result jsonb;
begin
  if not (better_supabase.actor_may_write()) then
    raise exception 'This session is read-only' using errcode = '42501', hint = 'ACTOR_READ_ONLY';
  end if;
  if not (coalesce(nullif((select auth.jwt()) ->> 'role', ''), session_user::text) in ('service_role', 'postgres', 'supabase_admin') or (score_lead.tenant is not null and coalesce(better_supabase.can('tenant', score_lead.tenant, 'scores.compute'), false))) then
    raise exception 'not allowed to score leads' using errcode = '42501', hint = 'LEAD_SCORING_FORBIDDEN';
  end if;
  if score_lead.subject_type not in ('person') then
    raise exception 'lead scoring doesn''t score this subject type' using errcode = '22023', hint = 'LEAD_SCORING_SUBJECT';
  end if;
  v_fields := "better_supabase"."lead_fields"(score_lead.tenant, score_lead.subject_type, score_lead.subject_id);
  if v_fields is null then
    raise exception 'no such record' using errcode = 'P0002', hint = 'LEAD_SCORING_NOT_FOUND';
  end if;
  v_result := "better_supabase"."compute_score"(score_lead.tenant, 'lead', score_lead.subject_type, score_lead.subject_id, v_fields);
  insert into "better_supabase"."lead_score_checks" ("tenant_id", "subject_type", "subject_id", "stale", "checked_at")
  values (score_lead.tenant, score_lead.subject_type, score_lead.subject_id, false, now())
  on conflict on constraint "lead_score_checks_pkey" do update
    set "stale" = false, "checked_at" = now();
  return v_result;
end;
$$;
revoke execute on function "better_supabase"."score_lead"(uuid, text, text) from public, anon;
grant execute on function "better_supabase"."score_lead"(uuid, text, text) to authenticated, service_role;

-- Scores up to max_count records that changed or weren't scored for
-- rescoreAfterHours, whose organization has an enabled lead model. Records
-- that were removed lose their check row. Returns how many it scored.
create or replace function "better_supabase"."score_leads_due"(max_count integer default 100)
returns integer
language plpgsql
security definer
set search_path = ''
as $$
#variable_conflict use_variable
declare
  v_check "better_supabase"."lead_score_checks";
  v_count integer := 0;
begin
  if not (coalesce(nullif((select auth.jwt()) ->> 'role', ''), session_user::text) in ('service_role', 'postgres', 'supabase_admin')) then
    raise exception 'not allowed to score leads' using errcode = '42501', hint = 'LEAD_SCORING_FORBIDDEN';
  end if;
  insert into "better_supabase"."lead_score_checks" ("tenant_id", "subject_type", "subject_id")
  select x."tenant_id", 'person', x."id"::text
  from "better_supabase"."people" x
  where x."archived_at" is null
    and not exists (select 1 from "better_supabase"."lead_score_checks" k where k."subject_type" = 'person' and k."subject_id" = x."id"::text)
  limit least(greatest(coalesce(max_count, 100), 1), 1000)
  on conflict do nothing;
  for v_check in
    select k.* from "better_supabase"."lead_score_checks" k
    where (k."stale" or k."checked_at" < now() - interval '24 hours')
      and exists (
        select 1 from "better_supabase"."score_models" m
        where m."tenant_id" = k."tenant_id" and m."key" = 'lead'
          and m."subject_type" = k."subject_type" and m."enabled"
      )
    order by k."checked_at" nulls first
    limit least(greatest(coalesce(max_count, 100), 1), 1000)
    for update of k skip locked
  loop
    if "better_supabase"."lead_fields"(v_check."tenant_id", v_check."subject_type", v_check."subject_id") is null then
      delete from "better_supabase"."lead_score_checks" k where k."subject_type" = v_check."subject_type" and k."subject_id" = v_check."subject_id";
      continue;
    end if;
    perform "better_supabase"."score_lead"(v_check."tenant_id", v_check."subject_type", v_check."subject_id");
    v_count := v_count + 1;
  end loop;
  return v_count;
end;
$$;
revoke execute on function "better_supabase"."score_leads_due"(integer) from public, anon, authenticated;
grant execute on function "better_supabase"."score_leads_due"(integer) to service_role;


create or replace function "better_supabase"."lead_person_changed"()
returns trigger
language plpgsql
security definer
set search_path = ''
as $$
begin
  insert into "better_supabase"."lead_score_checks" ("tenant_id", "subject_type", "subject_id")
  values (new."tenant_id", 'person', new."id"::text)
  on conflict ("subject_type", "subject_id") do update set "stale" = true;
  return null;
end;
$$;
revoke execute on function "better_supabase"."lead_person_changed"() from public, anon, authenticated;
drop trigger if exists "bs_person_changed" on "better_supabase"."people";
create trigger "bs_person_changed"
  after insert or update on "better_supabase"."people"
  for each row execute function "better_supabase"."lead_person_changed"();

create schema if not exists better_supabase;
create table if not exists better_supabase.modules (
  name text primary key,
  version integer not null,
  mode text not null,
  installed_at timestamptz not null default now(),
  updated_at timestamptz not null default now()
);
alter table better_supabase.modules enable row level security;
revoke all on better_supabase.modules from anon, authenticated;
grant select on better_supabase.modules to service_role;