Supabase + n8n: Build a Serverless Lead Database in 1 Hour

Sources:
  Retell AI  ──────────────────────────────────────────┐
  Typeform   ─────────────────────────── n8n Webhook   ├──► Supabase Postgres
  Meta Ads   ──────────────────────────── (normalize,  │
  Web Form   ─────────────────────────── dedup, upsert)┘
  Any webhook ──────────────────────────────────────────┘

-- Enable UUID generation
create extension if not exists "uuid-ossp";

create table if not exists leads (
  id              uuid primary key default uuid_generate_v4(),

  -- Contact identity
  first_name      text,
  last_name       text,
  email           text,
  phone           text,

  -- Source metadata
  source          text not null,        -- 'retell' | 'typeform' | 'meta_ads' | 'web_form'
  source_id       text,                 -- original ID from the source system
  campaign        text,                 -- ad campaign name if applicable

  -- Lead quality
  status          text default 'new',   -- 'new' | 'contacted' | 'qualified' | 'lost' | 'won'
  score           int,                  -- 0-10 AI score (set by n8n + Claude)
  qualified       boolean default false,

  -- Voice call specifics (Retell)
  call_duration_s int,                  -- call length in seconds
  call_transcript text,                 -- truncated transcript
  call_recording  text,                 -- recording URL

  -- Deduplication tracking
  touch_count     int default 1,        -- incremented on each upsert
  first_seen_at   timestamptz default now(),
  last_seen_at    timestamptz default now(),

  -- Metadata
  raw_payload     jsonb,                -- full original webhook payload
  notes           text,
  created_at      timestamptz default now(),
  updated_at      timestamptz default now()
);

-- Indexes for fast dedup lookups
create index if not exists leads_email_idx on leads (email) where email is not null;
create index if not exists leads_phone_idx on leads (phone) where phone is not null;
create index if not exists leads_source_idx on leads (source);
create index if not exists leads_status_idx on leads (status);
create index if not exists leads_created_at_idx on leads (created_at desc);

-- Auto-update updated_at
create or replace function update_updated_at_column()
returns trigger as $$
begin
  new.updated_at = now();
  return new;
end;
$$ language plpgsql;

create trigger update_leads_updated_at
  before update on leads
  for each row execute function update_updated_at_column();

-- Enable RLS (locks table to all roles by default)
alter table leads enable row level security;

-- service_role bypasses RLS automatically in Supabase
-- Use it exclusively in n8n server-side requests

-- Optional: allow authenticated dashboard users to read (not write)
create policy "dashboard_read_leads"
  on leads for select
  using (auth.role() = 'authenticated');

-- For the n8n service account, use service_role key only
-- Never use anon key for server-side operations

// n8n Function node: Normalize lead from any source
const body = $input.first().json.body || $input.first().json;
const source = body.source || body._source || 'unknown';

let lead = {
  source,
  raw_payload: body,
  status: 'new',
  first_seen_at: new Date().toISOString(),
  last_seen_at: new Date().toISOString(),
};

if (source === 'retell') {
  const call = body.call || body;
  lead.phone = call.from_number?.replace(/\D/g, '').replace(/^1/, '') || null;
  lead.first_name = call.extracted_data?.first_name || null;
  lead.last_name  = call.extracted_data?.last_name  || null;
  lead.email      = call.extracted_data?.email       || null;
  lead.call_duration_s = call.call_duration_s || null;
  lead.call_transcript = (call.transcript || '').slice(0, 2000);
  lead.call_recording  = call.recording_url  || null;
  lead.qualified  = call.extracted_data?.qualified === true;
  lead.campaign   = call.metadata?.campaign  || null;

} else if (source === 'typeform') {
  const answers = body.form_response?.answers || [];
  const getField = (ref) => answers.find(a => a.field?.ref === ref)?.text || null;
  lead.first_name = getField('first_name');
  lead.last_name  = getField('last_name');
  lead.email = getField('email') || answers.find(a => a.type === 'email')?.email || null;
  lead.phone = getField('phone') || answers.find(a => a.type === 'phone_number')?.phone_number || null;
  lead.source_id  = body.form_response?.token || null;
  lead.campaign   = body.form_response?.hidden?.campaign || null;

} else if (source === 'meta_ads') {
  const data = body.entry?.[0]?.changes?.[0]?.value || body;
  const fields = {};
  (data.field_data || []).forEach(f => { fields[f.name] = f.values?.[0] || null; });
  lead.first_name = fields.first_name || fields.full_name?.split(' ')[0] || null;
  lead.last_name  = fields.last_name  || fields.full_name?.split(' ')[1] || null;
  lead.email      = fields.email       || null;
  lead.phone      = fields.phone_number?.replace(/\D/g, '') || null;
  lead.source_id  = data.leadgen_id    || null;
  lead.campaign   = data.campaign_name || null;

} else if (source === 'web_form') {
  lead.first_name = body.first_name || body.name?.split(' ')[0] || null;
  lead.last_name  = body.last_name  || body.name?.split(' ')[1] || null;
  lead.email      = body.email       || null;
  lead.phone      = body.phone?.replace(/\D/g, '') || null;
  lead.campaign   = body.campaign    || body.utm_campaign || null;
  lead.source_id  = body.form_id     || null;
}

// Normalize phone: strip to 10 digits (US)
if (lead.phone) {
  lead.phone = lead.phone.replace(/^1/, '').slice(-10);
}

return [{ json: lead }];

// HTTP Request node config for dedup check
Method: GET
URL: https://YOUR_PROJECT.supabase.co/rest/v1/leads

// Query parameters:
select: id,touch_count
or: (email.eq.{{ $json.email }},phone.eq.{{ $json.phone }})
limit: 1

// Headers:
apikey: YOUR_SERVICE_ROLE_KEY
Authorization: Bearer YOUR_SERVICE_ROLE_KEY
Content-Type: application/json

// HTTP Request node: INSERT
Method: POST
URL: https://YOUR_PROJECT.supabase.co/rest/v1/leads

// Headers:
apikey: YOUR_SERVICE_ROLE_KEY
Authorization: Bearer YOUR_SERVICE_ROLE_KEY
Content-Type: application/json
Prefer: return=representation

// Body (JSON):
{
  "first_name":      "{{ $node['Normalize'].json.first_name }}",
  "last_name":       "{{ $node['Normalize'].json.last_name }}",
  "email":           "{{ $node['Normalize'].json.email }}",
  "phone":           "{{ $node['Normalize'].json.phone }}",
  "source":          "{{ $node['Normalize'].json.source }}",
  "source_id":       "{{ $node['Normalize'].json.source_id }}",
  "campaign":        "{{ $node['Normalize'].json.campaign }}",
  "qualified":       {{ $node['Normalize'].json.qualified }},
  "call_duration_s": {{ $node['Normalize'].json.call_duration_s ?? 'null' }},
  "call_transcript": "{{ $node['Normalize'].json.call_transcript }}",
  "raw_payload":     {{ JSON.stringify($node['Normalize'].json.raw_payload) }},
  "status":          "new"
}

// HTTP Request node: UPDATE (duplicate lead — increment touch_count)
Method: PATCH
URL: https://YOUR_PROJECT.supabase.co/rest/v1/leads?id=eq.{{ $node['Dedup Check'].json[0].id }}

// Headers: same as INSERT

// Body:
{
  "last_seen_at": "{{ new Date().toISOString() }}",
  "touch_count":  {{ ($node['Dedup Check'].json[0].touch_count || 1) + 1 }},
  "source":       "{{ $node['Normalize'].json.source }}",
  "campaign":     "{{ $node['Normalize'].json.campaign }}"
}

// Retell call_analyzed webhook payload (abbreviated)
{
  "event": "call_analyzed",
  "call": {
    "call_id": "call_abc123",
    "from_number": "+14695551234",
    "to_number": "+14695559999",
    "call_duration_s": 187,
    "recording_url": "https://...",
    "transcript": "Agent: Hi, this is...\nUser: Yeah I need a roof...",
    "extracted_data": {
      // Fields from your Retell agent's extraction config
      "first_name": "John",
      "last_name": "Martinez",
      "email": "john@example.com",
      "service_needed": "roof replacement",
      "qualified": true,
      "urgency": "high"
    },
    "metadata": {
      "campaign": "storm-damage-2026"
    }
  },
  "source": "retell"
}

-- Add unique constraints for upsert (partial — NULLs don't conflict)
create unique index leads_email_unique
  on leads (email) where email is not null;

create unique index leads_phone_unique
  on leads (phone) where phone is not null;

-- In your Supabase REST POST request, add:
-- Header:  Prefer: resolution=merge-duplicates
-- Param:   on_conflict=email
--
-- This performs: INSERT ... ON CONFLICT (email) DO UPDATE
-- Supabase handles it automatically

// n8n Function node: Build Telegram alert for qualified leads
const lead = $node['Normalize'].json;
const isNew = $node['Dedup Check'].json.length === 0;

if (!lead.qualified && (lead.score || 0) < 7) {
  return []; // Skip low-quality leads
}

const name = [lead.first_name, lead.last_name].filter(Boolean).join(' ') || 'Unknown';
const contact = [lead.email, lead.phone].filter(Boolean).join(' / ') || 'No contact';
const tag = isNew ? '🆕 New Lead' : '🔁 Return Lead';

const text = [
  tag,
  '👤 ' + name,
  '📞 ' + contact,
  '📍 Source: ' + lead.source,
  '🎯 Campaign: ' + (lead.campaign || 'none'),
  '✅ Qualified: ' + (lead.qualified ? 'Yes' : 'No'),
  '⏱ Call: ' + (lead.call_duration_s ? lead.call_duration_s + 's' : 'N/A'),
].join('\n');

return [{ json: { text } }];

// Follow with Telegram node:
// Chat ID: your Telegram chat ID
// Text: {{ $json.text }}

// HTTP Request node: Claude Haiku lead scoring
Method: POST
URL: https://api.anthropic.com/v1/messages

// Headers:
x-api-key: YOUR_ANTHROPIC_API_KEY
anthropic-version: 2023-06-01
Content-Type: application/json

// Body:
{
  "model": "claude-haiku-4-5-20251001",
  "max_tokens": 100,
  "messages": [{
    "role": "user",
    "content": "Score this lead 0-10 for a roofing company. Return only JSON: {"score": N, "reason": "one sentence"}\n\nName={{ $json.first_name }} {{ $json.last_name }}, Source={{ $json.source }}, Qualified={{ $json.qualified }}, Call={{ $json.call_duration_s }}s, Campaign={{ $json.campaign }}"
  }]
}

// Parse: JSON.parse($json.content[0].text).score
// Add to lead object before Supabase insert

-- This week's qualified leads by source
select source, count(*) as count, round(avg(score), 1) as avg_score
from leads
where qualified = true
  and created_at > date_trunc('week', now())
group by source order by count desc;

-- High-value leads not yet contacted
select first_name, last_name, email, phone, source, score, campaign
from leads
where score >= 7 and status = 'new'
  and created_at > now() - interval '7 days'
order by score desc, created_at desc;

-- Return lead audit (touched more than once)
select email, phone, touch_count,
       array_agg(distinct source) as sources,
       max(last_seen_at) as last_seen
from leads
where touch_count > 1
group by email, phone, touch_count
order by touch_count desc;

-- Monthly lead volume by campaign
select date_trunc('month', created_at) as month,
       campaign, source, count(*) as leads
from leads where created_at > now() - interval '6 months'
group by 1, 2, 3 order by 1 desc, 4 desc;