import OpenAI from 'openai'; import { query, queryOne } from '../db'; import { logger } from '../logger'; import { compileTemplate, TemplateData } from './handlebars'; const MODEL = process.env.OPENAI_ENRICHMENT_MODEL || 'gpt-4o'; export interface CampaignSettingsRow { enrichment_tone: string | null; enrichment_instructions: string | null; } /** Build OpenAI client; throws if key missing when called. */ function getClient(): OpenAI { const key = process.env.OPENAI_API_KEY; if (!key) throw new Error('OPENAI_API_KEY is not set'); return new OpenAI({ apiKey: key }); } /** Run Handlebars merge for one contact to get “original” email before LLM rewrite. */ export function buildOriginalEmail( compiledHtml: string, subject: string, contact: { email: string; first_name: string | null; last_name: string | null; custom_fields: Record; enrichment_data?: Record; }, companyName: string ): { original_subject: string; original_body: string } { const templateData: TemplateData = { ...contact.custom_fields, ...(contact.enrichment_data || {}), first_name: contact.first_name, last_name: contact.last_name, email: contact.email, unsubscribe_url: '#', company_name: companyName, }; return { original_subject: compileTemplate(subject, templateData), original_body: compileTemplate(compiledHtml, templateData), }; } /** * Call GPT-4o to rewrite the email as a personal 1:1 message; returns HTML body + subject. */ export async function enrichEmailWithOpenAI( originalSubject: string, originalBodyHtml: string, settings: CampaignSettingsRow, contactSummary: string ): Promise<{ enriched_subject: string; enriched_body: string }> { const client = getClient(); const tone = settings.enrichment_tone || 'professional and warm'; const instructions = settings.enrichment_instructions || ''; const system = `You rewrite marketing-style emails into short, personal one-to-one messages. Preserve factual claims and links from the original when appropriate, but change tone to feel like a real person wrote it (not a bulk marketing blast). Output valid JSON only with keys: enriched_subject (string), enriched_body (string). The enriched_body must be a complete HTML fragment suitable for email (no / wrapper required; you may use

, ,
, simple inline styles). Do not include unsubscribe links or List-Unsubscribe wording unless the original explicitly requires it for legal reasons — the product handles opt-out via reply.`; const user = `Tone: ${tone} ${instructions ? `Extra instructions from the sender:\n${instructions}\n` : ''} Recipient / context (JSON):\n${contactSummary} Original subject:\n${originalSubject} Original body (HTML):\n${originalBodyHtml} Return JSON: {"enriched_subject":"...","enriched_body":"..."}`; const completion = await client.chat.completions.create({ model: MODEL, response_format: { type: 'json_object' }, messages: [ { role: 'system', content: system }, { role: 'user', content: user }, ], temperature: 0.7, max_tokens: 4096, }); const raw = completion.choices[0]?.message?.content; if (!raw) throw new Error('Empty OpenAI response'); const parsed = JSON.parse(raw) as { enriched_subject?: string; enriched_body?: string }; if (!parsed.enriched_subject || !parsed.enriched_body) { throw new Error('Invalid enrichment JSON from model'); } return { enriched_subject: parsed.enriched_subject.trim(), enriched_body: parsed.enriched_body.trim(), }; } /** Keyword pass for unsubscribe intent. */ const UNSUB_KEYWORDS = /\b(unsubscribe|opt\s*out|remove\s*me|stop\s*email|stop\s*mailing)\b/i; export function keywordUnsubscribeIntent(body: string): boolean { return UNSUB_KEYWORDS.test(body || ''); } /** * LLM fallback: returns 'unsubscribe' | 'reply' | 'other' */ export async function classifyReplyWithOpenAI(subject: string, body: string): Promise<'unsubscribe' | 'reply' | 'other'> { const client = getClient(); const completion = await client.chat.completions.create({ model: MODEL, response_format: { type: 'json_object' }, messages: [ { role: 'system', content: 'Classify the email reply. Return JSON: {"intent":"unsubscribe"|"reply"|"other"}. unsubscribe = user wants no more email. reply = normal conversational reply. other = unclear.', }, { role: 'user', content: `Subject: ${subject}\n\nBody:\n${body.slice(0, 8000)}`, }, ], temperature: 0, max_tokens: 64, }); const raw = completion.choices[0]?.message?.content; if (!raw) return 'other'; try { const j = JSON.parse(raw) as { intent?: string }; const i = j.intent; if (i === 'unsubscribe' || i === 'reply' || i === 'other') return i; } catch { /* ignore */ } return 'other'; } /** Keyword match first; LLM for ambiguous wording. */ export async function classifyReplyIntent( subject: string, body: string ): Promise<'unsubscribe' | 'reply' | 'other'> { if (keywordUnsubscribeIntent(body) || keywordUnsubscribeIntent(subject)) return 'unsubscribe'; return classifyReplyWithOpenAI(subject, body); } /** Optional LLM polish for auto-reply unsubscribe message. */ export async function enrichPlainTextReply(template: string, context: Record): Promise { const client = getClient(); const completion = await client.chat.completions.create({ model: MODEL, response_format: { type: 'json_object' }, messages: [ { role: 'system', content: 'You improve short email replies. Output JSON {"text":"..."} only. Keep it brief, polite, and human.', }, { role: 'user', content: `Template:\n${template}\n\nContext:\n${JSON.stringify(context)}`, }, ], temperature: 0.5, max_tokens: 512, }); const raw = completion.choices[0]?.message?.content; if (!raw) return template; try { const j = JSON.parse(raw) as { text?: string }; if (j.text?.trim()) return j.text.trim(); } catch { /* ignore */ } return template; } export async function loadCompanyName(): Promise { const row = await queryOne<{ value: unknown }>('SELECT value FROM app_settings WHERE key = $1', ['company_name']); return (row?.value as string) || 'My Company'; } /** * Background job: enrich all subscribed contacts on the campaign list. */ export async function runEnrichmentJob(campaignId: string): Promise { const settings = await queryOne<{ enrichment_enabled: boolean } & CampaignSettingsRow>( `SELECT enrichment_enabled, enrichment_tone, enrichment_instructions FROM campaign_settings WHERE campaign_id = $1`, [campaignId] ); if (!settings) return; if (!settings.enrichment_enabled) { logger.warn({ campaignId }, '[enrichment] disabled'); return; } const campaign = await queryOne<{ id: string; subject: string; list_id: string; compiled_html: string | null; }>( `SELECT c.id, c.subject, c.list_id, t.compiled_html FROM campaigns c LEFT JOIN templates t ON t.id = c.template_id WHERE c.id = $1`, [campaignId] ); if (!campaign?.compiled_html) { logger.error({ campaignId }, '[enrichment] no template HTML'); return; } const campaignRow = campaign; const settingsRow = settings; const companyName = await loadCompanyName(); interface ContactRow { id: string; email: string; first_name: string | null; last_name: string | null; custom_fields: Record; enrichment_data: Record; } const contacts = await query( `SELECT c.id, c.email, c.first_name, c.last_name, c.custom_fields, c.enrichment_data FROM list_contacts lc JOIN contacts c ON c.id = lc.contact_id WHERE lc.list_id = $1 AND c.subscribed = true`, [campaign.list_id] ); const CONCURRENCY = 5; let idx = 0; async function worker() { while (idx < contacts.length) { const i = idx++; const contact = contacts[i]; const { original_subject, original_body } = buildOriginalEmail( campaignRow.compiled_html!, campaignRow.subject, contact, companyName ); await query( `INSERT INTO campaign_enriched_emails (campaign_id, contact_id, original_subject, original_body, status) VALUES ($1, $2, $3, $4, 'pending_review') ON CONFLICT (campaign_id, contact_id) DO UPDATE SET original_subject = EXCLUDED.original_subject, original_body = EXCLUDED.original_body, status = 'pending_review', enriched_subject = NULL, enriched_body = NULL, reviewed_at = NULL, sent_at = NULL, outbound_message_id = NULL, error_message = NULL, updated_at = now()`, [campaignId, contact.id, original_subject, original_body] ); try { const summary = JSON.stringify({ email: contact.email, first_name: contact.first_name, last_name: contact.last_name, custom_fields: contact.custom_fields, enrichment_data: contact.enrichment_data || {}, }); const { enriched_subject, enriched_body } = await enrichEmailWithOpenAI( original_subject, original_body, settingsRow, summary ); await query( `UPDATE campaign_enriched_emails SET enriched_subject = $1, enriched_body = $2, status = 'pending_review', error_message = NULL, updated_at = now() WHERE campaign_id = $3 AND contact_id = $4`, [enriched_subject, enriched_body, campaignId, contact.id] ); } catch (err) { const msg = err instanceof Error ? err.message : String(err); logger.error({ email: contact.email, campaignId, msg }, '[enrichment] contact failed'); await query( `UPDATE campaign_enriched_emails SET error_message = $1, status = 'failed', updated_at = now() WHERE campaign_id = $2 AND contact_id = $3`, [msg, campaignId, contact.id] ); } } } await Promise.all(Array.from({ length: Math.min(CONCURRENCY, contacts.length) }, () => worker())); }