Database

Create a Supabase project, write the migration, push it with the Supabase CLI, and save every event and message.

By the end of this page every email the poller sees is saved in Supabase as an event, a ticket, and a message. A unique Gmail message ID makes sure nothing is saved or answered twice.

Create the Supabase project

Log in to Supabase and click New project. The free plan is enough.
Name it ai-customer-support. Generate a database password and save it in your password manager.
Pick the region closest to your server.
Click Create new project and wait until it is ready.

Free projects pause after 7 days without activity. A running poller keeps it active.

Install the Supabase CLI

brew install supabase/tap/supabase

Link the project

supabase login
supabase init
supabase link --project-ref <project-ref>

The project ref is the part of the dashboard URL after /project/. link asks for the database password.

Write the migration

supabase migration new create_tables

This creates an empty file in supabase/migrations/. Paste this SQL into it:

supabase/migrations/20261028100000_create_tables.sql

create table events (
  id bigint generated always as identity primary key,
  gmail_message_id text not null unique,
  status text not null default 'processing'
    check (status in ('processing', 'processed', 'skipped', 'failed')),
  attempts int not null default 1,
  error text,
  result jsonb,
  created_at timestamptz not null default now(),
  processed_at timestamptz
);

create table tickets (
  id bigint generated always as identity primary key,
  gmail_thread_id text not null unique,
  customer_email text not null,
  subject text not null default '',
  category text,
  status text not null default 'open'
    check (status in ('open', 'answered', 'escalated')),
  proposed_action text,
  created_at timestamptz not null default now(),
  updated_at timestamptz not null default now()
);

create table messages (
  id bigint generated always as identity primary key,
  ticket_id bigint not null references tickets (id) on delete cascade,
  gmail_message_id text not null unique,
  direction text not null check (direction in ('in', 'out')),
  from_address text not null,
  body text not null,
  created_at timestamptz not null default now()
);

create index messages_ticket_id_idx on messages (ticket_id);

create table knowledge_base (
  id bigint generated always as identity primary key,
  section text not null,
  title text not null,
  content text not null,
  updated_at timestamptz not null default now(),
  unique (section, title)
);

alter table events enable row level security;
alter table tickets enable row level security;
alter table messages enable row level security;
alter table knowledge_base enable row level security;

Your file name has a different timestamp. The knowledge base stores one row per entry instead of one column per section, so a new section is new data, not a new migration.

Push it

supabase db push

RLS is on and there are no policies, so the public Data API cannot read these tables. The backend connects as the postgres user, which bypasses RLS. The dashboard never talks to Supabase directly.

Add the connection string

In the Supabase dashboard, click Connect and copy the Session pooler string. Replace [YOUR-PASSWORD] with your database password, and set it as DATABASE_URL in .env.

.env

DATABASE_URL=postgresql://postgres.<project-ref>:<password>@aws-0-<region>.pooler.supabase.com:5432/postgres

Use the session pooler on port 5432, not the direct connection. The direct host only has an IPv6 address, and many servers and home networks only have IPv4.

If your password contains characters such as @, #, / or %, the URL breaks. Generate a password with only letters and numbers, or URL-encode those characters.

Write the database helpers

app/db.py

import psycopg
from psycopg.rows import dict_row
from psycopg.types.json import Jsonb

from app.config import settings


def connect() -> psycopg.Connection:
    return psycopg.connect(settings.database_url, autocommit=True, row_factory=dict_row)


def start_event(gmail_message_id: str) -> dict | None:
    """Claim a message. Returns None when it is already handled or in progress."""
    with connect() as conn:
        return conn.execute(
            """
            insert into events (gmail_message_id) values (%s)
            on conflict (gmail_message_id) do update
              set attempts = events.attempts + 1, status = 'processing', error = null
              where events.status = 'failed'
            returning id, attempts
            """,
            (gmail_message_id,),
        ).fetchone()


def finish_event(event_id: int, status: str, result: dict) -> None:
    with connect() as conn:
        conn.execute(
            "update events set status = %s, result = %s, processed_at = now() where id = %s",
            (status, Jsonb(result), event_id),
        )


def fail_event(event_id: int, error: str) -> None:
    with connect() as conn:
        conn.execute(
            "update events set status = 'failed', error = %s where id = %s",
            (error, event_id),
        )


def upsert_ticket(gmail_thread_id: str, customer_email: str, subject: str) -> int:
    with connect() as conn:
        row = conn.execute(
            """
            insert into tickets (gmail_thread_id, customer_email, subject)
            values (%s, %s, %s)
            on conflict (gmail_thread_id) do update set updated_at = now()
            returning id
            """,
            (gmail_thread_id, customer_email, subject),
        ).fetchone()
        return row["id"]


def update_ticket(
    ticket_id: int, status: str, category: str | None = None, proposed_action: str | None = None
) -> None:
    with connect() as conn:
        conn.execute(
            """
            update tickets
            set status = %s, category = coalesce(%s, category),
                proposed_action = %s, updated_at = now()
            where id = %s
            """,
            (status, category, proposed_action, ticket_id),
        )


def save_message(
    ticket_id: int, gmail_message_id: str, direction: str, from_address: str, body: str
) -> None:
    with connect() as conn:
        conn.execute(
            """
            insert into messages (ticket_id, gmail_message_id, direction, from_address, body)
            values (%s, %s, %s, %s, %s)
            on conflict (gmail_message_id) do nothing
            """,
            (ticket_id, gmail_message_id, direction, from_address, body),
        )

start_event is the duplicate check. The insert only succeeds for a new message ID, or for one whose last try failed. Anything else returns None, and the pipeline skips it.

Save every message

Replace app/pipeline.py. It now claims the event, saves the ticket and the message, and then labels the email. The highlighted lines changed.

app/pipeline.py

import logging
from email.utils import parseaddr

from app import db
from app.gmail import PROCESSED_LABEL, Gmail, get_body, get_headers

logger = logging.getLogger(__name__)
MAX_ATTEMPTS = 3


def handle_message(gmail: Gmail, message_id: str) -> None:
    event = db.start_event(message_id)
    if event is None:
        gmail.add_label(message_id, PROCESSED_LABEL)
        return
    try:
        status, result = process(gmail, message_id)
        gmail.add_label(message_id, PROCESSED_LABEL)
        db.finish_event(event["id"], status, result)
        logger.info("%s %s %s", message_id, status, result)
    except Exception as error:
        db.fail_event(event["id"], str(error))
        if event["attempts"] >= MAX_ATTEMPTS:
            # Give up, so the message does not loop forever.
            gmail.add_label(message_id, PROCESSED_LABEL)
        raise


def process(gmail: Gmail, message_id: str) -> tuple[str, dict]:
    message = gmail.get_message(message_id)
    headers = get_headers(message)
    sender = parseaddr(headers.get("from", ""))[1]

    ticket_id = db.upsert_ticket(message["threadId"], sender, headers.get("subject", ""))
    db.save_message(ticket_id, message_id, "in", sender, get_body(message))
    return "processed", {"action": "saved"}

Check

Start the poller and send an email to the demo address.

uv run python -m app.poller

In the Supabase Table Editor, events has a row with status processed, and tickets and messages each have one row for your email. Then reply to that email from your personal inbox. The same ticket gets a second message, because both share the Gmail thread ID.

Database

On this page