-- Ultron master schema (Supabase / Postgres)

-- 1) Master leads table (deduped by Google place_id)
create table if not exists leads (
  place_id      text primary key,
  source        text default 'Google Maps',
  lead_type     text default 'Business',
  company       text,
  category      text,
  phone         text,
  email         text,
  website       text,
  instagram     text,
  facebook      text,
  tiktok        text,
  snapchat      text,
  whatsapp      text,
  youtube       text,
  twitter       text,
  rating        numeric,
  reviews       integer,
  newest_review text,
  oldest_review text,
  freshness     text,
  city          text,
  district      text,
  country       text default 'Saudi Arabia',
  address       text,
  lat           double precision,
  lng           double precision,
  maps_url      text,
  keyword       text,
  first_seen    timestamptz default now(),
  last_seen     timestamptz default now()
);
create index if not exists leads_city_idx       on leads (city);
create index if not exists leads_category_idx   on leads (category);
create index if not exists leads_first_seen_idx on leads (first_seen desc);
create index if not exists leads_freshness_idx  on leads (freshness);

-- email verification columns
alter table leads add column if not exists email_status     text;      -- valid|invalid|catch-all|risky|disposable|unknown
alter table leads add column if not exists email_verified   boolean;
alter table leads add column if not exists email_role       boolean;
alter table leads add column if not exists email_checked_at timestamptz;
create index if not exists leads_email_verified_idx on leads (email_verified);

-- review volume tier (count-based, distinct from age-based freshness)
alter table leads add column if not exists review_tier text;

-- Phase 3 (web enrichment) attempt marker: set once a lead's website has been crawled
-- for email + socials, so the enricher never re-crawls the same site every cron tick.
alter table leads add column if not exists web_enriched_at timestamptz;
create index if not exists leads_web_enrich_idx on leads (web_enriched_at) where website is not null;

-- Person-level fields (added 2026-07-16 for the 'bigasscrm' 46k Apollo-CRM import). ADDITIVE ONLY —
-- the Maps scraper never writes these; they carry the extra columns the imported CRM has. Never drop.
alter table leads add column if not exists full_name     text;
alter table leads add column if not exists first_name    text;
alter table leads add column if not exists last_name     text;
alter table leads add column if not exists job_title     text;
alter table leads add column if not exists seniority     text;
alter table leads add column if not exists department    text;
alter table leads add column if not exists linkedin      text;
alter table leads add column if not exists company_size  text;   -- Apollo also uses ranges ("11-50")
alter table leads add column if not exists lead_type     text;   -- already exists (default 'Business'); guarded no-op
alter table leads add column if not exists priority      text;
alter table leads add column if not exists notes         text;
alter table leads add column if not exists alt_emails    text;
alter table leads add column if not exists alt_phones    text;
alter table leads add column if not exists followers     integer;
alter table leads add column if not exists state         text;
alter table leads add column if not exists source_detail text;   -- import provenance (e.g. Apollo source file list)
alter table leads add column if not exists date_added    text;   -- raw import date string (first_seen gets the parsed ts)
create index if not exists leads_source_idx on leads (source);   -- powers the Platform filter / get_sources

-- 2) Coverage / work queue (grid tiles + saturation tracking)
create table if not exists scan_ledger (
  id           bigserial primary key,
  category     text not null,
  city         text not null,
  district     text default '',
  tile_lat     double precision,
  tile_lng     double precision,
  tile_zoom    integer,
  status       text default 'pending',   -- pending | saturated | done
  captured     integer default 0,
  new_found    integer default 0,
  last_scraped timestamptz,
  created_at   timestamptz default now(),
  unique (category, city, district, tile_lat, tile_lng, tile_zoom)
);
create index if not exists ledger_status_idx on scan_ledger (status);

-- 3) Live heartbeat for the dashboard
create table if not exists scraper_status (
  id              text primary key default 'main',
  state           text default 'idle',   -- running | idle | error
  current_keyword text,
  current_city    text,
  current_tile    text,
  note            text,
  updated_at      timestamptz default now()
);
insert into scraper_status (id, state) values ('main', 'idle')
  on conflict (id) do nothing;

-- 4) VPS heartbeat (written by scripts/heartbeat.js, read by the dashboard)
create table if not exists vps_status (
  id            text primary key default 'main',
  cpu_load      text,
  cpu_cores     integer,
  mem_used_pct  integer,
  mem_used_gb   text,
  mem_total_gb  text,
  disk          text,
  uptime_h      text,
  updated_at    timestamptz default now()
);
insert into vps_status (id) values ('main') on conflict (id) do nothing;

-- 5) One-call dashboard stats (returns JSON; called via REST rpc from the dashboard)
create or replace function get_dashboard_stats(p_range text default 'all')
returns json language sql stable as $$
  select json_build_object(
    'total',      (select count(*) from leads),
    'today',      (select count(*) from leads where first_seen >= date_trunc('day', now())),
    'hour',       (select count(*) from leads where first_seen >= now() - interval '1 hour'),
    'withEmail',  (select count(*) from leads where email is not null and email <> ''),
    'verifiedEmails', (select count(*) from leads where email_verified = true),
    'withPhone',  (select count(*) from leads where phone is not null and phone <> ''),
    'byCategory', (select coalesce(json_agg(x),'[]') from (select coalesce(category,'—') k, count(*) c from leads group by 1 order by c desc limit 12) x),
    'byCity',     (select coalesce(json_agg(x),'[]') from (select coalesce(city,'—') k, count(*) c from leads group by 1 order by c desc limit 12) x),
    'byFreshness',(select coalesce(json_agg(x),'[]') from (select coalesce(freshness,'UNKNOWN') k, count(*) c from leads group by 1 order by c desc) x),
    'byReviewTier',(select coalesce(json_agg(x),'[]') from (select coalesce(review_tier,'—') k, count(*) c from leads group by 1 order by min(reviews)) x),
    'recent',     (select coalesce(json_agg(r),'[]') from (
        select place_id, company, category, city, phone, email, email_status, rating, freshness, to_char(first_seen,'YYYY-MM-DD HH24:MI') ts
        from leads
        where case p_range
          when 'today' then first_seen >= date_trunc('day', now())
          when '7d'    then first_seen >= now() - interval '7 days'
          else true end
        order by first_seen desc limit 60
      ) r)
  );
$$;

-- 6) Coverage heatmap: leads bucketed into ~0.05° (~5km) cells for the pixel map
create or replace function get_coverage(p_category text default 'all')
returns json language sql stable as $$
  select coalesce(json_agg(x),'[]') from (
    select round((floor(lat/0.05)*0.05)::numeric, 3) la,
           round((floor(lng/0.05)*0.05)::numeric, 3) ln,
           count(*) c
    from leads
    where lat is not null and lng is not null
      and (p_category is null or p_category = 'all' or category = p_category)
    group by 1, 2
  ) x;
$$;

-- Individual business points for the pinpoint map
create or replace function get_points(p_category text default 'all', lim int default 5000)
returns json language sql stable as $$
  select coalesce(json_agg(x), '[]') from (
    select place_id, company, category, city, phone, lat, lng, coalesce(email_verified, false) v
    from leads
    where lat is not null and lng is not null
      and (p_category = 'all' or category = p_category)
    limit lim) x;
$$;

-- Real scanned tiles (ground truth, unlike get_coverage's lead-density heatmap) for the map overlay
create or replace function get_tiles(p_category text default 'all', p_city text default 'all', lim int default 20000)
returns json language sql stable as $$
  select coalesce(json_agg(x), '[]') from (
    select tile_lat la, tile_lng ln, tile_zoom z, status
    from scan_ledger
    where tile_lat is not null and tile_lng is not null
      and (p_category = 'all' or category = p_category)
      and (p_city = 'all' or city = p_city)
    limit lim) x;
$$;

-- Ops / coverage summary (scan_ledger rollup + live status)
create or replace function get_ops()
returns json language sql stable as $$
  select json_build_object(
    'ledger', (select coalesce(json_agg(x), '[]') from (
      select category, city, count(*)::int tiles,
             sum(case when status = 'done' then 1 else 0 end)::int done,
             sum(case when status = 'saturated' then 1 else 0 end)::int saturated,
             sum(coalesce(new_found,0))::int found,
             to_char(max(last_scraped), 'YYYY-MM-DD HH24:MI') last_scraped
      from scan_ledger group by 1, 2 order by max(last_scraped) desc nulls last) x),
    'status', (select row_to_json(s) from (select *, to_char(updated_at,'HH24:MI:SS') hb from scraper_status where id='main') s),
    'collected', (select count(*)::int from leads),
    'aged', (select count(*)::int from leads where freshness is not null),
    'verified', (select count(*)::int from leads where email_verified = true),
    'toVerify', (select count(*)::int from leads where email is not null and email <> '' and email_checked_at is null),
    'done', (select coalesce(json_agg(distinct (category||'|'||city)),'[]') from scan_ledger)
  );
$$;

-- 7) search_leads is defined at the END of this file (section 9) because it references the
--    CRM tables (lead_clients / contacts / app_users) created in section 8 below, and a
--    `language sql` function body is validated against existing objects at creation time.

-- Single lead full detail (for the click-through modal)
create or replace function get_lead(p_id text)
returns json language sql stable as $$
  select row_to_json(x) from (
    select company, category, phone, email, email_status, email_verified, email_role, website,
           instagram, facebook, tiktok, snapchat, whatsapp, youtube, twitter,
           rating, reviews, review_tier, freshness, newest_review, oldest_review,
           city, district, country, address, lat, lng, maps_url, keyword, source,
           to_char(first_seen,'YYYY-MM-DD HH24:MI') first_seen, to_char(last_seen,'YYYY-MM-DD HH24:MI') last_seen
    from leads where place_id = p_id
  ) x;
$$;

-- Distinct lead sources / platforms (powers the Leads-page Platform dropdown). e.g. 'Google Maps', 'bigasscrm'.
create or replace function get_sources()
returns json language sql stable as $$
  select coalesce(json_agg(src order by src), '[]')
  from (select distinct coalesce(nullif(source, ''), 'Google Maps') src from leads) t;
$$;

-- ============================================================
-- 8) CRM layer — team accounts, clients, assignments, notes, contact log
--    (auth is custom: passwords are scrypt-hashed in the dashboard server)
-- ============================================================
create extension if not exists pgcrypto;

-- Team accounts + roles
create table if not exists app_users (
  id         uuid primary key default gen_random_uuid(),
  email      text unique not null,
  name       text,
  role       text default 'agent',        -- admin | agent
  pw_hash    text,
  pw_salt    text,
  active     boolean default true,
  created_at timestamptz default now()
);

-- Revocable login sessions (validated by the dashboard server)
create table if not exists app_sessions (
  token      text primary key,
  user_id    uuid references app_users(id) on delete cascade,
  created_at timestamptz default now(),
  expires_at timestamptz,
  last_seen  timestamptz default now()
);
create index if not exists app_sessions_user_idx on app_sessions (user_id);

-- Client profiles (the agency's customers)
create table if not exists clients (
  id            uuid primary key default gen_random_uuid(),
  name          text not null,
  company       text,
  contact_name  text,
  contact_phone text,
  contact_email text,
  color         text,                     -- chart hue, e.g. #38bdf8
  notes         text,
  created_by    uuid references app_users(id) on delete set null,
  archived      boolean default false,
  created_at    timestamptz default now()
);
-- Client context (2026-07-19, must live HERE — get_client_overview below references them):
-- a brief every employee reads first + the client-provided shared-storage link.
alter table clients add column if not exists brief text;
alter table clients add column if not exists storage_url text;

-- Lead <-> client assignment (many-to-many; a lead can serve several clients)
create table if not exists lead_clients (
  place_id    text references leads(place_id) on delete cascade,
  client_id   uuid references clients(id) on delete cascade,
  assigned_by uuid references app_users(id) on delete set null,
  assigned_at timestamptz default now(),
  primary key (place_id, client_id)
);
create index if not exists lead_clients_client_idx on lead_clients (client_id);

-- User <-> client team membership (many-to-many)
create table if not exists user_clients (
  user_id   uuid references app_users(id) on delete cascade,
  client_id uuid references clients(id) on delete cascade,
  added_by  uuid references app_users(id) on delete set null,
  added_at  timestamptz default now(),
  primary key (user_id, client_id)
);

-- Per-lead notes timeline
create table if not exists lead_notes (
  id         bigserial primary key,
  place_id   text references leads(place_id) on delete cascade,
  client_id  uuid references clients(id) on delete set null,
  author_id  uuid references app_users(id) on delete set null,
  body       text not null,
  created_at timestamptz default now()
);
create index if not exists lead_notes_place_idx on lead_notes (place_id, created_at desc);
-- Rich notes (2026-07-19): attachments = [{key,name,mime,size}] in the private crm-files bucket.
alter table lead_notes add column if not exists attachments jsonb default '[]';

-- Contact / touch log — the dedup + analytics core
create table if not exists contacts (
  id           bigserial primary key,
  place_id     text references leads(place_id) on delete cascade,
  client_id    uuid references clients(id) on delete cascade,
  channel      text,                       -- call | whatsapp
  contacted_by uuid references app_users(id) on delete set null,
  contacted_at timestamptz default now(),
  note         text
);
create index if not exists contacts_place_client_idx on contacts (place_id, client_id, contacted_at desc);
create index if not exists contacts_client_idx on contacts (client_id, contacted_at desc);
create index if not exists contacts_by_idx on contacts (contacted_by, contacted_at desc);

-- Clients the current user may act for (admin => all; agent => memberships) — powers the switcher
create or replace function get_my_clients(p_user uuid, p_admin boolean default false)
returns json language sql stable as $$
  select coalesce(json_agg(row_to_json(x)), '[]') from (
    select c.id, c.name, c.color
    from clients c
    where c.archived = false
      and (p_admin or exists (select 1 from user_clients uc where uc.client_id = c.id and uc.user_id = p_user))
    order by c.name
  ) x;
$$;

-- Full client list for the Clients page (+ counts + this user's membership)
create or replace function get_clients(p_user uuid default null, p_include_archived boolean default false)
returns json language sql stable as $$
  select coalesce(json_agg(row_to_json(x)), '[]') from (
    select c.id, c.name, c.company, c.contact_name, c.contact_phone, c.contact_email,
           c.color, c.notes, c.archived, to_char(c.created_at,'YYYY-MM-DD') created_at,
           coalesce(c.target_calls,0) target_calls, coalesce(c.target_whatsapp,0) target_whatsapp,
           (select count(*) from lead_clients lc where lc.client_id = c.id)::int lead_count,
           (select count(*) from user_clients uc where uc.client_id = c.id)::int member_count,
           (select coalesce(json_agg(u.name order by u.name),'[]')
              from user_clients uc join app_users u on u.id = uc.user_id where uc.client_id = c.id) members,
           exists(select 1 from user_clients uc where uc.client_id = c.id and uc.user_id = p_user) is_member
    from clients c
    where (p_include_archived or c.archived = false)
    order by c.archived, c.name
  ) x;
$$;

-- CRM sidecar for the lead-detail modal: assignments + notes timeline + contact history
create or replace function get_lead_crm(p_id text)
returns json language sql stable as $$
  select json_build_object(
    'assignments', (select coalesce(json_agg(row_to_json(a)),'[]') from (
        select c.id, c.name, c.color
        from lead_clients lc join clients c on c.id = lc.client_id
        where lc.place_id = p_id and c.archived = false
        order by c.name) a),
    'notes', (select coalesce(json_agg(row_to_json(nn)),'[]') from (
        select ln.id, ln.body, coalesce(ln.attachments, '[]'::jsonb) attachments,
               coalesce(u.name, u.email, 'admin') author,
               c.name client, to_char(ln.created_at,'YYYY-MM-DD HH24:MI') at
        from lead_notes ln
        left join app_users u on u.id = ln.author_id
        left join clients c on c.id = ln.client_id
        where ln.place_id = p_id
        order by ln.created_at desc) nn),
    'contacts', (select coalesce(json_agg(row_to_json(cc)),'[]') from (
        select ct.channel, coalesce(u.name, u.email, 'admin') by_name,
               c.name client, c.color color, to_char(ct.contacted_at,'YYYY-MM-DD HH24:MI') at
        from contacts ct
        left join app_users u on u.id = ct.contacted_by
        left join clients c on c.id = ct.client_id
        where ct.place_id = p_id
        order by ct.contacted_at desc limit 50) cc)
  );
$$;

-- Per-client per-day contact counts for the dashboard line chart
create or replace function get_contact_stats(p_range text default '30d')
returns json language sql stable as $$
  select coalesce(json_agg(row_to_json(x)),'[]') from (
    select c.id client_id, c.name, c.color,
           to_char(date_trunc('day', ct.contacted_at),'YYYY-MM-DD') d, count(*)::int cnt
    from contacts ct join clients c on c.id = ct.client_id
    where ct.contacted_at >= case p_range
        when 'today' then date_trunc('day', now())
        when '7d'  then now() - interval '7 days'
        when '30d' then now() - interval '30 days'
        when '90d' then now() - interval '90 days'
        else now() - interval '3650 days' end
    group by c.id, c.name, c.color, date_trunc('day', ct.contacted_at)
    order by 4
  ) x;
$$;

-- Team roster + activity for the admin Team page (never returns password material)
create or replace function get_users()
returns json language sql stable as $$
  select coalesce(json_agg(row_to_json(x)),'[]') from (
    select u.id, u.email, u.name, u.role, u.active, to_char(u.created_at,'YYYY-MM-DD') created_at,
           (select count(*) from user_clients uc where uc.user_id = u.id)::int client_count,
           (select coalesce(json_agg(json_build_object('id',c.id,'name',c.name) order by c.name),'[]')
              from user_clients uc join clients c on c.id = uc.client_id where uc.user_id = u.id) clients,
           (select count(*) from contacts ct where ct.contacted_by = u.id)::int contacts_total,
           (select count(*) from contacts ct where ct.contacted_by = u.id and ct.channel='call')::int calls,
           (select count(*) from contacts ct where ct.contacted_by = u.id and ct.channel='whatsapp')::int whatsapps,
           (select to_char(max(ct.contacted_at),'YYYY-MM-DD HH24:MI') from contacts ct where ct.contacted_by = u.id) last_at
    from app_users u
    order by u.role, u.name nulls last, u.email
  ) x;
$$;

-- Admin: per-salesperson activity + recent contact feed (who / when / how)
create or replace function get_team_activity(p_range text default '30d')
returns json language sql stable as $$
  with rng as (select case p_range
      when 'today' then date_trunc('day', now())
      when '7d'  then now() - interval '7 days'
      when '30d' then now() - interval '30 days'
      when '90d' then now() - interval '90 days'
      else now() - interval '3650 days' end ts)
  select json_build_object(
    'byUser', (select coalesce(json_agg(row_to_json(x)),'[]') from (
        select u.id user_id, coalesce(u.name, u.email) name, u.role,
               count(ct.id)::int total,
               sum(case when ct.channel='call' then 1 else 0 end)::int calls,
               sum(case when ct.channel='whatsapp' then 1 else 0 end)::int whatsapps,
               to_char(max(ct.contacted_at),'YYYY-MM-DD HH24:MI') last_at
        from app_users u
        left join contacts ct on ct.contacted_by = u.id and ct.contacted_at >= (select ts from rng)
        group by u.id, coalesce(u.name, u.email), u.role
        order by total desc) x),
    'recent', (select coalesce(json_agg(row_to_json(y)),'[]') from (
        select coalesce(u.name, u.email, 'admin') by_name, ct.channel,
               l.company, cl.name client, to_char(ct.contacted_at,'YYYY-MM-DD HH24:MI') at
        from contacts ct
        left join app_users u on u.id = ct.contacted_by
        left join leads l on l.place_id = ct.place_id
        left join clients cl on cl.id = ct.client_id
        where ct.contacted_at >= (select ts from rng)
        order by ct.contacted_at desc limit 100) y)
  );
$$;

-- ============================================================
-- 9) Searchable / filterable leads for the Leads page (+ export)
--    Defined here (after section 8) so it can reference lead_clients / contacts / app_users.
--    p_client: when set, each row carries the latest contact for THAT client (dedup badge).
--    p_assigned_only: restrict to leads assigned to p_client.
-- ============================================================
drop function if exists search_leads(text, text, text, text, text, int, int);
drop function if exists search_leads(text, text, text, text, text, text, uuid, boolean, int, int);
create or replace function search_leads(
  q text default '', p_category text default 'all', p_city text default 'all',
  p_fresh text default 'all', p_verified text default 'all', p_phone text default 'all',
  p_source text default 'all', p_client uuid default null, p_assigned_only boolean default false,
  lim int default 100, off int default 0)
returns json language sql stable as $$
  with f as (
    select l.* from leads l
    where (q = '' or l.company ilike '%'||q||'%' or l.phone ilike '%'||q||'%'
           or l.email ilike '%'||q||'%' or l.category ilike '%'||q||'%' or l.city ilike '%'||q||'%')
      and (p_category = 'all' or l.category = p_category)
      and (p_city = 'all' or l.city = p_city)
      -- platform / source filter: 'all' = every source, else exact source (coalesced to 'Google Maps')
      and (p_source = 'all' or coalesce(nullif(l.source, ''), 'Google Maps') = p_source)
      and (p_fresh = 'all' or l.freshness = p_fresh)
      and (p_verified = 'all'
           or (p_verified = 'yes' and l.email_verified = true)
           or (p_verified = 'no' and l.email_verified is distinct from true))
      -- 'yes' = any phone · 'mobile' = Saudi mobile (05… / 9665…), i.e. WhatsApp-able
      and (p_phone = 'all'
           or (p_phone = 'yes' and l.phone is not null and l.phone <> '')
           or (p_phone = 'mobile' and regexp_replace(coalesce(l.phone,''),'[^0-9]','','g') ~ '^(0?5|9665)[0-9]{8}$'))
      and (not p_assigned_only or p_client is null
           or exists(select 1 from lead_clients lc where lc.place_id = l.place_id and lc.client_id = p_client))
  )
  select json_build_object(
    'total', (select count(*) from f),
    'rows', (select coalesce(json_agg(r), '[]') from (
      select f.place_id, f.company, f.category, f.city, f.district, f.country, f.phone, f.email, f.email_status, f.email_verified,
             f.website, f.instagram, f.facebook, f.tiktok, f.snapchat, f.youtube, f.twitter, f.whatsapp,
             f.rating, f.reviews, f.review_tier, f.freshness, f.oldest_review, f.newest_review, f.address, f.lat, f.lng, f.maps_url,
             f.source, f.keyword, to_char(f.first_seen,'YYYY-MM-DD HH24:MI') ts,
             lc.channel last_contact_channel, lc.by_name last_contact_by,
             to_char(lc.contacted_at,'YYYY-MM-DD HH24:MI') last_contact_at
      from f
      left join lateral (
        select ct.channel, ct.contacted_at, coalesce(u.name, u.email, 'admin') by_name
        from contacts ct left join app_users u on u.id = ct.contacted_by
        where p_client is not null and ct.place_id = f.place_id and ct.client_id = p_client
        order by ct.contacted_at desc limit 1
      ) lc on true
      order by f.first_seen desc limit lim offset off) r)
  );
$$;

-- ============================================================
-- 10) Password auth via pgcrypto bcrypt — SQL-resettable (added 2026-07-09)
--     Reset ANY user's password directly in the Supabase SQL editor:
--       update app_users set pw_hash = crypt('NEW_PASSWORD', gen_salt('bf')) where email = 'x@y.z';
--     (pw_salt column is now unused — bcrypt embeds its salt inside pw_hash.)
-- ============================================================
-- Login check: returns the user row iff the bcrypt password matches and the account is active.
create or replace function verify_login(p_email text, p_password text)
returns json language sql stable as $$
  select row_to_json(x) from (
    select id, email, name, role
    from app_users
    where email = lower(p_email) and active = true and pw_hash = crypt(p_password, pw_hash)
  ) x;
$$;

-- Set/reset a password (used by the Team page's reset action).
create or replace function set_app_password(p_id uuid, p_password text)
returns void language sql volatile as $$
  update app_users set pw_hash = crypt(p_password, gen_salt('bf')) where id = p_id;
$$;

-- Create a user with a bcrypt-hashed password (used by the Team page's add-member action).
create or replace function create_app_user(p_email text, p_name text, p_role text, p_password text)
returns json language plpgsql volatile as $$
declare r app_users;
begin
  insert into app_users (email, name, role, pw_hash, active)
  values (lower(p_email), nullif(p_name,''), coalesce(nullif(p_role,''),'agent'), crypt(p_password, gen_salt('bf')), true)
  returning * into r;
  return json_build_object('id', r.id, 'email', r.email, 'name', r.name, 'role', r.role, 'active', r.active);
exception when unique_violation then
  return json_build_object('error', 'email exists');
end;
$$;

-- ============================================================
-- 11) Client workspace: per-channel targets, auto-assign, overview + filterable leads (2026-07-09)
-- ============================================================
alter table clients add column if not exists target_calls    integer default 0;
alter table clients add column if not exists target_whatsapp integer default 0;

-- Any lead that has been contacted or client-noted BELONGS to that client (auto-assign).
-- ONE-TIME backfill — MUST NOT re-run on every db-setup, or it resurrects leads the user
-- deliberately removed from a pipeline (a lead with an old note kept coming back). Guarded
-- by a marker table so it fires exactly once.
do $$
begin
  if to_regclass('public._lc_backfill_done') is null then
    insert into lead_clients (place_id, client_id)
      select distinct place_id, client_id from contacts where client_id is not null on conflict do nothing;
    insert into lead_clients (place_id, client_id)
      select distinct place_id, client_id from lead_notes where client_id is not null on conflict do nothing;
    create table public._lc_backfill_done ();
  end if;
end $$;

-- KPIs + targets + per-salesperson + daily series for the client workspace header.
create or replace function get_client_overview(p_id uuid, p_range text default '30d')
returns json language sql stable as $$
  with rng as (select case p_range
      when 'today' then date_trunc('day', now())
      when '7d'  then now() - interval '7 days'
      when '90d' then now() - interval '90 days'
      when 'all' then now() - interval '3650 days'
      else now() - interval '30 days' end ts),
  cf as (select * from contacts where client_id = p_id and contacted_at >= (select ts from rng))
  select json_build_object(
    'client', (select row_to_json(c) from (
        select id, name, color, company, coalesce(target_calls,0) target_calls, coalesce(target_whatsapp,0) target_whatsapp,
               brief, storage_url
        from clients where id = p_id) c),
    'totals', json_build_object(
      'leads',       (select count(*) from lead_clients where client_id = p_id),
      'contacts',    (select count(*) from contacts where client_id = p_id),
      'salespeople', (select count(distinct contacted_by) from contacts where client_id = p_id)),
    'range', json_build_object(
      'contacts',  (select count(*) from cf),
      'calls',     (select count(*) from cf where channel = 'call'),
      'whatsapps', (select count(*) from cf where channel = 'whatsapp')),
    'today', json_build_object(
      'calls',     (select count(*) from contacts where client_id = p_id and channel='call' and contacted_at >= date_trunc('day',now())),
      'whatsapps', (select count(*) from contacts where client_id = p_id and channel='whatsapp' and contacted_at >= date_trunc('day',now()))),
    'byUser', (select coalesce(json_agg(row_to_json(u)),'[]') from (
        select cf.contacted_by user_id, coalesce(a.name, a.email, 'admin') name,
               count(cf.id)::int total,
               sum(case when cf.channel='call' then 1 else 0 end)::int calls,
               sum(case when cf.channel='whatsapp' then 1 else 0 end)::int whatsapps,
               to_char(max(cf.contacted_at),'YYYY-MM-DD HH24:MI') last_at
        from cf left join app_users a on a.id = cf.contacted_by
        group by cf.contacted_by, coalesce(a.name, a.email, 'admin')
        order by total desc) u),
    'days', (select coalesce(json_agg(row_to_json(dd)),'[]') from (
        select to_char(date_trunc('day', contacted_at),'YYYY-MM-DD') d,
               sum(case when channel='call' then 1 else 0 end)::int calls,
               sum(case when channel='whatsapp' then 1 else 0 end)::int whatsapps
        from cf group by 1 order by 1) dd)
  );
$$;

-- The client's leads with who-contacted-whom, filterable by salesperson / channel / date / search.
create or replace function get_client_leads(
  p_id uuid, p_user uuid default null, p_channel text default 'all',
  p_from text default null, p_to text default null, q text default '')
returns json language sql stable as $$
  select coalesce(json_agg(row_to_json(x)),'[]') from (
    select l.place_id, l.company, l.category, l.city, l.phone, l.freshness, l.email, l.email_status,
           lc.by_name last_contact_by, to_char(lc.contacted_at,'YYYY-MM-DD HH24:MI') last_contact_at, lc.channel last_contact_channel,
           (select count(*) from contacts ct where ct.place_id = l.place_id and ct.client_id = p_id)::int contact_count,
           (select count(*) from lead_notes ln where ln.place_id = l.place_id)::int note_count
    from lead_clients a
    join leads l on l.place_id = a.place_id
    left join lateral (
      select ct.channel, ct.contacted_at, coalesce(u.name, u.email, 'admin') by_name
      from contacts ct left join app_users u on u.id = ct.contacted_by
      where ct.place_id = a.place_id and ct.client_id = p_id
        and (p_user is null or ct.contacted_by = p_user)
        and (p_channel = 'all' or ct.channel = p_channel)
        and (p_from is null or ct.contacted_at >= p_from::timestamptz)
        and (p_to   is null or ct.contacted_at <  ((p_to::date) + 1))
      order by ct.contacted_at desc limit 1
    ) lc on true
    where a.client_id = p_id
      and (q = '' or l.company ilike '%'||q||'%' or l.phone ilike '%'||q||'%' or l.category ilike '%'||q||'%')
      and ((p_user is null and p_channel = 'all' and p_from is null and p_to is null) or lc.contacted_at is not null)
    order by lc.contacted_at desc nulls last, l.company
    limit 2000
  ) x;
$$;

-- ============================================================
-- 12) WhatsApp CRM layer (2026-07-16, see CRM-PLAN.md)
--     Identity spine (public.lead_phones) + conversation store in a SEPARATE schema `wa`
--     (kept out of the crowded public schema; PostgREST does not expose `wa` — the dashboard
--     reads it only through the public RPCs below). Synced hourly from Ahmed's Mac by
--     claude-voice/whatsapp-reader/ultron_sync.js.
-- ============================================================

-- Canonical phone: digits only, Saudi mobiles normalized to 9665XXXXXXXX.
create or replace function norm_phone(p text) returns text language sql immutable as $$
  select case
    when d ~ '^9665[0-9]{8}$'   then d
    when d ~ '^009665[0-9]{8}$' then substr(d, 3)
    when d ~ '^05[0-9]{8}$'     then '966' || substr(d, 2)
    when d ~ '^5[0-9]{8}$'      then '966' || d
    when d = ''                 then null
    else d
  end
  from (select regexp_replace(coalesce(p,''), '[^0-9]', '', 'g') d) t;
$$;

-- One row per phone number; a lead can own many numbers. THE join key for WhatsApp↔Ultron.
create table if not exists lead_phones (
  phone      text primary key,            -- canonical (norm_phone) digits
  place_id   text not null references leads(place_id) on delete cascade,
  label      text default 'primary',      -- primary | alt | whatsapp
  source     text,                        -- backfill:leads.phone | backfill:alt_phones | whatsapp
  created_at timestamptz default now()
);
create index if not exists lead_phones_place_idx on lead_phones(place_id);

-- Idempotent backfills (on conflict do nothing → safe on every db-setup run).
insert into lead_phones (phone, place_id, label, source)
  select norm_phone(l.phone), l.place_id, 'primary', 'backfill:leads.phone'
  from leads l
  where norm_phone(l.phone) ~ '^[0-9]{9,15}$'
  on conflict (phone) do nothing;
insert into lead_phones (phone, place_id, label, source)
  select norm_phone(x), l.place_id, 'alt', 'backfill:alt_phones'
  from leads l, regexp_split_to_table(coalesce(l.alt_phones,''), '[,;/|\s]+') x
  where norm_phone(x) ~ '^[0-9]{9,15}$'
  on conflict (phone) do nothing;

create schema if not exists wa;

-- One row per WhatsApp DM peer. visibility='private' (personal.json denylist) is NEVER
-- returned by any RPC — Ahmed's personal chats stay off the team dashboard.
create table if not exists wa.threads (
  peer_number  text primary key,          -- WhatsApp JID user part (may be a LID)
  peer_phone   text,                      -- canonical digits when known
  place_id     text references leads(place_id) on delete set null,
  is_group     boolean default false,
  display_name text,                      -- WhatsApp push name
  owner_number text,                      -- which of our numbers holds this thread
  visibility   text default 'team',       -- team | private
  last_msg_ts  timestamptz,
  msg_count    integer default 0,
  synced_at    timestamptz default now()
);
create index if not exists wa_threads_place_idx on wa.threads(place_id);

-- Verbatim messages (voice-note transcripts arrive pre-merged into body as "[voice note] …").
create table if not exists wa.messages (
  id             bigserial primary key,
  peer_number    text not null references wa.threads(peer_number) on delete cascade,
  wa_id          text not null,           -- WhatsApp message id (unique per chat)
  from_me        boolean not null,
  sender         text,
  body           text,
  media_type     text,                    -- image | video | audio | document | sticker | …
  media_hash     text,                    -- sha256 → wa.media / storage bucket object
  media_filename text,
  quoted_text    text,
  ts             timestamptz not null,
  unique (peer_number, wa_id)
);
create index if not exists wa_messages_peer_ts_idx on wa.messages(peer_number, ts desc);

-- Mirror of the local media table (binaries live in the private `wa-media` storage bucket).
create table if not exists wa.media (
  hash        text primary key,
  storage_url text,
  mime        text,
  filename    text,
  size        bigint,
  kind        text,                       -- document | image | video | audio | voice
  transcript  text,
  first_seen  timestamptz default now()
);

-- Lead detail modal: all team-visible threads + messages for one lead (newest first; UI reverses).
create or replace function get_lead_conversation(p_place_id text, lim int default 300)
returns json language sql stable as $$
  select json_build_object(
    'threads', (select coalesce(json_agg(row_to_json(t)),'[]') from (
        select peer_number, peer_phone, display_name, is_group,
               to_char(last_msg_ts,'YYYY-MM-DD HH24:MI') last_msg_at, msg_count
        from wa.threads
        where place_id = p_place_id and visibility = 'team'
        order by last_msg_ts desc) t),
    'messages', (select coalesce(json_agg(row_to_json(m)),'[]') from (
        select m.peer_number, m.wa_id, m.from_me, m.sender, m.body,
               m.media_type, m.media_hash, m.media_filename, m.quoted_text,
               to_char(m.ts,'YYYY-MM-DD HH24:MI') ts,
               md.mime media_mime, md.filename media_name, md.size media_size
        from wa.messages m
        join wa.threads t on t.peer_number = m.peer_number
        left join wa.media md on md.hash = m.media_hash
        where t.place_id = p_place_id and t.visibility = 'team'
        order by m.ts desc limit lim) m)
  );
$$;

-- Media lookup for the signed-URL route (/api/wa-media). Never exposes private-thread bodies.
create or replace function get_wa_media(p_hash text)
returns json language sql stable as $$
  select row_to_json(x) from (
    select hash, storage_url, mime, filename, size, kind from wa.media where hash = p_hash) x;
$$;

-- PostgREST's role needs explicit access to the wa schema: any public RPC that references
-- wa.* (search_leads p_has_wa, get_lead_conversation, get_wa_media) FAILS with
-- "permission denied for schema wa" without these. anon/authenticated stay locked out.
grant usage on schema wa to service_role;
grant select on all tables in schema wa to service_role;
alter default privileges in schema wa grant select on tables to service_role;

-- ============================================================
-- 13) Platform-aware filtration (2026-07-16, see CRM-PLAN.md Phase 4)
--     Two platform buckets TODAY: 'Google Maps' and 'bigasscrm' — WhatsApp-sourced leads FOLD INTO
--     'bigasscrm'; any future source keeps its own name. Bucketing is QUERY-LEVEL ONLY — leads.source
--     values are never rewritten. Adds leads.industry (restored from the dropped bigasscrm xlsx column,
--     backfilled by scripts/backfill-industry.py) + get_filter_options + per-platform search_leads params.
--
--     PLATFORM-GROUP EXPRESSION (single source of truth, used by every function below):
--       case when coalesce(nullif(l.source,''),'Google Maps') in ('bigasscrm','WhatsApp')
--            then 'bigasscrm' else coalesce(nullif(l.source,''),'Google Maps') end
-- ============================================================

-- Restored "Industry / Business Type" column. Additive; the scraper never writes it.
alter table leads add column if not exists industry text;
create index if not exists leads_industry_idx on leads (industry);

-- ---- v2 normalization helpers (2026-07-18, data-profile-driven) ----
-- The bigasscrm import is dirty in KNOWN ways (profiled 2026-07-18): seniority has case/underscore
-- dupes + comma-joined multi-values; state has 5 Province/Region/diacritic dupe spellings;
-- company_size mixes raw headcounts ("1300") with Apollo band strings ("51 - 200"). These helpers
-- are the single source of truth for both the option lists AND the search matching — change together.

-- Saudi region dupes fold into one canonical name.
create or replace function norm_state(s text) returns text language sql immutable as $$
  select case
    when s is null or trim(s) = '' then null
    when trim(s) in ('Riyadh Region', 'Ar Riyāḑ') then 'Riyadh Province'
    when trim(s) = 'Makkah Region' then 'Makkah Province'
    when trim(s) = 'Medina Province' then 'Al Madinah Province'
    else trim(s) end;
$$;

-- Any company_size encoding → one of 7 canonical headcount bands (first number wins:
-- "51 - 200" → 51 → '51-200'; "1300" → '1001-5000'). Null if no digits.
create or replace function size_band(s text) returns text language sql immutable as $$
  select case
    when v is null then null
    when v <= 10 then '1-10'
    when v <= 50 then '11-50'
    when v <= 200 then '51-200'
    when v <= 500 then '201-500'
    when v <= 1000 then '501-1000'
    when v <= 5000 then '1001-5000'
    else '5000+' end
  from (select ((regexp_match(coalesce(s, ''), '\d+'))[1])::bigint v) n;
$$;

-- Maps review count → the same 5 bands review_tier uses (review_tier itself is ~99% unbackfilled,
-- so the volume filter computes from `reviews` directly and ignores the sparse column).
create or replace function vol_band(n integer) returns text language sql immutable as $$
  select case
    when n is null then null
    when n < 5 then '0-5'
    when n < 50 then '5-50'
    when n < 100 then '50-100'
    when n < 500 then '100-500'
    else '500+' end;
$$;

-- Distinct PLATFORM GROUPS (WhatsApp folds into bigasscrm). Same function name → the /api/sources route
-- and the Leads-page Platform dropdown stay unchanged; they now see exactly the two buckets.
create or replace function get_sources()
returns json language sql stable as $$
  select coalesce(json_agg(src order by src), '[]')
  from (
    select distinct
      case when coalesce(nullif(l.source, ''), 'Google Maps') in ('bigasscrm', 'WhatsApp')
           then 'bigasscrm'
           else coalesce(nullif(l.source, ''), 'Google Maps') end src
    from leads l
  ) t;
$$;

-- Filter-dropdown options, scoped to one platform group (or 'all'). Each list = distinct non-null values
-- ordered by frequency desc, capped ~200. Categories are deduped case/whitespace-insensitively (group by
-- lower(trim); the most-frequent original casing wins). The RPC returns every list; the UI decides which
-- to show per platform (Industry/Seniority for bigasscrm; Category/City for Maps; State for both).
create or replace function get_filter_options(p_source text default 'all')
returns json language sql stable as $$
  with scoped as (
    select l.category, l.city, l.industry, l.seniority, l.state, l.job_title, l.department, l.company_size
    from leads l
    where p_source = 'all'
       or (case when coalesce(nullif(l.source, ''), 'Google Maps') in ('bigasscrm', 'WhatsApp')
                then 'bigasscrm'
                else coalesce(nullif(l.source, ''), 'Google Maps') end) = p_source
  )
  select json_build_object(
    -- Maps scraping leaks Google UI prompt strings ("Add hours", "Add website") into category —
    -- exclude the 'add %' family from the dropdown.
    'categories', (
      select coalesce(json_agg(val order by cnt desc), '[]') from (
        select (array_agg(category order by c desc))[1] val, sum(c) cnt
        from (
          select category, lower(trim(category)) k, count(*) c
          from scoped where category is not null and trim(category) <> ''
            and category not ilike 'add %'
          group by category, lower(trim(category))
        ) s
        group by k order by sum(c) desc limit 200
      ) t
    ),
    -- min-count 5 hides bigasscrm's junk city values ("8844", "RGHA3454", one-off districts);
    -- rare real cities stay reachable through the free-text search.
    'cities', (
      select coalesce(json_agg(val order by cnt desc), '[]') from (
        select city val, count(*) cnt from scoped
        where city is not null and trim(city) <> '' group by city
        having count(*) >= 5 order by count(*) desc limit 200
      ) t
    ),
    -- case/underscore-deduped ("Coffee shop" + "coffee_shop" merge; most-frequent casing wins)
    'industries', (
      select coalesce(json_agg(val order by cnt desc), '[]') from (
        select (array_agg(industry order by c desc))[1] val, sum(c) cnt
        from (
          select industry, lower(replace(trim(industry), '_', ' ')) k, count(*) c
          from scoped where industry is not null and trim(industry) <> ''
          group by industry, lower(replace(trim(industry), '_', ' '))
        ) s
        group by k order by sum(c) desc limit 200
      ) t
    ),
    -- split comma-joined multi-values ("Senior, Manager") into tokens, fold case/underscores →
    -- ~14 clean entries instead of 24 dirty ones. search_leads matches by token-contains.
    'seniorities', (
      select coalesce(json_agg(val order by cnt desc), '[]') from (
        select initcap(tok) val, count(*) cnt from (
          select trim(t) tok from scoped,
            lateral unnest(string_to_array(lower(replace(seniority, '_', ' ')), ',')) t
          where seniority is not null and trim(seniority) <> ''
        ) s where tok <> '' group by tok order by count(*) desc limit 30
      ) t
    ),
    'states', (
      select coalesce(json_agg(val order by cnt desc), '[]') from (
        select norm_state(state) val, count(*) cnt from scoped
        where norm_state(state) is not null group by 1 order by 2 desc limit 30
      ) t
    ),
    -- canonical headcount bands present in this platform's data, small→large
    'company_sizes', (
      select coalesce(json_agg(val order by ord), '[]') from (
        select size_band(company_size) val,
               array_position(array['1-10','11-50','51-200','201-500','501-1000','1001-5000','5000+'],
                              size_band(company_size)) ord
        from scoped where size_band(company_size) is not null group by 1
      ) t
    ),
    -- job titles are messy free text → same case-insensitive dedup as categories
    'job_titles', (
      select coalesce(json_agg(val order by cnt desc), '[]') from (
        select (array_agg(job_title order by c desc))[1] val, sum(c) cnt
        from (
          select job_title, lower(trim(job_title)) k, count(*) c
          from scoped where job_title is not null and trim(job_title) <> ''
          group by job_title, lower(trim(job_title))
        ) s
        group by k order by sum(c) desc limit 200
      ) t
    ),
    'departments', (
      select coalesce(json_agg(val order by cnt desc), '[]') from (
        select department val, count(*) cnt from scoped
        where department is not null and trim(department) <> '' group by department order by count(*) desc limit 200
      ) t
    )
  );
$$;

-- search_leads v4 (2026-07-18): adds p_state / p_size / p_rating / p_vol / p_presence, and matches
-- seniority/industry/state through the SAME normalization the option lists use (norm_state, size_band,
-- token-contains for seniority — so picking "Manager" also hits "manager" and "Senior, Manager").
-- p_source compares against the PLATFORM GROUP; q also matches full_name + job_title; returned rows
-- carry the person + industry + company_size + source_detail fields (for the CRM export).
-- Drop EVERY older overload: PostgREST resolves RPCs by name, and leftover legacy signatures
-- (found live: a 6-param and an 8-param from early dashboard versions) make calls ambiguous/stale.
drop function if exists search_leads(text, text, text, text, int, int);
drop function if exists search_leads(text, text, text, text, text, text, int, int);
drop function if exists search_leads(text, text, text, text, text, text, text, uuid, boolean, int, int);
drop function if exists search_leads(text, text, text, text, text, text, text, text, text, text, uuid, boolean, int, int);
drop function if exists search_leads(text, text, text, text, text, text, text, text, text, text, text, text, uuid, boolean, int, int);
create or replace function search_leads(
  q text default '', p_category text default 'all', p_city text default 'all',
  p_fresh text default 'all', p_verified text default 'all', p_phone text default 'all',
  p_source text default 'all', p_industry text default 'all', p_seniority text default 'all',
  p_job_title text default 'all', p_department text default 'all',
  p_state text default 'all', p_size text default 'all',
  p_rating text default 'all', p_vol text default 'all', p_presence text default 'all',
  p_has_wa text default 'all', p_client uuid default null, p_assigned_only boolean default false,
  lim int default 100, off int default 0)
returns json language sql stable as $$
  with f as (
    select l.* from leads l
    where (q = '' or l.company ilike '%'||q||'%' or l.phone ilike '%'||q||'%'
           or l.email ilike '%'||q||'%' or l.category ilike '%'||q||'%' or l.city ilike '%'||q||'%'
           or l.full_name ilike '%'||q||'%' or l.job_title ilike '%'||q||'%')
      and (p_category = 'all' or l.category = p_category)
      and (p_city = 'all' or l.city = p_city)
      -- platform / source filter compares against the PLATFORM GROUP (WhatsApp folds into bigasscrm)
      and (p_source = 'all'
           or (case when coalesce(nullif(l.source, ''), 'Google Maps') in ('bigasscrm', 'WhatsApp')
                    then 'bigasscrm'
                    else coalesce(nullif(l.source, ''), 'Google Maps') end) = p_source)
      -- industry via the same case/underscore fold the options use
      and (p_industry = 'all'
           or lower(replace(trim(l.industry), '_', ' ')) = lower(replace(trim(p_industry), '_', ' ')))
      -- seniority token-contains on the normalized string: "Manager" hits manager/Manager/"Senior, Manager"
      and (p_seniority = 'all'
           or lower(replace(l.seniority, '_', ' ')) like '%'||lower(replace(p_seniority, '_', ' '))||'%')
      -- job titles / departments compared case-insensitively (options are case-deduped)
      and (p_job_title = 'all' or lower(trim(l.job_title)) = lower(trim(p_job_title)))
      and (p_department = 'all' or lower(trim(l.department)) = lower(trim(p_department)))
      and (p_state = 'all' or norm_state(l.state) = p_state)
      and (p_size = 'all' or size_band(l.company_size) = p_size)
      -- Maps quality filters: minimum rating ('4.5'/'4.0'/'3.5') + review-volume band
      and (p_rating = 'all' or (l.rating is not null and l.rating >= p_rating::numeric))
      and (p_vol = 'all' or vol_band(l.reviews) = p_vol)
      -- web presence (works for both platforms)
      and (p_presence = 'all'
           or (p_presence = 'website' and coalesce(l.website, '') <> '')
           or (p_presence = 'no_website' and coalesce(l.website, '') = '')
           or (p_presence = 'instagram' and coalesce(l.instagram, '') <> '')
           or (p_presence = 'no_instagram' and coalesce(l.instagram, '') = ''))
      -- has WhatsApp conversation history (team-visible threads only)
      and (p_has_wa = 'all'
           or (p_has_wa = 'yes' and     exists (select 1 from wa.threads t where t.place_id = l.place_id and t.visibility = 'team'))
           or (p_has_wa = 'no'  and not exists (select 1 from wa.threads t where t.place_id = l.place_id and t.visibility = 'team')))
      and (p_fresh = 'all' or l.freshness = p_fresh)
      and (p_verified = 'all'
           or (p_verified = 'yes' and l.email_verified = true)
           or (p_verified = 'no' and l.email_verified is distinct from true))
      -- 'yes' = any phone · 'mobile' = Saudi mobile (05… / 9665…), i.e. WhatsApp-able
      and (p_phone = 'all'
           or (p_phone = 'yes' and l.phone is not null and l.phone <> '')
           or (p_phone = 'mobile' and regexp_replace(coalesce(l.phone,''),'[^0-9]','','g') ~ '^(0?5|9665)[0-9]{8}$'))
      and (not p_assigned_only or p_client is null
           or exists(select 1 from lead_clients lc where lc.place_id = l.place_id and lc.client_id = p_client))
  )
  select json_build_object(
    'total', (select count(*) from f),
    'rows', (select coalesce(json_agg(r), '[]') from (
      select f.place_id, f.company, f.category, f.city, f.district, f.country, f.phone, f.email, f.email_status, f.email_verified,
             f.website, f.instagram, f.facebook, f.tiktok, f.snapchat, f.youtube, f.twitter, f.whatsapp,
             f.rating, f.reviews, f.review_tier, f.freshness, f.oldest_review, f.newest_review, f.address, f.lat, f.lng, f.maps_url,
             f.source, f.keyword, to_char(f.first_seen,'YYYY-MM-DD HH24:MI') ts,
             f.full_name, f.job_title, f.seniority, f.department, f.industry, f.state, f.linkedin,
             f.company_size, f.source_detail,
             lc.channel last_contact_channel, lc.by_name last_contact_by,
             to_char(lc.contacted_at,'YYYY-MM-DD HH24:MI') last_contact_at
      from f
      left join lateral (
        select ct.channel, ct.contacted_at, coalesce(u.name, u.email, 'admin') by_name
        from contacts ct left join app_users u on u.id = ct.contacted_by
        where p_client is not null and ct.place_id = f.place_id and ct.client_id = p_client
        order by ct.contacted_at desc limit 1
      ) lc on true
      order by f.first_seen desc limit lim offset off) r)
  );
$$;

-- ============================================================
-- 14) CRM funnel layer (2026-07-18, CRM-PLAN "RESTRUCTURE" R1-R3)
--     LEAD-CENTRIC (Close model): no deals table. The stage lives on lead_clients — the
--     "this lead worked for this client" membership row IS the opportunity (Attio list-entry
--     pattern), so one lead can sit at different stages for different clients for free.
-- ============================================================

create table if not exists pipelines (
  id         uuid primary key default gen_random_uuid(),
  client_id  uuid not null references clients(id) on delete cascade,
  name       text not null default 'Sales',
  is_default boolean not null default true,
  created_at timestamptz default now()
);
create index if not exists pipelines_client_idx on pipelines(client_id);

create table if not exists pipeline_stages (
  id          uuid primary key default gen_random_uuid(),
  pipeline_id uuid not null references pipelines(id) on delete cascade,
  name        text not null,
  position    int  not null,
  kind        text not null default 'open' check (kind in ('open','won','lost')),
  color       text,
  wa_template text,                        -- optional prefilled WhatsApp opener for this stage
  unique (pipeline_id, position)
);

alter table lead_clients add column if not exists pipeline_id      uuid references pipelines(id);
alter table lead_clients add column if not exists stage_id         uuid references pipeline_stages(id);
alter table lead_clients add column if not exists entered_stage_at timestamptz default now();
alter table lead_clients add column if not exists board_position   double precision default 0;
alter table lead_clients add column if not exists owner_id         uuid references app_users(id);

-- Stage-change log — powers BOTH the lead timeline and conversion analytics. Append-only.
create table if not exists lead_stage_history (
  id            bigserial primary key,
  place_id      text not null,
  client_id     uuid not null references clients(id) on delete cascade,
  pipeline_id   uuid not null,
  from_stage_id uuid,                      -- null = first entry into the pipeline
  to_stage_id   uuid not null,
  changed_by    uuid,
  changed_at    timestamptz default now()
);
create index if not exists lsh_client_idx on lead_stage_history(client_id, changed_at desc);
create index if not exists lsh_place_idx  on lead_stage_history(place_id, changed_at desc);

-- Follow-ups (Pipedrive activity-based selling): "no answer, retry Sunday" is a TASK, not a stage.
create table if not exists tasks (
  id          bigserial primary key,
  place_id    text not null,
  client_id   uuid references clients(id) on delete set null,
  assigned_to uuid not null references app_users(id) on delete cascade,
  title       text not null,
  type        text not null default 'followup' check (type in ('call','whatsapp','followup','meeting')),
  due_at      timestamptz not null,
  done_at     timestamptz,
  created_by  uuid,
  created_at  timestamptz default now()
);
create index if not exists tasks_open_due_idx on tasks (assigned_to, due_at) where done_at is null;

-- The 7-stage Saudi WhatsApp-first outbound template, seeded once per client. Idempotent.
create or replace function ensure_default_pipeline(p_client uuid) returns uuid language plpgsql as $$
declare pid uuid;
begin
  select id into pid from pipelines where client_id = p_client and is_default limit 1;
  if pid is null then
    insert into pipelines(client_id) values (p_client) returning id into pid;
    -- v3 theme palette (2026-07-19): Attio-style stage dots, NO blue (Ahmed's call)
    insert into pipeline_stages(pipeline_id, name, position, kind, color) values
      (pid, 'New',             0, 'open', '#9ba1ab'),
      (pid, 'Contacted',       1, 'open', '#e2a33c'),
      (pid, 'Replied',         2, 'open', '#d98ac2'),
      (pid, 'Interested',      3, 'open', '#b06bff'),
      (pid, 'Meeting / Quote', 4, 'open', '#f97316'),
      (pid, 'Won',             5, 'won',  '#46c173'),
      (pid, 'Lost',            6, 'lost', '#e5484d');
  end if;
  return pid;
end $$;

-- Seed existing clients + drop every unstaged assignment into its client's New column.
do $$
declare c record; pid uuid; ns uuid;
begin
  for c in select id from clients loop
    pid := ensure_default_pipeline(c.id);
    select id into ns from pipeline_stages where pipeline_id = pid and position = 0;
    update lead_clients set pipeline_id = pid, stage_id = ns
      where client_id = c.id and stage_id is null;
  end loop;
end $$;

-- Every future assignment lands in the default pipeline at New automatically — keeps ALL write
-- paths (dashboard assign, imports, wa sync sparse leads) correct without repeating the logic.
create or replace function lead_clients_default_stage() returns trigger language plpgsql as $$
declare pid uuid; ns uuid;
begin
  if new.stage_id is null then
    pid := ensure_default_pipeline(new.client_id);
    select id into ns from pipeline_stages where pipeline_id = pid and position = 0;
    new.pipeline_id := pid; new.stage_id := ns; new.entered_stage_at := now();
  end if;
  return new;
end $$;
drop trigger if exists lead_clients_default_stage_t on lead_clients;
create trigger lead_clients_default_stage_t before insert on lead_clients
  for each row execute function lead_clients_default_stage();

-- Atomic stage move: validates the stage belongs to the row's pipeline, logs history, stamps
-- entered_stage_at. Creates the assignment if missing (drag onto a board a lead wasn't on yet).
create or replace function set_lead_stage(p_place_id text, p_client uuid, p_stage uuid, p_user uuid default null)
returns json language plpgsql as $$
declare lc record;
begin
  select * into lc from lead_clients where place_id = p_place_id and client_id = p_client;
  if not found then
    -- adding a lead to a funnel self-assigns it to the person who added it
    insert into lead_clients(place_id, client_id, assigned_by, owner_id) values (p_place_id, p_client, p_user, p_user);
    if p_user is not null then
      insert into lead_assignees(place_id, client_id, user_id, added_by)
        values (p_place_id, p_client, p_user, p_user) on conflict do nothing;
    end if;
    select * into lc from lead_clients where place_id = p_place_id and client_id = p_client;
  end if;
  if not exists (select 1 from pipeline_stages where id = p_stage and pipeline_id = lc.pipeline_id) then
    return json_build_object('ok', false, 'error', 'stage not in pipeline');
  end if;
  if lc.stage_id is distinct from p_stage then
    insert into lead_stage_history(place_id, client_id, pipeline_id, from_stage_id, to_stage_id, changed_by)
      values (p_place_id, p_client, lc.pipeline_id, lc.stage_id, p_stage, p_user);
    -- client-centric rule (Ahmed 2026-07-19): whoever works a lead first owns it
    update lead_clients set stage_id = p_stage, entered_stage_at = now(),
        owner_id = coalesce(owner_id, p_user)
      where place_id = p_place_id and client_id = p_client;
  end if;
  return json_build_object('ok', true);
end $$;

-- Kanban board for one client: ordered stages, per-stage totals, top 100 cards per column.
-- plpgsql (volatile) on purpose: it self-heals a brand-new client by ensuring the pipeline first.
create or replace function get_pipeline_board(p_client uuid)
returns json language plpgsql as $$
declare pid uuid;
begin
  pid := ensure_default_pipeline(p_client);
  return (select json_build_object(
    'pipeline_id', pid,
    'stages', (select coalesce(json_agg(s order by s.position), '[]') from (
      select st.id, st.name, st.position, st.kind, st.color,
        (select count(*) from lead_clients lc where lc.client_id = p_client and lc.stage_id = st.id) total,
        (select coalesce(json_agg(x), '[]') from (
           select l.place_id, coalesce(nullif(l.full_name,''), l.company, l.place_id) name,
                  nullif(l.company, '') company, nullif(l.full_name, '') person,
                  l.phone, l.city, l.freshness,
                  (regexp_replace(coalesce(l.phone,''),'[^0-9]','','g') ~ '^(0?5|9665)[0-9]{8}$') has_wa,
                  coalesce(u.name, u.email) owner,
                  (select coalesce(json_agg(coalesce(au.name, au.email)), '[]')
                     from lead_assignees la join app_users au on au.id = la.user_id
                     where la.place_id = l.place_id and la.client_id = p_client) assignees,
                  (select count(*) from lead_notes ln where ln.place_id = l.place_id) notes_n,
                  (select count(*) from tasks tk where tk.place_id = l.place_id and tk.done_at is null) tasks_n,
                  greatest(0, extract(epoch from now() - lc.entered_stage_at) / 86400)::int days
           from lead_clients lc
           join leads l on l.place_id = lc.place_id
           left join app_users u on u.id = lc.owner_id
           where lc.client_id = p_client and lc.stage_id = st.id
           order by lc.board_position, lc.entered_stage_at desc limit 100) x) leads
      from pipeline_stages st where st.pipeline_id = pid) s)));
end $$;

-- Lead modal stage context: for each client this lead is assigned to → the stage list + current
-- stage. Backfills any pre-trigger unstaged rows on read.
create or replace function get_lead_stagectx(p_place_id text)
returns json language plpgsql as $$
begin
  update lead_clients lc set pipeline_id = p.id, stage_id = ps.id, entered_stage_at = now()
    from pipelines p
    join pipeline_stages ps on ps.pipeline_id = p.id and ps.position = 0
    where lc.place_id = p_place_id and lc.stage_id is null
      and p.client_id = lc.client_id and p.is_default;
  return (select coalesce(json_agg(x), '[]') from (
    select lc.client_id, cl.name client, cl.color, lc.stage_id, lc.owner_id,
      (select coalesce(json_agg(json_build_object('id', ps.id, 'name', ps.name, 'kind', ps.kind)
              order by ps.position), '[]')
         from pipeline_stages ps where ps.pipeline_id = lc.pipeline_id) stages,
      (select coalesce(json_agg(json_build_object('user_id', la.user_id, 'name', coalesce(au.name, au.email))), '[]')
         from lead_assignees la join app_users au on au.id = la.user_id
         where la.place_id = lc.place_id and la.client_id = lc.client_id) assignees
    from lead_clients lc join clients cl on cl.id = lc.client_id
    where lc.place_id = p_place_id) x);
end $$;

-- My Day: the agent's open tasks bucketed overdue / today (Riyadh) / upcoming.
create or replace function get_my_tasks(p_user uuid)
returns json language sql stable as $$
  with t as (
    select t.id, t.place_id, t.title, t.type, t.client_id,
           coalesce(nullif(l.full_name,''), l.company, t.place_id) lead_name, l.phone,
           cl.name client, t.due_at, to_char(t.due_at at time zone 'Asia/Riyadh', 'YYYY-MM-DD HH24:MI') due
    from tasks t
    left join leads l on l.place_id = t.place_id
    left join clients cl on cl.id = t.client_id
    where t.assigned_to = p_user and t.done_at is null
  ), b as (
    select (date_trunc('day', now() at time zone 'Asia/Riyadh')) d0
  )
  select json_build_object(
    'overdue',  (select coalesce(json_agg(t order by t.due_at), '[]') from t, b
                 where (t.due_at at time zone 'Asia/Riyadh') < b.d0),
    'today',    (select coalesce(json_agg(t order by t.due_at), '[]') from t, b
                 where (t.due_at at time zone 'Asia/Riyadh') >= b.d0
                   and (t.due_at at time zone 'Asia/Riyadh') < b.d0 + interval '1 day'),
    'upcoming', (select coalesce(json_agg(t order by t.due_at), '[]') from (
                   select t.* from t, b where (t.due_at at time zone 'Asia/Riyadh') >= b.d0 + interval '1 day'
                   order by t.due_at limit 20) t));
$$;

-- ============================================================
-- 15) Multi-user WhatsApp accounts (2026-07-18, CRM-PLAN R4) — the ultron-wa VPS service.
--     Each employee links their own number(s) by QR from /connections; one Baileys worker per
--     number; threads/messages become per-account (composite identity). The Mac reader keeps
--     writing through a synthetic wa_accounts row.
-- ============================================================

create table if not exists public.wa_accounts (
  id                     uuid primary key default gen_random_uuid(),
  user_id                uuid references app_users(id) on delete cascade,   -- null = legacy/synthetic
  phone                  text,               -- canonical digits, null until first connect resolves self
  label                  text,
  status                 text not null default 'new',
      -- new | linking | connected | disconnected | logged_out | banned
  auto_reply             boolean not null default false,
  last_qr_at             timestamptz,
  last_connected_at      timestamptz,
  last_seen              timestamptz,        -- worker heartbeat (60s)
  last_disconnect_reason text,
  created_at             timestamptz default now()
);
create index if not exists wa_accounts_user_idx   on public.wa_accounts(user_id);
create index if not exists wa_accounts_status_idx on public.wa_accounts(status);

-- Per-number sharing (2026-07-27): the user who ADDED a number (wa_accounts.user_id = owner) may
-- invite other teammates to it with a role. viewer = see threads/conversations only; manager = full
-- control (send, campaigns, warmup, autoreply, link/unlink). The owner keeps everything; admins are
-- owner-equivalent on every number. Read by get_wa_accounts (effective-role join) + share routes.
create table if not exists public.wa_account_shares (
  account_id uuid not null references public.wa_accounts(id) on delete cascade,
  user_id    uuid not null references app_users(id)         on delete cascade,
  role       text not null default 'viewer' check (role in ('viewer','manager')),
  created_by uuid references app_users(id) on delete set null,
  created_at timestamptz default now(),
  primary key (account_id, user_id)
);
create index if not exists wa_account_shares_user_idx on public.wa_account_shares(user_id);

-- Baileys auth state in Postgres (creds + ALL signal keys — losing keys while keeping creds
-- = the Bad MAC decrypt loop). BufferJSON-serialized values.
create table if not exists public.wa_auth_state (
  account_id uuid not null references public.wa_accounts(id) on delete cascade,
  key_type   text not null,   -- creds | pre-key | session | sender-key | sender-key-memory
                              -- | app-state-sync-key | app-state-sync-version
  key_id     text not null,
  value      jsonb not null,
  updated_at timestamptz default now(),
  primary key (account_id, key_type, key_id)
);

-- DB-mediated pairing (2026-07-19): WhatsApp rejects device REGISTRATION from the VPS
-- and from proxy pools, so pairing runs on a residential machine (the Mac "link bridge").
-- The dashboard writes a link request onto the account row; the bridge polls for it,
-- publishes the QR / pairing code back onto the row; the dashboard polls it to render.
-- After the scan the bridge stores the session in wa_auth_state and flips status to
-- 'disconnected' — the VPS reconciler takes over within 60s.
alter table public.wa_accounts add column if not exists link_method       text;   -- qr | code
alter table public.wa_accounts add column if not exists link_phone        text;   -- for pairing-code linking
alter table public.wa_accounts add column if not exists link_qr           text;   -- QR payload, or 'CODE:XXXXXXXX'
alter table public.wa_accounts add column if not exists link_qr_at        timestamptz;
alter table public.wa_accounts add column if not exists link_requested_at timestamptz;

-- Bridge liveness (single row) — the Numbers page warns when the Mac is off.
create table if not exists public.wa_bridge (
  id        int primary key default 1,
  host      text,
  last_seen timestamptz
);

-- Shared LID→phone map (was per-Mac sqlite; workers on the VPS all learn into one table).
create table if not exists public.wa_lid_map (
  lid        text primary key,
  phone      text,
  learned_at timestamptz default now()
);

alter table wa.threads  add column if not exists owner_account_id   uuid references public.wa_accounts(id);
alter table wa.threads  add column if not exists stage              text default 'new';
alter table wa.threads  add column if not exists last_inbound_ts    timestamptz;
alter table wa.threads  add column if not exists last_outbound_ts   timestamptz;
alter table wa.threads  add column if not exists auto_reply_last_ts timestamptz;  -- 'infinity' = peer opted out
alter table wa.messages add column if not exists owner_account_id   uuid;

-- Backfill: one synthetic wa_accounts row per legacy owner_number (the Mac reader era), then
-- stamp both tables. Idempotent — only touches rows still missing owner_account_id.
do $$
declare r record; acc uuid;
begin
  for r in select distinct coalesce(owner_number, 'unknown') own
           from wa.threads where owner_account_id is null loop
    select id into acc from public.wa_accounts where coalesce(phone, label) = r.own limit 1;
    if acc is null then
      -- status 'new' so the ultron-wa reconciler never forks a worker for the Mac's number
      insert into public.wa_accounts(phone, label, status)
        values (case when r.own ~ '^[0-9]+$' then r.own end, 'mac-reader:'||r.own, 'new')
        returning id into acc;
    end if;
    update wa.threads set owner_account_id = acc
      where coalesce(owner_number, 'unknown') = r.own and owner_account_id is null;
  end loop;
  update wa.messages m set owner_account_id = t.owner_account_id
    from wa.threads t
    where m.owner_account_id is null and m.peer_number = t.peer_number;
end $$;

-- Repivot identity to (owner_account_id, peer_number). Guarded: only runs while the old
-- single-column PK is still in place, so re-running db-setup is safe. Constraint names are
-- discovered from the catalog (they vary: threads_pkey on live, not wa_threads_pkey).
do $$
declare pk_name text; r record;
begin
  select c.conname into pk_name from pg_constraint c
    join pg_class t on t.oid = c.conrelid
    join pg_namespace n on n.oid = t.relnamespace
    where n.nspname = 'wa' and t.relname = 'threads' and c.contype = 'p'
      and array_length(c.conkey, 1) = 1;
  if pk_name is not null then
    alter table wa.threads  alter column owner_account_id set not null;
    alter table wa.messages alter column owner_account_id set not null;
    -- messages constraints that reference the old identity: every FK, and the (peer, wa_id) unique
    for r in select c.conname, c.contype from pg_constraint c
             join pg_class t on t.oid = c.conrelid
             join pg_namespace n on n.oid = t.relnamespace
             where n.nspname = 'wa' and t.relname = 'messages' and c.contype in ('f', 'u') loop
      execute format('alter table wa.messages drop constraint %I', r.conname);
    end loop;
    execute format('alter table wa.threads drop constraint %I', pk_name);
    alter table wa.threads add primary key (owner_account_id, peer_number);
    alter table wa.messages add constraint wa_messages_owner_peer_wa_uk
      unique (owner_account_id, peer_number, wa_id);
    alter table wa.messages add constraint wa_messages_thread_fk
      foreign key (owner_account_id, peer_number)
      references wa.threads(owner_account_id, peer_number) on delete cascade;
  end if;
end $$;
create index if not exists wa_messages_owner_peer_ts_idx on wa.messages(owner_account_id, peer_number, ts desc);

-- get_lead_conversation v2: same shape the dashboard already renders, plus per-thread employee
-- attribution (which teammate's number holds the relationship) and the thread stage.
create or replace function get_lead_conversation(p_place_id text, lim int default 300)
returns json language sql stable as $$
  select json_build_object(
    'threads', (select coalesce(json_agg(row_to_json(t)),'[]') from (
        select th.peer_number, th.peer_phone, th.display_name, th.is_group, th.stage,
               th.owner_account_id, coalesce(u.name, u.email, th.owner_number) owner_name,
               to_char(th.last_msg_ts,'YYYY-MM-DD HH24:MI') last_msg_at, th.msg_count
        from wa.threads th
        left join public.wa_accounts a on a.id = th.owner_account_id
        left join app_users u on u.id = a.user_id
        where th.place_id = p_place_id and th.visibility = 'team'
        order by th.last_msg_ts desc) t),
    'messages', (select coalesce(json_agg(row_to_json(m)),'[]') from (
        select m.peer_number, m.owner_account_id, m.wa_id, m.from_me, m.sender, m.body,
               m.media_type, m.media_hash, m.media_filename, m.quoted_text,
               to_char(m.ts,'YYYY-MM-DD HH24:MI') ts,
               md.mime media_mime, md.filename media_name, md.size media_size
        from wa.messages m
        join wa.threads t on t.owner_account_id = m.owner_account_id and t.peer_number = m.peer_number
        left join wa.media md on md.hash = m.media_hash
        where t.place_id = p_place_id and t.visibility = 'team'
        order by m.ts desc limit lim) m)
  );
$$;

-- WhatsApp accounts a user can see for the /connections page: numbers they OWN plus numbers SHARED
-- to them (each row carries the caller's effective role — owner|manager|viewer; admins pass p_all and
-- are owner-equivalent on all). is_owner flags rows the caller may share/manage access on.
create or replace function get_wa_accounts(p_user uuid, p_all boolean default false)
returns json language sql stable as $$
  select coalesce(json_agg(row_to_json(x) order by x.created_at), '[]') from (
    select a.id, a.user_id, coalesce(u.name, u.email) user_name, a.phone, a.label, a.status,
           a.auto_reply, to_char(a.last_connected_at,'YYYY-MM-DD HH24:MI') last_connected_at,
           to_char(a.last_seen,'YYYY-MM-DD HH24:MI') last_seen, a.last_disconnect_reason, a.created_at,
           case when p_all or a.user_id = p_user then 'owner' else coalesce(s.role, 'viewer') end role,
           (p_all or a.user_id = p_user) is_owner
    from public.wa_accounts a
    left join app_users u on u.id = a.user_id
    left join public.wa_account_shares s on s.account_id = a.id and s.user_id = p_user
    where a.label not like 'mac-reader:%'
      and (p_all or a.user_id = p_user or s.user_id is not null)
  ) x;
$$;

-- Unified lead activity feed (Close/HubSpot one-timeline pattern): touches + notes + stage
-- changes + team-visible WhatsApp messages, newest first. p_client scopes the client-lane
-- entries (contacts, stage moves); notes and messages are lead-global.
create or replace function get_lead_timeline(p_place_id text, p_client uuid default null, lim int default 100)
returns json language sql stable as $$
  select coalesce(json_agg(e order by e.ts desc), '[]') from (
    select * from (
      select c.contacted_at ts, 'contact' type, c.channel, c.note body,
             coalesce(u.name, u.email, '—') actor, cl.name client
      from contacts c
      left join app_users u on u.id = c.contacted_by
      left join clients cl on cl.id = c.client_id
      where c.place_id = p_place_id and (p_client is null or c.client_id = p_client)
      union all
      select n.created_at, 'note', null, n.body, coalesce(u.name, u.email, '—'), cl.name
      from lead_notes n
      left join app_users u on u.id = n.author_id
      left join clients cl on cl.id = n.client_id
      where n.place_id = p_place_id
      union all
      select h.changed_at, 'stage', null, coalesce(fs.name, '—') || ' → ' || ts2.name,
             coalesce(u.name, u.email, '—'), cl.name
      from lead_stage_history h
      left join pipeline_stages fs on fs.id = h.from_stage_id
      join pipeline_stages ts2 on ts2.id = h.to_stage_id
      left join app_users u on u.id = h.changed_by
      left join clients cl on cl.id = h.client_id
      where h.place_id = p_place_id and (p_client is null or h.client_id = p_client)
      union all
      select m.ts, 'message', 'whatsapp',
             left(coalesce(nullif(m.body, ''), '[' || coalesce(m.media_type, 'media') || ']'), 300),
             case when m.from_me then coalesce(u.name, u.email, t.owner_number, 'us') else coalesce(t.display_name, 'them') end,
             null
      from wa.messages m
      join wa.threads t on t.owner_account_id = m.owner_account_id and t.peer_number = m.peer_number
      left join public.wa_accounts a on a.id = t.owner_account_id
      left join app_users u on u.id = a.user_id
      where t.place_id = p_place_id and t.visibility = 'team'
    ) all_events order by ts desc limit lim
  ) e;
$$;

-- ============================================================
-- 16) Client-centric command center + editable pipelines (2026-07-19, Ahmed's restructure)
--     The dashboard is no longer about raw leads — it answers "what did we get for each
--     CLIENT, at which stage, and did each EMPLOYEE actually work today?". Every employee
--     sees every funnel (get_my_clients now returns all active clients for everyone), and
--     stages are editable from the UI (set_pipeline_stages).
-- ============================================================

-- Multiple employees can work one lead-in-a-funnel (owner_id stays = first-touch owner).
create table if not exists lead_assignees (
  place_id  text not null,
  client_id uuid not null references clients(id) on delete cascade,
  user_id   uuid not null references app_users(id) on delete cascade,
  added_by  uuid,
  added_at  timestamptz default now(),
  primary key (place_id, client_id, user_id)
);
create index if not exists lead_assignees_user_idx on lead_assignees (user_id);

-- Everyone sees every client's funnel (membership table kept for other uses).
create or replace function get_my_clients(p_user uuid, p_admin boolean default false)
returns json language sql stable as $$
  select coalesce(json_agg(row_to_json(x)), '[]') from (
    select c.id, c.name, c.color from clients c
    where c.archived = false order by c.name
  ) x;
$$;

-- One call powers the home page: per-client funnel snapshot + per-employee activity.
-- "Today" = the Riyadh working day.
create or replace function get_command_center()
returns json language sql stable as $$
  with today as (select (now() at time zone 'Asia/Riyadh')::date d)
  select json_build_object(
    'clients', (select coalesce(json_agg(row_to_json(c2)), '[]') from (
      select c.id, c.name, c.color,
        (select count(*) from lead_clients lc where lc.client_id = c.id) total,
        (select count(*) from lead_clients lc join pipeline_stages ps on ps.id = lc.stage_id
          where lc.client_id = c.id and ps.kind = 'won') won,
        (select count(*) from contacts ct where ct.client_id = c.id
          and ct.contacted_at > now() - interval '7 days') touches_7d,
        (select count(*) from lead_stage_history h where h.client_id = c.id
          and h.changed_at > now() - interval '7 days') moves_7d,
        (select coalesce(json_agg(s order by s.position), '[]') from (
          select ps.id, ps.name, ps.position, ps.kind, ps.color,
            (select count(*) from lead_clients lc where lc.client_id = c.id and lc.stage_id = ps.id) total
          from pipeline_stages ps join pipelines p on p.id = ps.pipeline_id
          where p.client_id = c.id and p.is_default) s) stages
      from clients c where c.archived = false order by c.name) c2),
    'team', (select coalesce(json_agg(row_to_json(t)), '[]') from (
      select u.id, coalesce(u.name, u.email) name, u.role,
        (select count(*) from contacts ct where ct.contacted_by = u.id
          and (ct.contacted_at at time zone 'Asia/Riyadh')::date = (select d from today)
          and ct.channel = 'call') calls_today,
        (select count(*) from contacts ct where ct.contacted_by = u.id
          and (ct.contacted_at at time zone 'Asia/Riyadh')::date = (select d from today)
          and ct.channel = 'whatsapp') wa_today,
        (select count(*) from lead_stage_history h where h.changed_by = u.id
          and (h.changed_at at time zone 'Asia/Riyadh')::date = (select d from today)) moves_today,
        (select count(*) from tasks t2 where t2.assigned_to = u.id
          and t2.done_at is not null
          and (t2.done_at at time zone 'Asia/Riyadh')::date = (select d from today)) tasks_done_today,
        (select count(*) from contacts ct where ct.contacted_by = u.id
          and ct.contacted_at > now() - interval '7 days') touches_7d,
        (select count(*) from lead_stage_history h where h.changed_by = u.id
          and h.changed_at > now() - interval '7 days') moves_7d,
        (select count(*) from tasks t2 where t2.assigned_to = u.id and t2.done_at is null) open_tasks,
        (select count(*) from tasks t2 where t2.assigned_to = u.id and t2.done_at is null
          and t2.due_at < now()) overdue_tasks
      from app_users u where u.active = true order by u.name) t)
  );
$$;

-- Editable pipeline: upsert/rename/recolor/reorder stages in one atomic call; deletes
-- only stages holding zero leads (raises otherwise → whole call rolls back to safety).
-- p_stages = [{id?, name, kind, color}] in the desired order.
create or replace function set_pipeline_stages(p_client uuid, p_stages jsonb)
returns json language plpgsql as $$
declare pid uuid; s jsonb; keep uuid[] := '{}'; sid uuid; i int := 0; blocked text;
begin
  pid := ensure_default_pipeline(p_client);
  update pipeline_stages set position = position + 1000 where pipeline_id = pid;
  for s in select * from jsonb_array_elements(p_stages) loop
    sid := nullif(s->>'id', '')::uuid;
    if sid is null then
      insert into pipeline_stages(pipeline_id, name, position, kind, color)
        values (pid, coalesce(nullif(trim(s->>'name'), ''), 'Stage'), i,
                case when s->>'kind' in ('open','won','lost') then s->>'kind' else 'open' end,
                nullif(s->>'color', ''))
        returning id into sid;
    else
      update pipeline_stages set
          name = coalesce(nullif(trim(s->>'name'), ''), name),
          position = i,
          kind = case when s->>'kind' in ('open','won','lost') then s->>'kind' else kind end,
          color = coalesce(nullif(s->>'color', ''), color)
        where id = sid and pipeline_id = pid;
    end if;
    keep := array_append(keep, sid); i := i + 1;
  end loop;
  select string_agg(ps.name, ', ') into blocked
    from pipeline_stages ps
    where ps.pipeline_id = pid and ps.position >= 1000 and not (ps.id = any(keep))
      and exists (select 1 from lead_clients lc where lc.stage_id = ps.id);
  if blocked is not null then
    raise exception 'Move the leads out of: % first', blocked;
  end if;
  delete from pipeline_stages ps
    where ps.pipeline_id = pid and ps.position >= 1000 and not (ps.id = any(keep));
  return json_build_object('ok', true);
exception when others then
  return json_build_object('ok', false, 'error', SQLERRM);
end $$;

-- ============================================================
-- 17) COMPANY-FIRST STRUCTURE (2026-07-19, Ahmed's GO — Attio spec §10 with the dry-run fixes)
--     leads stay untouched; companies is a rollup layer + leads.company_id pointer.
--     The rollup DO-block runs ONLY while companies is empty (idempotent db-setup).
--     Dry-run traps baked in: aggregator-domain blocklist, bare-first-name guard,
--     city normalization, Arabic normalization, extended suffix list.
-- ============================================================

create table if not exists companies (
  company_id   bigserial primary key,
  company_key  text unique,             -- the grouping key (domain, name:city, or solo:place_id)
  name         text,
  name_norm    text,
  domain       text,
  industry     text, category text, company_size text,
  city         text, district text, country text, address text,
  lat double precision, lng double precision,
  maps_url     text,
  rating       numeric, reviews integer, review_tier text,
  source       text,
  created_at   timestamptz default now()
);
create index if not exists companies_name_norm_idx on companies (name_norm);
create index if not exists companies_domain_idx on companies (domain);
alter table leads add column if not exists company_id bigint references companies(company_id);
create index if not exists leads_company_idx on leads (company_id);

-- normalizers (shared by rollup + future upserts)
create or replace function norm_company_name(t text) returns text language sql immutable as $$
  select btrim(regexp_replace(
    regexp_replace(
      regexp_replace(
        translate(lower(coalesce(t,'')), 'أإآةىٱ', 'اااهيا'),
        '[ً-ْ]', '', 'g'),
      '\y(llc|ltd|co|company|corp|est|establishment|trading|group|holding|international|intl|saudi|arabia|ksa|شركة|مؤسسة|مجموعة|القابضة|المحدودة|السعودية|فرع|للتجارة|التجارية|العالمية|المتحدة|الوطنية)\y',
      ' ', 'g'),
    '[^a-z0-9؀-ۿ]+', ' ', 'g'));
$$;
create or replace function norm_company_domain(u text) returns text language sql immutable as $$
  select case when d = '' then null
              -- socials + aggregators/hosts are NOT company identities (dry-run: mahally.com
              -- alone fake-merged 456 unrelated stores)
              when d ~ '^(instagram|wa|api|facebook|tiktok|linktr|snapchat|twitter|x)\.' then null
              when d in ('mahally.com','franchisecenter.sa','confidential.careers','sites.google.com',
                         'drive.google.com','business.google.com','google.com','linktr.ee',
                         'americanarestaurants.com','ihg.com','wa.me','linktr.ee') then null
              when d ~ '(blogspot|wordpress|wixsite|godaddysites|webs)\.' then null
              else d end
  from (select nullif(regexp_replace(
          regexp_replace(lower(coalesce(u,'')), '^https?://(www\.)?', ''),
          '/.*$', ''), '') d) x;
$$;
create or replace function norm_company_city(c text) returns text language sql immutable as $$
  select case when v in ('makkah','mecca') then 'mecca'
              when v in ('madinah','medina','al madinah') then 'medina'
              else nullif(v,'') end
  from (select btrim(regexp_replace(lower(coalesce(c,'')), '[0-9]+', '', 'g')) v) x;
$$;

-- the rollup: 60k leads → ~40k companies. Runs once (companies empty), ~seconds.
do $$
declare n_companies int; n_linked int;
begin
  if exists (select 1 from companies limit 1) then return; end if;

  create temp table _keyed on commit drop as
  select place_id,
    case
      -- usable domain = strongest identity
      when norm_company_domain(website) is not null then 'd:' || norm_company_domain(website)
      -- bare first-name "companies" (bigasscrm trap: Mohammed ×442) never group by name
      when array_length(regexp_split_to_array(norm_company_name(company), '\s+'), 1) < 2
        then 'solo:' || place_id
      else 'n:' || norm_company_name(company) || ':' || coalesce(norm_company_city(city), '')
    end company_key
  from leads
  where coalesce(company, '') <> '';

  insert into companies (company_key, name, name_norm, domain, industry, category, company_size,
                         city, district, country, address, lat, lng, maps_url, rating, reviews, review_tier, source)
  select distinct on (k.company_key)
    k.company_key, l.company, norm_company_name(l.company), norm_company_domain(l.website),
    l.industry, l.category, l.company_size, l.city, l.district, l.country, l.address,
    l.lat, l.lng, l.maps_url, l.rating, l.reviews, l.review_tier, l.source
  from _keyed k join leads l on l.place_id = k.place_id
  order by k.company_key,
           (l.website is not null) desc, (l.rating is not null) desc,
           l.rating desc nulls last, l.reviews desc nulls last;

  update leads l set company_id = c.company_id
  from _keyed k join companies c on c.company_key = k.company_key
  where l.place_id = k.place_id and l.company_id is null;

  select count(*) into n_companies from companies;
  select count(*) into n_linked from leads where company_id is not null;
  raise notice 'company rollup: % companies, % leads linked', n_companies, n_linked;
end $$;

-- Companies page: search + filters + counts
create or replace function search_companies(
  q text default '', p_city text default 'all', p_cat text default 'all',
  lim int default 100, off int default 0)
returns json language sql stable as $$
  with f as (
    select c.* from companies c
    where (q = '' or c.name ilike '%'||q||'%' or c.domain ilike '%'||q||'%' or c.name_norm ilike '%'||norm_company_name(q)||'%')
      and (p_city = 'all' or c.city = p_city)
      and (p_cat = 'all' or c.category = p_cat or c.industry = p_cat)
      and c.company_key not like 'solo:%'   -- solo rows are people without a real employer
  )
  select json_build_object(
    'total', (select count(*) from f),
    'rows', (select coalesce(json_agg(r), '[]') from (
      select f.company_id, f.name, f.domain, f.industry, f.category, f.city, f.rating, f.reviews,
        (select count(*) from leads l where l.company_id = f.company_id) leads_n,
        (select count(*) from leads l where l.company_id = f.company_id
           and coalesce(l.full_name,'') <> '') people_n,
        (select l.phone from leads l where l.company_id = f.company_id and coalesce(l.phone,'') <> ''
           order by (coalesce(l.full_name,'') <> '') desc limit 1) phone
      from f
      order by (select count(*) from leads l where l.company_id = f.company_id) desc, f.name
      limit lim offset off) r));
$$;

create or replace function get_company(p_id bigint)
returns json language sql stable as $$
  select json_build_object(
    'company', (select row_to_json(c) from (
      select company_id, name, domain, industry, category, company_size, city, district,
             address, maps_url, rating, reviews, source from companies where company_id = p_id) c),
    'people', (select coalesce(json_agg(row_to_json(p2)), '[]') from (
      select place_id, full_name, job_title, seniority, department, phone, email, linkedin
      from leads where company_id = p_id and coalesce(full_name,'') <> ''
      order by (coalesce(job_title,'') <> '') desc, full_name limit 200) p2),
    'locations', (select coalesce(json_agg(row_to_json(l2)), '[]') from (
      select place_id, company, address, city, district, phone, rating, reviews, freshness
      from leads where company_id = p_id and coalesce(full_name,'') = ''
      order by reviews desc nulls last limit 200) l2));
$$;

create or replace function get_company_filters()
returns json language sql stable as $$
  select json_build_object(
    'cities', (select coalesce(json_agg(v), '[]') from (
      select city v from companies where coalesce(city,'') <> '' and company_key not like 'solo:%'
      group by city order by count(*) desc limit 60) x),
    'categories', (select coalesce(json_agg(v), '[]') from (
      select coalesce(category, industry) v from companies
      where coalesce(coalesce(category, industry),'') <> '' and company_key not like 'solo:%'
      group by 1 order by count(*) desc limit 100) x));
$$;

-- ============================================================
-- 19) Password-reset OTPs (2026-07-19) — emailed from Ahmed's Google Workspace via SMTP
--     (GMAIL_USER + GMAIL_APP_PASSWORD env), no paid mail service. Hashed, 10-min TTL,
--     single-use, 5 tries max.
-- ============================================================
create table if not exists password_otps (
  id         bigserial primary key,
  email      text not null,
  code_hash  text not null,
  expires_at timestamptz not null,
  tries      int default 0,
  used       boolean default false,
  created_at timestamptz default now()
);
create index if not exists password_otps_email_idx on password_otps (email, created_at desc);

-- ============================================================
-- 18) Next-actions queue (2026-07-19) — "what should I do now?"
--     Manual tasks (get_my_tasks) + AUTO follow-ups: my leads in an open stage past the
--     first column, quiet for 2+ days, with NO open task already planned (Pipedrive rule:
--     a planned lead needs no nagging). New-stage leads generate nothing — too early.
-- ============================================================
create or replace function get_next_actions(p_user uuid)
returns json language sql stable as $$
  select json_build_object(
    'tasks', get_my_tasks(p_user),
    'followups', (select coalesce(json_agg(row_to_json(f)), '[]') from (
      with mine as (
        select lc.place_id, lc.client_id, lc.stage_id, lc.entered_stage_at
        from lead_clients lc
        where (lc.owner_id = p_user
               or exists (select 1 from lead_assignees la
                          where la.place_id = lc.place_id and la.client_id = lc.client_id and la.user_id = p_user))
      )
      select l.place_id,
        coalesce(nullif(l.full_name,''), l.company, l.place_id) lead_name,
        cl.name client, ps.name stage_name, ps.color stage_color,
        greatest(0, extract(epoch from now() - greatest(
          coalesce((select max(ct.contacted_at) from contacts ct where ct.place_id = m.place_id), 'epoch'::timestamptz),
          coalesce((select max(wm.ts) from wa.messages wm
                     join wa.threads wt on wt.owner_account_id = wm.owner_account_id and wt.peer_number = wm.peer_number
                     where wt.place_id = m.place_id and wm.from_me), 'epoch'::timestamptz),
          m.entered_stage_at)) / 86400)::int days_silent,
        greatest(0, extract(epoch from now() - m.entered_stage_at) / 86400)::int days_in_stage
      from mine m
      join pipeline_stages ps on ps.id = m.stage_id and ps.kind = 'open' and ps.position > 0
      join clients cl on cl.id = m.client_id and cl.archived = false
      join leads l on l.place_id = m.place_id
      where not exists (select 1 from tasks t where t.place_id = m.place_id and t.done_at is null)
        and greatest(
          coalesce((select max(ct.contacted_at) from contacts ct where ct.place_id = m.place_id), 'epoch'::timestamptz),
          coalesce((select max(wm.ts) from wa.messages wm
                     join wa.threads wt on wt.owner_account_id = wm.owner_account_id and wt.peer_number = wm.peer_number
                     where wt.place_id = m.place_id and wm.from_me), 'epoch'::timestamptz),
          m.entered_stage_at) < now() - interval '2 days'
      order by 6 desc limit 30) f));
$$;

-- ============================================================
-- 20) WhatsApp messenger (2026-07-20) — the two-pane /whatsapp page.
--   get_wa_threads       : LEFT pane. One row per (owner_account_id, peer_number)
--                          across the user's accounts (admin: all), CRM name-matched
--                          (peer number → lead_phones/place_id → leads.full_name/company).
--   get_wa_conversation  : RIGHT pane. Message stream + resolved header for one thread.
-- Both HARD-FILTER wa.threads.visibility='team' (private/personal threads never leak).
-- Ownership scoping mirrors get_wa_accounts: (p_all OR a.user_id = p_user). The server
-- still re-checks account ownership before calling get_wa_conversation.
-- ============================================================

create or replace function get_wa_threads(
  p_user uuid, p_all boolean default false,
  p_account uuid default null, p_search text default null, lim int default 300)
returns json language sql stable as $$
  select coalesce(json_agg(row_to_json(x) order by x.last_epoch desc nulls last), '[]') from (
    select
      th.owner_account_id,
      a.label   account_label,
      a.phone   account_phone,
      coalesce(u.name, u.email, a.label, 'me') owner_user,
      th.peer_number,
      th.peer_phone,
      rp.place_id,
      (rp.place_id is not null) is_known,
      th.is_group,
      th.stage,
      th.display_name,
      coalesce(nullif(l.full_name,''), nullif(l.company,''), nullif(th.display_name,''),
               th.peer_phone, th.peer_number) name,
      lm.body       last_body,
      lm.media_type last_media,
      lm.from_me    last_from_me,
      th.msg_count,
      extract(epoch from th.last_msg_ts)::bigint last_epoch,
      to_char(th.last_msg_ts, 'YYYY-MM-DD HH24:MI') last_ts
    from wa.threads th
    join public.wa_accounts a on a.id = th.owner_account_id
    left join app_users u on u.id = a.user_id
    left join lateral (
      select coalesce(
        th.place_id,
        (select lp.place_id from public.lead_phones lp where lp.phone = th.peer_phone limit 1),
        (select lp.place_id from public.wa_lid_map lm
           join public.lead_phones lp on lp.phone = lm.phone
          where lm.lid = th.peer_number limit 1)
      ) place_id
    ) rp on true
    left join leads l on l.place_id = rp.place_id
    left join lateral (
      select m.body, m.media_type, m.from_me
      from wa.messages m
      where m.owner_account_id = th.owner_account_id and m.peer_number = th.peer_number
      order by m.ts desc limit 1
    ) lm on true
    where th.visibility = 'team'
      and (p_all or a.user_id = p_user)
      and (p_account is null or th.owner_account_id = p_account)
      and (coalesce(p_search,'') = '' or
           coalesce(l.full_name,'') || ' ' || coalesce(l.company,'') || ' ' ||
           coalesce(th.display_name,'') || ' ' || coalesce(th.peer_phone,'') || ' ' ||
           coalesce(th.peer_number,'') ilike '%' || p_search || '%')
    order by th.last_msg_ts desc nulls last
    limit lim
  ) x;
$$;

create or replace function get_wa_conversation(p_account uuid, p_peer text, lim int default 400)
returns json language sql stable as $$
  with th as (
    select t.*,
      coalesce(
        t.place_id,
        (select lp.place_id from public.lead_phones lp where lp.phone = t.peer_phone limit 1),
        (select lp.place_id from public.wa_lid_map lm
           join public.lead_phones lp on lp.phone = lm.phone
          where lm.lid = t.peer_number limit 1)
      ) rplace
    from wa.threads t
    where t.owner_account_id = p_account and t.peer_number = p_peer and t.visibility = 'team'
  )
  select json_build_object(
    'header', (select row_to_json(h) from (
      select t.peer_number, t.peer_phone, t.rplace place_id, t.display_name, t.is_group, t.stage,
             (t.rplace is not null) is_known,
             coalesce(nullif(l.full_name,''), nullif(l.company,''), nullif(t.display_name,''),
                      t.peer_phone, t.peer_number) name,
             a.id account_id, a.label account_label, a.phone account_phone, a.auto_reply,
             coalesce(u.name, u.email, a.label, 'me') owner_user
      from th t
      join public.wa_accounts a on a.id = t.owner_account_id
      left join app_users u on u.id = a.user_id
      left join leads l on l.place_id = t.rplace
    ) h),
    'messages', (select coalesce(json_agg(row_to_json(m) order by m.epoch asc), '[]') from (
      select x.wa_id, x.from_me, x.sender, x.body, x.media_type, x.media_hash, x.media_filename,
             x.quoted_text, extract(epoch from x.ts)::bigint epoch,
             to_char(x.ts,'YYYY-MM-DD HH24:MI') ts,
             md.mime media_mime, md.filename media_name, md.size media_size
      from (
        select m.* from wa.messages m
        where m.owner_account_id = p_account and m.peer_number = p_peer
          and exists (select 1 from th)
        order by m.ts desc limit lim
      ) x
      left join wa.media md on md.hash = x.media_hash
    ) m)
  );
$$;

-- People search for the WhatsApp "new message" box — match ANY WhatsApp-able number in
-- the Ultron DB (Saudi mobile in lead_phones), by name / company / number. Returns the
-- canonical phone so a picked person can be messaged directly.
create or replace function wa_people_search(q text, lim int default 10)
returns json language sql stable as $$
  select coalesce(json_agg(row_to_json(x)), '[]') from (
    select distinct on (lp.phone)
      l.place_id,
      coalesce(nullif(l.full_name,''), nullif(l.company,''), lp.phone) name,
      l.company, lp.phone
    from public.lead_phones lp
    join leads l on l.place_id = lp.place_id
    where lp.phone like '9665%'
      and (coalesce(l.full_name,'') || ' ' || coalesce(l.company,'') || ' ' || lp.phone)
          ilike '%' || coalesce(q,'') || '%'
    order by lp.phone, (nullif(l.full_name,'') is not null) desc
    limit lim
  ) x;
$$;

-- ============================================================
-- 21) Contact people / associates (2026-07-20) — attach one or more CONTACT PERSONS
-- (each itself a lead) to a company/branch lead. PURE LINK TABLE: removing an associate
-- never deletes the underlying lead row. Many-to-many. Ahmed's ask: a Maps branch's phone
-- is the branch number, not a person — so let a real contact person be attached.
-- ============================================================
create table if not exists public.lead_associates (
  company_place_id text not null references leads(place_id) on delete cascade,
  person_place_id  text not null references leads(place_id) on delete cascade,
  role             text,
  created_by       uuid references app_users(id),
  created_at       timestamptz default now(),
  primary key (company_place_id, person_place_id),
  check (company_place_id <> person_place_id)
);
create index if not exists lead_associates_person_idx on public.lead_associates(person_place_id);

-- General people/lead search for the contact-person picker. Deliberately NOT
-- wa_people_search — that one only returns Saudi mobiles (WhatsApp scope), which made
-- the "add a contact person" picker look broken because most searches returned nothing.
-- trigram index so the picker's ILIKE '%q%' is index-assisted, not a 60k-row seq scan
create index if not exists leads_people_search_trgm on leads using gin (
  (coalesce(full_name,'') || ' ' || coalesce(company,'') || ' ' || coalesce(phone,'')) gin_trgm_ops);

create or replace function people_search(q text, lim int default 10)
returns json language sql stable as $$
  select coalesce(json_agg(row_to_json(x)), '[]') from (
    select l.place_id,
           coalesce(nullif(l.full_name,''), nullif(l.company,''), l.place_id) name,
           l.company, l.phone, l.job_title
    from leads l
    where coalesce(q,'') <> ''
      and (coalesce(l.full_name,'') || ' ' || coalesce(l.company,'') || ' ' || coalesce(l.phone,''))
          ilike '%' || q || '%'
    order by (nullif(l.full_name,'') is not null) desc, l.company
    limit lim
  ) x;
$$;

create or replace function get_lead_associates(p_place_id text)
returns json language sql stable as $$
  select coalesce(json_agg(row_to_json(x) order by x.created_at), '[]') from (
    select a.person_place_id place_id,
           coalesce(nullif(l.full_name,''), nullif(l.company,''), l.place_id) name,
           l.phone, l.email, a.role, to_char(a.created_at,'YYYY-MM-DD') created_at
    from public.lead_associates a
    join leads l on l.place_id = a.person_place_id
    where a.company_place_id = p_place_id
  ) x;
$$;

-- ============================================================
-- 22) Company merge / detach (2026-07-20) — human-in-the-loop branch grouping.
-- Exact name+city grouping (§17) misses fuzzy variants (KTA / كيتا / Kayta). Rather than a
-- risky blind auto-merge on 46k companies, we SUGGEST similar companies (pg_trgm, same city)
-- and let Ahmed merge with one click — and detach a wrongly-grouped branch back out. All of
-- it just re-points leads.company_id, so nothing is destroyed and every move is reversible.
-- ============================================================
create extension if not exists pg_trgm;
create index if not exists companies_name_trgm on companies using gin (name_norm gin_trgm_ops);
create index if not exists companies_nameraw_trgm on companies using gin (name gin_trgm_ops);

-- companies that look like branches of p_id: similar normalized name, same city
create or replace function company_merge_suggestions(p_id bigint, p_limit int default 10)
returns json language sql stable as $$
  with base as (select name_norm, city from companies where company_id = p_id)
  select coalesce(json_agg(row_to_json(x) order by x.sim desc, x.lead_count desc), '[]') from (
    select c.company_id, c.name, c.city,
           (select count(*) from leads l where l.company_id = c.company_id)::int lead_count,
           round(similarity(c.name_norm, b.name_norm)::numeric, 2) sim
    from companies c, base b
    where c.company_id <> p_id
      and norm_company_city(c.city) is not distinct from norm_company_city(b.city)
      and b.name_norm <> '' and similarity(c.name_norm, b.name_norm) > 0.5
    order by sim desc, lead_count desc
    limit p_limit
  ) x;
$$;

-- merge p_merge INTO p_keep: re-point its leads, backfill missing keeper fields, drop the emptied one
create or replace function merge_companies(p_keep bigint, p_merge bigint)
returns json language plpgsql as $$
begin
  if p_keep = p_merge or p_keep is null or p_merge is null then return json_build_object('error','pick two different companies'); end if;
  update leads set company_id = p_keep where company_id = p_merge;
  update companies k set
    domain=coalesce(k.domain,m.domain), industry=coalesce(k.industry,m.industry),
    category=coalesce(k.category,m.category), company_size=coalesce(k.company_size,m.company_size),
    city=coalesce(k.city,m.city), rating=coalesce(k.rating,m.rating), reviews=coalesce(k.reviews,m.reviews)
  from companies m where k.company_id=p_keep and m.company_id=p_merge;
  delete from companies where company_id = p_merge;
  return json_build_object('ok', true);
end $$;

-- detach one lead (branch/person) into its OWN fresh company — undo a wrong grouping
create or replace function detach_lead_company(p_place_id text)
returns json language plpgsql as $$
declare new_id bigint; l record;
begin
  select * into l from leads where place_id = p_place_id;
  if not found then return json_build_object('error','no such lead'); end if;
  insert into companies (company_key, name, name_norm, category, city, district, country, address,
                         lat, lng, maps_url, rating, reviews, source)
  values ('solo:'||p_place_id, l.company, norm_company_name(l.company), l.category, l.city, l.district,
          l.country, l.address, l.lat, l.lng, l.maps_url, l.rating, l.reviews, coalesce(l.source,'Manual'))
  on conflict (company_key) do update set name=excluded.name
  returning company_id into new_id;
  update leads set company_id = new_id where place_id = p_place_id;
  return json_build_object('ok', true, 'company_id', new_id);
end $$;

-- Pipeline add-box: search COMPANIES (not branch-leads), so "ktaykit" is ONE result, not
-- five identical branches. Returns a representative lead (place_id) to drop into the stage —
-- prefer a real location/branch with the most reviews.
create or replace function pipeline_company_search(q text, lim int default 12)
returns json language sql stable as $$
  with dq as (select nullif(regexp_replace(coalesce(q,''),'\D','','g'),'') d)
  select coalesce(json_agg(row_to_json(x) order by x.reviews desc nulls last), '[]') from (
    select c.company_id, c.name, c.city, c.category, c.reviews,
      (select count(*) from leads l where l.company_id = c.company_id)::int lead_count,
      (select l.place_id from leads l where l.company_id = c.company_id
        order by (coalesce(l.full_name,'') = '') desc, l.reviews desc nulls last, l.place_id limit 1) rep_place_id
    from companies c, dq
    where c.name_norm <> ''
      and ( c.name ilike '%'||q||'%'
         or c.name_norm ilike '%'||norm_company_name(q)||'%'
         or (dq.d is not null and length(dq.d) >= 5
             and exists(select 1 from leads l where l.company_id = c.company_id and l.phone like '%'||dq.d||'%')) )
    order by c.reviews desc nulls last
    limit lim
  ) x;
$$;

-- ============================================================
-- 21) WhatsApp CAMPAIGNS + WARMUP ("Smartlead for WhatsApp", Milestone 1, 2026-07-26)
--     A paced outbound layer that sits ON TOP of the ultron-wa transport (supervisor→worker
--     doSend). NOTHING here touches auth/link/receive. A campaign = client + numbers + a
--     recipient queue + a message template + a Smartlead-style schedule; the ultron-wa
--     `pacer.js` loop drips sends within the window under per-number caps. `warmup.js` keeps
--     aged numbers healthy (maintenance) or ramps new ones. ADDITIVE ONLY (create-if-not-exists).
-- ============================================================

-- A campaign targets ONE client, runs across one-or-more of that team's WhatsApp numbers.
create table if not exists public.wa_campaigns (
  id                    uuid primary key default gen_random_uuid(),
  client_id             uuid references clients(id) on delete cascade,
  name                  text not null,
  status                text not null default 'draft',   -- draft | running | paused | done
  -- message content (spintax {a|b} + {name}/{company} vars are rendered by the pacer)
  message_template      text,
  media_key             text,                             -- storage key in the crm-files bucket (PDF/img)
  media_type            text,                             -- document | image
  media_name            text,
  -- Smartlead-style schedule (sane middle-ground defaults; the form pre-fills these)
  send_window_start     text not null default '09:00',    -- HH:MM, campaign timezone
  send_window_end       text not null default '16:00',
  timezone              text not null default 'Asia/Riyadh',
  active_days           int[] not null default '{0,1,2,3,4}',  -- 0=Sun … 6=Sat (Saudi work week Sun-Thu)
  sends_per_hour        int  not null default 8,
  daily_cap_per_number  int  not null default 40,
  jitter                boolean not null default true,    -- human pacing on the per-number cooldown
  created_by            uuid references app_users(id) on delete set null,
  created_at            timestamptz default now(),
  started_at            timestamptz
);
create index if not exists wa_campaigns_client_idx on public.wa_campaigns(client_id);
create index if not exists wa_campaigns_status_idx on public.wa_campaigns(status);

-- M:N — a campaign uses one or more numbers; a number can serve several campaigns.
create table if not exists public.wa_campaign_numbers (
  campaign_id uuid references public.wa_campaigns(id) on delete cascade,
  account_id  uuid references public.wa_accounts(id)  on delete cascade,
  primary key (campaign_id, account_id)
);
create index if not exists wa_campaign_numbers_acc_idx on public.wa_campaign_numbers(account_id);

-- THE send queue. One row per (campaign, lead). Canonical phone frozen at add-time.
create table if not exists public.wa_campaign_recipients (
  id                  bigserial primary key,
  campaign_id         uuid references public.wa_campaigns(id) on delete cascade,
  place_id            text references leads(place_id) on delete cascade,
  phone               text not null,                       -- canonical (norm_phone → 9665XXXXXXXX)
  status              text not null default 'queued',       -- queued|sent|failed|replied|skipped|opted_out
  assigned_account_id uuid references public.wa_accounts(id) on delete set null,
  scheduled_for       timestamptz,                          -- earliest send time (null = asap)
  sent_at             timestamptz,
  wa_id               text,                                 -- Baileys message id of the send
  error               text,
  attempts            int not null default 0,
  created_at          timestamptz default now(),
  unique (campaign_id, place_id)
);
create index if not exists wa_camp_recip_due_idx  on public.wa_campaign_recipients(campaign_id, status, scheduled_for);
create index if not exists wa_camp_recip_acc_idx  on public.wa_campaign_recipients(assigned_account_id, sent_at);
create index if not exists wa_camp_recip_sent_idx on public.wa_campaign_recipients(status, sent_at);

-- Per-number warmup config. Default = maintenance (light) so AGED numbers don't waste capacity.
create table if not exists public.wa_warmup_config (
  account_id     uuid primary key references public.wa_accounts(id) on delete cascade,
  enabled        boolean not null default true,
  mode           text not null default 'maintenance',   -- maintenance | ramp
  msgs_per_day   int  not null default 4,               -- target for maintenance mode
  ramp_start_day date,                                   -- day 1 of a ramp curve (mode='ramp')
  updated_at     timestamptz default now()
);

-- Warmup partners: the friendly numbers / other pool numbers a warmup account chats with.
create table if not exists public.wa_warmup_peers (
  id         bigserial primary key,
  account_id uuid references public.wa_accounts(id) on delete cascade,
  peer_phone text not null,                              -- canonical digits
  label      text,
  created_at timestamptz default now(),
  unique (account_id, peer_phone)
);

-- Campaign list for the /campaigns page (client-scoped by team membership; admins see all).
create or replace function get_campaigns(p_user uuid, p_admin boolean default false)
returns json language sql stable as $$
  select coalesce(json_agg(row_to_json(x) order by x.created_at desc), '[]') from (
    select c.id, c.name, c.status, c.client_id, cl.name client_name, cl.color client_color,
           c.sends_per_hour, c.daily_cap_per_number, c.send_window_start, c.send_window_end,
           to_char(c.created_at,'YYYY-MM-DD HH24:MI') created_at,
           (select count(*) from public.wa_campaign_recipients r where r.campaign_id = c.id)::int total,
           (select count(*) from public.wa_campaign_recipients r where r.campaign_id = c.id and r.status='sent')::int sent,
           (select count(*) from public.wa_campaign_recipients r where r.campaign_id = c.id and r.status='replied')::int replied,
           (select count(*) from public.wa_campaign_recipients r where r.campaign_id = c.id and r.status='queued')::int queued,
           (select count(*) from public.wa_campaign_numbers n where n.campaign_id = c.id)::int numbers
    from public.wa_campaigns c
    left join clients cl on cl.id = c.client_id
    where p_admin
       or exists (select 1 from user_clients uc where uc.client_id = c.client_id and uc.user_id = p_user)
  ) x;
$$;

-- Recipient status tally (queued/sent/replied/failed/…) for a campaign.
create or replace function campaign_stats(p_id uuid)
returns json language sql stable as $$
  select coalesce(json_object_agg(status, cnt), '{}'::json) from (
    select status, count(*)::int cnt from public.wa_campaign_recipients
    where campaign_id = p_id group by status
  ) s;
$$;

-- One campaign in full: config + attached numbers + status counts + a recipient sample.
create or replace function get_campaign(p_id uuid)
returns json language sql stable as $$
  select json_build_object(
    'campaign', (select row_to_json(c) from (
        select c.*, cl.name client_name, cl.color client_color
        from public.wa_campaigns c left join clients cl on cl.id = c.client_id
        where c.id = p_id) c),
    'numbers', (select coalesce(json_agg(row_to_json(a)),'[]') from (
        select a.id, a.phone, a.label, a.status,
               coalesce(u.name, u.email) user_name
        from public.wa_campaign_numbers n
        join public.wa_accounts a on a.id = n.account_id
        left join app_users u on u.id = a.user_id
        where n.campaign_id = p_id order by a.created_at) a),
    'counts', campaign_stats(p_id),
    'recipients', (select coalesce(json_agg(row_to_json(r)),'[]') from (
        select r.id, r.place_id, r.phone, r.status, r.assigned_account_id,
               l.company, l.full_name,
               to_char(r.sent_at,'YYYY-MM-DD HH24:MI') sent_at, r.error
        from public.wa_campaign_recipients r
        left join leads l on l.place_id = r.place_id
        where r.campaign_id = p_id
        order by r.sent_at desc nulls last, r.id limit 300) r)
  );
$$;

-- Add recipients from a set of place_ids. Only Saudi-mobile (WhatsApp-able) leads are queued;
-- landlines/toll numbers are silently skipped. Idempotent (unique (campaign_id, place_id)).
-- Returns how many were newly queued.
create or replace function campaign_add_recipients(p_campaign uuid, p_place_ids text[])
returns json language plpgsql as $$
declare added int; n_mobile int; n_total int;
begin
  n_total := coalesce(array_length(p_place_ids, 1), 0);
  with cand as (
    select l.place_id, norm_phone(l.phone) ph
    from leads l
    where l.place_id = any(p_place_ids)
      and norm_phone(l.phone) ~ '^9665[0-9]{8}$'
  ), ins as (
    insert into public.wa_campaign_recipients (campaign_id, place_id, phone, status)
    select p_campaign, place_id, ph, 'queued' from cand
    on conflict (campaign_id, place_id) do nothing
    returning 1
  )
  select (select count(*) from cand), (select count(*) from ins) into n_mobile, added;
  return json_build_object('added', added, 'skipped_no_mobile', n_total - n_mobile);
end $$;

-- Reply detection: flip 'sent' recipients to 'replied' when the peer's thread on the number
-- that sent it has an inbound message AFTER our send. Called by the pacer periodically and by
-- the dashboard on demand. p_campaign null = sweep every running/paused campaign. Idempotent.
create or replace function campaign_mark_replies(p_campaign uuid default null)
returns json language plpgsql as $$
declare flipped int;
begin
  with upd as (
    update public.wa_campaign_recipients r
       set status = 'replied'
      from wa.threads t
     where r.status = 'sent'
       and r.sent_at is not null
       and r.assigned_account_id is not null
       and t.owner_account_id = r.assigned_account_id
       -- LID-aware (2026-07-29): a reply usually lands in the LID-keyed thread (peer_number = LID,
       -- peer_phone = canonical). Matching peer_number alone silently missed those replies.
       and (t.peer_number = r.phone or t.peer_phone = r.phone)
       and t.last_inbound_ts is not null
       and t.last_inbound_ts > r.sent_at
       and (p_campaign is null or r.campaign_id = p_campaign)
    returning 1
  )
  select count(*) into flipped from upd;
  return json_build_object('replied', flipped);
end $$;

-- Warmup config + peers for the numbers a user can act on: numbers they own + numbers shared to
-- them as MANAGER (admins: all real numbers). `configured` = a real config row exists (drives the
-- saved-state CARD vs the setup FORM in the UI).
create or replace function get_warmup(p_user uuid, p_admin boolean default false)
returns json language sql stable as $$
  select coalesce(json_agg(row_to_json(x) order by x.created_at), '[]') from (
    select a.id account_id, a.phone, a.label, a.status,
           coalesce(w.enabled, true) enabled,
           coalesce(w.mode, 'maintenance') mode,
           coalesce(w.msgs_per_day, 4) msgs_per_day,
           w.ramp_start_day, a.created_at,
           (w.account_id is not null) configured,
           (select coalesce(json_agg(json_build_object('id', p.id, 'peer_phone', p.peer_phone, 'label', p.label) order by p.id), '[]')
              from public.wa_warmup_peers p where p.account_id = a.id) peers
    from public.wa_accounts a
    left join public.wa_warmup_config w on w.account_id = a.id
    where (a.label is null or a.label not like 'mac-reader:%')
      and (p_admin or a.user_id = p_user
           or exists (select 1 from public.wa_account_shares s
                       where s.account_id = a.id and s.user_id = p_user and s.role = 'manager'))
  ) x;
$$;

-- ============================================================
-- 22) WhatsApp FLOWS + AI-REPLY BRAIN ("Smartlead for WhatsApp", Milestone 2, 2026-07-27)
--     A FLOW is a node-graph (JSON) belonging to a client that drives each lead through the
--     client's existing pipeline_stages as a WhatsApp conversation progresses. Sits ON TOP of the
--     M1 campaign/pacer layer + the ultron-wa transport — the engine lives in ultron-wa/flow.js
--     (mirrors pacer.js/autoreply.js) and drives sends via the SAME supervisor→worker doSend
--     primitive. ADDITIVE ONLY (create-if-not-exists / add-column-if-not-exists). Nothing here
--     touches M1, auth, link, or receive.
--
--     Node types (see flow.js): send · send_file · wait · wait_reply · ai · condition · set_stage ·
--     handoff. A run advances on a supervisor tick (wait_until elapsed) AND when the lead replies
--     (flow.consider hook next to autoreply.consider). "Negotiation needs a human" = a handoff node
--     that stops automation and creates a `tasks` row for the lead's owner.
-- ============================================================

-- A flow belongs to ONE client. graph = { entry: <nodeId>, nodes:[{id,type,config,x,y}], edges:[{from,to,branch}] }.
create table if not exists public.wa_flows (
  id          uuid primary key default gen_random_uuid(),
  client_id   uuid references clients(id) on delete cascade,
  name        text not null,
  graph       jsonb not null default '{"entry":null,"nodes":[],"edges":[]}'::jsonb,
  status      text not null default 'draft',    -- draft | active | archived
  created_by  uuid references app_users(id) on delete set null,
  created_at  timestamptz default now(),
  updated_at  timestamptz default now()
);
create index if not exists wa_flows_client_idx on public.wa_flows(client_id);
create index if not exists wa_flows_status_idx on public.wa_flows(status);

-- One run per (flow, lead). The live cursor: which node the lead sits at, when to next advance,
-- and per-lead variables (context — incl. the lead's last inbound message for the AI/condition nodes).
create table if not exists public.wa_flow_runs (
  id              uuid primary key default gen_random_uuid(),
  flow_id         uuid references public.wa_flows(id) on delete cascade,
  place_id        text references leads(place_id) on delete cascade,
  client_id       uuid references clients(id) on delete set null,
  current_node_id text,
  status          text not null default 'active',  -- active | waiting | done | handoff | stopped
  wait_until      timestamptz,                     -- earliest next tick (null = ready now)
  context         jsonb not null default '{}'::jsonb,
  account_id      uuid references public.wa_accounts(id) on delete set null,  -- number that owns the convo
  peer_phone      text,                            -- canonical digits (norm_phone → 9665XXXXXXXX)
  started_at      timestamptz default now(),
  updated_at      timestamptz default now(),
  unique (flow_id, place_id)
);
create index if not exists wa_flow_runs_due_idx  on public.wa_flow_runs(status, wait_until);
create index if not exists wa_flow_runs_peer_idx on public.wa_flow_runs(account_id, peer_phone, status);
create index if not exists wa_flow_runs_flow_idx on public.wa_flow_runs(flow_id, status);

-- Per-client knowledge base the AI node reads (FAQ / offer / tone). Explicit, separate from clients.brief.
alter table clients add column if not exists kb_doc text;

-- Optional M1↔M2 link: a campaign can name a flow to auto-start a run for a recipient when they reply.
alter table public.wa_campaigns add column if not exists flow_id uuid references public.wa_flows(id) on delete set null;

-- Flow list for the /flows page (client-scoped by team membership; admins see all).
create or replace function get_flows(p_user uuid, p_admin boolean default false)
returns json language sql stable as $$
  select coalesce(json_agg(row_to_json(x) order by x.updated_at desc), '[]') from (
    select f.id, f.name, f.status, f.client_id, cl.name client_name, cl.color client_color,
           to_char(f.updated_at,'YYYY-MM-DD HH24:MI') updated_at,
           coalesce(jsonb_array_length(f.graph->'nodes'), 0) nodes,
           (select count(*) from public.wa_flow_runs r where r.flow_id = f.id)::int runs_total,
           (select count(*) from public.wa_flow_runs r where r.flow_id = f.id and r.status in ('active','waiting'))::int runs_active,
           (select count(*) from public.wa_flow_runs r where r.flow_id = f.id and r.status='handoff')::int runs_handoff,
           (select count(*) from public.wa_flow_runs r where r.flow_id = f.id and r.status='done')::int runs_done
    from public.wa_flows f
    left join clients cl on cl.id = f.client_id
    where p_admin
       or exists (select 1 from user_clients uc where uc.client_id = f.client_id and uc.user_id = p_user)
  ) x;
$$;

-- One flow in full (the builder loads this): config + graph + run status counts.
create or replace function get_flow(p_id uuid)
returns json language sql stable as $$
  select json_build_object(
    'flow', (select row_to_json(c) from (
        select f.*, cl.name client_name, cl.color client_color, cl.kb_doc client_kb
        from public.wa_flows f left join clients cl on cl.id = f.client_id
        where f.id = p_id) c),
    'counts', (select coalesce(json_object_agg(status, cnt), '{}'::json) from (
        select status, count(*)::int cnt from public.wa_flow_runs where flow_id = p_id group by status) s)
  );
$$;

-- Runs monitor: which leads sit in which node right now (most-recently-updated first).
-- `trace` = the run's last ~12 node steps (context._trace, written by flow.js) — n8n-style
-- input/output visibility per node, added 2026-07-29.
create or replace function get_flow_runs(p_flow uuid, lim int default 300)
returns json language sql stable as $$
  select coalesce(json_agg(row_to_json(x) order by x.updated_at desc), '[]') from (
    select r.id, r.place_id, r.status, r.current_node_id, r.peer_phone,
           coalesce(nullif(l.full_name,''), l.company, r.place_id) lead_name,
           l.company, a.phone account_phone,
           to_char(r.wait_until,'YYYY-MM-DD HH24:MI') wait_until,
           to_char(r.updated_at,'YYYY-MM-DD HH24:MI') updated_at,
           coalesce(r.context->'_trace', '[]'::jsonb) trace
    from public.wa_flow_runs r
    left join leads l on l.place_id = r.place_id
    left join public.wa_accounts a on a.id = r.account_id
    where r.flow_id = p_flow
    order by r.updated_at desc limit lim
  ) x;
$$;

-- Start (or fetch the existing) run for a lead on a flow. Cursor = graph.entry (fallback: first node).
-- Idempotent via unique(flow_id, place_id) — a re-reply never double-starts. Called by flow.js.
create or replace function start_flow_run(p_flow uuid, p_place text, p_account uuid, p_peer text)
returns json language plpgsql as $$
declare entry text; existing public.wa_flow_runs; g jsonb; cid uuid; rid uuid;
begin
  select r.* into existing from public.wa_flow_runs r where r.flow_id = p_flow and r.place_id = p_place;
  if found then return json_build_object('ok', true, 'run_id', existing.id, 'created', false, 'status', existing.status); end if;
  select f.graph, f.client_id into g, cid from public.wa_flows f where f.id = p_flow;
  if g is null then return json_build_object('ok', false, 'error', 'flow not found'); end if;
  entry := nullif(g->>'entry','');
  if entry is null then entry := g->'nodes'->0->>'id'; end if;
  if entry is null then return json_build_object('ok', false, 'error', 'flow has no entry node'); end if;
  insert into public.wa_flow_runs (flow_id, place_id, client_id, current_node_id, status, account_id, peer_phone)
    values (p_flow, p_place, cid, entry, 'active', p_account, norm_phone(p_peer))
    on conflict (flow_id, place_id) do nothing
    returning id into rid;
  return json_build_object('ok', true, 'run_id', rid, 'created', rid is not null);
end $$;

-- ============================================================
-- §23 — Hosted MCP connector token (Attio-style "add a link")
-- Each app_user gets ONE revocable, regenerable token. A teammate's personal
-- connector URL = <dashboard-origin>/mcp/<mcp_token>. The dashboard mounts the
-- Ultron MCP over Streamable HTTP at /mcp/:token, resolves the token → app_user,
-- and scopes every tool to that user (admin => see-all, agent => their own
-- numbers/clients). Revoke = null the token (link 401s); regenerate = new random.
-- The token is a bearer secret, so it is NEVER exposed via any PostgREST RPC —
-- the dashboard reads/writes it server-side with the service key only.
-- ============================================================
alter table app_users add column if not exists mcp_token text;
create unique index if not exists app_users_mcp_token_idx on app_users (mcp_token) where mcp_token is not null;

-- ============================================================
-- §24 — LEAD LISTS + CAMPAIGN FUNNEL KPIs + PER-NUMBER REPLY FLOW (2026-07-29)
--     The "make it one product" pass. LISTS are the Apollo-style glue: select people on /leads
--     (checkboxes or a whole filter), save them as a named list, feed the list into a campaign.
--     FUNNEL RPCs answer the questions clients ask weekly (sent / replied / meetings / won)
--     without counting phone threads by hand. reply_flow_id is the inbound catch-all: the flow
--     that owns UNSOLICITED inbound messages on a number (flow.js spawnReplyFlow).
--     ADDITIVE ONLY.
-- ============================================================

create table if not exists public.lead_lists (
  id         uuid primary key default gen_random_uuid(),
  name       text not null,
  client_id  uuid references clients(id) on delete set null,   -- optional client tag
  created_by uuid references app_users(id) on delete set null,
  archived   boolean not null default false,
  created_at timestamptz default now(),
  updated_at timestamptz default now()
);
create table if not exists public.lead_list_members (
  list_id  uuid not null references public.lead_lists(id) on delete cascade,
  place_id text not null references leads(place_id) on delete cascade,
  added_by uuid references app_users(id) on delete set null,
  added_at timestamptz default now(),
  primary key (list_id, place_id)
);
create index if not exists lead_list_members_place_idx on public.lead_list_members(place_id);

-- The flow that owns UNSOLICITED inbound on this number (customer-service catch-all). Null = off.
alter table public.wa_accounts add column if not exists reply_flow_id uuid references public.wa_flows(id) on delete set null;

-- SOFT delete for numbers (2026-07-29, Ahmed's rule: NEVER delete messages/history). A deleted
-- number keeps its wa.threads/wa.messages and its phone+label, is hidden from every UI list, and
-- the reconciler won't run it. Restore = deleted_at back to null (same uuid → history intact).
alter table public.wa_accounts add column if not exists deleted_at timestamptz;

-- What deleting a number touches — powers the confirm modal. Messages/threads are KEPT (informational);
-- campaigns listed get detached, and a running campaign whose ONLY sender this is gets auto-paused.
create or replace function wa_account_impact(p_id uuid)
returns json language sql stable as $$
  select json_build_object(
    'phone',    (select phone from public.wa_accounts where id = p_id),
    'label',    (select label from public.wa_accounts where id = p_id),
    'threads',  (select count(*)::int from wa.threads t where t.owner_account_id = p_id),
    'messages', (select count(*)::int from wa.messages m where m.owner_account_id = p_id),
    'live_runs',(select count(*)::int from public.wa_flow_runs r where r.account_id = p_id and r.status in ('active','waiting')),
    'is_reply_brain', exists (select 1 from public.wa_accounts a where a.id = p_id and a.reply_flow_id is not null),
    'campaigns', (select coalesce(json_agg(json_build_object(
          'id', c.id, 'name', c.name, 'status', c.status,
          'sole_sender', (select count(*) from public.wa_campaign_numbers n3 where n3.campaign_id = c.id) = 1
        )), '[]')
        from public.wa_campaign_numbers n join public.wa_campaigns c on c.id = n.campaign_id
       where n.account_id = p_id)
  );
$$;

-- When a recipient row was claimed for sending (2026-07-29): lets the pacer's orphan-reaper requeue
-- only STALE claims (>5min) instead of racing rows that are legitimately mid-send on a slow proxy.
alter table public.wa_campaign_recipients add column if not exists claimed_at timestamptz;

-- v2 of campaign_add_recipients (overrides §21): ONE recipient per human PHONE. The same phone often
-- exists under several place_ids (one owner, many Maps places / a wa: sparse lead) — queueing each
-- place double-messaged real people (live incident 2026-07-29: one phone queued 9×). Also never queue
-- our OWN infrastructure numbers, nor a phone this campaign already holds in any status.
create or replace function campaign_add_recipients(p_campaign uuid, p_place_ids text[])
returns json language plpgsql as $$
declare added int; n_mobile int; n_total int;
begin
  n_total := coalesce(array_length(p_place_ids, 1), 0);
  with cand0 as (
    select l.place_id, norm_phone(l.phone) ph
    from leads l
    where l.place_id = any(p_place_ids)
      and norm_phone(l.phone) ~ '^9665[0-9]{8}$'
  ), cand as (
    select distinct on (ph) place_id, ph from cand0
    where not exists (select 1 from public.wa_accounts a where a.phone = cand0.ph)
      and not exists (select 1 from public.wa_campaign_recipients r
                       where r.campaign_id = p_campaign and r.phone = cand0.ph)
    order by ph, place_id
  ), ins as (
    insert into public.wa_campaign_recipients (campaign_id, place_id, phone, status)
    select p_campaign, place_id, ph, 'queued' from cand
    on conflict (campaign_id, place_id) do nothing
    returning 1
  )
  select (select count(*) from cand0), (select count(*) from ins) into n_mobile, added;
  return json_build_object('added', added, 'skipped_no_mobile', n_total - n_mobile);
end $$;

-- v2 of get_warmup (overrides §21): deleted numbers must not appear in the warmup list.
create or replace function get_warmup(p_user uuid, p_admin boolean default false)
returns json language sql stable as $$
  select coalesce(json_agg(row_to_json(x) order by x.created_at), '[]') from (
    select a.id account_id, a.phone, a.label, a.status,
           coalesce(w.enabled, true) enabled,
           coalesce(w.mode, 'maintenance') mode,
           coalesce(w.msgs_per_day, 4) msgs_per_day,
           w.ramp_start_day, a.created_at,
           (w.account_id is not null) configured,
           (select coalesce(json_agg(json_build_object('id', p.id, 'peer_phone', p.peer_phone, 'label', p.label) order by p.id), '[]')
              from public.wa_warmup_peers p where p.account_id = a.id) peers
    from public.wa_accounts a
    left join public.wa_warmup_config w on w.account_id = a.id
    where (a.label is null or a.label not like 'mac-reader:%')
      and a.deleted_at is null
      and (p_admin or a.user_id = p_user
           or exists (select 1 from public.wa_account_shares s
                       where s.account_id = a.id and s.user_id = p_user and s.role = 'manager'))
  ) x;
$$;

-- Warmup partner options (2026-07-29, Ahmed's rule): EVERY real linked number in the system —
-- any user's, no share/invite needed — because only infra numbers may be warmup partners.
create or replace function get_warmup_partner_options()
returns json language sql stable as $$
  select coalesce(json_agg(row_to_json(x) order by x.label nulls last), '[]') from (
    select a.id, a.phone, a.label, coalesce(u.name, u.email) user_name, a.status
    from public.wa_accounts a
    left join app_users u on u.id = a.user_id
    where a.deleted_at is null and a.phone is not null
      and (a.label is null or a.label not like 'mac-reader:%')
  ) x;
$$;

-- v2 of get_wa_accounts (overrides the §15 definition — must live AFTER the alter above so a fresh
-- db-setup run parses): adds reply_flow_id + its flow name/status so /connections can show the
-- catch-all selector per number.
create or replace function get_wa_accounts(p_user uuid, p_all boolean default false)
returns json language sql stable as $$
  select coalesce(json_agg(row_to_json(x) order by x.created_at), '[]') from (
    select a.id, a.user_id, coalesce(u.name, u.email) user_name, a.phone, a.label, a.status,
           a.auto_reply, to_char(a.last_connected_at,'YYYY-MM-DD HH24:MI') last_connected_at,
           to_char(a.last_seen,'YYYY-MM-DD HH24:MI') last_seen, a.last_disconnect_reason, a.created_at,
           a.reply_flow_id, rf.name reply_flow_name, rf.status reply_flow_status,
           case when p_all or a.user_id = p_user then 'owner' else coalesce(s.role, 'viewer') end role,
           (p_all or a.user_id = p_user) is_owner
    from public.wa_accounts a
    left join app_users u on u.id = a.user_id
    left join public.wa_flows rf on rf.id = a.reply_flow_id
    left join public.wa_account_shares s on s.account_id = a.id and s.user_id = p_user
    where a.label not like 'mac-reader:%'
      and a.deleted_at is null
      and (p_all or a.user_id = p_user or s.user_id is not null)
  ) x;
$$;

-- All (non-archived) lists with member + WhatsApp-able counts. Team-wide visibility by design.
create or replace function get_lead_lists()
returns json language sql stable as $$
  select coalesce(json_agg(row_to_json(x) order by x.updated_at desc), '[]') from (
    select ll.id, ll.name, ll.client_id, cl.name client_name, cl.color client_color,
           to_char(ll.created_at,'YYYY-MM-DD') created_at,
           to_char(ll.updated_at,'YYYY-MM-DD HH24:MI') updated_at,
           coalesce(u.name, u.email) created_by_name,
           (select count(*) from public.lead_list_members m where m.list_id = ll.id)::int members,
           (select count(*) from public.lead_list_members m join leads l on l.place_id = m.place_id
             where m.list_id = ll.id and norm_phone(l.phone) ~ '^9665[0-9]{8}$')::int mobiles
    from public.lead_lists ll
    left join clients cl on cl.id = ll.client_id
    left join app_users u on u.id = ll.created_by
    where not ll.archived
  ) x;
$$;

-- Add explicit place_ids to a list (the checkbox path). Idempotent.
create or replace function list_add_members(p_list uuid, p_place_ids text[], p_user uuid default null)
returns json language plpgsql as $$
declare added int;
begin
  with ins as (
    insert into public.lead_list_members (list_id, place_id, added_by)
    select p_list, pid, p_user from unnest(p_place_ids) pid
    on conflict (list_id, place_id) do nothing
    returning 1)
  select count(*) into added from ins;
  update public.lead_lists set updated_at = now() where id = p_list;
  return json_build_object('added', coalesce(added, 0));
end $$;

-- Add EVERYTHING matching a /leads filter (the "select all N matching" path). Deliberately a thin
-- wrapper around search_leads itself — same params, same semantics, ZERO duplicated filter logic —
-- so the list always contains exactly what the page showed.
create or replace function list_add_from_filter(
  p_list uuid, p_user uuid default null,
  q text default '', p_category text default 'all', p_city text default 'all',
  p_fresh text default 'all', p_verified text default 'all', p_phone text default 'all',
  p_source text default 'all', p_industry text default 'all', p_seniority text default 'all',
  p_job_title text default 'all', p_department text default 'all',
  p_state text default 'all', p_size text default 'all',
  p_rating text default 'all', p_vol text default 'all', p_presence text default 'all',
  p_has_wa text default 'all', p_client uuid default null, p_assigned_only boolean default false)
returns json language plpgsql as $$
declare res json; added int;
begin
  res := search_leads(q, p_category, p_city, p_fresh, p_verified, p_phone, p_source, p_industry,
                      p_seniority, p_job_title, p_department, p_state, p_size, p_rating, p_vol,
                      p_presence, p_has_wa, p_client, p_assigned_only, 500000, 0);
  with ids as (select (r->>'place_id') pid from json_array_elements(res->'rows') r),
  ins as (
    insert into public.lead_list_members (list_id, place_id, added_by)
    select p_list, pid, p_user from ids where pid is not null
    on conflict (list_id, place_id) do nothing
    returning 1)
  select count(*) into added from ins;
  update public.lead_lists set updated_at = now() where id = p_list;
  return json_build_object('added', coalesce(added, 0), 'matched', json_array_length(res->'rows'));
end $$;

-- Feed a whole list into a campaign (mobile-only + dedupe rules live in campaign_add_recipients).
create or replace function campaign_add_from_list(p_campaign uuid, p_list uuid)
returns json language sql as $$
  select campaign_add_recipients(p_campaign,
    coalesce((select array_agg(place_id) from public.lead_list_members where list_id = p_list), '{}'));
$$;

-- ONE run in full (2026-07-29): the run + its whole trace + the flow's graph — powers the
-- campaign "lead flow" modal (green/red node tree, per-node in/out, retry-from-break).
create or replace function get_flow_run(p_run uuid)
returns json language sql stable as $$
  select json_build_object(
    'run', (select row_to_json(x) from (
       select r.id, r.flow_id, r.place_id, r.client_id, r.status, r.current_node_id, r.peer_phone,
              r.account_id, to_char(r.wait_until,'YYYY-MM-DD HH24:MI') wait_until,
              to_char(r.updated_at,'YYYY-MM-DD HH24:MI') updated_at,
              coalesce(r.context->'_trace','[]'::jsonb) trace,
              coalesce(nullif(l.full_name,''), l.company, r.peer_phone) lead_name
         from public.wa_flow_runs r
         left join leads l on l.place_id = r.place_id
        where r.id = p_run) x),
    'flow', (select json_build_object('id', f.id, 'name', f.name, 'status', f.status, 'graph', f.graph)
               from public.wa_flow_runs r join public.wa_flows f on f.id = r.flow_id
              where r.id = p_run)
  );
$$;

-- ---- FUNNEL KPIs -------------------------------------------------------------------------------
-- The numbers clients actually ask for. Recipient statuses come from the campaign queue; live
-- conversation states come from the campaign flow's runs; meetings/won come from REAL pipeline
-- stage moves (lead_stage_history) after the lead was messaged — plus the ai_agent's own outcome.
create or replace function campaign_funnel(p_id uuid)
returns json language sql stable as $$
  with c as (select flow_id, client_id from public.wa_campaigns where id = p_id),
  r as (select rr.*, fr.status run_status, fr.context run_ctx
          from public.wa_campaign_recipients rr
          left join c on true
          left join public.wa_flow_runs fr on fr.flow_id = c.flow_id and fr.place_id = rr.place_id
         where rr.campaign_id = p_id),
  moved as (select distinct h.place_id, ps.kind, lower(ps.name) sname
              from lead_stage_history h
              join pipeline_stages ps on ps.id = h.to_stage_id, c
             where h.client_id = c.client_id
               and exists (select 1 from public.wa_campaign_recipients rr
                            where rr.campaign_id = p_id and rr.place_id = h.place_id
                              and rr.sent_at is not null and h.changed_at >= rr.sent_at))
  select json_build_object(
    'total',           (select count(*) from r),
    'queued',          (select count(*) from r where status = 'queued'),
    'sent',            (select count(*) from r where status in ('sent','replied')),
    'replied',         (select count(*) from r where status = 'replied'),
    'failed',          (select count(*) from r where status = 'failed'),
    'opted_out',       (select count(*) from r where status = 'opted_out'),
    'in_conversation', (select count(*) from r where run_status in ('active','waiting')),
    'handoff',         (select count(*) from r where run_status = 'handoff'),
    'meetings',        (select count(distinct place_id) from moved where sname like 'meeting%'),
    'won',             (select count(*) from (
                          select place_id from moved where kind = 'won'
                          union
                          select place_id from r where run_status = 'done' and run_ctx->>'_agent_outcome' = 'won') w),
    'lost',            (select count(*) from r where run_status = 'done' and run_ctx->>'_agent_outcome' = 'lost'),
    'nurture',         (select count(*) from r where run_status = 'done' and run_ctx->>'_agent_outcome' = 'nurture'),
    'daily',           (select coalesce(json_agg(row_to_json(d) order by d.day), '[]') from (
                          select to_char(sent_at at time zone 'Asia/Riyadh','YYYY-MM-DD') as day,
                                 count(*)::int sent,
                                 (count(*) filter (where status = 'replied'))::int replied
                            from public.wa_campaign_recipients
                           where campaign_id = p_id and sent_at is not null
                             and sent_at >= now() - interval '30 days'
                           group by 1) d)
  );
$$;

-- Read a list's members as leads-page rows (same shape as search_leads rows, contact columns nulled)
-- so /leads can VIEW a saved list via its List dropdown. Newest-added first.
create or replace function get_list_leads(p_list uuid, lim int default 100, off int default 0)
returns json language sql stable as $$
  select json_build_object(
    'total', (select count(*) from public.lead_list_members m where m.list_id = p_list),
    'rows', (select coalesce(json_agg(r), '[]') from (
      select l.place_id, l.company, l.category, l.city, l.district, l.country, l.phone, l.email,
             l.email_status, l.email_verified, l.website, l.instagram, l.facebook, l.tiktok,
             l.snapchat, l.youtube, l.twitter, l.whatsapp, l.rating, l.reviews, l.review_tier,
             l.freshness, l.oldest_review, l.newest_review, l.address, l.lat, l.lng, l.maps_url,
             l.source, l.keyword, to_char(l.first_seen,'YYYY-MM-DD HH24:MI') ts,
             l.full_name, l.job_title, l.seniority, l.department, l.industry, l.state, l.linkedin,
             l.company_size, l.source_detail,
             null::text last_contact_channel, null::text last_contact_by, null::text last_contact_at
      from public.lead_list_members m
      join leads l on l.place_id = m.place_id
      where m.list_id = p_list
      order by m.added_at desc
      limit lim offset off) r)
  );
$$;

-- v2 of get_campaign (overrides §21 — must live here AFTER §22's wa_flow_runs exists for fresh runs):
-- each recipient row now carries its FLOW-RUN progress — run_status, the node it sits at, and the
-- last trace step — so the campaign detail answers "where exactly did this lead stop" (2026-07-29).
create or replace function get_campaign(p_id uuid)
returns json language sql stable as $$
  select json_build_object(
    'campaign', (select row_to_json(c) from (
        select c.*, cl.name client_name, cl.color client_color
        from public.wa_campaigns c left join clients cl on cl.id = c.client_id
        where c.id = p_id) c),
    'numbers', (select coalesce(json_agg(row_to_json(a)),'[]') from (
        select a.id, a.phone, a.label, a.status,
               coalesce(u.name, u.email) user_name
        from public.wa_campaign_numbers n
        join public.wa_accounts a on a.id = n.account_id
        left join app_users u on u.id = a.user_id
        where n.campaign_id = p_id order by a.created_at) a),
    'counts', campaign_stats(p_id),
    'recipients', (select coalesce(json_agg(row_to_json(r)),'[]') from (
        select r.id, r.place_id, r.phone, r.status, r.assigned_account_id,
               l.company, l.full_name,
               to_char(r.sent_at,'YYYY-MM-DD HH24:MI') sent_at, r.error,
               fr.status run_status, fr.current_node_id run_node, fr.id run_id,
               coalesce(fr.context->'_trace'->-1, null) last_trace
        from public.wa_campaign_recipients r
        left join leads l on l.place_id = r.place_id
        left join public.wa_campaigns c2 on c2.id = r.campaign_id
        left join public.wa_flow_runs fr on fr.flow_id = c2.flow_id and fr.place_id = r.place_id
        where r.campaign_id = p_id
        order by r.sent_at desc nulls last, r.id limit 300) r)
  );
$$;

-- Per-client rollup across ALL its campaigns in a window — the weekly client report in one call.
create or replace function client_funnel(p_client uuid, p_days int default 30)
returns json language sql stable as $$
  with camp as (select id, flow_id from public.wa_campaigns where client_id = p_client),
  r as (select rr.*, c.flow_id
          from public.wa_campaign_recipients rr join camp c on c.id = rr.campaign_id
         where rr.sent_at is null or rr.sent_at >= now() - (p_days || ' days')::interval),
  runs as (select r.place_id, fr.status run_status, fr.context run_ctx
             from r join public.wa_flow_runs fr on fr.flow_id = r.flow_id and fr.place_id = r.place_id),
  moved as (select distinct h.place_id, ps.kind, lower(ps.name) sname
              from lead_stage_history h
              join pipeline_stages ps on ps.id = h.to_stage_id
             where h.client_id = p_client
               and h.changed_at >= now() - (p_days || ' days')::interval
               and exists (select 1 from r where r.place_id = h.place_id and r.sent_at is not null))
  select json_build_object(
    'days',            p_days,
    'campaigns',       (select count(*) from camp),
    'sent',            (select count(*) from r where status in ('sent','replied')),
    'replied',         (select count(*) from r where status = 'replied'),
    'in_conversation', (select count(*) from runs where run_status in ('active','waiting')),
    'handoff',         (select count(*) from runs where run_status = 'handoff'),
    'meetings',        (select count(distinct place_id) from moved where sname like 'meeting%'),
    'won',             (select count(*) from (
                          select place_id from moved where kind = 'won'
                          union
                          select place_id from runs where run_status = 'done' and run_ctx->>'_agent_outcome' = 'won') w)
  );
$$;

-- ============================================================
-- §25 — LEAD-LEVEL AI MEMORY (2026-07-30): durable, PHONE-anchored memory the ai_agent reads across
-- ALL of a lead's conversations (any sender number / campaign / flow / client), each row tagged with
-- provenance so the AI knows "this was for Client A, via +9665…, 12 days ago" vs the current chat.
-- Layer 1 (this conversation) still lives on wa_flow_runs.context._memory; this is Layer 2.
-- ============================================================
create table if not exists public.lead_memory (
  id            bigserial primary key,
  phone         text not null,                 -- canonical lead phone = the memory anchor (survives place_id dupes)
  place_id      text,
  kind          text not null default 'fact',  -- 'fact' | 'summary'
  mkey          text,                          -- fact key (budget/needs/…); null for summaries
  body          text not null,
  run_id        uuid,
  flow_id       uuid,
  client_id     uuid,
  client_name   text,                          -- denormalized provenance label
  account_id    uuid,
  account_phone text,                          -- sender number (provenance)
  created_at    timestamptz default now()
);
create index if not exists lead_memory_phone_idx on public.lead_memory(phone, created_at desc);

-- ============================================================
-- §26 — LINKEDIN OUTREACH LAYER (M1, 2026-07-30). "Smartlead for LinkedIn" on the Unipile
-- transport (api-first managed LinkedIn API; hosted-auth/cookie connect, per-account sticky
-- proxy, new_relation + message webhooks). Mirrors the WhatsApp split: public li_* tables are
-- PostgREST-visible config/queue; the `li` schema (threads/messages/profiles) is deliberately
-- NOT exposed — dashboard reads via RPCs, the VPS engine (ultron-li/) writes via direct pg,
-- the Railway webhook receiver writes via the li_ingest_event RPC.
-- A campaign = client + accounts + recipient queue + a FIXED 4-touch sequence (the researched
-- LinkedIn-safe playbook): optional invite note → opener after accept → 2 follow-ups; a reply
-- stops the sequence. No flow graph in M1 (LinkedIn tolerates max ~4 touches; a rigid cadence
-- is the product-correct shape here, unlike WhatsApp).
-- ============================================================

-- Canonical LinkedIn person-URL slug: 'https://www.linkedin.com/in/Foo-Bar/?x=1' → 'foo-bar'.
-- Returns null for non-person URLs (company pages can't be invited).
create or replace function li_public_id(u text) returns text language sql immutable as $$
  select nullif(lower(regexp_replace(
           (regexp_match(coalesce(u,''), '/in/([^/?#]+)', 'i'))[1], '%[0-9A-Fa-f]{2}', '', 'g')), '')
$$;

-- One row per connected LinkedIn identity (Unipile "account"). user_id = the teammate who owns it.
create table if not exists public.li_accounts (
  id                 uuid primary key default gen_random_uuid(),
  user_id            uuid references app_users(id) on delete set null,
  unipile_account_id text unique,
  display_name       text,
  public_identifier  text,                          -- linkedin.com/in/<this>
  provider_id        text,                          -- own LinkedIn internal member id
  status             text not null default 'pending', -- pending|connected|credentials|disconnected|deleted
  premium            boolean not null default false,  -- paid LinkedIn → 300-char notes, higher caps
  daily_invite_cap   int not null default 15,         -- safe envelope: 15-25/day warmed
  weekly_invite_cap  int not null default 90,         -- LinkedIn's rolling ~100/week, with margin
  daily_msg_cap      int not null default 50,
  last_seen          timestamptz,
  last_error         text,
  deleted_at         timestamptz,
  created_at         timestamptz default now()
);
-- DIY (voyager) transport: cookie-based auth instead of Unipile. li_at is the session cookie,
-- csrf_token = the JSESSIONID value (ajax:…), both pasted from the user's own browser. own_provider_id
-- = the account's own member URN (captured on validate). Routed through the Evomi Saudi residential
-- proxy with a per-account sticky session. Additive so an existing DB upgrades cleanly.
alter table public.li_accounts add column if not exists li_at          text;
alter table public.li_accounts add column if not exists csrf_token     text;
alter table public.li_accounts add column if not exists user_agent     text;
alter table public.li_accounts add column if not exists own_provider_id text;
alter table public.li_accounts add column if not exists cookie_added_at timestamptz;

create schema if not exists li;

-- provider_id → lead resolution (the lead_phones of LinkedIn).
create table if not exists li.profiles (
  provider_id       text primary key,
  public_identifier text,
  place_id          text references leads(place_id) on delete set null,
  name              text,
  headline          text,
  updated_at        timestamptz default now()
);
create index if not exists li_profiles_pub_idx on li.profiles(public_identifier);

create table if not exists li.threads (
  owner_account_id uuid not null,
  peer_provider_id text not null,
  chat_id          text,                            -- Unipile chat id (once a chat exists)
  place_id         text,
  peer_name        text,
  last_msg_ts      timestamptz,
  last_inbound_ts  timestamptz,
  last_outbound_ts timestamptz,
  visibility       text not null default 'team',
  primary key (owner_account_id, peer_provider_id)
);

create table if not exists li.messages (
  id               bigserial primary key,
  owner_account_id uuid not null,
  peer_provider_id text not null,
  provider_msg_id  text,
  direction        text not null,                   -- in | out
  body             text,
  sent_at          timestamptz default now(),
  unique (owner_account_id, peer_provider_id, provider_msg_id)
);
create index if not exists li_messages_thread_idx on li.messages(owner_account_id, peer_provider_id, sent_at);

-- Raw webhook payload log — cheap insurance while the Unipile integration beds in.
create table if not exists li.webhook_log (
  id bigserial primary key, kind text, payload jsonb, created_at timestamptz default now()
);

-- Campaigns: client + accounts + queue + the fixed sequence + Smartlead-style schedule.
create table if not exists public.li_campaigns (
  id                uuid primary key default gen_random_uuid(),
  client_id         uuid references clients(id) on delete cascade,
  name              text not null,
  status            text not null default 'draft',  -- draft | running | paused | done
  invite_note       text,                            -- null/'' = BLANK invite (research default;
                                                     -- notes don't lift acceptance, only replies)
  opener            text,                            -- sent 24-48h-style after accept (required to start)
  followup1         text, followup1_days int not null default 4,
  followup2         text, followup2_days int not null default 6,
  withdraw_days     int not null default 21,         -- withdraw stale pending invites (hygiene)
  send_window_start text not null default '09:00',
  send_window_end   text not null default '17:00',
  timezone          text not null default 'Asia/Riyadh',
  active_days       int[] not null default '{0,1,2,3,4}',  -- Sun-Thu
  invites_per_day   int not null default 12,         -- per account, inside its own cap too
  created_by        uuid references app_users(id) on delete set null,
  created_at        timestamptz default now(),
  started_at        timestamptz
);
create index if not exists li_campaigns_client_idx on public.li_campaigns(client_id);

create table if not exists public.li_campaign_accounts (
  campaign_id uuid references public.li_campaigns(id) on delete cascade,
  account_id  uuid references public.li_accounts(id)  on delete cascade,
  primary key (campaign_id, account_id)
);

-- The queue. linkedin_url frozen at add-time; provider_id resolved lazily by the engine.
-- Status walk: queued → invited → accepted → messaged → replied | done (sequence exhausted)
--            | failed | skipped | withdrawn.  touches = sequence messages sent (0=none yet).
create table if not exists public.li_campaign_recipients (
  id                  bigserial primary key,
  campaign_id         uuid references public.li_campaigns(id) on delete cascade,
  place_id            text references leads(place_id) on delete cascade,
  linkedin_url        text not null,
  public_identifier   text,                          -- li_public_id(linkedin_url), frozen
  provider_id         text,
  status              text not null default 'queued',
  assigned_account_id uuid references public.li_accounts(id) on delete set null,
  invited_at          timestamptz,
  accepted_at         timestamptz,
  replied_at          timestamptz,
  touches             int not null default 0,
  next_action_at      timestamptz,                   -- when the next sequence touch is due
  claimed_at          timestamptz,
  attempts            int not null default 0,
  error               text,
  notified_at         timestamptz,                   -- reply Telegram-ping sent
  created_at          timestamptz default now(),
  unique (campaign_id, place_id)
);
create index if not exists li_recip_due_idx on public.li_campaign_recipients(campaign_id, status, next_action_at);
create index if not exists li_recip_acc_idx on public.li_campaign_recipients(assigned_account_id, invited_at);

-- Accounts visible to a user (own; admins all), with live usage counts for the caps UI.
create or replace function get_li_accounts(p_user uuid, p_all boolean default false)
returns json language sql stable as $$
  select coalesce(json_agg(row_to_json(x) order by x.created_at), '[]') from (
    select a.id, a.user_id, coalesce(u.name, u.email) user_name, a.unipile_account_id,
           a.display_name, a.public_identifier, a.status, a.premium,
           a.daily_invite_cap, a.weekly_invite_cap, a.daily_msg_cap, a.last_error,
           to_char(a.last_seen,'YYYY-MM-DD HH24:MI') last_seen, a.created_at,
           (select count(*) from public.li_campaign_recipients r
             where r.assigned_account_id = a.id and r.invited_at >= date_trunc('day', now() at time zone 'Asia/Riyadh') at time zone 'Asia/Riyadh')::int invites_today,
           (select count(*) from public.li_campaign_recipients r
             where r.assigned_account_id = a.id and r.invited_at >= now() - interval '7 days')::int invites_7d
    from public.li_accounts a
    left join app_users u on u.id = a.user_id
    where a.deleted_at is null and (p_all or a.user_id = p_user)
  ) x;
$$;

create or replace function get_li_campaigns(p_user uuid, p_admin boolean default false)
returns json language sql stable as $$
  select coalesce(json_agg(row_to_json(x) order by x.created_at desc), '[]') from (
    select c.id, c.name, c.status, c.client_id, cl.name client_name, cl.color client_color,
           c.invites_per_day, c.send_window_start, c.send_window_end,
           to_char(c.created_at,'YYYY-MM-DD HH24:MI') created_at,
           (select count(*) from public.li_campaign_recipients r where r.campaign_id = c.id)::int total,
           (select count(*) from public.li_campaign_recipients r where r.campaign_id = c.id and r.status='invited')::int invited,
           (select count(*) from public.li_campaign_recipients r where r.campaign_id = c.id and r.status in ('accepted','messaged'))::int accepted,
           (select count(*) from public.li_campaign_recipients r where r.campaign_id = c.id and r.status='replied')::int replied,
           (select count(*) from public.li_campaign_recipients r where r.campaign_id = c.id and r.status='queued')::int queued,
           (select count(*) from public.li_campaign_accounts n where n.campaign_id = c.id)::int accounts
    from public.li_campaigns c
    left join clients cl on cl.id = c.client_id
    where p_admin
       or exists (select 1 from user_clients uc where uc.client_id = c.client_id and uc.user_id = p_user)
  ) x;
$$;

create or replace function li_campaign_stats(p_id uuid)
returns json language sql stable as $$
  select coalesce(json_object_agg(status, cnt), '{}'::json) from (
    select status, count(*)::int cnt from public.li_campaign_recipients
    where campaign_id = p_id group by status
  ) s;
$$;

create or replace function get_li_campaign(p_id uuid)
returns json language sql stable as $$
  select json_build_object(
    'campaign', (select row_to_json(c) from (
        select c.*, cl.name client_name, cl.color client_color
        from public.li_campaigns c left join clients cl on cl.id = c.client_id
        where c.id = p_id) c),
    'accounts', (select coalesce(json_agg(row_to_json(a)),'[]') from (
        select a.id, a.display_name, a.public_identifier, a.status,
               coalesce(u.name, u.email) user_name
        from public.li_campaign_accounts n
        join public.li_accounts a on a.id = n.account_id
        left join app_users u on u.id = a.user_id
        where n.campaign_id = p_id order by a.created_at) a),
    'counts', li_campaign_stats(p_id),
    'recipients', (select coalesce(json_agg(row_to_json(r)),'[]') from (
        select r.id, r.place_id, r.linkedin_url, r.status, r.touches, r.error,
               l.company, l.full_name, l.job_title,
               to_char(r.invited_at,'YYYY-MM-DD HH24:MI') invited_at,
               to_char(r.accepted_at,'YYYY-MM-DD HH24:MI') accepted_at,
               to_char(r.replied_at,'YYYY-MM-DD HH24:MI') replied_at
        from public.li_campaign_recipients r
        left join leads l on l.place_id = r.place_id
        where r.campaign_id = p_id
        order by r.replied_at desc nulls last, r.invited_at desc nulls last, r.id limit 400) r)
  );
$$;

-- Queue recipients from place_ids. Only leads with a PERSON LinkedIn URL qualify. ONE row per
-- human profile per campaign (v3.3 lesson: dedupe on the identity key, not just place_id) and
-- never a slug already queued in ANY status on this campaign.
create or replace function li_campaign_add_recipients(p_campaign uuid, p_place_ids text[])
returns json language plpgsql as $$
declare added int; n_li int; n_total int;
begin
  n_total := coalesce(array_length(p_place_ids, 1), 0);
  with cand as (
    select distinct on (li_public_id(l.linkedin))
           l.place_id, l.linkedin, li_public_id(l.linkedin) slug
    from leads l
    where l.place_id = any(p_place_ids) and li_public_id(l.linkedin) is not null
    order by li_public_id(l.linkedin), l.place_id
  ), fresh as (
    select c.* from cand c
    where not exists (select 1 from public.li_campaign_recipients r
                       where r.campaign_id = p_campaign and r.public_identifier = c.slug)
  ), ins as (
    insert into public.li_campaign_recipients (campaign_id, place_id, linkedin_url, public_identifier, status)
    select p_campaign, place_id, linkedin, slug, 'queued' from fresh
    on conflict (campaign_id, place_id) do nothing
    returning 1
  )
  select (select count(*) from cand), (select count(*) from ins) into n_li, added;
  return json_build_object('added', added, 'skipped_no_linkedin', n_total - n_li);
end $$;

create or replace function li_campaign_add_from_list(p_campaign uuid, p_list uuid)
returns json language sql as $$
  select li_campaign_add_recipients(p_campaign,
    (select coalesce(array_agg(place_id), '{}') from public.lead_list_members where list_id = p_list));
$$;

-- Reply sweep (belt to the webhook): any inbound on the assigned account's thread with this
-- peer after our invite → replied (an inbound also implies the invite was accepted).
create or replace function li_campaign_mark_replies(p_campaign uuid default null)
returns json language plpgsql as $$
declare flipped int;
begin
  with upd as (
    update public.li_campaign_recipients r
       set status = 'replied', replied_at = coalesce(r.replied_at, t.last_inbound_ts),
           accepted_at = coalesce(r.accepted_at, t.last_inbound_ts), next_action_at = null
      from li.threads t
     where r.status in ('invited','accepted','messaged','done')
       and r.provider_id is not null and r.assigned_account_id is not null
       and t.owner_account_id = r.assigned_account_id and t.peer_provider_id = r.provider_id
       and t.last_inbound_ts is not null and t.last_inbound_ts > coalesce(r.invited_at, r.created_at)
       and (p_campaign is null or r.campaign_id = p_campaign)
    returning 1
  )
  select count(*) into flipped from upd;
  return json_build_object('replied', flipped);
end $$;

-- The weekly-report numbers: full funnel for one LinkedIn campaign.
create or replace function li_campaign_funnel(p_id uuid)
returns json language sql stable as $$
  select json_build_object(
    'queued',    (select count(*) from public.li_campaign_recipients where campaign_id=p_id and status='queued'),
    'invited',   (select count(*) from public.li_campaign_recipients where campaign_id=p_id and invited_at is not null),
    'accepted',  (select count(*) from public.li_campaign_recipients where campaign_id=p_id and accepted_at is not null),
    'replied',   (select count(*) from public.li_campaign_recipients where campaign_id=p_id and status='replied'),
    'done',      (select count(*) from public.li_campaign_recipients where campaign_id=p_id and status='done'),
    'failed',    (select count(*) from public.li_campaign_recipients where campaign_id=p_id and status='failed'),
    'withdrawn', (select count(*) from public.li_campaign_recipients where campaign_id=p_id and status='withdrawn'),
    'accept_rate', (select case when count(*) filter (where invited_at is not null) = 0 then null
        else round(100.0 * count(*) filter (where accepted_at is not null)
                   / count(*) filter (where invited_at is not null)) end
        from public.li_campaign_recipients where campaign_id=p_id),
    'daily', (select coalesce(json_agg(row_to_json(d) order by d.dt), '[]') from (
        select to_char(invited_at,'MM-DD') dt, count(*)::int invited,
               count(*) filter (where accepted_at is not null)::int accepted
        from public.li_campaign_recipients
        where campaign_id=p_id and invited_at > now() - interval '30 days'
        group by 1) d)
  );
$$;

-- Inbox: thread list across the accounts a user may see (own; admins/p_all everything).
create or replace function get_li_threads(
  p_user uuid, p_all boolean default false, p_account uuid default null,
  p_search text default '', lim int default 80)
returns json language sql stable as $$
  select coalesce(json_agg(row_to_json(x)), '[]') from (
    select t.owner_account_id, t.peer_provider_id, t.chat_id,
           coalesce(p.name, t.peer_name) peer_name, p.headline,
           coalesce(t.place_id, p.place_id) place_id, l.company, l.full_name,
           a.display_name account_name, a.public_identifier account_pub,
           to_char(t.last_msg_ts,'YYYY-MM-DD HH24:MI') last_msg_ts,
           (t.last_inbound_ts is not null and t.last_inbound_ts >= coalesce(t.last_outbound_ts,'-infinity')) awaiting_us,
           (select m.body from li.messages m
             where m.owner_account_id = t.owner_account_id and m.peer_provider_id = t.peer_provider_id
             order by m.sent_at desc limit 1) last_body
    from li.threads t
    join public.li_accounts a on a.id = t.owner_account_id and a.deleted_at is null
    left join li.profiles p on p.provider_id = t.peer_provider_id
    left join leads l on l.place_id = coalesce(t.place_id, p.place_id)
    where t.visibility = 'team'
      and (p_all or a.user_id = p_user)
      and (p_account is null or t.owner_account_id = p_account)
      and (coalesce(p_search,'') = '' or coalesce(p.name, t.peer_name) ilike '%'||p_search||'%'
           or l.company ilike '%'||p_search||'%' or l.full_name ilike '%'||p_search||'%')
    order by t.last_msg_ts desc nulls last limit lim
  ) x;
$$;

create or replace function get_li_conversation(p_account uuid, p_peer text, lim int default 300)
returns json language sql stable as $$
  select json_build_object(
    'thread', (select row_to_json(t) from (
        select t.owner_account_id, t.peer_provider_id, t.chat_id,
               coalesce(p.name, t.peer_name) peer_name, p.headline, p.public_identifier,
               coalesce(t.place_id, p.place_id) place_id
        from li.threads t left join li.profiles p on p.provider_id = t.peer_provider_id
        where t.owner_account_id = p_account and t.peer_provider_id = p_peer) t),
    'messages', (select coalesce(json_agg(row_to_json(m) order by m.sent_at), '[]') from (
        select m.direction, m.body, to_char(m.sent_at,'YYYY-MM-DD HH24:MI') sent_at
        from li.messages m
        where m.owner_account_id = p_account and m.peer_provider_id = p_peer
        order by m.sent_at desc limit lim) m)
  );
$$;

-- Outbound recorder (dashboard manual sends + the engine's sequence sends share this path).
create or replace function li_record_outbound(
  p_account uuid, p_peer text, p_chat text, p_msg_id text, p_body text,
  p_peer_name text default null)
returns void language plpgsql as $$
begin
  insert into li.threads as t (owner_account_id, peer_provider_id, chat_id, peer_name, last_msg_ts, last_outbound_ts)
  values (p_account, p_peer, p_chat, p_peer_name, now(), now())
  on conflict (owner_account_id, peer_provider_id) do update
    set chat_id = coalesce(excluded.chat_id, t.chat_id),
        peer_name = coalesce(t.peer_name, excluded.peer_name),
        last_msg_ts = now(), last_outbound_ts = now();
  insert into li.messages (owner_account_id, peer_provider_id, provider_msg_id, direction, body)
  values (p_account, p_peer, p_msg_id, 'out', p_body)
  on conflict do nothing;
end $$;

-- Webhook ingester — the Railway receiver normalizes Unipile payloads into these three kinds.
--  'account':  {unipile_account_id, status, name?, public_identifier?, error?}
--  'relation': {unipile_account_id, provider_id, public_identifier?, name?, headline?}  (invite accepted)
--  'message':  {unipile_account_id, provider_id, direction in|out, msg_id?, chat_id?, body?, peer_name?}
create or replace function li_ingest_event(p_kind text, p_payload jsonb)
returns json language plpgsql as $$
declare acc public.li_accounts; pid text; rep int := 0;
begin
  insert into li.webhook_log (kind, payload) values (p_kind, p_payload);

  if p_kind = 'account' then
    update public.li_accounts
       set status = coalesce(p_payload->>'status', status),
           display_name = coalesce(nullif(p_payload->>'name',''), display_name),
           public_identifier = coalesce(nullif(p_payload->>'public_identifier',''), public_identifier),
           last_error = p_payload->>'error', last_seen = now()
     where unipile_account_id = p_payload->>'unipile_account_id';
    return json_build_object('ok', true);
  end if;

  -- Resolve the account by our internal uuid (the voyager polling path passes account_id) OR by the
  -- Unipile account id (the webhook path). Either identifies the same li_accounts row.
  select * into acc from public.li_accounts
   where deleted_at is null
     and ( ((p_payload->>'account_id') is not null and id = (p_payload->>'account_id')::uuid)
        or ((p_payload->>'unipile_account_id') is not null and unipile_account_id = p_payload->>'unipile_account_id') );
  if acc.id is null then return json_build_object('ok', false, 'error', 'unknown account'); end if;
  pid := p_payload->>'provider_id';
  if pid is null or pid = '' then return json_build_object('ok', false, 'error', 'no provider_id'); end if;

  if p_kind = 'relation' then
    insert into li.profiles as p (provider_id, public_identifier, name, headline)
    values (pid, nullif(p_payload->>'public_identifier',''), nullif(p_payload->>'name',''), nullif(p_payload->>'headline',''))
    on conflict (provider_id) do update
      set public_identifier = coalesce(excluded.public_identifier, p.public_identifier),
          name = coalesce(excluded.name, p.name),
          headline = coalesce(excluded.headline, p.headline), updated_at = now();
    -- flip the matching invite(s) on this account: invited → accepted, opener due in 1-4h
    update public.li_campaign_recipients r
       set status = 'accepted', accepted_at = now(),
           provider_id = coalesce(r.provider_id, pid),
           next_action_at = now() + (interval '1 hour') * (1 + random() * 3)
     where r.assigned_account_id = acc.id and r.status = 'invited'
       and (r.provider_id = pid
            or (r.provider_id is null and r.public_identifier = lower(coalesce(p_payload->>'public_identifier',''))));
    -- attach the lead to the profile once known
    update li.profiles p set place_id = coalesce(p.place_id, r.place_id)
      from public.li_campaign_recipients r
     where p.provider_id = pid and r.provider_id = pid and r.assigned_account_id = acc.id;
    insert into li.threads as t (owner_account_id, peer_provider_id, peer_name, place_id)
    values (acc.id, pid, nullif(p_payload->>'name',''), null)
    on conflict (owner_account_id, peer_provider_id) do update
      set peer_name = coalesce(t.peer_name, excluded.peer_name);
    return json_build_object('ok', true);
  end if;

  if p_kind = 'message' then
    insert into li.threads as t (owner_account_id, peer_provider_id, chat_id, peer_name, place_id,
                                 last_msg_ts, last_inbound_ts, last_outbound_ts)
    values (acc.id, pid, nullif(p_payload->>'chat_id',''), nullif(p_payload->>'peer_name',''),
            (select place_id from li.profiles where provider_id = pid), now(),
            case when p_payload->>'direction' = 'in' then now() end,
            case when p_payload->>'direction' = 'out' then now() end)
    on conflict (owner_account_id, peer_provider_id) do update
      set chat_id = coalesce(excluded.chat_id, t.chat_id),
          peer_name = coalesce(t.peer_name, excluded.peer_name),
          place_id = coalesce(t.place_id, excluded.place_id),
          last_msg_ts = now(),
          last_inbound_ts = case when p_payload->>'direction' = 'in' then now() else t.last_inbound_ts end,
          last_outbound_ts = case when p_payload->>'direction' = 'out' then now() else t.last_outbound_ts end;
    insert into li.messages (owner_account_id, peer_provider_id, provider_msg_id, direction, body)
    values (acc.id, pid, nullif(p_payload->>'msg_id',''), coalesce(p_payload->>'direction','in'), p_payload->>'body')
    on conflict do nothing;
    if p_payload->>'direction' = 'in' then
      with upd as (
        update public.li_campaign_recipients r
           set status = 'replied', replied_at = coalesce(r.replied_at, now()),
               accepted_at = coalesce(r.accepted_at, now()), next_action_at = null
         where r.assigned_account_id = acc.id and r.provider_id = pid
           and r.status in ('invited','accepted','messaged','done')
        returning 1)
      select count(*) into rep from upd;
    end if;
    return json_build_object('ok', true, 'replied', rep);
  end if;

  return json_build_object('ok', false, 'error', 'unknown kind');
end $$;

-- PostgREST's service_role needs access to the `li` schema for the public RPCs that reference it
-- (get_li_threads/get_li_conversation READ; li_ingest_event/li_record_outbound WRITE via the webhook +
-- manual-send paths). Same discipline as the `wa` grants above — anon/authenticated stay locked out.
grant usage on schema li to service_role;
grant select, insert, update, delete on all tables in schema li to service_role;
grant usage, select on all sequences in schema li to service_role;
alter default privileges in schema li grant select, insert, update, delete on tables to service_role;
alter default privileges in schema li grant usage, select on sequences to service_role;


-- ============================================================
-- §27 — MAHALLY PLATFORM (2026-07-30). Salla's multi-vendor marketplace (mahally.com) as a
--     lead source: 44.7k Saudi online stores, each with a WhatsApp-able phone that Mahally's
--     own UI never renders. Scraper = src/scrape-mahally-stores.js.
--
--     PLATFORM BUCKET: nothing to add — get_sources()/search_leads bucket 'bigasscrm'+'WhatsApp'
--     together and pass ANY other source through under its own name, so source='Mahally'
--     becomes its own Platform automatically.
--
--     WHY DEDICATED COLUMNS (not JSONB): every one of these is a filter Ahmed wants on the
--     leads page, and the whole point is filtering. Columns index; JSONB paths don't, cheaply.
--     All are NULL for non-Mahally rows, which costs nothing in Postgres.
--
--     THE COMMERCIAL ANGLES these unlock (Ahmed's own framing, 2026-07-30):
--       payment_rails / has_installment → stores on NO installment, or only on the expensive
--         big two (tabby/tamara) and not the cheaper rails (mispay/madfu/emkan) = the BNPL pitch.
--       delivery_avg_hours              → slow fulfilment = the courier pitch.
--       trust_score / rating            → low score = the reputation-marketing pitch.
--       store_joined_at                 → just-opened store = needs everything.
--       gmv_tier                        → Salla's own revenue tier 1-5; tier 5 is ~242 products
--                                         in the entire 12.1M catalog. Can they afford us?
--       store_images                    → 5 sample product photos; judge photography quality.
--       store_hash                      → Salla CDN hash. A candidate domain that embeds this
--                                         hash IS this store (verified 2026-07-30 on
--                                         kawnroaster.com) — the key for the future email pass.
-- ============================================================
alter table leads add column if not exists company_en         text;
alter table leads add column if not exists main_activity      text;
alter table leads add column if not exists cr_number          text;
alter table leads add column if not exists store_id           bigint;
alter table leads add column if not exists store_hash         text;
alter table leads add column if not exists store_logo         text;
alter table leads add column if not exists store_images       text[];
alter table leads add column if not exists store_joined_at    timestamptz;
alter table leads add column if not exists trust_score        numeric;
alter table leads add column if not exists gmv_tier           smallint;
alter table leads add column if not exists product_count      integer;
alter table leads add column if not exists payment_rails      text[];
alter table leads add column if not exists delivery_avg_hours numeric;
alter table leads add column if not exists has_installment    boolean;
alter table leads add column if not exists has_express        boolean;
alter table leads add column if not exists made_in_ksa        boolean;
alter table leads add column if not exists is_charity         boolean;
alter table leads add column if not exists is_rising          boolean;
alter table leads add column if not exists verified_store     boolean;
alter table leads add column if not exists has_cr             boolean;
alter table leads add column if not exists has_address        boolean;
alter table leads add column if not exists has_return_policy  boolean;
alter table leads add column if not exists top_category       text;

create unique index if not exists leads_store_id_idx    on leads (store_id) where store_id is not null;
create index if not exists leads_store_hash_idx         on leads (store_hash) where store_hash is not null;
create index if not exists leads_gmv_tier_idx           on leads (gmv_tier) where gmv_tier is not null;
create index if not exists leads_trust_score_idx        on leads (trust_score) where trust_score is not null;
create index if not exists leads_store_joined_idx       on leads (store_joined_at) where store_joined_at is not null;
create index if not exists leads_product_count_idx      on leads (product_count) where product_count is not null;
create index if not exists leads_delivery_idx           on leads (delivery_avg_hours) where delivery_avg_hours is not null;
-- GIN so "uses tabby" / "uses any cheap rail" are indexed array-containment tests, not scans.
create index if not exists leads_payment_rails_idx      on leads using gin (payment_rails);

-- Delivery speed → the 4 bands the UI offers (avg_hours is a float; nobody filters on 110.55).
create or replace function delivery_band(h numeric) returns text language sql immutable as $$
  select case
    when h is null then null
    when h <= 24  then 'same/next day'
    when h <= 72  then '1-3 days'
    when h <= 168 then '3-7 days'
    else 'over a week' end;
$$;

-- Store age → bands. 'brand new' is the money bucket: opened inside 90 days.
create or replace function store_age_band(t timestamptz) returns text language sql immutable as $$
  select case
    when t is null then null
    when t > now() - interval '90 days'  then 'brand new (<3mo)'
    when t > now() - interval '1 year'   then 'under 1 year'
    when t > now() - interval '3 years'  then '1-3 years'
    else 'established (3y+)' end;
$$;

-- The cheap-BNPL opportunity flag, as a function so the UI and any query agree on the definition:
--   'none'      = sells with no installment option at all
--   'big2only'  = tabby/tamara only — paying the expensive cut, no cheap rail
--   'has cheap' = already on mispay/madfu/emkan
create or replace function bnpl_gap(rails text[], has_inst boolean) returns text language sql immutable as $$
  select case
    when rails is null or array_length(rails,1) is null then case when has_inst then null else 'none' end
    when not (rails && array['tabby_installment','tamara_installment','mispay_installment',
                             'madfu_installment','emkan_installment']) then 'none'
    when rails && array['mispay_installment','madfu_installment','emkan_installment'] then 'has cheap'
    else 'big2only' end;
$$;

-- Mahally-specific filter dropdowns. Mirrors get_filter_options' shape so the leads page can
-- swap option sources per platform. Counts gate out junk values the same way §13 does.
create or replace function get_mahally_filter_options()
returns json language sql stable as $$
  with m as (select * from leads where source = 'Mahally')
  select json_build_object(
    'categories',   (select coalesce(json_agg(v order by v), '[]') from (
                       select distinct trim(c) v from m, unnest(string_to_array(m.category, ',')) c
                       where coalesce(trim(c),'') <> '' group by trim(c) having count(*) >= 3) t),
    'activities',   (select coalesce(json_agg(v order by v), '[]') from (
                       select distinct main_activity v from m where coalesce(main_activity,'') <> '') t),
    'cities',       (select coalesce(json_agg(v order by v), '[]') from (
                       select city v from m where coalesce(city,'') <> '' group by city having count(*) >= 3) t),
    'top_categories',(select coalesce(json_agg(v order by v), '[]') from (
                       select distinct top_category v from m where coalesce(top_category,'') <> '') t),
    'payment_rails',(select coalesce(json_agg(v order by v), '[]') from (
                       select distinct r v from m, unnest(m.payment_rails) r) t),
    'gmv_tiers',    (select coalesce(json_agg(v order by v), '[]') from (
                       select distinct gmv_tier v from m where gmv_tier is not null) t),
    'delivery_bands',   json_build_array('same/next day','1-3 days','3-7 days','over a week'),
    'store_age_bands',  json_build_array('brand new (<3mo)','under 1 year','1-3 years','established (3y+)'),
    'bnpl_gaps',        json_build_array('none','big2only','has cheap'),
    'trust_bands',      json_build_array('under 3','3 to 4','4 to 4.5','4.5 to 5','5'),
    'product_bands',    json_build_array('1-10','11-50','51-200','201-1000','1000+'),
    'total',        (select count(*) from m),
    'with_phone',   (select count(*) from m where coalesce(phone,'') <> '')
  );
$$;

-- A ready-made "who should we pitch" view over the Mahally rows. Not used by the dashboard yet;
-- it's the shape the filter UI will query and a convenient handle for ad-hoc MCP/SQL questions.
create or replace view mahally_prospects as
  select l.place_id, l.store_id, l.company, l.company_en, l.phone, l.city, l.category,
         l.main_activity, l.top_category, l.rating, l.reviews, l.trust_score, l.gmv_tier,
         l.product_count, l.payment_rails, l.has_installment, l.has_express,
         l.delivery_avg_hours, delivery_band(l.delivery_avg_hours) as delivery,
         l.store_joined_at, store_age_band(l.store_joined_at)      as store_age,
         bnpl_gap(l.payment_rails, l.has_installment)              as bnpl,
         l.verified_store, l.has_cr, l.has_address, l.has_return_policy,
         l.made_in_ksa, l.is_charity, l.is_rising, l.cr_number, l.store_hash,
         l.store_logo, l.store_images, l.company_id, l.first_seen
  from leads l
  where l.source = 'Mahally';


-- ============================================================
-- §28 — MAHALLY FILTERS on search_leads (2026-07-30). Ahmed's ask: "mahally deserves better
--     detailed really long filtration buttons" — Maps gets 5 filters, Mahally gets 18.
--
--     SIGNATURE DISCIPLINE: `create or replace` with a different arg count makes an OVERLOAD, not a
--     replacement, and two overloads that both have defaults make an old-style call AMBIGUOUS. So the
--     previous signature is DROPPED explicitly first (same pattern as §7/§13 above). Both statements
--     must run in ONE transaction or the dashboard 404s in between.
--
--     Every existing predicate below is preserved byte-for-byte; the Mahally block is purely additive
--     and every new param defaults to 'all', so existing dashboard calls behave identically.
-- ============================================================

-- Band helpers — the UI dropdowns in get_mahally_filter_options() declare these exact labels, so
-- options and matching share one definition (same rule as norm_state/size_band/vol_band in §13).
create or replace function product_band(n integer) returns text language sql immutable as $$
  select case
    when n is null then null
    when n <= 10   then '1-10'
    when n <= 50   then '11-50'
    when n <= 200  then '51-200'
    when n <= 1000 then '201-1000'
    else '1000+' end;
$$;

create or replace function trust_band(t numeric) returns text language sql immutable as $$
  select case
    when t is null then null
    when t <  3    then 'under 3'
    when t <  4    then '3 to 4'
    when t <  4.5  then '4 to 4.5'
    when t <  5    then '4.5 to 5'
    else '5' end;
$$;

drop function if exists search_leads(text, text, text, text, text, text, text, text, text, text, text, text, text, text, text, text, text, uuid, boolean, int, int);

create or replace function search_leads(
  q text default '', p_category text default 'all', p_city text default 'all',
  p_fresh text default 'all', p_verified text default 'all', p_phone text default 'all',
  p_source text default 'all', p_industry text default 'all', p_seniority text default 'all',
  p_job_title text default 'all', p_department text default 'all',
  p_state text default 'all', p_size text default 'all',
  p_rating text default 'all', p_vol text default 'all', p_presence text default 'all',
  p_has_wa text default 'all',
  -- ---- Mahally (source='Mahally') filters; all default 'all' = no-op ----
  p_activity text default 'all',        -- main_activity
  p_top_cat text default 'all',         -- dominant product category
  p_gmv text default 'all',             -- MINIMUM Salla GMV/revenue tier (1-5)
  p_products text default 'all',        -- product_band()
  p_store_age text default 'all',       -- store_age_band(); 'brand new (<3mo)' is the money bucket
  p_trust text default 'all',           -- trust_band()
  p_delivery text default 'all',        -- delivery_band()
  p_bnpl text default 'all',            -- bnpl_gap(): 'none' | 'big2only' | 'has cheap'
  p_rail text default 'all',            -- a specific payment rail present in payment_rails[]
  p_express text default 'all',         -- all|yes|no
  p_ksa text default 'all',             -- made_in_ksa  all|yes|no
  p_store_verified text default 'all',  -- Mahally "موّثق" badge (NOT p_verified, which is email)
  p_has_cr text default 'all',          -- has commercial registration badge
  p_has_addr text default 'all',        -- has official address badge
  p_returns text default 'all',         -- has return/exchange policy
  p_charity text default 'all',         -- all|yes|no  (no = exclude charities from a sales list)
  p_rising text default 'all',          -- all|yes|no
  p_client uuid default null, p_assigned_only boolean default false,
  lim int default 100, off int default 0)
returns json language sql stable as $$
  with f as (
    select l.* from leads l
    where (q = '' or l.company ilike '%'||q||'%' or l.phone ilike '%'||q||'%'
           or l.email ilike '%'||q||'%' or l.category ilike '%'||q||'%' or l.city ilike '%'||q||'%'
           or l.full_name ilike '%'||q||'%' or l.job_title ilike '%'||q||'%'
           or l.company_en ilike '%'||q||'%')
      -- Mahally packs several tags into `category`, so fall back to a contains test for it.
      and (p_category = 'all' or l.category = p_category
           or (l.source = 'Mahally' and l.category ilike '%'||p_category||'%'))
      and (p_city = 'all' or l.city = p_city)
      -- platform / source filter compares against the PLATFORM GROUP (WhatsApp folds into bigasscrm)
      and (p_source = 'all'
           or (case when coalesce(nullif(l.source, ''), 'Google Maps') in ('bigasscrm', 'WhatsApp')
                    then 'bigasscrm'
                    else coalesce(nullif(l.source, ''), 'Google Maps') end) = p_source)
      -- industry via the same case/underscore fold the options use
      and (p_industry = 'all'
           or lower(replace(trim(l.industry), '_', ' ')) = lower(replace(trim(p_industry), '_', ' ')))
      -- seniority token-contains on the normalized string: "Manager" hits manager/Manager/"Senior, Manager"
      and (p_seniority = 'all'
           or lower(replace(l.seniority, '_', ' ')) like '%'||lower(replace(p_seniority, '_', ' '))||'%')
      -- job titles / departments compared case-insensitively (options are case-deduped)
      and (p_job_title = 'all' or lower(trim(l.job_title)) = lower(trim(p_job_title)))
      and (p_department = 'all' or lower(trim(l.department)) = lower(trim(p_department)))
      and (p_state = 'all' or norm_state(l.state) = p_state)
      and (p_size = 'all' or size_band(l.company_size) = p_size)
      -- Maps quality filters: minimum rating ('4.5'/'4.0'/'3.5') + review-volume band
      and (p_rating = 'all' or (l.rating is not null and l.rating >= p_rating::numeric))
      and (p_vol = 'all' or vol_band(l.reviews) = p_vol)
      -- web presence (works for both platforms)
      and (p_presence = 'all'
           or (p_presence = 'website' and coalesce(l.website, '') <> '')
           or (p_presence = 'no_website' and coalesce(l.website, '') = '')
           or (p_presence = 'instagram' and coalesce(l.instagram, '') <> '')
           or (p_presence = 'no_instagram' and coalesce(l.instagram, '') = ''))
      -- has WhatsApp conversation history (team-visible threads only)
      and (p_has_wa = 'all'
           or (p_has_wa = 'yes' and     exists (select 1 from wa.threads t where t.place_id = l.place_id and t.visibility = 'team'))
           or (p_has_wa = 'no'  and not exists (select 1 from wa.threads t where t.place_id = l.place_id and t.visibility = 'team')))
      and (p_fresh = 'all' or l.freshness = p_fresh)
      and (p_verified = 'all'
           or (p_verified = 'yes' and l.email_verified = true)
           or (p_verified = 'no' and l.email_verified is distinct from true))
      -- 'yes' = any phone · 'mobile' = Saudi mobile (05… / 9665…), i.e. WhatsApp-able
      and (p_phone = 'all'
           or (p_phone = 'yes' and l.phone is not null and l.phone <> '')
           or (p_phone = 'mobile' and regexp_replace(coalesce(l.phone,''),'[^0-9]','','g') ~ '^(0?5|9665)[0-9]{8}$'))
      -- ---- Mahally filters ----
      and (p_activity = 'all' or l.main_activity = p_activity)
      and (p_top_cat = 'all' or l.top_category = p_top_cat)
      and (p_gmv = 'all' or (l.gmv_tier is not null and l.gmv_tier >= p_gmv::smallint))
      and (p_products = 'all' or product_band(l.product_count) = p_products)
      and (p_store_age = 'all' or store_age_band(l.store_joined_at) = p_store_age)
      and (p_trust = 'all' or trust_band(l.trust_score) = p_trust)
      and (p_delivery = 'all' or delivery_band(l.delivery_avg_hours) = p_delivery)
      and (p_bnpl = 'all' or bnpl_gap(l.payment_rails, l.has_installment) = p_bnpl)
      and (p_rail = 'all' or l.payment_rails @> array[p_rail])
      and (p_express = 'all' or (p_express = 'yes' and l.has_express is true)
                             or (p_express = 'no'  and l.has_express is distinct from true))
      and (p_ksa = 'all' or (p_ksa = 'yes' and l.made_in_ksa is true)
                         or (p_ksa = 'no'  and l.made_in_ksa is distinct from true))
      and (p_store_verified = 'all' or (p_store_verified = 'yes' and l.verified_store is true)
                                    or (p_store_verified = 'no'  and l.verified_store is distinct from true))
      and (p_has_cr = 'all' or (p_has_cr = 'yes' and l.has_cr is true)
                            or (p_has_cr = 'no'  and l.has_cr is distinct from true))
      and (p_has_addr = 'all' or (p_has_addr = 'yes' and l.has_address is true)
                              or (p_has_addr = 'no'  and l.has_address is distinct from true))
      and (p_returns = 'all' or (p_returns = 'yes' and l.has_return_policy is true)
                             or (p_returns = 'no'  and l.has_return_policy is distinct from true))
      and (p_charity = 'all' or (p_charity = 'yes' and l.is_charity is true)
                             or (p_charity = 'no'  and l.is_charity is distinct from true))
      and (p_rising = 'all' or (p_rising = 'yes' and l.is_rising is true)
                            or (p_rising = 'no'  and l.is_rising is distinct from true))
      and (not p_assigned_only or p_client is null
           or exists(select 1 from lead_clients lc where lc.place_id = l.place_id and lc.client_id = p_client))
  )
  select json_build_object(
    'total', (select count(*) from f),
    'rows', (select coalesce(json_agg(r), '[]') from (
      select f.place_id, f.company, f.category, f.city, f.district, f.country, f.phone, f.email, f.email_status, f.email_verified,
             f.website, f.instagram, f.facebook, f.tiktok, f.snapchat, f.youtube, f.twitter, f.whatsapp,
             f.rating, f.reviews, f.review_tier, f.freshness, f.oldest_review, f.newest_review, f.address, f.lat, f.lng, f.maps_url,
             f.source, f.keyword, to_char(f.first_seen,'YYYY-MM-DD HH24:MI') ts,
             f.full_name, f.job_title, f.seniority, f.department, f.industry, f.state, f.linkedin,
             f.company_size, f.source_detail,
             -- Mahally columns (null for other platforms)
             f.company_en, f.main_activity, f.top_category, f.store_id, f.store_hash, f.store_logo,
             f.trust_score, f.gmv_tier, f.product_count, f.payment_rails, f.delivery_avg_hours,
             f.has_installment, f.has_express, f.made_in_ksa, f.is_charity, f.is_rising,
             f.verified_store, f.has_cr, f.has_address, f.has_return_policy, f.cr_number,
             to_char(f.store_joined_at,'YYYY-MM-DD') store_joined,
             store_age_band(f.store_joined_at) store_age,
             delivery_band(f.delivery_avg_hours) delivery,
             bnpl_gap(f.payment_rails, f.has_installment) bnpl,
             lc.channel last_contact_channel, lc.by_name last_contact_by,
             to_char(lc.contacted_at,'YYYY-MM-DD HH24:MI') last_contact_at
      from f
      left join lateral (
        select ct.channel, ct.contacted_at, coalesce(u.name, u.email, 'admin') by_name
        from contacts ct left join app_users u on u.id = ct.contacted_by
        where p_client is not null and ct.place_id = f.place_id and ct.client_id = p_client
        order by ct.contacted_at desc limit 1
      ) lc on true
      order by f.first_seen desc limit lim offset off) r)
  );
$$;

-- ============================================================
-- §27 — LOCAL CONNECTORS (2026-07-30): each teammate runs a small "Ultron Connector" app on their OWN
-- device; a number PAIRS + RUNS + SENDS from that device's own residential IP (near the phone) instead
-- of the shared VPS proxy — the ban-safe architecture (Ahmed's call after 966590725575 got banned right
-- after a cold campaign send from a Riyadh-ish proxy while the phone lives in Khobar). Ultron stays the
-- brain (CRM/flows/campaigns); the connector is the hands. SECURITY: the connector holds its Baileys
-- session LOCALLY and talks to Ultron over HTTPS with a per-connector TOKEN — NO database keys ever ship
-- to a teammate's laptop. Additive; existing VPS-hosted numbers (connector_id null) are untouched.
-- ============================================================
create table if not exists public.wa_connectors (
  id          uuid primary key default gen_random_uuid(),
  user_id     uuid references app_users(id) on delete cascade,
  name        text,                          -- device label ("Ahmed's MacBook")
  platform    text,                          -- darwin | win32 | linux
  token       text unique not null,          -- the secret the connector app authenticates with
  version     text,
  last_seen   timestamptz,
  created_at  timestamptz default now()
);
create index if not exists wa_connectors_user_idx on public.wa_connectors(user_id);

-- A number is served EITHER by the VPS (connector_id null → the existing proxy worker) OR by a local
-- connector (connector_id set → pairs/runs/sends on that device). One switch, no other code forks.
alter table public.wa_accounts add column if not exists connector_id uuid references public.wa_connectors(id) on delete set null;
create index if not exists wa_accounts_connector_idx on public.wa_accounts(connector_id);

-- The SEND QUEUE for local connectors. The pacer/flow/manual-send write here when an account is
-- connector-served; the owning connector polls, sends from its own IP, and reports the result.
create table if not exists public.wa_outbox (
  id            bigserial primary key,
  account_id    uuid not null references public.wa_accounts(id) on delete cascade,
  connector_id  uuid references public.wa_connectors(id) on delete cascade,
  peer_number   text,                         -- thread peer key (LID-safe)
  peer_phone    text,                         -- canonical phone (delivery target)
  is_group      boolean not null default false,
  body          text,
  media_key     text, media_type text,        -- crm-files key + document|image (connector signs its own URL)
  status        text not null default 'queued', -- queued | claimed | sent | failed
  claimed_at    timestamptz,
  wa_id         text, error text, attempts int not null default 0,
  created_at    timestamptz default now(), sent_at timestamptz
);
create index if not exists wa_outbox_due_idx on public.wa_outbox(connector_id, status, created_at);
create index if not exists wa_outbox_acct_idx on public.wa_outbox(account_id, sent_at);

-- Connectors a user can see (own; admins all) + the numbers each one serves + live/offline.
create or replace function get_wa_connectors(p_user uuid, p_admin boolean default false)
returns json language sql stable as $$
  select coalesce(json_agg(row_to_json(x) order by x.created_at), '[]') from (
    select c.id, c.name, c.platform, c.version, c.created_at,
           to_char(c.last_seen,'YYYY-MM-DD HH24:MI') last_seen,
           (c.last_seen is not null and c.last_seen > now() - interval '3 minutes') online,
           coalesce(u.name, u.email) user_name,
           (select coalesce(json_agg(json_build_object('id', a.id, 'phone', a.phone, 'label', a.label, 'status', a.status) order by a.created_at), '[]')
              from public.wa_accounts a where a.connector_id = c.id and a.deleted_at is null) accounts
    from public.wa_connectors c
    left join app_users u on u.id = c.user_id
    where p_admin or c.user_id = p_user
  ) x;
$$;

-- Store an inbound/echo message a connector reports (mirrors the VPS worker's thread-upsert + msg-insert,
-- and resolves the lead via lead_phones). SECURITY DEFINER so it can write the `wa` schema without granting
-- service_role blanket write on it. Called only via the token-authed /api/connector/inbound route.
create or replace function connector_ingest_message(p_account uuid, p_payload jsonb)
returns json language plpgsql security definer set search_path = public, wa as $$
declare
  v_peer  text := p_payload->>'peer_number';
  v_group boolean := coalesce((p_payload->>'is_group')::boolean, false);
  v_fromme boolean := coalesce((p_payload->>'from_me')::boolean, false);
  v_waid  text := nullif(p_payload->>'wa_id','');
  v_body  text := p_payload->>'body';
  v_push  text := nullif(p_payload->>'push_name','');
  v_ts    timestamptz := to_timestamp(coalesce((p_payload->>'ts')::double precision, extract(epoch from now())));
  v_canon text := case when v_group then null else norm_phone(v_peer) end;
  v_place text; v_owner text;
begin
  if v_peer is null or v_peer = '' then return json_build_object('ok', false); end if;
  if v_canon is not null then select place_id into v_place from public.lead_phones where phone = v_canon limit 1; end if;
  select phone into v_owner from public.wa_accounts where id = p_account;
  insert into wa.threads
    (owner_account_id, peer_number, owner_number, peer_phone, place_id, display_name, is_group,
     visibility, last_msg_ts, last_inbound_ts, last_outbound_ts, msg_count)
  values (p_account, v_peer, v_owner, v_canon, v_place, case when v_fromme then null else v_push end, v_group,
     'team', v_ts, case when v_fromme then null else v_ts end, case when v_fromme then v_ts else null end, 0)
  on conflict (owner_account_id, peer_number) do update set
    owner_number     = coalesce(wa.threads.owner_number, excluded.owner_number),
    peer_phone       = coalesce(wa.threads.peer_phone, excluded.peer_phone),
    place_id         = coalesce(wa.threads.place_id, excluded.place_id),
    display_name     = coalesce(excluded.display_name, wa.threads.display_name),
    last_msg_ts      = greatest(wa.threads.last_msg_ts, excluded.last_msg_ts),
    last_inbound_ts  = greatest(wa.threads.last_inbound_ts, excluded.last_inbound_ts),
    last_outbound_ts = greatest(wa.threads.last_outbound_ts, excluded.last_outbound_ts);
  if v_waid is not null then
    insert into wa.messages (owner_account_id, peer_number, wa_id, from_me, body, media_type, media_hash, quoted_text, ts)
    values (p_account, v_peer, v_waid, v_fromme, v_body, null, null, null, v_ts)
    on conflict (owner_account_id, peer_number, wa_id) do nothing;
  end if;
  return json_build_object('ok', true, 'place_id', v_place);
end $$;

-- ============================================================
-- ALWAYS LAST: force PostgREST to reload its schema cache after every db-setup run.
-- Without this, dropped/replaced RPC signatures go stale and the dashboard 404s
-- (bit us 2026-07-16 — see CRM-PLAN.md Phase 4 postmortem).
-- ============================================================
notify pgrst, 'reload schema';
