PLAN_CRM_EMAIL_THREADS.md64.4 KBView on GitHub # Plan: `crm_email_threads` — Thread-Level Postgres Table
## Current State
### What's missing
`crm_email_events` is per-message. There is no thread-level row in PostgreSQL. This causes three compounding problems:
**1. `listThreads` always calls the Gmail API**
`listThreadsViaProvider` (`aws-primary-mail-runtime.ts:552`) unconditionally calls `manager.list()` on every inbox load to get folder membership and sort order. The old Cloudflare Durable Object served this from local SQLite with zero external calls when nothing had changed. That capability is gone.
**2. Thread metadata is re-aggregated from scratch on every request**
`getThreadPreviewSummaries` (`aws-primary-mail-runtime.ts:316`) issues one SQL query that returns ALL message rows for ALL threads on the current page, then runs a JavaScript loop to aggregate subject, snippet, sender, labels, messageCount, hasDraft, and participants. For a 50-thread page with 5 messages each, that's ~250 rows fetched and processed in memory every single time.
**3. Labels are inaccurate**
Labels are stored per-message in `crm_email_events.labels`. The aggregation takes labels only from the **newest non-draft message**. A thread where only older messages are unread will appear as read. Thread-level label state (STARRED, IMPORTANT, UNREAD applied to the thread) is never stored as a unit.
### Write path today (what must be kept in sync)
| Trigger | Function | File | Storage written |
|---|---|---|---|
| Thread sync (new message) | `findOrCreateEmailEvents()` | `services/crm/email-events.ts:370` | `crm_email_events`, `crm_events` |
| Thread sync (full) | `syncThreadViaProvider()` | `aws-primary-mail-runtime.ts:644` | S3 snapshot |
| Label change (user action) | `patchStoredThreadLabels()` | `aws-primary-mail-runtime.ts:433` | S3 snapshot only |
| Label change (Gmail push) | `agent.modifyThreadLabelsInDB()` in pipeline | `pipelines.ts:1189` | S3 snapshot only |
| Sync workflow (background) | `syncThreads()` via `SyncThreadsWorkflow` | `workflows/sync-threads-workflow.ts:61` | S3 + `crm_email_events` |
### Read path today
| Consumer | Reads from | Round-trips |
|---|---|---|
| `listThreads` (no query) | Gmail API (thread IDs) + `crm_email_events` (metadata) | 1 Gmail + 1 DB |
| `listThreads` (search) | Gmail API + `crm_conversations` → `crm_email_events` | 1 Gmail + 2 DB |
| `mail.get` | S3 (full blob) + `crm_events` (conversationId) | 1 S3 + 1 DB |
| Thread metadata | S3 snapshot | 1 S3 |
### Critical files
| File | Role |
|---|---|
| `apps/server/src/lib/aws-primary-mail-runtime.ts` | Core runtime: list, sync, label patch, snapshot load |
| `apps/server/src/services/crm/email-events.ts` | Write path: inserts `crm_email_events` rows per message |
| `apps/server/src/pipelines.ts` | Gmail push notification handler: label changes |
| `apps/server/src/workflows/sync-threads-workflow.ts` | Background sync workflow |
| `apps/server/src/db/crm-schema.ts` | Schema file to add the new table |
| `apps/server/src/db/migrations/` | Migration directory (next: `0075_add_crm_email_threads.sql`) |
| `apps/server/src/services/mail/list-threads-read-model.ts` | Read model entry point |
---
## Proposed Changes
### New table: `crm_email_threads`
One row per thread per user. Updated on every sync and every label change. Enables pure-DB `listThreads`.
```typescript
// apps/server/src/db/crm-schema.ts
export const crmEmailThreads = pgTable(
'crm_email_threads',
{
// Identity
threadId: text('thread_id').notNull(), // Gmail thread ID
connectionId: text('connection_id').notNull() // FK → connection.id
.references(() => connection.id, { onDelete: 'cascade' }),
userId: text('user_id').notNull() // FK → user.id
.references(() => user.id, { onDelete: 'cascade' }),
// Thread-level label state (union across all messages)
labels: text('labels').array().notNull().default([]),
hasUnread: boolean('has_unread').notNull().default(false),
isStarred: boolean('is_starred').notNull().default(false),
// Display metadata (from newest non-draft message)
subject: text('subject').notNull().default(''),
snippet: text('snippet'),
fromEmail: text('from_email').notNull().default(''),
fromName: text('from_name'),
latestMessageAt: timestamp('latest_message_at').notNull(),
// Counts
messageCount: integer('message_count').notNull().default(0),
hasDraft: boolean('has_draft').notNull().default(false),
// CRM link
conversationId: uuid('conversation_id')
.references(() => crmConversations.id, { onDelete: 'set null' }),
// Sync state
historyId: text('history_id'), // Gmail historyId at time of last sync
createdAt: timestamp('created_at').notNull().defaultNow(),
updatedAt: timestamp('updated_at').notNull().defaultNow(),
},
(table) => [
// Primary key: one row per (thread, connection) — a thread can exist across
// multiple connections if user has multiple Gmail accounts
primaryKey({ columns: [table.threadId, table.connectionId] }),
index('idx_crm_email_threads_user_id').on(table.userId),
index('idx_crm_email_threads_connection_id').on(table.connectionId),
index('idx_crm_email_threads_conversation_id').on(table.conversationId),
// Primary query pattern: inbox sorted by recency
index('idx_crm_email_threads_user_latest').on(
table.userId,
table.latestMessageAt.desc(),
),
// Label filter (WHERE 'INBOX' = ANY(labels))
index('idx_crm_email_threads_labels').using('gin', table.labels),
// Unread filter
index('idx_crm_email_threads_unread').on(table.userId, table.hasUnread),
],
);
```
### New helper: `upsertEmailThread()`
A single function that computes and writes the thread row. Called from every write path.
```typescript
// apps/server/src/services/crm/email-threads.ts (new file)
export async function upsertEmailThread(
db: DB,
params: {
threadId: string;
connectionId: string;
userId: string;
messages: ParsedMessage[]; // all messages in thread, including drafts
conversationId?: string | null;
historyId?: string | null;
},
): Promise<void> {
const { threadId, connectionId, userId, messages, conversationId, historyId } = params;
// Union all labels across all messages — this is the thread-level label set
const allLabels = new Set<string>();
for (const msg of messages) {
for (const tag of msg.tags ?? []) {
allLabels.add(tag.name);
}
}
// Newest non-draft for display fields
const nonDrafts = messages.filter((m) => !m.isDraft);
const latest = nonDrafts.at(-1); // last element = most recent (messages are oldest-first)
await db
.insert(crmEmailThreads)
.values({
threadId,
connectionId,
userId,
labels: Array.from(allLabels),
hasUnread: allLabels.has('UNREAD'),
isStarred: allLabels.has('STARRED'),
subject: latest?.subject ?? '',
snippet: latest?.snippet ?? undefined,
fromEmail: latest?.sender.email ?? '',
fromName: latest?.sender.name ?? undefined,
latestMessageAt: latest ? new Date(latest.receivedOn) : new Date(),
messageCount: nonDrafts.length,
hasDraft: messages.some((m) => m.isDraft),
conversationId: conversationId ?? null,
historyId: historyId ?? null,
})
.onConflictDoUpdate({
target: [crmEmailThreads.threadId, crmEmailThreads.connectionId],
set: {
labels: sql`excluded.labels`,
hasUnread: sql`excluded.has_unread`,
isStarred: sql`excluded.is_starred`,
subject: sql`excluded.subject`,
snippet: sql`excluded.snippet`,
fromEmail: sql`excluded.from_email`,
fromName: sql`excluded.from_name`,
latestMessageAt: sql`excluded.latest_message_at`,
messageCount: sql`excluded.message_count`,
hasDraft: sql`excluded.has_draft`,
// Only update conversationId if the incoming value is non-null
// (don't wipe a linked conversation on a partial sync)
conversationId: sql`COALESCE(excluded.conversation_id, crm_email_threads.conversation_id)`,
historyId: sql`excluded.history_id`,
updatedAt: sql`now()`,
},
});
}
```
### Label-only patch: `patchEmailThreadLabels()`
For label changes that don't re-sync the full thread content:
```typescript
export async function patchEmailThreadLabels(
db: DB,
threadId: string,
connectionId: string,
addLabels: string[],
removeLabels: string[],
): Promise<void> {
// Remove then add — avoids read-modify-write race
await db
.update(crmEmailThreads)
.set({
labels: sql`
array(
SELECT unnest(labels || ${addLabels}::text[])
EXCEPT
SELECT unnest(${removeLabels}::text[])
)
`,
hasUnread: sql`
CASE
WHEN 'UNREAD' = ANY(${addLabels}::text[]) THEN true
WHEN 'UNREAD' = ANY(${removeLabels}::text[]) THEN false
ELSE has_unread
END
`,
isStarred: sql`
CASE
WHEN 'STARRED' = ANY(${addLabels}::text[]) THEN true
WHEN 'STARRED' = ANY(${removeLabels}::text[]) THEN false
ELSE is_starred
END
`,
updatedAt: sql`now()`,
})
.where(
and(
eq(crmEmailThreads.threadId, threadId),
eq(crmEmailThreads.connectionId, connectionId),
),
);
}
```
---
## Phased Implementation Plan
### Phase 1 — Schema & Migration (no functional change)
**Goal:** Get the table into the DB. Nothing reads from it yet. Safe to deploy at any time.
**Steps:**
1. Add `crmEmailThreads` table definition to `apps/server/src/db/crm-schema.ts`.
- Add the Drizzle relations entry to `crmConversationsRelations` (many threads per conversation).
2. Run `pnpm drizzle-kit generate` from `apps/server/` to generate:
```
apps/server/src/db/migrations/0075_add_crm_email_threads.sql
```
3. Inspect the generated SQL. It should contain:
```sql
CREATE TABLE "crm_email_threads" (
"thread_id" text NOT NULL,
"connection_id" text NOT NULL,
"user_id" text NOT NULL,
"labels" text[] NOT NULL DEFAULT '{}',
"has_unread" boolean NOT NULL DEFAULT false,
"is_starred" boolean NOT NULL DEFAULT false,
"subject" text NOT NULL DEFAULT '',
"snippet" text,
"from_email" text NOT NULL DEFAULT '',
"from_name" text,
"latest_message_at" timestamp NOT NULL,
"message_count" integer NOT NULL DEFAULT 0,
"has_draft" boolean NOT NULL DEFAULT false,
"conversation_id" uuid,
"history_id" text,
"created_at" timestamp NOT NULL DEFAULT now(),
"updated_at" timestamp NOT NULL DEFAULT now(),
PRIMARY KEY ("thread_id", "connection_id")
);
ALTER TABLE "crm_email_threads"
ADD CONSTRAINT "crm_email_threads_connection_id_fk"
FOREIGN KEY ("connection_id") REFERENCES "connection"("id") ON DELETE CASCADE;
ALTER TABLE "crm_email_threads"
ADD CONSTRAINT "crm_email_threads_user_id_fk"
FOREIGN KEY ("user_id") REFERENCES "user"("id") ON DELETE CASCADE;
ALTER TABLE "crm_email_threads"
ADD CONSTRAINT "crm_email_threads_conversation_id_fk"
FOREIGN KEY ("conversation_id") REFERENCES "crm_conversations"("id") ON DELETE SET NULL;
CREATE INDEX "idx_crm_email_threads_user_id" ON "crm_email_threads" ("user_id");
CREATE INDEX "idx_crm_email_threads_connection_id" ON "crm_email_threads" ("connection_id");
CREATE INDEX "idx_crm_email_threads_conversation_id" ON "crm_email_threads" ("conversation_id");
CREATE INDEX "idx_crm_email_threads_user_latest"
ON "crm_email_threads" ("user_id", "latest_message_at" DESC);
CREATE INDEX "idx_crm_email_threads_labels"
ON "crm_email_threads" USING GIN ("labels");
CREATE INDEX "idx_crm_email_threads_unread"
ON "crm_email_threads" ("user_id", "has_unread");
```
4. Apply the migration in staging: `pnpm db:migrate` (or however migrations are run in this repo).
**Verification:** `SELECT count(*) FROM crm_email_threads;` → 0. Table exists, no data yet.
---
### Phase 2 — Write Path: Populate on Every Sync
**Goal:** Every thread sync now also upserts `crm_email_threads`. The read path is unchanged — the table is written but not yet read from.
**Steps:**
#### 2a. Create `apps/server/src/services/crm/email-threads.ts`
New file containing `upsertEmailThread()` and `patchEmailThreadLabels()` as defined above.
#### 2b. Call `upsertEmailThread()` at the end of `findOrCreateEmailEvents()`
**File:** `apps/server/src/services/crm/email-events.ts`
`findOrCreateEmailEvents` already has all the data needed: `messages`, `threadId`, `userId`, `conversationId`. At the end of the function, after all inserts, add:
```typescript
await upsertEmailThread(db, {
threadId,
connectionId, // needs to be threaded in as a new param
userId,
messages,
conversationId,
});
```
**Note:** `connectionId` is not currently a param of `findOrCreateEmailEvents`. It needs to be added. Callers are:
- `apps/server/src/pipelines.ts` — already has `connectionId` in scope
- `apps/server/src/workflows/sync-threads-workflow.ts` — already has `connectionId` in scope
#### 2c. Call `patchEmailThreadLabels()` inside `patchStoredThreadLabels()`
**File:** `apps/server/src/lib/aws-primary-mail-runtime.ts:433`
After the S3 label patch, add a DB patch:
```typescript
await withDb(async (db) => {
await patchEmailThreadLabels(db, threadId, connectionId, addLabels, removeLabels);
});
```
#### 2d. Call `patchEmailThreadLabels()` in the Gmail push label handler
**File:** `apps/server/src/pipelines.ts:1189`
After `agent.modifyThreadLabelsInDB(threadId, addLabels, removeLabels)`, add:
```typescript
await withDb(async (db) => {
await patchEmailThreadLabels(db, threadId, connectionId, addLabels, removeLabels);
});
```
#### 2e. Propagate `conversationId` updates
When `findOrCreateConversation()` links a thread to a conversation, call:
```typescript
await db
.update(crmEmailThreads)
.set({ conversationId, updatedAt: sql`now()` })
.where(eq(crmEmailThreads.threadId, threadId));
```
This already happens transitively via `findOrCreateEmailEvents` calling `upsertEmailThread` with the resolved `conversationId`, but an explicit update covers the case where a conversation is linked after initial sync.
**Verification after Phase 2:**
```sql
-- After syncing a few threads, spot-check:
SELECT thread_id, labels, has_unread, subject, message_count, latest_message_at
FROM crm_email_threads
WHERE user_id = '<your-user-id>'
ORDER BY latest_message_at DESC
LIMIT 20;
```
---
### Phase 3 — Backfill Existing Threads
**Goal:** Populate `crm_email_threads` for all threads that already exist in `crm_email_events` before Phase 2 was deployed.
This is a one-time operation. It can run as a background job or a SQL migration. Given the potentially large row count (many messages per thread), running as a background script is safer than a blocking migration.
#### Option A: SQL migration (simpler, may be slow on large datasets)
Add as `0076_backfill_crm_email_threads.sql`:
```sql
-- Backfill crm_email_threads from existing crm_email_events rows.
-- Uses DISTINCT ON to get the newest non-draft message per (threadId, connectionId).
-- Labels are the union of all messages per thread.
INSERT INTO crm_email_threads (
thread_id, connection_id, user_id,
labels, has_unread, is_starred,
subject, snippet, from_email, from_name,
latest_message_at, message_count, has_draft,
conversation_id,
created_at, updated_at
)
SELECT
e.thread_id,
conn.id AS connection_id,
ev.user_id,
-- Union of all labels for this thread
ARRAY(
SELECT DISTINCT unnest(agg.all_labels)
FROM (
SELECT ARRAY_AGG(label) AS all_labels
FROM (
SELECT unnest(ee2.labels) AS label
FROM crm_email_events ee2
INNER JOIN crm_events ev2 ON ee2.event_id = ev2.id
WHERE ee2.thread_id = e.thread_id AND ev2.user_id = ev.user_id
) flattened
) agg
) AS labels,
-- hasUnread: any message has UNREAD label
EXISTS (
SELECT 1 FROM crm_email_events ee2
INNER JOIN crm_events ev2 ON ee2.event_id = ev2.id
WHERE ee2.thread_id = e.thread_id
AND ev2.user_id = ev.user_id
AND 'UNREAD' = ANY(ee2.labels)
) AS has_unread,
'STARRED' = ANY(
ARRAY(
SELECT DISTINCT unnest(ee2.labels)
FROM crm_email_events ee2
INNER JOIN crm_events ev2 ON ee2.event_id = ev2.id
WHERE ee2.thread_id = e.thread_id AND ev2.user_id = ev.user_id
)
) AS is_starred,
-- Display fields from newest non-draft message
e.subject,
e.snippet,
e.from_email,
e.from_name,
ev.occurred_at AS latest_message_at,
-- message_count: count of non-draft messages
(
SELECT COUNT(*)
FROM crm_email_events ee2
INNER JOIN crm_events ev2 ON ee2.event_id = ev2.id
WHERE ee2.thread_id = e.thread_id
AND ev2.user_id = ev.user_id
AND ee2.is_draft = false
) AS message_count,
-- hasDraft: any draft exists
EXISTS (
SELECT 1 FROM crm_email_events ee2
INNER JOIN crm_events ev2 ON ee2.event_id = ev2.id
WHERE ee2.thread_id = e.thread_id
AND ev2.user_id = ev.user_id
AND ee2.is_draft = true
) AS has_draft,
ev.conversation_id,
now(), now()
FROM (
-- Newest non-draft message per (thread_id, user_id)
SELECT DISTINCT ON (ee.thread_id, ev.user_id)
ee.thread_id, ee.subject, ee.snippet, ee.from_email, ee.from_name,
ev.user_id, ev.occurred_at, ev.conversation_id
FROM crm_email_events ee
INNER JOIN crm_events ev ON ee.event_id = ev.id
WHERE ee.is_draft = false
ORDER BY ee.thread_id, ev.user_id, ev.occurred_at DESC
) e
INNER JOIN crm_events ev ON ev.user_id = e.user_id
INNER JOIN "connection" conn ON conn.user_id = e.user_id
ON CONFLICT (thread_id, connection_id) DO NOTHING;
```
**Note on connection join:** A user may have multiple Gmail connections. The backfill joins on `user_id` → `connection` which may produce multiple rows per thread if the user has multiple connections. The `ON CONFLICT DO NOTHING` handles this gracefully — only the first connection's row is kept. A more precise backfill would pass `connectionId` through `crm_email_events` (which doesn't currently store it — see Phase 4 note below).
#### Option B: Runtime backfill script
Write a one-off tRPC admin procedure or direct DB script that:
1. Pages through distinct `(threadId, userId)` pairs in `crm_email_events`
2. For each, calls `upsertEmailThread()` with data reconstructed from existing rows
3. Runs in batches of 500, with progress logging
**Recommendation:** Option B is safer for production — it respects rate limits, can be paused, and reuses validated application logic.
---
### Phase 4 — Read Path: Replace `getThreadPreviewSummaries`
**Goal:** `listThreads` reads thread metadata from `crm_email_threads` (a single PK lookup) instead of scanning and aggregating `crm_email_events` rows.
**Steps:**
Replace the body of `getThreadPreviewSummaries` in `aws-primary-mail-runtime.ts`:
```typescript
async function getThreadPreviewSummaries(
userId: string,
threadIds: string[],
): Promise<Map<string, ThreadPreviewSummary>> {
if (!threadIds.length) return new Map();
return await withDb(async (db) => {
const rows = await db
.select()
.from(crmEmailThreads)
.where(
and(
eq(crmEmailThreads.userId, userId),
inArray(crmEmailThreads.threadId, threadIds),
),
);
const summaries = new Map<string, ThreadPreviewSummary>();
for (const row of rows) {
summaries.set(row.threadId, {
conversationId: row.conversationId ?? null,
latestReceivedOn: row.latestMessageAt.toISOString(),
sender: { email: row.fromEmail, name: row.fromName ?? undefined },
subject: row.subject,
snippet: row.snippet ?? undefined,
labels: (row.labels ?? []).map((label) => ({
id: label,
name: label,
type: GMAIL_SYSTEM_LABELS.has(label) ? 'system' : 'user',
})),
messageCount: row.messageCount,
hasDraft: row.hasDraft,
participants: [], // Phase 4 note: participants not stored yet — see below
});
}
return summaries;
});
}
```
**Phase 4 note — participants:** `participants` (the max-2 unique senders array) is not stored in `crm_email_threads` as currently designed. Options:
- Add `participants jsonb` column to `crm_email_threads` (simplest, denormalized)
- Compute from `crm_email_events` for the missing threads only (keeps schema leaner)
- Accept that participants are empty until a follow-up phase
Recommendation: Add `participants jsonb NOT NULL DEFAULT '[]'` to the table in Phase 1 and populate it in `upsertEmailThread()`.
**Verification after Phase 4:**
- Load inbox. Confirm thread list renders correctly.
- Confirm labels, unread state, subject, snippet match what Gmail shows.
- Compare DB query count before/after — should drop from N_messages to N_threads for the metadata fetch.
---
### Phase 5 — Skip the Gmail API Call (Pure DB `listThreads`)
**Goal:** When the local data is fresh, `listThreads` serves the thread list entirely from PostgreSQL, eliminating the Gmail API round-trip for the non-search path.
This is the most impactful change but also the most complex, because Gmail is currently the authority on folder membership and sort order. We can restore that authority to PostgreSQL because `crm_email_threads.labels` now mirrors Gmail's label state and `latestMessageAt` gives us sort order.
**Steps:**
#### 5a. Change `listThreadsViaProvider` to query DB first
```typescript
async function listThreadsViaProvider(
connectionId: string,
params: { folder?: string; q?: string; maxResults?: number; pageToken?: string; labelIds?: string[] },
traceContext?: TraceContext,
): Promise<IGetThreadsResponse> {
const record = await getConnectionRecord(connectionId);
const labelFilter = folderToGmailLabel(params.folder ?? 'inbox', params.labelIds);
// --- DB-first path (no search query) ---
if (!params.q) {
const { threads, nextPageToken } = await queryThreadsFromDB(
record.userId,
connectionId,
labelFilter,
params.maxResults ?? defaultPageSize,
params.pageToken,
);
// If we got a full page, return it — DB is the authority
if (threads.length >= (params.maxResults ?? defaultPageSize)) {
return { threads, nextPageToken };
}
// Underfilled: fall through to Gmail for a sync, then re-query
}
// --- Gmail path (search query, or underfilled DB) ---
const manager = connectionToMailManager(record);
const listResponse = await manager.list({ ... }, traceContext);
const threadIds = listResponse.threads.map((t) => t.id);
const previewSummaries = await getThreadPreviewSummaries(record.userId, threadIds);
// ... rest of existing merge logic
}
```
#### 5b. Add `queryThreadsFromDB()`
```typescript
async function queryThreadsFromDB(
userId: string,
connectionId: string,
labelFilter: { includeLabels: string[]; requireAll: boolean },
limit: number,
pageToken?: string,
): Promise<{ threads: IGetThreadsResponse['threads']; nextPageToken?: string }> {
return withDb(async (db) => {
// cursor-based pagination: pageToken is an ISO timestamp of latestMessageAt
const cursorDate = pageToken ? new Date(pageToken) : undefined;
const rows = await db
.select()
.from(crmEmailThreads)
.where(
and(
eq(crmEmailThreads.userId, userId),
eq(crmEmailThreads.connectionId, connectionId),
// Folder filter: check if label is present in the array
...labelFilter.includeLabels.map(
(label) => sql`${label} = ANY(${crmEmailThreads.labels})`,
),
cursorDate
? lt(crmEmailThreads.latestMessageAt, cursorDate)
: undefined,
),
)
.orderBy(desc(crmEmailThreads.latestMessageAt))
.limit(limit + 1); // fetch one extra to detect next page
const hasMore = rows.length > limit;
const pageRows = hasMore ? rows.slice(0, limit) : rows;
const threads = pageRows.map((row) => buildThreadPreviewFromSummary(row.threadId, {
conversationId: row.conversationId ?? null,
latestReceivedOn: row.latestMessageAt.toISOString(),
sender: { email: row.fromEmail, name: row.fromName ?? undefined },
subject: row.subject,
snippet: row.snippet ?? undefined,
labels: (row.labels ?? []).map((l) => ({
id: l, name: l,
type: GMAIL_SYSTEM_LABELS.has(l) ? 'system' : 'user',
})),
messageCount: row.messageCount,
hasDraft: row.hasDraft,
participants: (row.participants as Sender[]) ?? [],
}));
const nextPageToken=[redacted]
? pageRows.at(-1)!.latestMessageAt.toISOString()
: undefined;
return { threads, nextPageToken };
});
}
```
**Note:** The existing `pageToken` is a Gmail API cursor (opaque string from Gmail). Changing it to a timestamp cursor is a breaking change for any in-flight pagination. Consider naming this differently internally and returning the Gmail-style token format to clients.
#### 5c. Store `connectionId` on `crm_email_events`
The backfill in Phase 3 exposed a gap: `crm_email_events` doesn't store `connectionId`. This makes it impossible to precisely scope a thread to the connection it came from when a user has multiple Gmail accounts. The `crm_email_threads` primary key is `(threadId, connectionId)`, so `connectionId` must be available at write time.
In Phase 2b, `findOrCreateEmailEvents` already receives `connectionId` as a new parameter. The thread row is written correctly from that point. The backfill (Phase 3) must handle the multi-connection case carefully.
**Verification after Phase 5:**
- Load inbox on a fresh session. Confirm no Gmail API call is made (check server logs / tracing).
- Load inbox after receiving a new email (push notification should have updated `crm_email_threads`). Confirm the new thread appears.
- Test folder switching (SENT, TRASH, custom labels). Confirm label filter works.
- Test pagination: scroll past 50 threads. Confirm cursor works correctly.
---
## Rollback Plan
Each phase is independently deployable and independently rollback-safe:
| Phase | Rollback |
|---|---|
| 1 (schema) | Drop the table. No existing code reads from it. |
| 2 (writes) | Remove `upsertEmailThread` calls. Table goes stale but nothing breaks — reads still come from `crm_email_events`. |
| 3 (backfill) | No rollback needed — backfill is additive. |
| 4 (reads from new table) | Revert `getThreadPreviewSummaries` to the old aggregation query. |
| 5 (skip Gmail call) | Revert `listThreadsViaProvider` to always call Gmail first. |
---
## Data Integrity Risks
**Label drift.** If a label change arrives via Gmail push but `patchEmailThreadLabels()` fails (DB error), `crm_email_threads.labels` will be out of sync with Gmail. The next full thread sync will correct this, but until then the thread may appear in the wrong folder. Mitigation: wrap the DB patch and S3 patch together, log failures, and rely on the background sync as the correction mechanism.
**Multi-connection edge case.** If a user adds a second Gmail account and threads from both accounts share the same `threadId` (unlikely per Gmail's guarantees, but the schema handles it via the composite PK), both connections will have their own row. The read path must filter by `connectionId`, not just `userId`.
**Stale backfill.** Threads that were synced before Phase 2 is deployed will have their initial data from the Phase 3 backfill. Any label changes between the backfill run and the first full re-sync will not be reflected. The background sync cycle (incremental history sync) will correct these within the next sync interval.
**Participants column.** If `participants` is added as a jsonb column but `upsertEmailThread()` computes it incorrectly (wrong ordering, wrong deduplication), it will silently serve wrong data. Add a spot-check assertion in the upsert function during development.
---
## Verification Steps (end-to-end)
After all phases are deployed:
1. **Inbox load time:** Measure p50/p99 latency for `listThreads` before and after. Expect a reduction in p99 from eliminating the Gmail round-trip.
2. **Label accuracy:**
```sql
-- Find threads where crm_email_threads.has_unread disagrees with any
-- message having UNREAD in crm_email_events
SELECT t.thread_id, t.has_unread AS thread_unread,
bool_or('UNREAD' = ANY(e.labels)) AS any_message_unread
FROM crm_email_threads t
JOIN crm_email_events e ON e.thread_id = t.thread_id
JOIN crm_events ev ON e.event_id = ev.id AND ev.user_id = t.user_id
GROUP BY t.thread_id, t.has_unread
HAVING t.has_unread != bool_or('UNREAD' = ANY(e.labels));
```
Should return 0 rows.
3. **Thread count:** Distinct threadIds in `crm_email_threads` should approximately equal distinct threadIds in `crm_email_events` for the same user.
4. **Pagination completeness:** Paginate through the full inbox. Confirm total thread count matches Gmail's reported count.
5. **Push notification sync:** Mark a thread as read in Gmail (another client). Within the push notification window, reload the inbox. Confirm `has_unread` is updated and thread no longer appears in unread filter.
---
## `listThreads` — Inbox Load Flow (Phases 1–4 Implemented)
Step-by-step trace for `mail.listThreads` with `folder="inbox"`, no search query.
```
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
listThreads — inbox load (no search query)
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
┌─────────────────────────────────────────────────────────────────────┐
│ STEP 1 — Frontend calls tRPC │
│ trpc/routes/mail.ts → listThreads procedure │
└─────────────────────────────────────────────────────────────────────┘
Input payload:
{
folder: "inbox",
q: "", ← empty = no search
maxResults: 50,
cursor: "", ← first page
labelIds: [],
requireAllLabels: true
}
Because q === "" → takes the non-search branch →
calls listThreadsReadModel(...)
┌─────────────────────────────────────────────────────────────────────┐
│ STEP 2 — Read model thin wrapper │
│ services/mail/list-threads-read-model.ts │
└─────────────────────────────────────────────────────────────────────┘
listThreadsReadModel
→ getThreadsFromDB (server-utils.ts)
→ getAwsPrimaryThreadsFromDB (aws-primary-mail-runtime.ts)
→ listThreadsViaProvider ← all work happens here
┌─────────────────────────────────────────────────────────────────────┐
│ STEP 3 — Gmail API call (still required — Phase 5 not yet built) │
│ listThreadsViaProvider → manager.list() │
└─────────────────────────────────────────────────────────────────────┘
Gmail returns thread IDs + pagination token, sorted by Gmail's
notion of recency. No message content — just identifiers.
listResponse from Gmail API:
{
threads: [
{ id: "18a3f...", historyId: "93201" },
{ id: "18a3d...", historyId: "93185" },
{ id: "18a31...", historyId: "93102" },
... (50 entries)
],
nextPageToken=[redacted]
}
threadIds = ["18a3f...", "18a3d...", "18a31...", ...] ← 50 IDs
┌─────────────────────────────────────────────────────────────────────┐
│ STEP 4 — DB lookup (the new part) │
│ getThreadPreviewSummaries(userId, threadIds) │
│ → SELECT * FROM crm_email_threads │
│ WHERE user_id = $1 AND thread_id = ANY($2) │
└─────────────────────────────────────────────────────────────────────┘
One query, up to 50 rows via the (user_id, thread_id) index.
No message rows scanned. No per-thread aggregation.
Raw DB rows returned (one per thread that has been synced):
┌──────────────┬────────────┬──────────────────────────┬─────────────────────┐
│ thread_id │ has_unread │ subject │ latest_message_at │
├──────────────┼────────────┼──────────────────────────┼─────────────────────┤
│ "18a3f..." │ true │ "Q3 proposal" │ 2026-04-05 14:23:00 │
│ "18a3d..." │ false │ "Re: onboarding call" │ 2026-04-04 09:11:00 │
│ "18a31..." │ true │ "Invoice #4821" │ 2026-04-03 17:45:00 │
│ ... │
└─────────────────────────────────────────────────────────────────────────────┘
Also stored per row (not shown above):
labels: ["INBOX", "UNREAD"]
from_email: "<email>"
from_name: "Alice Chen"
snippet: "Hey Jesse, wanted to follow up on..."
message_count: 3
has_draft: false
participants: [{"email":"<email>","name":"Alice Chen"},...]
conversation_id: "uuid-conv-123" ← null if not CRM-linked
Returns: Map<threadId, ThreadPreviewSummary> (e.g. 48 of 50 found)
┌─────────────────────────────────────────────────────────────────────┐
│ STEP 5 — Detect missing threads │
│ missingThreadIds = threadIds not found in crm_email_threads │
└─────────────────────────────────────────────────────────────────────┘
Detection (aws-primary-mail-runtime.ts):
const missingThreadIds = threadIds.filter((threadId) => {
const preview = previewSummaries.get(threadId);
return !preview || (!preview.latestReceivedOn && !preview.hasDraft);
});
A thread is "missing" if:
• it has no row in crm_email_threads at all, OR
• its row exists but has no latestReceivedOn and no hasDraft
(i.e. it was inserted as an empty placeholder)
Example: Gmail returned 50 IDs. DB returned 48.
missingThreadIds = ["18a0a...", "18a0b..."] ← 2 new threads
┌─────────────────────────────────────────────────────────────────────┐
│ STEP 5a — Fallback: S3 read (per missing thread) │
│ loadThreadSnapshot(connectionId, threadId) │
└─────────────────────────────────────────────────────────────────────┘
For each missing thread ID, in parallel:
readThreadFromCurrentBucket(connectionId, threadId)
→ S3 GET: s3://THREADS_BUCKET/{connectionId}/{threadId}.json
┌──────────────┬─────────────────────────────────────────────────┐
│ "18a0a..." │ S3 HIT → JSON blob returned, no Gmail call │
│ "18a0b..." │ S3 MISS → falls through to Step 5b │
└──────────────┴─────────────────────────────────────────────────┘
S3 HIT shape (current label state, includes drafts):
{
messages: [ { messageId, subject, sender, tags, receivedOn, isDraft, ... } ],
labels: [ { id: "INBOX", name: "INBOX" }, { id: "UNREAD", name: "UNREAD" } ],
hasUnread: true,
conversationId: null
}
→ buildThreadPreview() shapes it for the response. Done for this thread.
┌─────────────────────────────────────────────────────────────────────┐
│ STEP 5b — Fallback: Gmail API fetch (S3 miss only) │
│ fetchAndStoreThreadFromProvider(connectionId, threadId) │
└─────────────────────────────────────────────────────────────────────┘
Only reached if S3 had no snapshot (brand-new thread, never fetched).
manager.get(threadId)
→ Gmail API: GET /gmail/v1/users/me/threads/{threadId}
→ returns full thread with all messages and current labels
Immediately writes back to S3:
storeThreadInCurrentBucket(connectionId, threadId, enrichedThread)
→ S3 PUT: s3://THREADS_BUCKET/{connectionId}/{threadId}.json
After this write:
• S3 now has the snapshot → next listThreads call is a 5a HIT
• crm_email_threads: ✗ still missing
• crm_email_events: ✗ still missing
→ buildThreadPreview() shapes it for the response.
┌──────────────────────────────────────────────────────────────────┐
│ IMPORTANT: listThreads does NOT write to Postgres for │
│ missing threads. S3 is the only write that happens here. │
│ │
│ crm_email_threads and crm_email_events are only populated by: │
│ • Gmail push notification → ZERO_WORKFLOW → syncThreads() │
│ → createAndLinkEvents() → findOrCreateEmailEvents() │
│ → upsertEmailThread() │
│ • Background SyncThreadsWorkflow (scheduled) │
│ │
│ Until one of those fires, every subsequent listThreads call for │
│ this thread goes: DB miss → S3 hit (fast, no Gmail call). │
└──────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────────┐
│ STEP 6 — Shape each thread into the wire format │
│ buildThreadPreviewFromSummary(threadId, summary) │
└─────────────────────────────────────────────────────────────────────┘
For each of the 50 thread IDs (preserving Gmail's sort order):
{
id: "18a3f...",
historyId: null,
$raw: {
conversationId: "uuid-conv-123",
latestReceivedOn: "2026-04-05T14:23:00.000Z",
sender: { email: "<email>", name: "Alice Chen" },
subject: "Q3 proposal",
snippet: "Hey Jesse, wanted to follow up on...",
labels: [
{ id: "INBOX", name: "INBOX", type: "system" },
{ id: "UNREAD", name: "UNREAD", type: "system" }
],
messageCount: 3,
hasDraft: false,
participants: [
{ email: "<email>", name: "Alice Chen" },
{ email: "<email>", name: "Bob Smith" }
]
}
}
┌─────────────────────────────────────────────────────────────────────┐
│ STEP 7 — tRPC response to frontend │
└─────────────────────────────────────────────────────────────────────┘
{
threads: [ ...50 thread objects in Gmail sort order ],
nextPageToken=[redacted]
}
Frontend renders inbox from $raw fields on each thread.
"UNREAD" in labels → bold row.
conversationId present → CRM link indicator shown.
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
HOW crm_email_threads STAYS CURRENT
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
Event: new message arrives (Gmail push)
─────────────────────────────────────────────────────────────────────
Gmail push → pipelines.ts ZERO_WORKFLOW
Stage 4: for each new threadId in the history payload,
agent.syncThreads({ threadIds, source: 'background-sync' })
→ fetchAndStoreThreadFromProvider(connectionId, threadId)
Gmail API: GET /threads/{threadId} ← full thread fetch
S3 PUT: {connectionId}/{threadId}.json ← snapshot written
→ createAndLinkEvents()
→ findOrCreateEmailEvents()
INSERT crm_events + crm_email_events ← one row per message
(skips drafts)
→ upsertEmailThread()
INSERT/UPDATE crm_email_threads ← thread row written
labels: union of all message tags (from ParsedMessage)
hasUnread: 'UNREAD' in union
subject/snippet: from newest non-draft
latestMessageAt: newest non-draft receivedOn
participants: up to 2 unique senders
After this point:
• S3: has snapshot ✓
• crm_email_events: has one row per non-draft message ✓
• crm_email_threads: has one row for this thread ✓
• Next listThreads: DB hit in Step 4, no fallback needed ✓
Event: user marks thread read / archives
─────────────────────────────────────────────────────────────────────
User action in client
→ mail.modifyLabels tRPC
→ modifyAwsPrimaryThreadLabels()
→ Gmail API (applies label change on provider)
→ syncThreadViaProvider() (refreshes S3 snapshot)
→ patchStoredThreadLabels() ← writes S3
→ patchEmailThreadLabels() ← writes crm_email_threads
SQL: labels = (labels ∪ addLabels) \ removeLabels
hasUnread flipped if UNREAD in either set
Event: Gmail push for label change (e.g. read on mobile)
─────────────────────────────────────────────────────────────────────
Gmail push notification
→ pipelines.ts ZERO_WORKFLOW Stage 7
→ agent.modifyThreadLabelsInDB() ← writes S3
→ patchEmailThreadLabels() ← writes crm_email_threads
same SQL array op as above
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
WHAT'S NOT YET DONE (Phase 5 from the plan)
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
Step 3 above still calls Gmail API unconditionally.
Phase 5 would short-circuit that: if crm_email_threads has a
full page for this folder, return it directly and skip Gmail.
The Gmail call currently provides two things we still need:
- authoritative sort order
- pagination token (Gmail opaque cursor)
Until Phase 5 lands, the Gmail round-trip remains on every
inbox load. The speedup from Phases 1-4 is in Step 4 only:
up to 50 PK lookups on crm_email_threads instead of scanning
all crm_email_events rows for those threads.
```
---
## All Email Operations — `crm_email_threads` Update Matrix
All 21 user-facing email mutations audited. For each: does it currently update `crm_email_threads`, and if not, what needs to be added?
### How most operations flow
Operations that use `modifyLabelsWithSync` already write to `crm_email_threads` via `patchEmailThreadLabels()` — as long as Phase 2c has been deployed (the call to `patchEmailThreadLabels` inside `patchStoredThreadLabels`). These are **✅ covered**.
Operations that do NOT go through `modifyLabelsWithSync` require explicit work — marked **⚠️ gap** below.
### Operation matrix
| Operation | tRPC route | Via `modifyLabelsWithSync`? | `crm_email_threads` fields affected | Status |
|---|---|---|---|---|
| Mark as read | `mail.markAsRead` | ✅ yes | `labels` (remove UNREAD), `hasUnread → false` | ✅ covered |
| Mark as unread | `mail.markAsUnread` | ✅ yes | `labels` (add UNREAD), `hasUnread → true` | ✅ covered |
| Mark as important | `mail.markAsImportant` | ✅ yes | `labels` (add IMPORTANT) | ✅ covered |
| Modify labels (generic) | `mail.modifyLabels` | ✅ yes | `labels`, `hasUnread`, `isStarred` | ✅ covered |
| Toggle star | `mail.toggleStar` | ✅ yes | `labels`, `isStarred` | ✅ covered |
| Toggle important | `mail.toggleImportant` | ✅ yes | `labels` | ✅ covered |
| Bulk star | `mail.bulkStar` | ✅ yes | `labels`, `isStarred → true` | ✅ covered |
| Bulk mark important | `mail.bulkMarkImportant` | ✅ yes | `labels` | ✅ covered |
| Bulk unstar | `mail.bulkUnstar` | ✅ yes | `labels`, `isStarred → false` | ✅ covered |
| Bulk unmark important | `mail.bulkUnmarkImportant` | ✅ yes | `labels` | ✅ covered |
| Bulk delete (trash) | `mail.bulkDelete` | ✅ yes | `labels` (add TRASH, remove INBOX/SPAM) | ✅ covered |
| Bulk archive | `mail.bulkArchive` | ✅ yes | `labels` (remove INBOX) | ✅ covered |
| Bulk mute | `mail.bulkMute` | ✅ yes | `labels` (add MUTE) | ✅ covered |
| Snooze threads | `mail.snoozeThreads` | ✅ yes | `labels` (add SNOOZED, remove INBOX) | ✅ covered |
| Unsnooze threads | `mail.unsnoozeThreads` | ✅ yes | `labels` (remove SNOOZED, add INBOX) | ✅ covered |
| Delete (permanent) | `mail.delete` | ❌ no | Should delete the row | ⚠️ gap |
| Delete all spam | `mail.deleteAllSpam` | ❌ unclear | Should delete rows with SPAM label | ⚠️ gap |
| Send (new/reply) | `mail.send` | ❌ indirect | `upsertEmailThread` not called directly | ⚠️ gap |
| Unsend (cancel scheduled) | `mail.unsend` | n/a | KV only — no thread row change needed | ✅ no-op |
| Set remind | `mail.setRemind` | n/a | KV/queue only — no thread row change needed | ✅ no-op |
| Cancel remind | `mail.cancelRemind` | n/a | KV only — no thread row change needed | ✅ no-op |
---
### Gap 1: `mail.delete` — permanent deletion not reflected in `crm_email_threads`
**File:** `apps/server/src/trpc/routes/mail.ts` (the `delete` mutation)
**Current behavior:** Deletes from the local runtime's thread cache (`exec("DELETE FROM threads WHERE thread_id = ?")`), then calls `stub.reloadFolder('bin')`. Does **not** touch `crm_email_threads`.
**Fix:** After the existing delete call, delete the row from `crm_email_threads`:
```typescript
await withDb(async (db) => {
await db
.delete(crmEmailThreads)
.where(
and(
eq(crmEmailThreads.threadId, input.id),
eq(crmEmailThreads.connectionId, activeConnection.id),
),
);
});
```
**Why this is safe:** Permanent deletion is irreversible by definition. Removing the DB row immediately is correct — there is no undo path that would need to restore it.
---
### Gap 2: `mail.deleteAllSpam` — bulk deletion of spam threads
**File:** `apps/server/src/trpc/routes/mail.ts` (the `deleteAllSpam` mutation)
**Implementation:** Delegates to a `deleteAllSpam(activeConnection.id)` utility in `apps/server/src/lib/server-utils.ts`.
**Current behavior:** Unknown — need to audit `deleteAllSpam()` to see if it calls `modifyLabelsWithSync` or makes a direct Gmail API call. Either way, there is no delete of `crm_email_threads` rows.
**Fix option A (if Gmail permanently deletes the messages):** Delete `crm_email_threads` rows for all threads with `SPAM` in `labels` for this connection:
```typescript
await withDb(async (db) => {
await db
.delete(crmEmailThreads)
.where(
and(
eq(crmEmailThreads.connectionId, activeConnection.id),
sql`'SPAM' = ANY(${crmEmailThreads.labels})`,
),
);
});
```
**Fix option B (if Gmail moves to TRASH first):** `modifyLabelsWithSync` with `addLabels: ['TRASH'], removeLabels: ['SPAM']` — which already flows through `patchEmailThreadLabels`. But this requires restructuring `deleteAllSpam` to go through the same path as individual deletes.
Audit `server-utils.ts:deleteAllSpam` first before implementing.
---
### Gap 3: `mail.send` — new thread row not created immediately
**File:** `apps/server/src/trpc/routes/mail.ts` (the `send` mutation)
**Current behavior:** `agent.stub.sendDraft()` / `agent.stub.send()` sends the email via the mail provider. The thread row in `crm_email_threads` is **not** created synchronously here — it is populated later when the Gmail push notification arrives and `upsertEmailThread()` runs via the sync workflow.
**Impact:** After sending, the sent thread will not appear in `crm_email_threads` until the next push notification sync. For most use cases this is acceptable since the user is navigating away from the compose view. However, if the user immediately navigates to the Sent folder, the thread may not appear (or may appear via the S3 fallback path instead of the DB).
**Fix:** After a successful send, call `upsertEmailThread()` with the returned `threadId` and available metadata. The full message content is not available at this point (we don't re-fetch), but we can write a minimal row:
```typescript
// After agent.stub.send() returns threadId, messageId:
await withDb(async (db) => {
await upsertEmailThread(db, {
threadId: result.threadId,
connectionId: activeConnection.id,
userId: session.user.id,
messages: [{
// Construct a minimal ParsedMessage from the send payload
subject: input.subject,
snippet: extractSnippet(input.message),
sender: { email: input.fromEmail, name: session.user.name },
receivedOn: new Date().toISOString(),
isDraft: false,
tags: [{ name: 'SENT' }],
}],
});
});
```
**Alternative (simpler, lower risk):** Don't create the row immediately. Instead, after the send call returns, fire a background sync for that `threadId` only:
```typescript
// Trigger an async re-sync so the thread row appears quickly
await agent.stub.syncThreads({ threadIds: [result.threadId], source: 'post-send' });
```
This reuses the existing `upsertEmailThread` call inside `syncThreads` and avoids constructing a partial `ParsedMessage` manually. Recommended approach.
---
### Summary: what to implement
| Gap | File to change | Change |
|---|---|---|
| `mail.delete` | `trpc/routes/mail.ts` | Add `db.delete(crmEmailThreads).where(threadId + connectionId)` after the existing delete |
| `mail.deleteAllSpam` | `lib/server-utils.ts` (first audit) | Add bulk delete from `crm_email_threads` WHERE SPAM in labels, OR restructure through `modifyLabelsWithSync` |
| `mail.send` | `trpc/routes/mail.ts` | After send returns, call `agent.stub.syncThreads({ threadIds: [threadId], source: 'post-send' })` |
---
## Label Normalisation — `crm_labels` + `crm_thread_labels`
### Problem with the current `text[]` approach
`crm_email_threads.labels` stores label **names** as a raw string array, e.g. `['INBOX', 'UNREAD', 'work']`. This has two concrete failure modes:
1. **Rename breaks the DB.** Gmail labels have a stable `id` (e.g. `Label_3691628123`) and a mutable `name`. If a user renames "work" → "clients" in Gmail settings, every row in `crm_email_threads` still says `"work"`. The next time Cedar needs to call Gmail with that label, `resolveLabelId("work")` fails — and currently the code **creates a new label called "work"**, duplicating it.
2. **No relational integrity.** There is no way to know which labels exist for a given connection, what colour they are, or when they were last seen. Querying "all threads in label X" requires a full array scan with no join path to label metadata.
### Proposed schema
Two new tables replace `crm_email_threads.labels text[]`.
#### `crm_labels` — one row per label per connection
```typescript
// apps/server/src/db/crm-schema.ts
export const crmLabels = pgTable(
'crm_labels',
{
// Gmail's stable label identifier.
// System labels use their name as the ID: 'INBOX', 'SENT', 'UNREAD', etc.
// User labels look like: 'Label_3691628123'
labelId: text('label_id').notNull(),
connectionId: text('connection_id').notNull()
.references(() => connection.id, { onDelete: 'cascade' }),
// Mutable display name — the only thing that changes on rename
name: text('name').notNull(),
type: text('type').notNull(), // 'system' | 'user'
color: jsonb('color'), // { backgroundColor: string, textColor: string } | null
createdAt: timestamp('created_at').notNull().defaultNow(),
updatedAt: timestamp('updated_at').notNull().defaultNow(),
},
(table) => [
primaryKey({ columns: [table.labelId, table.connectionId] }),
index('idx_crm_labels_connection_id').on(table.connectionId),
// Fast lookup of a label by name within a connection (used during sync)
index('idx_crm_labels_connection_name').on(table.connectionId, table.name),
],
);
```
#### `crm_thread_labels` — join table
```typescript
export const crmThreadLabels = pgTable(
'crm_thread_labels',
{
threadId: text('thread_id').notNull(),
connectionId: text('connection_id').notNull(),
labelId: text('label_id').notNull(),
},
(table) => [
primaryKey({ columns: [table.threadId, table.connectionId, table.labelId] }),
// FK to the thread
foreignKey({
columns: [table.threadId, table.connectionId],
foreignColumns: [crmEmailThreads.threadId, crmEmailThreads.connectionId],
}).onDelete('cascade'),
// FK to the label
foreignKey({
columns: [table.labelId, table.connectionId],
foreignColumns: [crmLabels.labelId, crmLabels.connectionId],
}).onDelete('cascade'),
// Primary query: "all threads for label X"
index('idx_crm_thread_labels_label').on(table.connectionId, table.labelId),
// Reverse: "all labels for thread Y"
index('idx_crm_thread_labels_thread').on(table.threadId, table.connectionId),
],
);
```
#### Remove `labels` from `crm_email_threads`
Drop the `labels text[]` column. Keep `hasUnread` and `isStarred` as denormalized booleans — these are read on every thread list render and are not worth a join.
```typescript
// Remove this line from crmEmailThreads:
labels: text('labels').array().notNull().default([]),
```
### How queries change
**List inbox threads (before):**
```sql
SELECT * FROM crm_email_threads
WHERE user_id = $1
AND connection_id = $2
AND 'INBOX' = ANY(labels)
ORDER BY latest_message_at DESC
LIMIT 30;
```
**List inbox threads (after):**
```sql
SELECT t.*
FROM crm_email_threads t
JOIN crm_thread_labels tl
ON tl.thread_id = t.thread_id
AND tl.connection_id = t.connection_id
WHERE t.user_id = $1
AND t.connection_id = $2
AND tl.label_id = 'INBOX'
ORDER BY t.latest_message_at DESC
LIMIT 30;
```
The join is indexed on `(connection_id, label_id)` — effectively a bitmap index on the label — and is fast for the same reason the GIN array index was.
**List threads with a user label (after):**
```sql
SELECT t.*
FROM crm_email_threads t
JOIN crm_thread_labels tl
ON tl.thread_id = t.thread_id
AND tl.connection_id = t.connection_id
WHERE t.connection_id = $1
AND tl.label_id = 'Label_3691628123'
ORDER BY t.latest_message_at DESC
LIMIT 30;
```
No name lookup needed — the labelId is stable even when the user renames it.
**Label rename (before):** had to touch every thread row's `labels` array.
**Label rename (after):**
```sql
UPDATE crm_labels
SET name = 'clients', updated_at = now()
WHERE label_id = 'Label_3691628123'
AND connection_id = $1;
```
One row. No thread rows touched.
**Get all labels for a thread:**
```sql
SELECT l.*
FROM crm_labels l
JOIN crm_thread_labels tl
ON tl.label_id = l.label_id
AND tl.connection_id = l.connection_id
WHERE tl.thread_id = $1
AND tl.connection_id = $2;
```
### Write path changes
#### `upsertEmailThread()` — label writes
Instead of writing `labels: Array.from(allLabels)` to the thread row, split the write:
1. **Upsert each label into `crm_labels`** (name is stable at sync time — it comes from the `labelLookup` map that `getThread()` builds from the Gmail labels API):
```typescript
for (const tag of allTags) {
await db
.insert(crmLabels)
.values({
labelId: tag.id,
connectionId,
name: tag.name,
type: tag.type,
color: tag.color ?? null,
})
.onConflictDoUpdate({
target: [crmLabels.labelId, crmLabels.connectionId],
set: {
name: sql`excluded.name`,
color: sql`excluded.color`,
updatedAt: sql`now()`,
},
});
}
```
2. **Delete stale join rows and insert current ones** (replace-style):
```typescript
await db
.delete(crmThreadLabels)
.where(
and(
eq(crmThreadLabels.threadId, threadId),
eq(crmThreadLabels.connectionId, connectionId),
),
);
await db
.insert(crmThreadLabels)
.values(
allTags.map((tag) => ({
threadId,
connectionId,
labelId: tag.id,
})),
)
.onConflictDoNothing();
```
#### `patchEmailThreadLabels()` — label changes
Instead of the SQL array arithmetic, write inserts/deletes to `crm_thread_labels`:
```typescript
export async function patchThreadLabels(
db: DB,
threadId: string,
connectionId: string,
addLabelIds: string[],
removeLabelIds: string[],
): Promise<void> {
await Promise.all([
// Remove
removeLabelIds.length > 0
? db
.delete(crmThreadLabels)
.where(
and(
eq(crmThreadLabels.threadId, threadId),
eq(crmThreadLabels.connectionId, connectionId),
inArray(crmThreadLabels.labelId, removeLabelIds),
),
)
: Promise.resolve(),
// Add (insert or ignore if already present)
addLabelIds.length > 0
? db
.insert(crmThreadLabels)
.values(addLabelIds.map((labelId) => ({ threadId, connectionId, labelId })))
.onConflictDoNothing()
: Promise.resolve(),
]);
// Keep the denormalized booleans on crm_email_threads in sync
if (addLabelIds.includes('UNREAD') || removeLabelIds.includes('UNREAD') ||
addLabelIds.includes('STARRED') || removeLabelIds.includes('STARRED')) {
await db
.update(crmEmailThreads)
.set({
hasUnread: sql`
EXISTS (
SELECT 1 FROM crm_thread_labels
WHERE thread_id = ${threadId}
AND connection_id = ${connectionId}
AND label_id = 'UNREAD'
)
`,
isStarred: sql`
EXISTS (
SELECT 1 FROM crm_thread_labels
WHERE thread_id = ${threadId}
AND connection_id = ${connectionId}
AND label_id = 'STARRED'
)
`,
updatedAt: sql`now()`,
})
.where(
and(
eq(crmEmailThreads.threadId, threadId),
eq(crmEmailThreads.connectionId, connectionId),
),
);
}
}
```
### Label sync on connection setup / periodic refresh
When a user connects their Gmail account (or on a periodic background job), sync the full label list into `crm_labels`:
```typescript
async function syncLabels(db: DB, connectionId: string, manager: GoogleMailManager) {
const gmailLabels = await manager.getUserLabels();
for (const label of gmailLabels) {
await db
.insert(crmLabels)
.values({
labelId: label.id,
connectionId,
name: label.name,
type: label.type,
color: label.color ?? null,
})
.onConflictDoUpdate({
target: [crmLabels.labelId, crmLabels.connectionId],
set: {
name: sql`excluded.name`,
color: sql`excluded.color`,
updatedAt: sql`now()`,
},
});
}
// Delete labels that no longer exist in Gmail
const activeIds = gmailLabels.map((l) => l.id);
await db
.delete(crmLabels)
.where(
and(
eq(crmLabels.connectionId, connectionId),
notInArray(crmLabels.labelId, activeIds),
),
);
// CASCADE on crm_thread_labels handles cleanup of join rows automatically
}
```
Call `syncLabels()`:
- On initial connection setup
- Inside the Gmail push notification handler when `labelIds` in the history payload include an ID not yet in `crm_labels`
- As a periodic background task (once per hour is sufficient — label renames are rare)
### `folderToLabelFilter` changes
The function currently converts a folder string to a label name. With the new schema it converts to a **label ID**. For system labels the ID equals the name (`'INBOX'`, `'SENT'`, etc.) so no change is needed there. For user labels, the caller must pass a `labelId` (not a name).
The existing `labels.list` tRPC endpoint already returns both `id` and `name` to the frontend. The frontend should send `labelId` (not `name`) when filtering by a user label — which is already how the `labelIds` input param on `mail.listThreads` works.
### Implementation phases
| Phase | Change | Risk |
|---|---|---|
| A — Schema | Add `crm_labels`, `crm_thread_labels` tables; drop `labels` column from `crm_email_threads` | Low (additive except for column drop) |
| B — Write path | Update `upsertEmailThread` and `patchEmailThreadLabels` to write to new tables | Medium |
| C — Label sync | Add `syncLabels()` call on connection setup and push notification | Low |
| D — Read path | Update `queryThreadsFromDB` to JOIN on `crm_thread_labels` | Medium |
| E — Backfill | For existing threads: insert label rows from the tag data in S3 snapshots | One-time |
Column drop in Phase A should be done in a separate migration from Phase B going live, to allow rollback without data loss.