From af0007fd9ea0f3e33af6226adddb4b040f4167d1 Mon Sep 17 00:00:00 2001 From: Dries Augustyns Date: Wed, 28 Oct 2026 19:02:49 +0100 Subject: [PATCH] perf(contacts): serve the contact field list from Redis GET /contacts/fields scanned every contact in the project on each request, re-reading the table once per custom field. At millions of contacts the segment builder took a minute to open (#487). The field names are now registered on write: every path that stores Contact.data adds its keys to a per-project sorted set in Redis, so a new field is offered immediately. Coverage and type are measured by a background contact-fields job in a single pass over the project's contacts, which also prunes keys no contact carries any more. The endpoint only reads Redis and queues a scan when the measurements are missing, older than an hour, or lack a field. Write-triggered scans are debounced by 10 seconds and a project is scanned at most every 5 minutes. A field not measured yet reports coverage as null. A stored JSON null no longer counts toward coverage, and a field holding mixed types reports as string. The segment builder loads its options through SWR. Based on the analysis and single-pass query from #490. Co-authored-by: Pau Sanchez Co-Authored-By: Claude Opus 5.5 --- apps/api/src/controllers/Contacts.ts | 8 +- .../controllers/__tests__/Workflows.test.ts | 6 +- apps/api/src/jobs/contact-fields-processor.ts | 45 +++ apps/api/src/jobs/worker.ts | 6 + apps/api/src/services/ContactFieldService.ts | 257 ++++++++++++++++++ apps/api/src/services/ContactService.ts | 124 ++------- apps/api/src/services/QueueService.ts | 19 ++ .../src/services/WorkflowExecutionService.ts | 5 + apps/api/src/services/WorkflowService.ts | 4 +- .../__tests__/ContactFieldService.test.ts | 133 +++++++++ .../services/__tests__/EventService.test.ts | 2 + .../__tests__/WorkflowEntryService.test.ts | 2 + .../__tests__/WorkflowService.test.ts | 2 + apps/api/src/services/keys.ts | 15 + .../src/components/DataManagementSettings.tsx | 11 +- .../components/EmailEditor/LogicMention.ts | 5 +- .../components/EmailEditor/VariableMention.ts | 4 +- .../src/components/SegmentFilterBuilder.tsx | 143 +++++----- apps/web/src/lib/hooks/useContacts.ts | 15 +- apps/web/src/lib/templateLint.ts | 2 +- packages/types/src/api/contact.ts | 24 ++ packages/types/src/jobs/maintenance.ts | 8 + 22 files changed, 630 insertions(+), 210 deletions(-) create mode 100644 apps/api/src/jobs/contact-fields-processor.ts create mode 100644 apps/api/src/services/ContactFieldService.ts create mode 100644 apps/api/src/services/__tests__/ContactFieldService.test.ts diff --git a/apps/api/src/controllers/Contacts.ts b/apps/api/src/controllers/Contacts.ts index f2e4a595a..560faa424 100644 --- a/apps/api/src/controllers/Contacts.ts +++ b/apps/api/src/controllers/Contacts.ts @@ -8,6 +8,7 @@ import signale from 'signale'; import {requireAuth, requireEmailVerified} from '../middleware/auth.js'; import {contactWriteRateLimit} from '../middleware/rateLimit.js'; import {HttpException} from '../exceptions/index.js'; +import {ContactFieldService} from '../services/ContactFieldService.js'; import {ContactService} from '../services/ContactService.js'; import {QueueService} from '../services/QueueService.js'; import {CatchAsync} from '../utils/asyncHandler.js'; @@ -92,12 +93,7 @@ export class Contacts { const auth = res.locals.auth; try { - const fieldsWithTypes = await ContactService.getAvailableFields(auth.projectId!); - - return res.status(200).json({ - fields: fieldsWithTypes, - count: fieldsWithTypes.length, - }); + return res.status(200).json(await ContactFieldService.list(auth.projectId!)); } catch (error) { signale.error('[CONTACTS] Failed to get available fields:', error); return res.status(500).json({ diff --git a/apps/api/src/controllers/__tests__/Workflows.test.ts b/apps/api/src/controllers/__tests__/Workflows.test.ts index 0077cfe15..07626e5d9 100644 --- a/apps/api/src/controllers/__tests__/Workflows.test.ts +++ b/apps/api/src/controllers/__tests__/Workflows.test.ts @@ -1,5 +1,6 @@ import {beforeEach, describe, expect, it} from 'vitest'; import {factories} from '../../../../../test/helpers'; +import {ContactService} from '../../services/ContactService'; import {EventService} from '../../services/EventService'; import {WorkflowService} from '../../services/WorkflowService'; @@ -16,8 +17,9 @@ describe('Workflows Controller', () => { // ======================================== describe('GET /workflows/fields', () => { it('should return contact and event fields', async () => { - const contact = await factories.createContact({ - projectId, + // Through the service: custom fields are registered by contact writes + const contact = await ContactService.create(projectId, { + email: 'john@example.com', data: { firstName: 'John', lastName: 'Doe', diff --git a/apps/api/src/jobs/contact-fields-processor.ts b/apps/api/src/jobs/contact-fields-processor.ts new file mode 100644 index 000000000..8caef61f3 --- /dev/null +++ b/apps/api/src/jobs/contact-fields-processor.ts @@ -0,0 +1,45 @@ +/** + * Contact Fields Worker + * Measures the coverage and type of a project's custom contact fields. Jobs are queued by + * ContactFieldService when a read finds the measurements missing or stale, or when a write + * introduces a new key. + */ + +import type {ContactFieldsJobData} from '@plunk/types'; +import {DelayedError, type Job, Worker} from 'bullmq'; +import signale from 'signale'; + +import {ContactFieldService, MIN_SCAN_INTERVAL_MS} from '../services/ContactFieldService.js'; +import {contactFieldsQueue} from '../services/QueueService.js'; + +async function processScan(job: Job, token?: string): Promise { + const {projectId} = job.data; + + // A project that keeps writing new keys would otherwise be rescanned every few seconds. + // Push the job back instead; it keeps its id, so further requests still collapse onto it. + const lastScannedAt = await ContactFieldService.lastScannedAt(projectId); + if (lastScannedAt !== null && Date.now() - lastScannedAt < MIN_SCAN_INTERVAL_MS) { + await job.moveToDelayed(lastScannedAt + MIN_SCAN_INTERVAL_MS, token); + throw new DelayedError(); + } + + await ContactFieldService.scan(projectId); +} + +export function createContactFieldsWorker(): Worker { + const worker = new Worker(contactFieldsQueue.name, processScan, { + connection: contactFieldsQueue.opts.connection, + // Each job is a full pass over a project's contacts; keep a couple at most in flight. + concurrency: 2, + }); + + worker.on('failed', (job, error) => { + signale.error(`[CONTACT-FIELDS] Job ${job?.id} failed:`, error); + }); + + worker.on('error', error => { + signale.error('[CONTACT-FIELDS] Worker error:', error); + }); + + return worker; +} diff --git a/apps/api/src/jobs/worker.ts b/apps/api/src/jobs/worker.ts index 44951edca..756ecd3fc 100644 --- a/apps/api/src/jobs/worker.ts +++ b/apps/api/src/jobs/worker.ts @@ -16,6 +16,7 @@ import {createBulkContactWorker} from './bulk-contact-processor.js'; import {createCampaignCancelCleanupWorker} from './campaign-cancel-cleanup-processor.js'; import {createCampaignWorker} from './campaign-processor.js'; import {createCardVerificationWorker} from './card-verification-processor.js'; +import {createContactFieldsWorker} from './contact-fields-processor.js'; import {createCampaignStatsSweepWorker} from './campaign-stats-sweep-processor.js'; import {createScheduledEmailSweepWorker} from './scheduled-email-sweep-processor.js'; import {createSnoozeSweepWorker} from './snooze-sweep-processor.js'; @@ -90,6 +91,11 @@ async function startWorkers() { workers.push({name: 'segment-count', worker: segmentCountWorker}); signale.success('[WORKER] Segment count worker started'); + // Start contact fields worker + const contactFieldsWorker = createContactFieldsWorker(); + workers.push({name: 'contact-fields', worker: contactFieldsWorker}); + signale.success('[WORKER] Contact fields worker started'); + // Start domain verification worker const domainVerificationWorker = createDomainVerificationWorker(); workers.push({name: 'domain-verification', worker: domainVerificationWorker}); diff --git a/apps/api/src/services/ContactFieldService.ts b/apps/api/src/services/ContactFieldService.ts new file mode 100644 index 000000000..fc0b7bc33 --- /dev/null +++ b/apps/api/src/services/ContactFieldService.ts @@ -0,0 +1,257 @@ +import type {ContactField, ContactFieldType, ContactFieldsResponse} from '@plunk/types'; +import signale from 'signale'; + +import {prisma} from '../database/prisma.js'; +import {redis} from '../database/redis.js'; +import {Keys} from './keys.js'; +import {contactFieldsQueue} from './QueueService.js'; + +/** + * Add keys to a project's field registry, scored by when Redis first saw them. + * Returns how many were new. Redis' clock rather than the caller's, so the scan's prune + * below compares timestamps from a single source. + */ +const RECORD_KEYS_SCRIPT = ` +local t = redis.call('TIME') +local now = tonumber(t[1]) * 1000 + math.floor(tonumber(t[2]) / 1000) +local added = 0 +for i = 1, #ARGV do + added = added + redis.call('ZADD', KEYS[1], 'NX', now, ARGV[i]) +end +return added +`; + +/** + * Store a scan's result. Keys found by the scan are added; keys first seen before the scan + * started that it did not find are gone from every contact and are pruned. Keys first seen + * during the scan are kept, since the scan may have missed the write that introduced them. + * + * KEYS[1] = field keys, KEYS[2] = field stats + * ARGV[1] = scan start (ms), ARGV[2] = stats JSON, ARGV[3] = TTL (s), ARGV[4..] = found keys + */ +const STORE_SCAN_SCRIPT = ` +local found = {} +for i = 4, #ARGV do + found[ARGV[i]] = true + redis.call('ZADD', KEYS[1], 'NX', ARGV[1], ARGV[i]) +end +for _, key in ipairs(redis.call('ZRANGEBYSCORE', KEYS[1], '-inf', '(' .. ARGV[1])) do + if not found[key] then + redis.call('ZREM', KEYS[1], key) + end +end +redis.call('SET', KEYS[2], ARGV[2], 'EX', ARGV[3]) +redis.call('EXPIRE', KEYS[1], ARGV[3]) +return 0 +`; + +interface ContactFieldCommands { + contactFieldsRecord(keysKey: string, ...dataKeys: string[]): Promise; + contactFieldsStoreScan( + keysKey: string, + statsKey: string, + scanStart: number, + stats: string, + ttl: number, + ...dataKeys: string[] + ): Promise; +} + +redis.defineCommand('contactFieldsRecord', {numberOfKeys: 1, lua: RECORD_KEYS_SCRIPT}); +redis.defineCommand('contactFieldsStoreScan', {numberOfKeys: 2, lua: STORE_SCAN_SCRIPT}); + +const client = redis as unknown as ContactFieldCommands; + +interface FieldStats { + scannedAt: number; + fields: Record; +} + +const STANDARD_FIELDS: ContactField[] = [ + {field: 'email', type: 'string', coverage: 100}, + {field: 'subscribed', type: 'boolean', coverage: 100}, + {field: 'createdAt', type: 'date', coverage: 100}, + {field: 'updatedAt', type: 'date', coverage: 100}, +]; + +const ISO_DATE = /^\d{4}-\d{2}-\d{2}(T\d{2}:\d{2}:\d{2}(\.\d{3})?Z?)?$/; + +/** Coverage drifts slowly, so a read re-measures it at most this often. */ +const STALE_AFTER_MS = 60 * 60 * 1000; + +/** Minimum gap between two scans of one project, however many new keys arrive. */ +export const MIN_SCAN_INTERVAL_MS = 5 * 60 * 1000; + +/** Lets a burst of writes that introduce keys (an import) collapse onto one scan. */ +const WRITE_DEBOUNCE_MS = 10 * 1000; + +/** Projects nobody looks at drop out of Redis; the next read rescans. */ +const TTL_SECONDS = 30 * 24 * 60 * 60; + +/** + * The fields a project's contacts can be filtered and personalised on. + * + * Discovering the keys inside `Contact.data` needs a full scan of the project's contacts, + * which takes seconds at millions of rows. So the key list is kept up to date on write, and + * coverage and type are measured by a background job and read from Redis. The endpoint never + * touches the contacts table. + */ +export class ContactFieldService { + public static async list(projectId: string): Promise { + const [keys, rawStats] = await Promise.all([ + redis.zrange(Keys.Contact.fieldKeys(projectId), 0, -1), + redis.get(Keys.Contact.fieldStats(projectId)), + ]); + const stats = this.parseStats(rawStats); + + if (!stats || Date.now() - stats.scannedAt > STALE_AFTER_MS || keys.some(key => !stats.fields[key])) { + await this.scheduleScan(projectId, 0); + } + + const customFields: ContactField[] = keys.map(key => ({ + field: `data.${key}`, + type: stats?.fields[key]?.type ?? 'string', + coverage: stats?.fields[key]?.coverage ?? null, + })); + const fields = [...STANDARD_FIELDS, ...customFields].sort((a, b) => a.field.localeCompare(b.field)); + + return {fields, count: fields.length, pending: !stats}; + } + + /** + * Register the `Contact.data` keys a write stored. A no-op for known keys; a new one is + * offered from the next read and gets its coverage measured shortly after. + * + * Never throws: a Redis blip must not fail a contact write, and the next scan picks up + * whatever was missed. + */ + public static async recordKeys(projectId: string, dataKeys: string[]): Promise { + if (dataKeys.length === 0) { + return; + } + + try { + const added = await client.contactFieldsRecord(Keys.Contact.fieldKeys(projectId), ...dataKeys); + + if (added > 0) { + await this.scheduleScan(projectId, WRITE_DEBOUNCE_MS); + } + } catch (error) { + signale.warn(`[CONTACT-FIELDS] Could not record fields for project ${projectId}:`, error); + } + } + + /** The keys of a `Contact.data` value, if it is an object. */ + public static keysOf(data: unknown): string[] { + return data !== null && typeof data === 'object' && !Array.isArray(data) ? Object.keys(data) : []; + } + + /** Stop offering a key that was removed from every contact. */ + public static async forget(projectId: string, key: string): Promise { + await redis.zrem(Keys.Contact.fieldKeys(projectId), key); + } + + /** When the project was last scanned, or null if never. */ + public static async lastScannedAt(projectId: string): Promise { + return this.parseStats(await redis.get(Keys.Contact.fieldStats(projectId)))?.scannedAt ?? null; + } + + /** + * Measure every `Contact.data` key of the project in one pass over its contacts, and store + * the result. Run by the contact-fields worker only. + */ + public static async scan(projectId: string): Promise { + const [seconds, micros] = await redis.time(); + const scanStart = Number(seconds) * 1000 + Math.floor(Number(micros) / 1000); + + const [total, rows] = await Promise.all([ + prisma.contact.count({where: {projectId}}), + prisma.$queryRaw< + Array<{ + key: string; + contacts: bigint; + type_min: string | null; + type_max: string | null; + sample_min: string | null; + sample_max: string | null; + }> + >` + SELECT + kv.key, + count(*) FILTER (WHERE jsonb_typeof(kv.value) <> 'null') AS contacts, + min(jsonb_typeof(kv.value)) FILTER (WHERE jsonb_typeof(kv.value) <> 'null') AS type_min, + max(jsonb_typeof(kv.value)) FILTER (WHERE jsonb_typeof(kv.value) <> 'null') AS type_max, + min(kv.value #>> '{}') FILTER (WHERE jsonb_typeof(kv.value) = 'string') AS sample_min, + max(kv.value #>> '{}') FILTER (WHERE jsonb_typeof(kv.value) = 'string') AS sample_max + FROM contacts c + CROSS JOIN LATERAL jsonb_each(CASE WHEN jsonb_typeof(c.data) = 'object' THEN c.data ELSE '{}'::jsonb END) kv + WHERE c."projectId" = ${projectId} + GROUP BY kv.key + `, + ]); + + const stats: FieldStats = {scannedAt: scanStart, fields: {}}; + + for (const row of rows) { + stats.fields[row.key] = { + type: this.inferType(row.type_min, row.type_max, row.sample_min, row.sample_max), + coverage: total > 0 ? Math.round((Number(row.contacts) / total) * 100) : 0, + }; + } + + await client.contactFieldsStoreScan( + Keys.Contact.fieldKeys(projectId), + Keys.Contact.fieldStats(projectId), + scanStart, + JSON.stringify(stats), + TTL_SECONDS, + ...rows.map(row => row.key), + ); + } + + /** + * A field whose non-null values all share one JSON type gets that type; a mix falls back + * to string, the only operator set that works across all of it. + */ + private static inferType( + typeMin: string | null, + typeMax: string | null, + sampleMin: string | null, + sampleMax: string | null, + ): ContactFieldType { + if (typeMin === null || typeMin !== typeMax) { + return 'string'; + } + + if (typeMin === 'boolean' || typeMin === 'number') { + return typeMin; + } + + if (typeMin === 'string' && sampleMin && sampleMax && ISO_DATE.test(sampleMin) && ISO_DATE.test(sampleMax)) { + return 'date'; + } + + return 'string'; + } + + /** Queue a scan, collapsing onto one already queued for the project. Never throws. */ + private static async scheduleScan(projectId: string, delay: number): Promise { + try { + await contactFieldsQueue.add('scan', {projectId}, {jobId: `contact-fields-${projectId}`, delay}); + } catch (error) { + signale.warn(`[CONTACT-FIELDS] Could not queue a field scan for project ${projectId}:`, error); + } + } + + private static parseStats(raw: string | null): FieldStats | null { + if (!raw) { + return null; + } + + try { + return JSON.parse(raw) as FieldStats; + } catch { + return null; + } + } +} diff --git a/apps/api/src/services/ContactService.ts b/apps/api/src/services/ContactService.ts index f651f6103..6f7159bc3 100644 --- a/apps/api/src/services/ContactService.ts +++ b/apps/api/src/services/ContactService.ts @@ -12,6 +12,7 @@ import signale from 'signale'; import {prisma} from '../database/prisma.js'; import {HttpException} from '../exceptions/index.js'; +import {ContactFieldService} from './ContactFieldService.js'; import {EventService} from './EventService.js'; export class ContactService { @@ -204,7 +205,7 @@ export class ContactService { data: {email: string; data?: Prisma.JsonValue; subscribed?: boolean}, ): Promise { try { - return await prisma.contact.create({ + const contact = await prisma.contact.create({ data: { projectId, email: this.normalizeEmail(data.email), @@ -212,6 +213,10 @@ export class ContactService { subscribed: data.subscribed ?? true, }, }); + + await ContactFieldService.recordKeys(projectId, ContactFieldService.keysOf(data.data)); + + return contact; } catch (error) { // Check if this is a unique constraint violation (P2002) if (error instanceof Error && 'code' in error && error.code === 'P2002') { @@ -306,6 +311,10 @@ export class ContactService { data: updateData, }); + if (data.data !== undefined) { + await ContactFieldService.recordKeys(projectId, ContactFieldService.keysOf(updated.data)); + } + // Track subscription event if status changed if (isSubscriptionChanging) { if (data.subscribed && !wasSubscribed) { @@ -386,6 +395,8 @@ export class ContactService { }, }); + await ContactFieldService.recordKeys(projectId, Object.keys(mergedData)); + // Track subscription event if status changed if (isSubscriptionChanging) { if (subscribed && !wasSubscribed) { @@ -405,7 +416,7 @@ export class ContactService { } } else { try { - return await prisma.contact.create({ + const created = await prisma.contact.create({ data: { projectId, email: normalizedEmail, @@ -413,6 +424,10 @@ export class ContactService { subscribed: subscribed ?? defaultSubscribed, }, }); + + await ContactFieldService.recordKeys(projectId, Object.keys(mergedData)); + + return created; } catch (error) { // Provide helpful error message for database/validation issues throw new HttpException( @@ -806,109 +821,6 @@ export class ContactService { return resumed; } - /** - * Get all available contact fields for a project - * Returns both standard fields and custom fields from the data JSON column - * Now includes type information inferred from actual data - * - * @param projectId - The project ID to filter contacts - * @returns Array of field objects with name and type - */ - public static async getAvailableFields( - projectId: string, - ): Promise> { - // Get total contact count for coverage calculation - const totalContacts = await prisma.contact.count({ - where: {projectId}, - }); - - // Standard fields with known types (always 100% coverage) - const standardFields = [ - {field: 'email', type: 'string' as const, coverage: 100}, - {field: 'subscribed', type: 'boolean' as const, coverage: 100}, - {field: 'createdAt', type: 'date' as const, coverage: 100}, - {field: 'updatedAt', type: 'date' as const, coverage: 100}, - ]; - - // Get custom fields from the data JSON column with type inference and coverage - // Use raw SQL to extract all keys, sample values, and contact counts from the JSON data column - const result = await prisma.$queryRaw< - Array<{key: string; sample_value: string; json_type: string; contact_count: bigint}> - >` - WITH field_keys AS ( - SELECT DISTINCT jsonb_object_keys(data) as key - FROM contacts - WHERE - "projectId" = ${projectId} - AND data IS NOT NULL - AND jsonb_typeof(data) = 'object' - ), - field_samples AS ( - SELECT - fk.key, - jsonb_typeof(c.data->fk.key) as json_type, - (c.data->>fk.key) as sample_value - FROM field_keys fk - CROSS JOIN LATERAL ( - SELECT data - FROM contacts - WHERE - "projectId" = ${projectId} - AND data ? fk.key - AND data->fk.key IS NOT NULL - LIMIT 1 - ) c - ), - field_counts AS ( - SELECT - fk.key, - COUNT(*) as contact_count - FROM field_keys fk - JOIN contacts c ON c."projectId" = ${projectId} - AND c.data ? fk.key - AND c.data->fk.key IS NOT NULL - GROUP BY fk.key - ) - SELECT - fs.key, - fs.sample_value, - fs.json_type, - fc.contact_count - FROM field_samples fs - JOIN field_counts fc ON fc.key = fs.key - `; - - // Infer types from JSON types and sample values, calculate coverage - const customFields = result.map(row => { - let type: 'string' | 'number' | 'boolean' | 'date' = 'string'; - - // PostgreSQL jsonb_typeof returns: "object", "array", "string", "number", "boolean", "null" - if (row.json_type === 'boolean') { - type = 'boolean'; - } else if (row.json_type === 'number') { - type = 'number'; - } else if (row.json_type === 'string' && row.sample_value) { - // Try to detect dates (ISO 8601 format) - const dateRegex = /^\d{4}-\d{2}-\d{2}(T\d{2}:\d{2}:\d{2}(\.\d{3})?Z?)?$/; - if (dateRegex.test(row.sample_value)) { - type = 'date'; - } - } - - // Calculate coverage percentage - const contactCount = Number(row.contact_count); - const coverage = totalContacts > 0 ? Math.round((contactCount / totalContacts) * 100) : 0; - - return { - field: `data.${row.key}`, - type, - coverage, - }; - }); - - return [...standardFields, ...customFields].sort((a, b) => a.field.localeCompare(b.field)); - } - /** * Get unique values for a contact field * Optimized for large datasets (1M+ contacts) - limits results and uses efficient queries @@ -1069,6 +981,8 @@ export class ContactService { AND data ? ${jsonField} `; + await ContactFieldService.forget(projectId, jsonField); + return {deletedFrom: result}; } diff --git a/apps/api/src/services/QueueService.ts b/apps/api/src/services/QueueService.ts index 34586e097..828e0d294 100644 --- a/apps/api/src/services/QueueService.ts +++ b/apps/api/src/services/QueueService.ts @@ -13,6 +13,7 @@ import type { ContactImportJobData, DomainVerificationJobData, CampaignCancelCleanupJobData, + ContactFieldsJobData, EmailBodyCleanupJobData, IdempotencyKeyCleanupJobData, MeterEventJobData, @@ -297,6 +298,24 @@ export const scheduledEmailSweepQueue = new Queue('s }, }); +/** + * Measures coverage and type of a project's custom contact fields. One job per project at a + * time: the job id is the project, so writes that discover new keys while a scan is queued + * collapse onto it. Completed and failed jobs are removed so the id frees up for the next one. + */ +export const contactFieldsQueue = new Queue('contact-fields', { + connection: redisConnection, + defaultJobOptions: { + attempts: 2, + backoff: { + type: 'exponential', + delay: 30000, + }, + removeOnComplete: true, + removeOnFail: true, + }, +}); + function emailPriorityFor(sourceType: EmailSourceType): number { switch (sourceType) { case EmailSourceType.TRANSACTIONAL: diff --git a/apps/api/src/services/WorkflowExecutionService.ts b/apps/api/src/services/WorkflowExecutionService.ts index 231199410..c43e36d42 100644 --- a/apps/api/src/services/WorkflowExecutionService.ts +++ b/apps/api/src/services/WorkflowExecutionService.ts @@ -18,6 +18,7 @@ import {prisma} from '../database/prisma.js'; import {HttpException} from '../exceptions/index.js'; import {DASHBOARD_URI} from '../app/constants.js'; +import {ContactFieldService} from './ContactFieldService.js'; import {EmailService} from './EmailService.js'; import {NtfyService} from './NtfyService.js'; import {QueueService} from './QueueService.js'; @@ -1247,6 +1248,10 @@ export class WorkflowExecutionService { }); } + if (hasDataUpdates) { + await ContactFieldService.recordKeys(execution.workflow.projectId, Object.keys(newData)); + } + if (subscriptionChanging) { const {EventService} = await import('./EventService.js'); await EventService.trackEvent( diff --git a/apps/api/src/services/WorkflowService.ts b/apps/api/src/services/WorkflowService.ts index d323d977a..253ea776e 100644 --- a/apps/api/src/services/WorkflowService.ts +++ b/apps/api/src/services/WorkflowService.ts @@ -15,7 +15,7 @@ import {prisma} from '../database/prisma.js'; import {BadRequest, ConflictError, HttpException} from '../exceptions/index.js'; import type {ListSort} from '../utils/listSort.js'; -import {ContactService} from './ContactService.js'; +import {ContactFieldService} from './ContactFieldService.js'; import {EventService} from './EventService.js'; import {NtfyService} from './NtfyService.js'; import {WorkflowEntryService} from './WorkflowEntryService.js'; @@ -1397,7 +1397,7 @@ export class WorkflowService { */ public static async getAvailableFields(projectId: string, eventName?: string) { // Get contact fields with types (standard + custom data fields) - const contactFieldsWithTypes = await ContactService.getAvailableFields(projectId); + const {fields: contactFieldsWithTypes} = await ContactFieldService.list(projectId); // Build typed field list with 'contact.' prefix const contactFields = contactFieldsWithTypes.map(f => ({ diff --git a/apps/api/src/services/__tests__/ContactFieldService.test.ts b/apps/api/src/services/__tests__/ContactFieldService.test.ts new file mode 100644 index 000000000..7d9348c12 --- /dev/null +++ b/apps/api/src/services/__tests__/ContactFieldService.test.ts @@ -0,0 +1,133 @@ +import {beforeEach, describe, expect, it, vi} from 'vitest'; +import {factories} from '../../../../../test/helpers'; +import {prisma} from '../../database/prisma'; +import {ContactFieldService} from '../ContactFieldService'; +import {ContactService} from '../ContactService'; +import {contactFieldsQueue} from '../QueueService'; + +vi.mock('../QueueService', async importOriginal => ({ + ...(await importOriginal()), + contactFieldsQueue: {add: vi.fn()}, +})); + +const queuedScans = () => vi.mocked(contactFieldsQueue.add).mock.calls; + +const customFields = async (projectId: string) => + (await ContactFieldService.list(projectId)).fields.filter(f => f.field.startsWith('data.')); + +describe('ContactFieldService', () => { + let projectId: string; + + beforeEach(async () => { + const {project} = await factories.createUserWithProject(); + projectId = project.id; + }); + + it('lists only the standard fields and queues a scan for a project never scanned', async () => { + const result = await ContactFieldService.list(projectId); + + expect(result.pending).toBe(true); + expect(result.fields.map(f => f.field)).toEqual(['createdAt', 'email', 'subscribed', 'updatedAt']); + expect(queuedScans()).toEqual([['scan', {projectId}, {jobId: `contact-fields-${projectId}`, delay: 0}]]); + }); + + it('measures coverage and type in a scan', async () => { + await factories.createContact({projectId, data: {plan: 'pro', seats: 3, trial: true, renewsAt: '2026-01-01'}}); + await factories.createContact({projectId, data: {plan: 'free', seats: '3', trial: null}}); + await factories.createContact({projectId, data: {plan: 'free'}}); + await factories.createContact({projectId}); + + await ContactFieldService.scan(projectId); + const result = await ContactFieldService.list(projectId); + + expect(result.pending).toBe(false); + expect(await customFields(projectId)).toEqual([ + {field: 'data.plan', type: 'string', coverage: 75}, + {field: 'data.renewsAt', type: 'date', coverage: 25}, + // A number on one contact and a string on another + {field: 'data.seats', type: 'string', coverage: 50}, + // A stored JSON null does not count + {field: 'data.trial', type: 'boolean', coverage: 25}, + ]); + }); + + it('does not queue another scan while the measurements are fresh and complete', async () => { + await factories.createContact({projectId, data: {plan: 'pro'}}); + await ContactFieldService.scan(projectId); + + await ContactFieldService.list(projectId); + + expect(queuedScans()).toHaveLength(0); + }); + + it('offers a field from the write that introduces it, before it is measured', async () => { + await factories.createContact({projectId, data: {plan: 'pro'}}); + await ContactFieldService.scan(projectId); + + await ContactService.upsert(projectId, 'new@example.com', {company: 'Acme'}); + + expect(await customFields(projectId)).toEqual([ + {field: 'data.company', type: 'string', coverage: null}, + {field: 'data.plan', type: 'string', coverage: 100}, + ]); + expect(queuedScans()[0]).toEqual(['scan', {projectId}, {jobId: `contact-fields-${projectId}`, delay: 10000}]); + }); + + it('does not queue a scan for writes of known fields', async () => { + await ContactService.create(projectId, {email: 'a@example.com', data: {plan: 'pro'}}); + await ContactFieldService.scan(projectId); + vi.mocked(contactFieldsQueue.add).mockClear(); + + const contact = await ContactService.create(projectId, {email: 'b@example.com', data: {plan: 'free'}}); + await ContactService.update(projectId, contact.id, {data: {plan: 'pro'}}); + await ContactService.upsert(projectId, 'c@example.com', {plan: 'free'}); + + expect(queuedScans()).toHaveLength(0); + }); + + it('prunes fields no contact carries any more', async () => { + const contact = await ContactService.create(projectId, {email: 'a@example.com', data: {plan: 'pro', legacy: 1}}); + await prisma.contact.update({where: {id: contact.id}, data: {data: {plan: 'pro'}}}); + await new Promise(resolve => setTimeout(resolve, 5)); + + await ContactFieldService.scan(projectId); + + expect((await customFields(projectId)).map(f => f.field)).toEqual(['data.plan']); + }); + + it('keeps a field first written while a scan is running', async () => { + await factories.createContact({projectId, data: {plan: 'pro'}}); + + const queryRaw = prisma.$queryRaw.bind(prisma); + vi.spyOn(prisma, '$queryRaw').mockImplementationOnce((async (...args: Parameters) => { + const rows = await queryRaw(...args); + // The write lands after the scan read past it + await ContactFieldService.recordKeys(projectId, ['company']); + return rows; + }) as typeof prisma.$queryRaw); + + await ContactFieldService.scan(projectId); + + expect((await customFields(projectId)).map(f => f.field)).toEqual(['data.company', 'data.plan']); + }); + + it('stops offering a deleted field', async () => { + await ContactService.create(projectId, {email: 'a@example.com', data: {plan: 'pro', legacy: 1}}); + await ContactFieldService.scan(projectId); + + await ContactService.deleteField(projectId, 'data.legacy'); + + expect((await customFields(projectId)).map(f => f.field)).toEqual(['data.plan']); + }); + + it('keeps projects apart', async () => { + const {project: other} = await factories.createUserWithProject(); + await ContactService.create(other.id, {email: 'a@example.com', data: {secret: 'x'}}); + await factories.createContact({projectId: other.id, data: {secret: 'x'}}); + await ContactFieldService.scan(other.id); + + await ContactFieldService.scan(projectId); + + expect(await customFields(projectId)).toEqual([]); + }); +}); diff --git a/apps/api/src/services/__tests__/EventService.test.ts b/apps/api/src/services/__tests__/EventService.test.ts index d7e66bf82..0e095b79c 100644 --- a/apps/api/src/services/__tests__/EventService.test.ts +++ b/apps/api/src/services/__tests__/EventService.test.ts @@ -9,6 +9,8 @@ vi.mock('../../database/redis', () => { const store = new Map(); return { redis: { + // ContactFieldService registers its Lua scripts at import + defineCommand: vi.fn(), get: vi.fn(async (key: string) => { const item = store.get(key); if (!item) return null; diff --git a/apps/api/src/services/__tests__/WorkflowEntryService.test.ts b/apps/api/src/services/__tests__/WorkflowEntryService.test.ts index 60689c278..81292b8e9 100644 --- a/apps/api/src/services/__tests__/WorkflowEntryService.test.ts +++ b/apps/api/src/services/__tests__/WorkflowEntryService.test.ts @@ -10,6 +10,8 @@ vi.mock('../../database/redis', () => { const store = new Map(); return { redis: { + // ContactFieldService registers its Lua scripts at import + defineCommand: vi.fn(), get: vi.fn(async (key: string) => { const item = store.get(key); if (!item) return null; diff --git a/apps/api/src/services/__tests__/WorkflowService.test.ts b/apps/api/src/services/__tests__/WorkflowService.test.ts index cee5a354c..4f9553362 100644 --- a/apps/api/src/services/__tests__/WorkflowService.test.ts +++ b/apps/api/src/services/__tests__/WorkflowService.test.ts @@ -9,6 +9,8 @@ vi.mock('../../database/redis', () => { const store = new Map(); return { redis: { + // ContactFieldService registers its Lua scripts at import + defineCommand: vi.fn(), get: vi.fn(async (key: string) => { const item = store.get(key); if (!item) return null; diff --git a/apps/api/src/services/keys.ts b/apps/api/src/services/keys.ts index a985911b2..78c956199 100644 --- a/apps/api/src/services/keys.ts +++ b/apps/api/src/services/keys.ts @@ -116,6 +116,21 @@ export const Keys = { return 'campaign:stats_dirty'; }, }, + Contact: { + /** + * Sorted set of the `Contact.data` keys a project uses, scored by when each was first + * seen. Contact writes add to it, so a new field is offered immediately; the background + * scan prunes keys no contact carries any more. See ContactFieldService. + */ + fieldKeys(projectId: string): string { + return `contact:field_keys:${projectId}`; + }, + + /** Coverage and type per data key, as measured by the last background scan. */ + fieldStats(projectId: string): string { + return `contact:field_stats:${projectId}`; + }, + }, Project: { id(id: string): string { return `project:id:${id}`; diff --git a/apps/web/src/components/DataManagementSettings.tsx b/apps/web/src/components/DataManagementSettings.tsx index bfdaea7b4..3decc07b3 100644 --- a/apps/web/src/components/DataManagementSettings.tsx +++ b/apps/web/src/components/DataManagementSettings.tsx @@ -17,18 +17,13 @@ import { TableRow, IconSpinner, } from '@plunk/ui'; +import type {ContactFieldsResponse} from '@plunk/types'; import {Database, Trash2, Zap} from 'lucide-react'; import {toast} from 'sonner'; import useSWR from 'swr'; import {useActiveProject} from '../lib/contexts/ActiveProjectProvider'; import {network} from '../lib/network'; -interface FieldData { - field: string; - type: 'string' | 'number' | 'boolean' | 'date'; - coverage: number; -} - interface FieldUsage { usedInSegments: Array<{id: string; name: string}>; usedInCampaigns: Array<{id: string; name: string}>; @@ -52,7 +47,7 @@ export function DataManagementSettings() { const [isDeleting, setIsDeleting] = useState(false); // Fetch contact fields - const {data: fieldsData, mutate: mutateFields} = useSWR<{fields: FieldData[]; count: number}>( + const {data: fieldsData, mutate: mutateFields} = useSWR( activeProject?.id ? `/contacts/fields` : null, ); @@ -169,7 +164,7 @@ export function DataManagementSettings() { {field.type} - {field.coverage}% + {field.coverage === null ? '—' : `${field.coverage}%`}