← back to Rentv 2026
pr-intelligence: lib (geo/taxonomy/normalize/dedupe/scoring/status), services, adapters, job queue + worker
14c23d86926e723bceecd1d2d529e99ae6b6cad4 · 2026-07-30 11:11:09 -0700 · Steve Abrams
Files touched
A src/pr/adapters/edgar.jsA src/pr/adapters/index.jsA src/pr/adapters/linkedin.jsA src/pr/adapters/manual.jsA src/pr/adapters/registry.jsA src/pr/adapters/rss.jsA src/pr/adapters/search.jsA src/pr/adapters/website.jsA src/pr/jobs/index.jsA src/pr/lib/dedupe.jsA src/pr/lib/geo.jsA src/pr/lib/normalize.jsA src/pr/lib/scoring.jsA src/pr/lib/status.jsA src/pr/lib/taxonomy.jsA src/pr/services/audit.jsA src/pr/services/campaigns.jsA src/pr/services/email-providers.jsA src/pr/services/importexport.jsA src/pr/services/letters.jsA src/pr/services/organizations.jsA src/pr/services/outreach.jsA src/pr/services/people.jsA src/pr/services/relationships.jsA src/pr/services/replies.jsA src/pr/services/runs.jsA src/pr/services/settings.jsA src/pr/services/sources.jsA src/pr/services/suppression.jsA src/pr/services/tasks.jsA src/pr/worker.js
Diff
commit 14c23d86926e723bceecd1d2d529e99ae6b6cad4
Author: Steve Abrams <steve@designerwallcoverings.com>
Date: Thu Jul 30 11:11:09 2026 -0700
pr-intelligence: lib (geo/taxonomy/normalize/dedupe/scoring/status), services, adapters, job queue + worker
---
src/pr/adapters/edgar.js | 78 ++++++
src/pr/adapters/index.js | 106 ++++++++
src/pr/adapters/linkedin.js | 56 +++++
src/pr/adapters/manual.js | 23 ++
src/pr/adapters/registry.js | 88 +++++++
src/pr/adapters/rss.js | 67 +++++
src/pr/adapters/search.js | 82 ++++++
src/pr/adapters/website.js | 143 +++++++++++
src/pr/jobs/index.js | 493 +++++++++++++++++++++++++++++++++++++
src/pr/lib/dedupe.js | 84 +++++++
src/pr/lib/geo.js | 50 ++++
src/pr/lib/normalize.js | 87 +++++++
src/pr/lib/scoring.js | 95 +++++++
src/pr/lib/status.js | 60 +++++
src/pr/lib/taxonomy.js | 150 +++++++++++
src/pr/services/audit.js | 41 +++
src/pr/services/campaigns.js | 80 ++++++
src/pr/services/email-providers.js | 149 +++++++++++
src/pr/services/importexport.js | 222 +++++++++++++++++
src/pr/services/letters.js | 166 +++++++++++++
src/pr/services/organizations.js | 240 ++++++++++++++++++
src/pr/services/outreach.js | 215 ++++++++++++++++
src/pr/services/people.js | 259 +++++++++++++++++++
src/pr/services/relationships.js | 63 +++++
src/pr/services/replies.js | 184 ++++++++++++++
src/pr/services/runs.js | 72 ++++++
src/pr/services/settings.js | 113 +++++++++
src/pr/services/sources.js | 71 ++++++
src/pr/services/suppression.js | 61 +++++
src/pr/services/tasks.js | 34 +++
src/pr/worker.js | 43 ++++
31 files changed, 3675 insertions(+)
diff --git a/src/pr/adapters/edgar.js b/src/pr/adapters/edgar.js
new file mode 100644
index 00000000..38f9bab4
--- /dev/null
+++ b/src/pr/adapters/edgar.js
@@ -0,0 +1,78 @@
+'use strict';
+// SEC EDGAR adapter — official government dataset for public companies & REITs.
+// Terms verified: EDGAR data is public-domain; SEC asks for a declared User-Agent
+// and ≤10 req/s (we run far below). https://www.sec.gov/os/accessing-edgar-data
+const { throttled, report, register } = require('./index');
+
+const UA = process.env.PR_EDGAR_UA || 'RENTV PR Research admin@rentv.example';
+
+async function edgarFetch(url) {
+ return throttled('edgar', async () => {
+ const r = await fetch(url, { headers: { 'User-Agent': UA, Accept: 'application/json' }, signal: AbortSignal.timeout(20000) });
+ if (!r.ok) throw new Error('EDGAR ' + r.status + ' for ' + url);
+ return r.json();
+ });
+}
+
+const adapter = register({
+ name: 'edgar',
+ kind: 'discovery',
+ configured: () => true, // public API; UA env optional but recommended
+ missingConfig: () => (process.env.PR_EDGAR_UA ? [] : ['PR_EDGAR_UA (recommended: your contact email in the UA per SEC guidance)']),
+ sourceMeta: () => ({
+ source_type: 'sec_edgar', is_primary_source: true,
+ license_or_usage_note: 'SEC EDGAR public-domain data; declared User-Agent; ≤10 req/s per SEC fair-access policy.',
+ }),
+
+ /** Company full-text search for CRE issuers (e.g. by SIC codes for real estate). */
+ async searchCompanies(query, { cursor } = {}) {
+ const from = cursor && cursor.from ? cursor.from : 0;
+ const url = `https://efts.sec.gov/LATEST/search-index?q=${encodeURIComponent(query)}&dateRange=custom&from=${from}`;
+ // The stable public endpoint is the full-text search UI API:
+ const searchUrl = `https://efts.sec.gov/LATEST/search-index?q=${encodeURIComponent(query)}`;
+ const errors = [];
+ let items = [];
+ try {
+ const j = await edgarFetch(`https://www.sec.gov/cgi-bin/browse-edgar?action=getcompany&company=${encodeURIComponent(query)}&type=10-K&dateb=&owner=include&count=40&output=atom`);
+ // browse-edgar with output=atom returns XML not JSON; fall through to tickers file below.
+ void j;
+ } catch { /* expected for atom output; use the tickers file */ }
+ try {
+ // company_tickers.json: complete public-company registry (name, ticker, CIK).
+ const all = await edgarFetch('https://www.sec.gov/files/company_tickers.json');
+ const q = query.toLowerCase();
+ items = Object.values(all)
+ .filter((c) => c.title && c.title.toLowerCase().includes(q))
+ .slice(from, from + 40)
+ .map((c) => ({
+ legal_name: c.title, ticker: c.ticker, cik: String(c.cik_str).padStart(10, '0'),
+ source_url: `https://www.sec.gov/cgi-bin/browse-edgar?action=getcompany&CIK=${c.cik_str}&type=10-K`,
+ }));
+ } catch (e) { errors.push({ error: e.message, url: searchUrl }); }
+ return report({
+ provider: 'edgar', query, errors, primary_source: true, confidence_recommendation: 90,
+ cursor: items.length === 40 ? { from: from + 40 } : null, items,
+ terms_note: adapter.sourceMeta().license_or_usage_note,
+ });
+ },
+
+ /** Company profile by CIK: addresses (state presence), SIC, business description. */
+ async companyProfile(cik) {
+ const padded = String(cik).replace(/\D/g, '').padStart(10, '0');
+ const errors = []; let items = [];
+ try {
+ const j = await edgarFetch(`https://data.sec.gov/submissions/CIK${padded}.json`);
+ items = [{
+ legal_name: j.name, cik: padded, sic: j.sic, sic_description: j.sicDescription,
+ state_of_incorporation: j.stateOfIncorporation,
+ business_address: j.addresses && j.addresses.business,
+ website: null, // EDGAR does not carry websites; website adapter fills this
+ source_url: `https://data.sec.gov/submissions/CIK${padded}.json`,
+ is_reit: /real estate investment trust|6798/i.test((j.sicDescription || '') + ' ' + (j.sic || '')),
+ }];
+ } catch (e) { errors.push({ error: e.message }); }
+ return report({ provider: 'edgar', query: 'CIK ' + padded, errors, primary_source: true, confidence_recommendation: 95, items, terms_note: adapter.sourceMeta().license_or_usage_note });
+ },
+});
+
+module.exports = adapter;
diff --git a/src/pr/adapters/index.js b/src/pr/adapters/index.js
new file mode 100644
index 00000000..5bd611fc
--- /dev/null
+++ b/src/pr/adapters/index.js
@@ -0,0 +1,106 @@
+'use strict';
+// Adapter registry + shared helpers. Every adapter implements:
+// name — provider key
+// kind — 'discovery' | 'enrichment' | 'import'
+// configured() — bool (env present)
+// missingConfig() — string[] of env vars still needed
+// sourceMeta() — { source_type, is_primary_source, license_or_usage_note }
+// …and its work methods return an AdapterReport:
+// { provider, query, at, result_count, rate_limit_state, cursor, errors:[],
+// terms_note, primary_source, confidence_recommendation, items:[…] }
+//
+// Throttling: each adapter runs through `throttled(name, fn)` — a per-provider
+// minimum-interval limiter so no source is hammered regardless of caller.
+
+const LAST_CALL = new Map();
+const MIN_INTERVAL_MS = { default: 1500, website: 2000, search: 1200, edgar: 350, fdic: 1000, rss: 2000 };
+
+async function throttled(provider, fn) {
+ const min = MIN_INTERVAL_MS[provider] != null ? MIN_INTERVAL_MS[provider] : MIN_INTERVAL_MS.default;
+ const last = LAST_CALL.get(provider) || 0;
+ const wait = last + min - Date.now();
+ if (wait > 0) await new Promise((r) => setTimeout(r, wait));
+ LAST_CALL.set(provider, Date.now());
+ return fn();
+}
+
+/** Fetch with UA, timeout, and robots-aware politeness (HEAD robots.txt check is
+ * overkill per request; we honor per-host disallow of our specific paths cheaply
+ * by caching robots.txt for an hour). */
+const robotsCache = new Map(); // host -> { at, disallow: [prefixes] }
+async function robotsAllows(url) {
+ try {
+ const u = new URL(url);
+ const key = u.origin;
+ let entry = robotsCache.get(key);
+ if (!entry || Date.now() - entry.at > 3600e3) {
+ const r = await fetch(key + '/robots.txt', { headers: { 'User-Agent': UA }, signal: AbortSignal.timeout(8000) });
+ const text = r.ok ? await r.text() : '';
+ const disallow = [];
+ let applies = false;
+ for (const line of text.split('\n')) {
+ const l = line.trim();
+ if (/^user-agent:\s*\*/i.test(l)) applies = true;
+ else if (/^user-agent:/i.test(l)) applies = false;
+ else if (applies) {
+ const m = l.match(/^disallow:\s*(\S+)/i);
+ if (m) disallow.push(m[1]);
+ }
+ }
+ entry = { at: Date.now(), disallow };
+ robotsCache.set(key, entry);
+ }
+ return !entry.disallow.some((p) => p !== '/' && u.pathname.startsWith(p)) &&
+ !entry.disallow.includes('/');
+ } catch { return true; } // robots unreachable → default allow (standard practice)
+}
+
+const UA = 'RENTV-PR-Research/1.0 (+https://rentv.agentabrams.com; press-list research; contact: admin)';
+
+async function politeFetch(url, { provider = 'default', asText = true } = {}) {
+ if (!(await robotsAllows(url))) {
+ const err = new Error('robots.txt disallows ' + url);
+ err.code = 'ROBOTS_DISALLOW';
+ throw err;
+ }
+ return throttled(provider, async () => {
+ const r = await fetch(url, { headers: { 'User-Agent': UA, Accept: 'text/html,application/json,application/xml;q=0.9,*/*;q=0.8' }, redirect: 'follow', signal: AbortSignal.timeout(20000) });
+ const body = asText ? await r.text() : await r.arrayBuffer();
+ return { status: r.status, ok: r.ok, body, headers: r.headers, finalUrl: r.url };
+ });
+}
+
+function report(base) {
+ return {
+ provider: base.provider, query: base.query || null, at: new Date().toISOString(),
+ result_count: (base.items || []).length, rate_limit_state: base.rate_limit_state || {},
+ cursor: base.cursor || null, errors: base.errors || [],
+ terms_note: base.terms_note || null, primary_source: !!base.primary_source,
+ confidence_recommendation: base.confidence_recommendation != null ? base.confidence_recommendation : 40,
+ items: base.items || [],
+ };
+}
+
+function stripTags(html) {
+ return String(html || '')
+ .replace(/<script[\s\S]*?<\/script>/gi, ' ').replace(/<style[\s\S]*?<\/style>/gi, ' ')
+ .replace(/<[^>]+>/g, ' ').replace(/ /g, ' ').replace(/&/g, '&')
+ .replace(/�?39;|’|‘/g, "'").replace(/"|“|”/g, '"')
+ .replace(/\s+/g, ' ').trim();
+}
+
+const registry = {};
+function register(adapter) { registry[adapter.name] = adapter; return adapter; }
+function getAdapter(name) { return registry[name] || null; }
+function listAdapters() {
+ return Object.values(registry).map((a) => ({
+ name: a.name, kind: a.kind, configured: a.configured(), missing_config: a.missingConfig(),
+ ...a.sourceMeta(),
+ }));
+}
+
+module.exports = { throttled, politeFetch, robotsAllows, report, stripTags, register, getAdapter, listAdapters, UA };
+
+// Register the concrete adapters (requires at the bottom to avoid circular imports).
+require('./website'); require('./edgar'); require('./search'); require('./registry');
+require('./rss'); require('./linkedin'); require('./manual');
diff --git a/src/pr/adapters/linkedin.js b/src/pr/adapters/linkedin.js
new file mode 100644
index 00000000..90106d2a
--- /dev/null
+++ b/src/pr/adapters/linkedin.js
@@ -0,0 +1,56 @@
+'use strict';
+// LinkedIn adapter — STRICTLY the permitted surface:
+// 1. locate PUBLICLY INDEXED profile/company URLs via the authorized search adapter
+// 2. store URL + indexed title + indexed snippet + retrieval date
+// 3. manual URL entry / user-supplied export import (handled by importexport service)
+// It never logs in, never automates a browser session, never reuses cookies, never
+// bypasses auth or rate limits, never sends connection requests, and an indexed
+// snippet is NEVER surfaced as a verified current title without corroboration.
+const { report, register } = require('./index');
+const search = require('./search');
+const { normalizeLinkedInUrl } = require('../lib/normalize');
+
+const adapter = register({
+ name: 'linkedin',
+ kind: 'enrichment',
+ configured: () => search.configured(),
+ missingConfig: () => (search.configured() ? [] : ['PR_EXA_API_KEY or PR_BRAVE_API_KEY (LinkedIn locate runs through the authorized search API)']),
+ sourceMeta: () => ({
+ source_type: 'search_api', is_primary_source: false,
+ license_or_usage_note: 'Publicly indexed LinkedIn URLs located via an authorized search API. No logged-in scraping, no browser automation, no cookies, no connection automation. Indexed titles require corroboration before display as verified.',
+ }),
+
+ /** Locate a person's public LinkedIn URL. → items: [{linkedin_url, indexed_title, indexed_snippet, source_url}] */
+ async locatePerson({ full_name, organization_name }) {
+ const rep = await search.findLinkedIn({ personName: full_name, companyName: organization_name, kind: 'in' });
+ const items = rep.items.map((x) => ({
+ linkedin_url: normalizeLinkedInUrl(x.url),
+ indexed_title: (x.title || '').replace(/\s*[|–-]\s*LinkedIn.*$/i, '').trim(),
+ indexed_snippet: (x.snippet || '').slice(0, 400),
+ source_url: x.url,
+ status_recommendation: 'found_uncorroborated',
+ })).filter((x) => x.linkedin_url && x.linkedin_url.includes('/in/'));
+ return report({
+ provider: 'linkedin(search)', query: rep.query, errors: rep.errors,
+ confidence_recommendation: 35, items,
+ terms_note: adapter.sourceMeta().license_or_usage_note,
+ });
+ },
+
+ /** Locate a company page URL. */
+ async locateCompany({ organization_name }) {
+ const rep = await search.findLinkedIn({ companyName: organization_name, kind: 'company' });
+ const items = rep.items.map((x) => ({
+ linkedin_company_url: normalizeLinkedInUrl(x.url),
+ indexed_title: (x.title || '').trim(),
+ source_url: x.url,
+ })).filter((x) => x.linkedin_company_url && x.linkedin_company_url.includes('/company/'));
+ return report({
+ provider: 'linkedin(search)', query: rep.query, errors: rep.errors,
+ confidence_recommendation: 40, items,
+ terms_note: adapter.sourceMeta().license_or_usage_note,
+ });
+ },
+});
+
+module.exports = adapter;
diff --git a/src/pr/adapters/manual.js b/src/pr/adapters/manual.js
new file mode 100644
index 00000000..75e132af
--- /dev/null
+++ b/src/pr/adapters/manual.js
@@ -0,0 +1,23 @@
+'use strict';
+// Manual research entry — the admin typed it in. Highest-trust human source; the
+// admin is asked (in the UI) to paste the URL they verified against.
+const { report, register } = require('./index');
+
+const adapter = register({
+ name: 'manual',
+ kind: 'import',
+ configured: () => true,
+ missingConfig: () => [],
+ sourceMeta: () => ({
+ source_type: 'manual', is_primary_source: false,
+ license_or_usage_note: 'Manually entered by an admin; verification URL recorded when provided.',
+ }),
+ async record({ note, url }) {
+ return report({
+ provider: 'manual', query: note || null, confidence_recommendation: 85,
+ items: [{ note, url }], terms_note: adapter.sourceMeta().license_or_usage_note,
+ });
+ },
+});
+
+module.exports = adapter;
diff --git a/src/pr/adapters/registry.js b/src/pr/adapters/registry.js
new file mode 100644
index 00000000..1b4784b2
--- /dev/null
+++ b/src/pr/adapters/registry.js
@@ -0,0 +1,88 @@
+'use strict';
+// Official-registry adapter. Implemented live: FDIC BankFind (banks) and NCUA (credit
+// unions) — both open federal APIs with documented public reuse. State licensing
+// sources (CA DRE/DFPI/DOI, AZ DIFI/ADRE) publish datasets whose exact export terms
+// vary; those flow in through the CSV import path with a per-file usage note rather
+// than a scraper (documented in docs/CRE_PR_DATA_SOURCES.md).
+const { throttled, report, register } = require('./index');
+
+const adapter = register({
+ name: 'registry',
+ kind: 'discovery',
+ configured: () => true, // FDIC/NCUA are open, key-free APIs
+ missingConfig: () => [],
+ sourceMeta: () => ({
+ source_type: 'gov_registry', is_primary_source: true,
+ license_or_usage_note: 'FDIC BankFind Suite API / NCUA data — official U.S. government datasets, public reuse.',
+ }),
+
+ /**
+ * FDIC BankFind: active banks by state. https://banks.data.fdic.gov/docs/
+ * Returns orgs with legal name, city, county, website when present.
+ */
+ async fdicBanks(state, { cursor } = {}) {
+ const offset = cursor && cursor.offset ? cursor.offset : 0;
+ const url = `https://banks.data.fdic.gov/api/institutions?filters=${encodeURIComponent(`STALP:${state} AND ACTIVE:1`)}&fields=NAME,CITY,COUNTY,STALP,WEBADDR,CERT,BKCLASS&limit=100&offset=${offset}&format=json`;
+ const errors = []; let items = []; let total = 0;
+ try {
+ const j = await throttled('fdic', async () => {
+ const r = await fetch(url, { headers: { Accept: 'application/json' }, signal: AbortSignal.timeout(20000) });
+ if (!r.ok) throw new Error('FDIC ' + r.status);
+ return r.json();
+ });
+ total = (j.totals && j.totals.count) || 0;
+ items = (j.data || []).map((d) => {
+ const x = d.data || d;
+ return {
+ legal_name: x.NAME, city: x.CITY, county: x.COUNTY, state: x.STALP,
+ website_url: x.WEBADDR ? (String(x.WEBADDR).startsWith('http') ? x.WEBADDR : 'https://' + x.WEBADDR) : null,
+ fdic_cert: x.CERT, bank_class: x.BKCLASS,
+ organization_type: 'bank',
+ source_url: `https://banks.data.fdic.gov/bankfind-suite/bankfind?name=${encodeURIComponent(x.NAME)}`,
+ };
+ });
+ } catch (e) { errors.push({ error: e.message, url }); }
+ const next = offset + items.length;
+ return report({
+ provider: 'registry:fdic', query: `FDIC active banks in ${state}`, errors, primary_source: true,
+ confidence_recommendation: 95, cursor: next < total ? { offset: next, total } : null, items,
+ terms_note: adapter.sourceMeta().license_or_usage_note,
+ });
+ },
+
+ /**
+ * NCUA credit-union directory by state (open data API).
+ * Falls back to reporting the manual-download path if the API shape changes.
+ */
+ async ncuaCreditUnions(state, { cursor } = {}) {
+ const page = cursor && cursor.page ? cursor.page : 0;
+ // NCUA's public mapping API (used by their own Find-a-Credit-Union tool).
+ const url = `https://mapping.ncua.gov/api/CreditUnion/Search?state=${encodeURIComponent(state)}&page=${page}`;
+ const errors = []; let items = [];
+ try {
+ const j = await throttled('fdic', async () => {
+ const r = await fetch(url, { headers: { Accept: 'application/json' }, signal: AbortSignal.timeout(20000) });
+ if (!r.ok) throw new Error('NCUA ' + r.status);
+ return r.json();
+ });
+ const rows = Array.isArray(j) ? j : (j.results || j.data || []);
+ items = rows.map((x) => ({
+ legal_name: x.CU_Name || x.cuName || x.name,
+ city: x.City || x.city, state,
+ website_url: x.WebsiteUrl || x.website || null,
+ charter: x.Charter_Number || x.charterNumber || null,
+ organization_type: 'credit_union',
+ source_url: 'https://mapping.ncua.gov/',
+ })).filter((x) => x.legal_name);
+ } catch (e) {
+ errors.push({ error: e.message, url, fallback: 'Download the NCUA credit-union directory CSV from ncua.gov and use the CSV import path.' });
+ }
+ return report({
+ provider: 'registry:ncua', query: `NCUA credit unions in ${state}`, errors, primary_source: true,
+ confidence_recommendation: 95, cursor: items.length >= 20 ? { page: page + 1 } : null, items,
+ terms_note: adapter.sourceMeta().license_or_usage_note,
+ });
+ },
+});
+
+module.exports = adapter;
diff --git a/src/pr/adapters/rss.js b/src/pr/adapters/rss.js
new file mode 100644
index 00000000..690faf24
--- /dev/null
+++ b/src/pr/adapters/rss.js
@@ -0,0 +1,67 @@
+'use strict';
+// RSS / newsroom-feed adapter: watches an organization's press feed for recent
+// releases (recent-activity signal + press-release evidence for people mentioned).
+const { politeFetch, report, stripTags, register } = require('./index');
+
+function parseFeed(xml) {
+ const items = [];
+ // RSS 2.0 <item> and Atom <entry>
+ for (const m of String(xml).matchAll(/<(item|entry)[\s>]([\s\S]*?)<\/\1>/gi)) {
+ const body = m[2];
+ const pick = (tag) => {
+ const mm = body.match(new RegExp(`<${tag}[^>]*>([\\s\\S]*?)<\\/${tag}>`, 'i'));
+ return mm ? stripTags(mm[1].replace(/<!\[CDATA\[([\s\S]*?)\]\]>/g, '$1')) : null;
+ };
+ const linkAttr = body.match(/<link[^>]*href=["']([^"']+)["']/i);
+ items.push({
+ title: pick('title'),
+ url: pick('link') || (linkAttr ? linkAttr[1] : null),
+ published_at: pick('pubDate') || pick('published') || pick('updated'),
+ summary: (pick('description') || pick('summary') || '').slice(0, 300),
+ });
+ if (items.length >= 50) break;
+ }
+ return items.filter((x) => x.title);
+}
+
+const adapter = register({
+ name: 'rss',
+ kind: 'enrichment',
+ configured: () => true,
+ missingConfig: () => [],
+ sourceMeta: () => ({
+ source_type: 'press_release', is_primary_source: true,
+ license_or_usage_note: 'Organization-published RSS/Atom feed; headlines + limited summaries stored.',
+ }),
+
+ /** Fetch a newsroom feed URL (or try common feed paths off a newsroom URL). */
+ async fetchFeed(feedOrPageUrl) {
+ const errors = []; let items = []; let used = feedOrPageUrl;
+ const candidates = [feedOrPageUrl];
+ if (!/\.(xml|rss|atom)([?#]|$)/i.test(feedOrPageUrl) && !/feed/i.test(feedOrPageUrl)) {
+ const base = feedOrPageUrl.replace(/\/$/, '');
+ candidates.push(base + '/feed', base + '/rss', base + '/feed.xml', base + '/rss.xml');
+ }
+ for (const url of candidates) {
+ try {
+ const r = await politeFetch(url, { provider: 'rss' });
+ if (!r.ok) continue;
+ const parsed = parseFeed(r.body);
+ if (parsed.length) { items = parsed; used = url; break; }
+ // maybe an HTML page advertising its feed
+ const alt = String(r.body).match(/<link[^>]*type=["']application\/(rss|atom)\+xml["'][^>]*href=["']([^"']+)["']/i);
+ if (alt) {
+ const feedUrl = new URL(alt[2], url).toString();
+ const rf = await politeFetch(feedUrl, { provider: 'rss' });
+ if (rf.ok) { items = parseFeed(rf.body); used = feedUrl; if (items.length) break; }
+ }
+ } catch (e) { errors.push({ url, error: e.message }); }
+ }
+ return report({
+ provider: 'rss', query: used, errors, primary_source: true, confidence_recommendation: 75,
+ items, terms_note: adapter.sourceMeta().license_or_usage_note,
+ });
+ },
+});
+
+module.exports = adapter;
diff --git a/src/pr/adapters/search.js b/src/pr/adapters/search.js
new file mode 100644
index 00000000..ccbe29bc
--- /dev/null
+++ b/src/pr/adapters/search.js
@@ -0,0 +1,82 @@
+'use strict';
+// Authorized web-search API adapter. We NEVER scrape search-result pages; this uses a
+// proper search API. Two backends are supported out of the box:
+// - Exa (PR_EXA_API_KEY) https://api.exa.ai/search
+// - Brave (PR_BRAVE_API_KEY) https://api.search.brave.com/res/v1/web/search
+// If neither key is configured, the adapter reports itself unconfigured and every
+// call returns an empty report with the missing-env note (jobs record this and move on).
+const { throttled, report, register } = require('./index');
+
+function backend() {
+ if (process.env.PR_EXA_API_KEY) return 'exa';
+ if (process.env.PR_BRAVE_API_KEY) return 'brave';
+ return null;
+}
+
+async function exaSearch(query, { numResults = 10, cursor } = {}) {
+ const r = await fetch('https://api.exa.ai/search', {
+ method: 'POST',
+ headers: { 'x-api-key': process.env.PR_EXA_API_KEY, 'Content-Type': 'application/json' },
+ body: JSON.stringify({ query, numResults, type: 'keyword', ...(cursor || {}) }),
+ signal: AbortSignal.timeout(20000),
+ });
+ if (!r.ok) throw new Error('exa ' + r.status + ': ' + (await r.text()).slice(0, 200));
+ const j = await r.json();
+ return (j.results || []).map((x) => ({ title: x.title, url: x.url, snippet: x.text ? x.text.slice(0, 300) : (x.snippet || ''), published_at: x.publishedDate || null }));
+}
+
+async function braveSearch(query, { numResults = 10, cursor } = {}) {
+ const offset = cursor && cursor.offset ? cursor.offset : 0;
+ const r = await fetch(`https://api.search.brave.com/res/v1/web/search?q=${encodeURIComponent(query)}&count=${numResults}&offset=${offset}`, {
+ headers: { 'X-Subscription-Token': process.env.PR_BRAVE_API_KEY, Accept: 'application/json' },
+ signal: AbortSignal.timeout(20000),
+ });
+ if (!r.ok) throw new Error('brave ' + r.status + ': ' + (await r.text()).slice(0, 200));
+ const j = await r.json();
+ return (((j.web || {}).results) || []).map((x) => ({ title: x.title, url: x.url, snippet: x.description || '', published_at: x.age || null }));
+}
+
+const adapter = register({
+ name: 'search',
+ kind: 'discovery',
+ configured: () => !!backend(),
+ missingConfig: () => (backend() ? [] : ['PR_EXA_API_KEY or PR_BRAVE_API_KEY (authorized web-search API)']),
+ sourceMeta: () => ({
+ source_type: 'search_api', is_primary_source: false,
+ license_or_usage_note: 'Authorized search API (' + (backend() || 'unconfigured') + '); indexed snippets only — never treated as verified titles without corroboration.',
+ }),
+
+ /** Run one query. Returns AdapterReport with items [{title,url,snippet,published_at}]. */
+ async search(query, opts = {}) {
+ const be = backend();
+ if (!be) {
+ return report({
+ provider: 'search', query, errors: [{ error: 'no search API key configured (PR_EXA_API_KEY or PR_BRAVE_API_KEY)', code: 'UNCONFIGURED' }],
+ confidence_recommendation: 0, items: [],
+ });
+ }
+ const errors = []; let items = [];
+ try {
+ items = await throttled('search', () => (be === 'exa' ? exaSearch(query, opts) : braveSearch(query, opts)));
+ } catch (e) { errors.push({ error: e.message }); }
+ return report({
+ provider: 'search:' + be, query, errors, primary_source: false,
+ confidence_recommendation: 40,
+ cursor: opts.cursor && items.length ? { offset: ((opts.cursor.offset || 0) + items.length) } : (items.length >= (opts.numResults || 10) ? { offset: items.length } : null),
+ items, terms_note: adapter.sourceMeta().license_or_usage_note,
+ });
+ },
+
+ /** Locate publicly indexed LinkedIn URLs for a person/company via site: queries. */
+ async findLinkedIn({ personName, companyName, kind = 'in' }) {
+ const q = kind === 'company'
+ ? `site:linkedin.com/company "${companyName}"`
+ : `site:linkedin.com/in "${personName}" "${companyName}"`;
+ const rep = await adapter.search(q, { numResults: 5 });
+ rep.items = rep.items.filter((x) => /linkedin\.com\/(in|company)\//i.test(x.url));
+ rep.result_count = rep.items.length;
+ return rep;
+ },
+});
+
+module.exports = adapter;
diff --git a/src/pr/adapters/website.js b/src/pr/adapters/website.js
new file mode 100644
index 00000000..ac0fb3d9
--- /dev/null
+++ b/src/pr/adapters/website.js
@@ -0,0 +1,143 @@
+'use strict';
+// Organization-website adapter (PRIMARY source). Verifies a website, finds
+// newsroom/press/team/leadership/contact pages, extracts general press emails,
+// phones, contact-form URLs, and public team names+titles from those pages.
+const { politeFetch, report, stripTags, register } = require('./index');
+const { normalizeDomain, normalizeEmail } = require('../lib/normalize');
+
+const PAGE_HINTS = {
+ press: /news(room)?|press|media(-|\s)?(center|centre|room|contact|kit)|announcements/i,
+ team: /team|people|leadership|our-people|professionals|staff|about-us\/(team|people|leadership)|management/i,
+ contact: /contact/i,
+};
+
+const EMAIL_RE = /\b[a-z0-9._%+-]+@[a-z0-9.-]+\.[a-z]{2,}\b/gi;
+const PRESSY_EMAIL = /press|media|pr@|news|communications|comms/i;
+const PHONE_RE = /\(?\d{3}\)?[\s.-]\d{3}[\s.-]\d{4}/g;
+
+function absolutize(href, base) {
+ try { return new URL(href, base).toString(); } catch { return null; }
+}
+
+function extractLinks(html, baseUrl) {
+ const links = [];
+ for (const m of String(html).matchAll(/<a\s[^>]*href=["']([^"'#]+)["'][^>]*>([\s\S]*?)<\/a>/gi)) {
+ const url = absolutize(m[1], baseUrl);
+ if (!url || !/^https?:/i.test(url)) continue;
+ links.push({ url, text: stripTags(m[2]).slice(0, 120) });
+ }
+ return links;
+}
+
+/** Titles worth extracting from team pages (kept intentionally broad; classify later). */
+const TITLE_HINT = /officer|director|president|principal|partner|manager|vp|vice president|head of|chief|founder|lead|communications|marketing|media|escrow|lending|relations/i;
+
+/** Extract (name, title) candidate pairs from a team/leadership page. Conservative:
+ * requires a plausible Name-Case name adjacent to a title-looking string. */
+function extractPeople(html) {
+ const text = String(html)
+ .replace(/<script[\s\S]*?<\/script>/gi, ' ').replace(/<style[\s\S]*?<\/style>/gi, ' ')
+ .replace(/<(h[1-6]|p|div|li|td|span|strong|em|a|br)[^>]*>/gi, '\n').replace(/<[^>]+>/g, ' ')
+ .replace(/ /g, ' ').replace(/&/g, '&');
+ const lines = text.split('\n').map((l) => l.replace(/\s+/g, ' ').trim()).filter(Boolean);
+ const people = [];
+ const NAME_RE = /^([A-Z][a-z]+(?:\s+[A-Z]\.?)?(?:\s+[A-Z][a-z'’-]+){1,2})$/;
+ for (let i = 0; i < lines.length; i++) {
+ const nm = lines[i].match(NAME_RE);
+ if (!nm) continue;
+ // look ahead up to 2 lines for a title
+ for (let j = 1; j <= 2 && i + j < lines.length; j++) {
+ const cand = lines[i + j];
+ if (cand.length > 90 || NAME_RE.test(cand)) break;
+ if (TITLE_HINT.test(cand)) { people.push({ full_name: nm[1], exact_title: cand }); break; }
+ }
+ if (people.length >= 60) break; // sane cap per page
+ }
+ // de-dupe by name
+ const seen = new Set();
+ return people.filter((p) => { const k = p.full_name.toLowerCase(); if (seen.has(k)) return false; seen.add(k); return true; });
+}
+
+const adapter = register({
+ name: 'website',
+ kind: 'enrichment',
+ configured: () => true,
+ missingConfig: () => [],
+ sourceMeta: () => ({
+ source_type: 'org_website', is_primary_source: true,
+ license_or_usage_note: 'Public organization website; facts extracted, limited excerpts stored. robots.txt honored; throttled.',
+ }),
+
+ /**
+ * Verify + profile an organization website.
+ * Returns AdapterReport with one item: { website_ok, press_page_url, newsroom_url,
+ * contact_form_url, general_press_email, general_press_phone, press_emails[],
+ * team_pages[], people[], evidence:[{url, excerpt, fields}] }
+ */
+ async profileOrganization(websiteUrl) {
+ const errors = [];
+ const item = { website_ok: false, press_page_url: null, newsroom_url: null, contact_form_url: null, general_press_email: null, general_press_phone: null, press_emails: [], team_pages: [], people: [], evidence: [] };
+ let home;
+ try {
+ home = await politeFetch(websiteUrl, { provider: 'website' });
+ item.website_ok = home.ok;
+ } catch (e) {
+ errors.push({ url: websiteUrl, error: e.message });
+ return report({ provider: 'website', query: websiteUrl, errors, primary_source: true, confidence_recommendation: 0, items: [item] });
+ }
+ if (!home.ok) {
+ errors.push({ url: websiteUrl, error: 'HTTP ' + home.status });
+ return report({ provider: 'website', query: websiteUrl, errors, primary_source: true, confidence_recommendation: 0, items: [item] });
+ }
+ const base = home.finalUrl || websiteUrl;
+ const links = extractLinks(home.body, base);
+ const domain = normalizeDomain(base);
+ const internal = links.filter((l) => normalizeDomain(l.url) === domain);
+ const firstMatch = (re) => { const l = internal.find((x) => re.test(x.url) || re.test(x.text)); return l ? l.url : null; };
+ item.press_page_url = firstMatch(PAGE_HINTS.press);
+ item.newsroom_url = internal.find((x) => /news(room)?/i.test(x.url))?.url || item.press_page_url;
+ item.contact_form_url = firstMatch(PAGE_HINTS.contact);
+ const teamUrl = firstMatch(PAGE_HINTS.team);
+ if (teamUrl) item.team_pages.push(teamUrl);
+
+ // Harvest emails/phones from home + press + contact pages (max 3 fetches beyond home).
+ const pagesToScan = [{ url: base, body: home.body }];
+ for (const u of [item.press_page_url, item.contact_form_url].filter(Boolean).slice(0, 2)) {
+ try { const r = await politeFetch(u, { provider: 'website' }); if (r.ok) pagesToScan.push({ url: u, body: r.body }); }
+ catch (e) { errors.push({ url: u, error: e.message }); }
+ }
+ for (const pg of pagesToScan) {
+ const emails = [...new Set((String(pg.body).match(EMAIL_RE) || []).map((e) => normalizeEmail(e)).filter(Boolean))]
+ .filter((e) => !/example\.|sentry|wixpress|schema\.org|\.png$|\.jpg$/i.test(e));
+ const pressEmails = emails.filter((e) => PRESSY_EMAIL.test(e));
+ item.press_emails.push(...pressEmails);
+ if (!item.general_press_email && pressEmails.length) item.general_press_email = pressEmails[0];
+ if (!item.general_press_email && emails.length && /contact/i.test(pg.url)) item.general_press_email = emails.find((e) => /^info@|^hello@|^contact@/.test(e)) || null;
+ if (!item.general_press_phone) {
+ const ph = String(pg.body).match(PHONE_RE);
+ if (ph) item.general_press_phone = ph[0];
+ }
+ const text = stripTags(pg.body);
+ item.evidence.push({ url: pg.url, excerpt: text.slice(0, 400), fields: pg.url === base ? ['website_url'] : ['press_page_url', 'general_press_email'] });
+ }
+ item.press_emails = [...new Set(item.press_emails)];
+
+ // Team page: extract public names + titles (public info from the org's own site).
+ if (teamUrl) {
+ try {
+ const r = await politeFetch(teamUrl, { provider: 'website' });
+ if (r.ok) {
+ item.people = extractPeople(r.body);
+ item.evidence.push({ url: teamUrl, excerpt: stripTags(r.body).slice(0, 400), fields: ['people'] });
+ }
+ } catch (e) { errors.push({ url: teamUrl, error: e.message }); }
+ }
+ return report({
+ provider: 'website', query: websiteUrl, errors, primary_source: true,
+ confidence_recommendation: 85, items: [item],
+ terms_note: adapter.sourceMeta().license_or_usage_note,
+ });
+ },
+});
+
+module.exports = adapter;
diff --git a/src/pr/jobs/index.js b/src/pr/jobs/index.js
new file mode 100644
index 00000000..610dddb4
--- /dev/null
+++ b/src/pr/jobs/index.js
@@ -0,0 +1,493 @@
+'use strict';
+// Database-backed job queue + the job handlers.
+//
+// Queue guarantees: idempotent enqueue (dedupe_key), checkpointed resume, retry with
+// exponential backoff, per-provider throttling (adapters), pausable, safe to rerun.
+// GEO GATE: discovery jobs for AZ refuse to enqueue while pr_settings.arizona_unlocked
+// is false — the mandated CA-first rollout is enforced in code, not convention.
+const db = require('../db');
+const runs = require('../services/runs');
+const settings = require('../services/settings');
+const audit = require('../services/audit');
+const organizations = require('../services/organizations');
+const people = require('../services/people');
+const relationships = require('../services/relationships');
+const sources = require('../services/sources');
+const suppression = require('../services/suppression');
+const replies = require('../services/replies');
+const geo = require('../lib/geo');
+const { classifyTitle } = require('../lib/taxonomy');
+const { normalizeDomain } = require('../lib/normalize');
+const adapters = require('../adapters');
+
+const DISCOVERY_JOBS = new Set(['discover-organizations', 'discover-people', 'discover-pr-agency-relationships']);
+
+// ── Queue primitives ─────────────────────────────────────────────────────────
+async function enqueue(job_type, payload = {}, { priority = 5, run_id = null, dedupe = true, run_after = null, actor = 'system' } = {}) {
+ // Arizona hard gate — broad AZ discovery cannot even be queued while locked.
+ if (DISCOVERY_JOBS.has(job_type) && payload.state === 'AZ') {
+ const unlocked = await settings.arizonaUnlocked();
+ if (!unlocked && !payload.admin_override) {
+ const err = new Error('Arizona research is locked until the California quality gate passes (see /admin/pr-intelligence/settings).');
+ err.code = 'AZ_LOCKED';
+ throw err;
+ }
+ }
+ const dedupe_key = dedupe ? `${job_type}:${JSON.stringify(payload)}`.slice(0, 500) : null;
+ const row = (await db.query(
+ `INSERT INTO pr_jobs (job_type, payload, dedupe_key, priority, run_id, run_after)
+ VALUES ($1,$2,$3,$4,$5,COALESCE($6, now()))
+ ON CONFLICT (dedupe_key) WHERE status IN ('queued','running','paused') DO NOTHING
+ RETURNING *`,
+ [job_type, JSON.stringify(payload), dedupe_key, priority, run_id, run_after])).rows[0];
+ if (row) await audit.log({ actor, action: 'job.enqueue', entity_type: 'job', entity_id: row.id, detail: { job_type, payload } });
+ return row || { deduped: true, job_type, payload };
+}
+
+/** Claim the next runnable job (FOR UPDATE SKIP LOCKED — safe with multiple workers). */
+async function claim() {
+ return db.tx(async (client) => {
+ const r = await client.query(
+ `SELECT * FROM pr_jobs WHERE status='queued' AND run_after <= now()
+ ORDER BY priority, id LIMIT 1 FOR UPDATE SKIP LOCKED`);
+ if (!r.rows.length) return null;
+ const job = r.rows[0];
+ await client.query(`UPDATE pr_jobs SET status='running', started_at=now(), attempts=attempts+1 WHERE id=$1`, [job.id]);
+ return { ...job, attempts: job.attempts + 1 };
+ });
+}
+
+async function complete(jobId, checkpoint) {
+ await db.query(`UPDATE pr_jobs SET status='completed', completed_at=now(), checkpoint=COALESCE($2, checkpoint) WHERE id=$1`,
+ [jobId, checkpoint ? JSON.stringify(checkpoint) : null]);
+}
+
+/** Failure → retry with exponential backoff until max_attempts, then failed. */
+async function fail(jobId, err, checkpoint) {
+ const job = await db.one('SELECT attempts, max_attempts FROM pr_jobs WHERE id=$1', [jobId]);
+ const msg = String(err && err.message || err).slice(0, 800);
+ if (job && job.attempts < job.max_attempts) {
+ const delaySec = Math.min(3600, Math.pow(2, job.attempts) * 30); // 60s, 120s, 240s… cap 1h
+ await db.query(
+ `UPDATE pr_jobs SET status='queued', last_error=$2, run_after = now() + ($3 || ' seconds')::interval,
+ checkpoint=COALESCE($4, checkpoint) WHERE id=$1`,
+ [jobId, msg, String(delaySec), checkpoint ? JSON.stringify(checkpoint) : null]);
+ return { retried: true, delaySec };
+ }
+ await db.query(`UPDATE pr_jobs SET status='failed', last_error=$2, completed_at=now(),
+ checkpoint=COALESCE($3, checkpoint) WHERE id=$1`,
+ [jobId, msg, checkpoint ? JSON.stringify(checkpoint) : null]);
+ return { retried: false };
+}
+
+async function setJobStatus(id, status, actor) {
+ if (!['queued', 'paused', 'cancelled'].includes(status)) throw new Error('only queued/paused/cancelled may be set manually');
+ const row = (await db.query(`UPDATE pr_jobs SET status=$2::pr_job_status WHERE id=$1 AND status IN ('queued','paused','running','failed') RETURNING *`, [id, status])).rows[0];
+ await audit.log({ actor, action: 'job.status', entity_type: 'job', entity_id: id, after: { status } });
+ return row;
+}
+
+async function listJobs({ status, limit = 100 } = {}) {
+ const where = []; const params = [];
+ if (status) { params.push(status); where.push(`status=$${params.length}::pr_job_status`); }
+ params.push(limit);
+ return db.rows(`SELECT * FROM pr_jobs ${where.length ? 'WHERE ' + where.join(' AND ') : ''} ORDER BY id DESC LIMIT $${params.length}`, params);
+}
+
+// ── Shared helpers for handlers ──────────────────────────────────────────────
+async function upsertOrgFromItem(item, { state, metro, sourceMeta, run_id, adapter, query }) {
+ const county = item.county || null;
+ const metroKey = metro || (county ? geo.metroForCounty(state, county) : null);
+ const res = await organizations.create({
+ legal_name: item.legal_name, display_name: item.display_name || item.legal_name,
+ website_url: item.website_url, organization_type: item.organization_type || 'brokerage',
+ state_presence: [state], metros: metroKey ? [metroKey] : [], counties: county ? [county] : [],
+ headquarters: item.city ? `${item.city}, ${state}` : null,
+ }, {
+ evidence: {
+ source: {
+ source_type: sourceMeta.source_type, source_name: adapter + (query ? ': ' + query : ''),
+ url: item.source_url || null, is_primary_source: sourceMeta.is_primary_source,
+ license_or_usage_note: sourceMeta.license_or_usage_note, adapter,
+ short_excerpt: item.snippet || item.summary || null,
+ },
+ fields: { organization: 60, website_url: item.website_url ? 60 : 0 },
+ },
+ actor: 'system:' + adapter,
+ });
+ return res;
+}
+
+// ── Job handlers ─────────────────────────────────────────────────────────────
+const handlers = {
+ /**
+ * discover-organizations: run the query matrix / a registry for {state, metro?, category?}.
+ * Checkpoint: { template_idx, cursor } — resumes mid-matrix and mid-pagination.
+ */
+ async 'discover-organizations'(job) {
+ const p = job.payload; const state = p.state || 'CA';
+ let run = p.run_id ? await runs.get(p.run_id) : null;
+ if (!run) {
+ run = await runs.create({ state, metro: p.metro, category: p.category, adapter: p.adapter || 'search+registry', actor: 'system' });
+ await db.query('UPDATE pr_jobs SET run_id=$2 WHERE id=$1', [job.id, run.id]);
+ }
+ await runs.setStatus(run.id, 'running');
+ const ck = { ...(job.checkpoint || {}) };
+ try {
+ // 1) Registry pass for bank/credit-union categories (or no category filter).
+ if (!ck.registry_done && (!p.category || ['bank', 'credit_union'].includes(p.category))) {
+ const reg = adapters.getAdapter('registry');
+ let cursor = ck.registry_cursor || null;
+ if (!p.category || p.category === 'bank') {
+ do {
+ const rep = await reg.fdicBanks(state, { cursor });
+ for (const item of rep.items) {
+ const r = await upsertOrgFromItem({ ...item, organization_type: 'bank' }, { state, metro: p.metro, sourceMeta: reg.sourceMeta(), adapter: 'registry:fdic', query: rep.query });
+ await runs.progress(run.id, { candidates: 1, created: r.created ? 1 : 0, updated: r.created ? 0 : 1, duplicates: r.matched === 'exact' ? 1 : 0 });
+ }
+ for (const e of rep.errors) await runs.progress(run.id, { error: e.error });
+ cursor = rep.cursor;
+ ck.registry_cursor = cursor;
+ await db.query('UPDATE pr_jobs SET checkpoint=$2 WHERE id=$1', [job.id, JSON.stringify(ck)]);
+ await runs.progress(run.id, { pages: 1, checkpoint: ck });
+ } while (cursor);
+ }
+ ck.registry_done = true;
+ }
+ // 2) Query-matrix pass through the authorized search API.
+ const search = adapters.getAdapter('search');
+ const tpls = await db.rows(
+ `SELECT * FROM pr_query_templates WHERE enabled=true AND target IN ('organization','press_contact')
+ AND (cardinality(states)=0 OR $1 = ANY(states))
+ AND (cardinality(org_types)=0 OR $2::text IS NULL OR $2 = ANY(org_types))
+ ORDER BY id`, [state, p.category || null]);
+ const metroLabel = p.metro ? geo.metroLabel(state, p.metro) : geo.STATES[state].label;
+ const stateLabel = geo.STATES[state].label;
+ for (let i = ck.template_idx || 0; i < tpls.length; i++) {
+ const t = tpls[i];
+ if (t.template.includes('[company name]')) continue; // company-scoped queries run in verify/person jobs
+ const q = t.template.replace(/\[metro\]/g, metroLabel).replace(/\[state\]/g, stateLabel);
+ const rep = await search.search(q, { numResults: 10 });
+ if (rep.errors.some((e) => e.code === 'UNCONFIGURED')) {
+ await runs.progress(run.id, { error: 'search API unconfigured — matrix skipped (set PR_EXA_API_KEY or PR_BRAVE_API_KEY)' });
+ break;
+ }
+ for (const item of rep.items) {
+ const domain = normalizeDomain(item.url);
+ if (!domain || /linkedin\.com|facebook\.com|wikipedia\.org|yelp\.com|indeed\.com|glassdoor|youtube\.com/.test(domain)) continue;
+ const name = (item.title || '').split(/[|–-]/)[0].trim();
+ if (!name || name.length < 3) continue;
+ const r = await organizations.create({
+ display_name: name, website_url: 'https://' + domain,
+ organization_type: p.category || (t.org_types && t.org_types[0]) || 'pr_agency',
+ state_presence: [state], metros: p.metro ? [p.metro] : [],
+ lifecycle_status: 'discovered',
+ }, {
+ evidence: {
+ source: { source_type: 'search_api', source_name: 'search: ' + q, url: item.url, short_excerpt: item.snippet, is_primary_source: false, license_or_usage_note: rep.terms_note, adapter: 'search' },
+ fields: { organization: rep.confidence_recommendation },
+ },
+ actor: 'system:search',
+ });
+ await runs.progress(run.id, { candidates: 1, created: r.created ? 1 : 0, duplicates: r.matched !== 'none' ? 1 : 0 });
+ }
+ ck.template_idx = i + 1;
+ await db.query('UPDATE pr_jobs SET checkpoint=$2 WHERE id=$1', [job.id, JSON.stringify(ck)]);
+ await runs.progress(run.id, { pages: 1, checkpoint: ck });
+ }
+ await runs.setStatus(run.id, 'completed');
+ return ck;
+ } catch (e) {
+ await runs.progress(run.id, { error: e.message, checkpoint: ck });
+ await runs.setStatus(run.id, 'failed');
+ const err = new Error(e.message); err.checkpoint = ck; throw err;
+ }
+ },
+
+ /** verify-organization: website profiling → press pages, press email, internal-PR flag, team pages. */
+ async 'verify-organization'(job) {
+ const orgId = job.payload.organization_id;
+ const org = await db.one('SELECT * FROM pr_organizations WHERE id=$1', [orgId]);
+ if (!org) return {};
+ if (!org.website_url) {
+ await organizations.update(orgId, { lifecycle_status: org.lifecycle_status === 'discovered' ? 'needs_review' : org.lifecycle_status }, 'system:verify');
+ return {};
+ }
+ const web = adapters.getAdapter('website');
+ const rep = await web.profileOrganization(org.website_url);
+ const item = rep.items[0] || {};
+ const patch = {};
+ if (item.website_ok) {
+ patch.verification_status = 'company_site_verified';
+ patch.last_verified_at = new Date().toISOString();
+ if (org.lifecycle_status === 'discovered') patch.lifecycle_status = 'needs_review';
+ }
+ if (item.press_page_url) patch.press_page_url = item.press_page_url;
+ if (item.newsroom_url) patch.newsroom_url = item.newsroom_url;
+ if (item.contact_form_url) patch.contact_form_url = item.contact_form_url;
+ if (item.general_press_email) patch.general_press_email = item.general_press_email;
+ if (item.general_press_phone) patch.general_press_phone = item.general_press_phone;
+ patch.internal_pr_status = item.press_emails.length || item.press_page_url ? 'has_internal_pr' : org.internal_pr_status;
+ await organizations.update(orgId, patch, 'system:verify-organization');
+ for (const ev of item.evidence || []) {
+ await organizations.attachEvidence(orgId, {
+ source: { source_type: 'org_website', source_name: 'Website: ' + (org.display_name || ''), url: ev.url, short_excerpt: ev.excerpt, is_primary_source: true, license_or_usage_note: web.sourceMeta().license_or_usage_note, adapter: 'website' },
+ fields: Object.fromEntries((ev.fields || []).map((f) => [f, 85])),
+ }, 'system:verify-organization');
+ }
+ // queue person discovery off the found team pages
+ if ((item.people || []).length || (item.team_pages || []).length) {
+ await enqueue('discover-people', { organization_id: orgId, people: item.people, team_pages: item.team_pages }, { priority: 6 });
+ }
+ return { verified: item.website_ok };
+ },
+
+ /** discover-people: persist team-page extracted names+titles with per-field evidence. */
+ async 'discover-people'(job) {
+ const { organization_id, people: found = [], team_pages = [] } = job.payload;
+ const org = await db.one('SELECT * FROM pr_organizations WHERE id=$1', [organization_id]);
+ if (!org) return {};
+ const web = adapters.getAdapter('website');
+ let list = found;
+ if (!list.length && team_pages.length) {
+ const rep = await web.profileOrganization(team_pages[0]);
+ list = (rep.items[0] || {}).people || [];
+ }
+ let created = 0;
+ for (const cand of list.slice(0, 60)) {
+ const cls = classifyTitle(cand.exact_title);
+ if (cls.department === 'other' && !/officer|president|principal|partner|director/i.test(cand.exact_title || '')) continue;
+ const r = await people.create({
+ full_name: cand.full_name, exact_title: cand.exact_title, organization_id,
+ state: (org.state_presence || [])[0] || null, metro: (org.metros || [])[0] || null,
+ lifecycle_status: 'discovered',
+ }, {
+ evidence: {
+ source: { source_type: 'org_website', source_name: 'Team page: ' + org.display_name, url: team_pages[0] || org.website_url, is_primary_source: true, license_or_usage_note: web.sourceMeta().license_or_usage_note, adapter: 'website' },
+ fields: { organization: 85, exact_title: 85 },
+ },
+ actor: 'system:discover-people',
+ });
+ if (r.created) created++;
+ }
+ return { created };
+ },
+
+ /** verify-person: locate public LinkedIn URL (authorized search) + corroboration bookkeeping. */
+ async 'verify-person'(job) {
+ const pid = job.payload.person_id;
+ const p = await db.one('SELECT p.*, o.display_name AS org_name FROM pr_people p LEFT JOIN pr_organizations o ON o.id=p.organization_id WHERE p.id=$1', [pid]);
+ if (!p) return {};
+ const li = adapters.getAdapter('linkedin');
+ if (!p.linkedin_url && li.configured() && p.org_name) {
+ const rep = await li.locatePerson({ full_name: p.full_name, organization_name: p.org_name });
+ const hit = rep.items[0];
+ if (hit) {
+ await people.update(pid, { linkedin_url: hit.linkedin_url, linkedin_status: 'found_uncorroborated' }, 'system:verify-person');
+ await db.query('UPDATE pr_people SET linkedin_indexed_title=$2, linkedin_indexed_snippet=$3 WHERE id=$1', [pid, hit.indexed_title, hit.indexed_snippet]);
+ await people.attachEvidence(pid, {
+ source: { source_type: 'search_api', source_name: 'LinkedIn locate (authorized search)', url: hit.source_url, short_excerpt: hit.indexed_snippet, is_primary_source: false, license_or_usage_note: li.sourceMeta().license_or_usage_note, adapter: 'linkedin' },
+ fields: { linkedin_url: 35 },
+ }, 'system:verify-person');
+ }
+ }
+ // Cross-check: company-site evidence for org+title already? then linkedin found → corroborated.
+ const ev = await sources.evidenceFor('person', pid);
+ const siteBacked = ev.some((e) => ['org_website', 'press_release', 'gov_registry', 'manual'].includes(e.source_type) && ['exact_title', 'organization'].includes(e.field_name));
+ if (p.linkedin_url && siteBacked && p.linkedin_status === 'found_uncorroborated') {
+ await people.update(pid, { linkedin_status: 'found_corroborated' }, 'system:verify-person');
+ }
+ if (siteBacked && p.lifecycle_status === 'discovered') {
+ await people.update(pid, { lifecycle_status: 'needs_verification' }, 'system:verify-person');
+ }
+ return { site_backed: siteBacked };
+ },
+
+ /** discover-pr-agency-relationships: public client/case-study pages on agency sites. */
+ async 'discover-pr-agency-relationships'(job) {
+ const state = job.payload.state || 'CA';
+ const agencies = await db.rows(
+ `SELECT * FROM pr_organizations WHERE organization_type IN ('pr_agency','marketing_agency','public_affairs')
+ AND $1 = ANY(state_presence) AND website_url IS NOT NULL AND lifecycle_status NOT IN ('duplicate','archived')
+ ORDER BY id LIMIT 25 OFFSET $2`, [state, (job.checkpoint || {}).offset || 0]);
+ const web = adapters.getAdapter('website');
+ const { politeFetch, stripTags } = adapters;
+ let linked = 0;
+ for (const agency of agencies) {
+ try {
+ const home = await politeFetch(agency.website_url, { provider: 'website' });
+ if (!home.ok) continue;
+ const m = String(home.body).match(/href=["']([^"']*(?:clients|work|case-stud|portfolio)[^"']*)["']/i);
+ if (!m) continue;
+ const clientsUrl = new URL(m[1], home.finalUrl || agency.website_url).toString();
+ const pg = await politeFetch(clientsUrl, { provider: 'website' });
+ if (!pg.ok) continue;
+ const text = stripTags(pg.body);
+ // match existing CRE orgs named on the clients page
+ const candidates = await db.rows(
+ `SELECT id, display_name FROM pr_organizations
+ WHERE organization_type NOT IN ('pr_agency','marketing_agency','public_affairs')
+ AND $1 = ANY(state_presence) AND lifecycle_status NOT IN ('duplicate','archived') LIMIT 500`, [state]);
+ for (const c of candidates) {
+ if (c.display_name.length >= 5 && text.toLowerCase().includes(c.display_name.toLowerCase())) {
+ await relationships.create({
+ source_organization_id: agency.id, target_organization_id: c.id,
+ relationship_type: 'agency_client', current_status: 'unknown', confidence: 55,
+ notes: 'Client named on public agency page',
+ source: { source_type: 'org_website', source_name: 'Agency client page: ' + agency.display_name, url: clientsUrl, short_excerpt: 'Client list names ' + c.display_name, is_primary_source: true, license_or_usage_note: web.sourceMeta().license_or_usage_note, adapter: 'website' },
+ actor: 'system:agency-relationships',
+ });
+ linked++;
+ }
+ }
+ } catch { /* per-agency errors don't stop the sweep */ }
+ }
+ const nextOffset = ((job.checkpoint || {}).offset || 0) + agencies.length;
+ if (agencies.length === 25) await enqueue('discover-pr-agency-relationships', { state, page: nextOffset }, { priority: 7 });
+ return { offset: nextOffset, linked };
+ },
+
+ /** refresh-stale-records: 90d high-priority / 180d others → mark stale + queue re-verify. */
+ async 'refresh-stale-records'() {
+ const hi = Number(await settings.get('refresh_days_high', 90));
+ const norm = Number(await settings.get('refresh_days_normal', 180));
+ const stalePeople = await db.rows(
+ `UPDATE pr_people SET lifecycle_status='stale'
+ WHERE lifecycle_status IN ('verified','ready_for_outreach')
+ AND last_verified_at IS NOT NULL
+ AND ((contact_priority_score >= 60 AND last_verified_at < now() - ($1 || ' days')::interval)
+ OR (contact_priority_score < 60 AND last_verified_at < now() - ($2 || ' days')::interval))
+ RETURNING id`, [String(hi), String(norm)]);
+ for (const r of stalePeople) await enqueue('verify-person', { person_id: r.id }, { priority: 7 });
+ const staleOrgs = await db.rows(
+ `SELECT id FROM pr_organizations
+ WHERE lifecycle_status IN ('verified','ready_for_outreach')
+ AND (last_verified_at IS NULL OR last_verified_at < now() - ($1 || ' days')::interval)
+ LIMIT 100`, [String(norm)]);
+ for (const r of staleOrgs) await enqueue('verify-organization', { organization_id: r.id }, { priority: 8 });
+ return { stale_people: stalePeople.length, reverify_orgs: staleOrgs.length };
+ },
+
+ /** detect-duplicates: sweep for uncertain matches and surface them (never auto-merge). */
+ async 'detect-duplicates'(job) {
+ const offset = (job.checkpoint || {}).offset || 0;
+ const orgs = await db.rows(`SELECT * FROM pr_organizations WHERE lifecycle_status NOT IN ('duplicate','archived') ORDER BY id LIMIT 50 OFFSET $1`, [offset]);
+ let flagged = 0;
+ for (const o of orgs) {
+ const matches = (await organizations.findMatches(o)).filter((m) => m.org.id !== o.id);
+ const uncertain = matches.filter((m) => m.verdict === 'uncertain');
+ if (uncertain.length) {
+ await audit.activity({ entity_type: 'organization', entity_id: o.id, activity: 'research', detail: { event: 'duplicate_candidates', candidates: uncertain.map((m) => ({ id: m.org.id, score: m.score, reasons: m.reasons })) }, actor: 'system:detect-duplicates' });
+ if (o.lifecycle_status === 'discovered') await organizations.update(o.id, { lifecycle_status: 'needs_review' }, 'system:detect-duplicates');
+ flagged++;
+ }
+ }
+ if (orgs.length === 50) await enqueue('detect-duplicates', { page: offset + 50 }, { priority: 8 });
+ return { offset: offset + orgs.length, flagged };
+ },
+
+ /** calculate-scores: refresh all scores (paged). */
+ async 'calculate-scores'(job) {
+ const offset = (job.checkpoint || {}).offset || 0;
+ const orgs = await db.rows(`SELECT id FROM pr_organizations ORDER BY id LIMIT 100 OFFSET $1`, [offset]);
+ for (const o of orgs) await organizations.refreshScores(o.id);
+ const ppl = await db.rows(`SELECT id FROM pr_people ORDER BY id LIMIT 100 OFFSET $1`, [offset]);
+ for (const p of ppl) await people.refreshScores(p.id);
+ if (orgs.length === 100 || ppl.length === 100) await enqueue('calculate-scores', { page: offset + 100 }, { priority: 9 });
+ return { offset: offset + 100 };
+ },
+
+ /** generate-draft: batch draft generation for campaign members without a draft. */
+ async 'generate-draft'(job) {
+ const outreach = require('../services/outreach');
+ const { campaign_id } = job.payload;
+ const members = await db.rows(
+ `SELECT m.person_id FROM pr_campaign_members m
+ WHERE m.campaign_id=$1 AND NOT EXISTS (
+ SELECT 1 FROM pr_outreach_messages om WHERE om.campaign_id=$1 AND om.person_id=m.person_id AND om.direction='outbound')
+ LIMIT 25`, [campaign_id]);
+ let drafted = 0; const skipped = [];
+ for (const m of members) {
+ try { await outreach.generateDraft({ person_id: m.person_id, campaign_id, actor: 'system:generate-draft' }); drafted++; }
+ catch (e) { skipped.push({ person_id: m.person_id, reason: e.message }); }
+ }
+ if (members.length === 25) await enqueue('generate-draft', { campaign_id, again: Date.now() }, { priority: 6, dedupe: false });
+ return { drafted, skipped };
+ },
+
+ /** sync-email-threads / detect-replies. */
+ async 'sync-email-threads'() { return replies.syncThreads({}); },
+ async 'detect-replies'() { return replies.syncThreads({}); },
+
+ /** create-follow-up-tasks: messages whose follow_up_at passed → task + status. */
+ async 'create-follow-up-tasks'() {
+ const tasksSvc = require('../services/tasks');
+ const due = await db.rows(
+ `UPDATE pr_outreach_messages SET status='follow_up_due'
+ WHERE direction='outbound' AND status='sent' AND follow_up_at IS NOT NULL AND follow_up_at <= now()
+ RETURNING id, person_id, subject`, []);
+ for (const m of due) {
+ await tasksSvc.create({ title: 'Follow up: ' + (m.subject || 'outreach #' + m.id), entity_type: 'message', entity_id: m.id, due_at: new Date().toISOString(), actor: 'system' });
+ if (m.person_id) await audit.activity({ entity_type: 'person', entity_id: m.person_id, activity: 'follow_up_due', detail: { message_id: m.id }, actor: 'system' });
+ }
+ return { created: due.length };
+ },
+
+ /** process-bounces / process-opt-outs: handled inline by reply sync; these sweep strays. */
+ async 'process-bounces'() {
+ const rows = await db.rows(
+ `SELECT m.id, m.person_id, p.public_work_email FROM pr_outreach_messages m
+ JOIN pr_people p ON p.id=m.person_id
+ WHERE m.status='bounced' AND NOT EXISTS (SELECT 1 FROM pr_suppression s WHERE s.person_id=m.person_id)`, []);
+ for (const r of rows) await suppression.add({ email: r.public_work_email, person_id: r.person_id, reason: 'bounce', source: 'sweep', permanent: false, actor: 'system' });
+ return { swept: rows.length };
+ },
+ async 'process-opt-outs'() {
+ const rows = await db.rows(
+ `SELECT m.id, m.person_id, p.public_work_email FROM pr_outreach_messages m
+ JOIN pr_people p ON p.id=m.person_id
+ WHERE m.status='opted_out' AND NOT EXISTS (SELECT 1 FROM pr_suppression s WHERE s.person_id=m.person_id)`, []);
+ for (const r of rows) await suppression.add({ email: r.public_work_email, person_id: r.person_id, reason: 'opt_out', source: 'sweep', permanent: true, actor: 'system' });
+ return { swept: rows.length };
+ },
+
+ /** generate-coverage-report: gap analysis by metro × category, saved to settings. */
+ async 'generate-coverage-report'(job) {
+ const state = job.payload.state || 'CA';
+ const metros = geo.metrosFor(state);
+ const { ORG_TYPE_KEYS } = require('../lib/taxonomy');
+ const grid = [];
+ for (const m of metros) {
+ const byType = await db.rows(
+ `SELECT organization_type, count(*)::int AS n,
+ count(*) FILTER (WHERE lifecycle_status IN ('verified','ready_for_outreach','contacted','active_relationship'))::int AS verified
+ FROM pr_organizations WHERE $1 = ANY(state_presence) AND $2 = ANY(metros) AND lifecycle_status NOT IN ('duplicate','archived')
+ GROUP BY organization_type`, [state, m.key]);
+ const map = Object.fromEntries(byType.map((r) => [r.organization_type, r]));
+ grid.push({ metro: m.key, label: m.label, coverage: ORG_TYPE_KEYS.map((t) => ({ type: t, count: (map[t] || {}).n || 0, verified: (map[t] || {}).verified || 0 })) });
+ }
+ const gaps = [];
+ for (const g of grid) for (const c of g.coverage) if (c.count === 0) gaps.push({ metro: g.metro, type: c.type });
+ const reportDoc = { state, generated_at: new Date().toISOString(), grid, gap_count: gaps.length, gaps: gaps.slice(0, 500) };
+ await settings.set('coverage_report_' + state, reportDoc, 'system:coverage');
+ return { gaps: gaps.length };
+ },
+};
+
+/** Run one claimed job through its handler with checkpoint-aware failure. */
+async function runJob(job) {
+ const h = handlers[job.job_type];
+ if (!h) { await fail(job.id, new Error('unknown job type ' + job.job_type)); return { ok: false }; }
+ try {
+ const ck = await h(job);
+ await complete(job.id, ck);
+ return { ok: true, checkpoint: ck };
+ } catch (e) {
+ const r = await fail(job.id, e, e.checkpoint);
+ return { ok: false, error: e.message, ...r };
+ }
+}
+
+const JOB_TYPES = Object.keys(handlers);
+
+module.exports = { enqueue, claim, complete, fail, runJob, listJobs, setJobStatus, handlers, JOB_TYPES, DISCOVERY_JOBS };
diff --git a/src/pr/lib/dedupe.js b/src/pr/lib/dedupe.js
new file mode 100644
index 00000000..42f0db6f
--- /dev/null
+++ b/src/pr/lib/dedupe.js
@@ -0,0 +1,84 @@
+'use strict';
+// Duplicate detection. Returns classified matches — 'exact' merges may be automated;
+// 'uncertain' matches are ALWAYS presented to the admin (never auto-merged).
+const { normalizeOrgName, normalizePersonName, normalizeDomain, normalizeEmail, normalizeLinkedInUrl } = require('./normalize');
+
+/** Token-set similarity (0..1) for normalized names. */
+function nameSimilarity(a, b) {
+ const A = new Set(String(a || '').split(' ').filter(Boolean));
+ const B = new Set(String(b || '').split(' ').filter(Boolean));
+ if (!A.size || !B.size) return 0;
+ let inter = 0;
+ for (const t of A) if (B.has(t)) inter++;
+ return inter / Math.max(A.size, B.size);
+}
+
+/**
+ * Classify a candidate organization against an existing row.
+ * Considers: normalized legal/brand name, domain, address/state, parent linkage.
+ * → { verdict: 'exact'|'uncertain'|'distinct', score, reasons[] }
+ */
+function classifyOrgMatch(candidate, existing) {
+ const reasons = [];
+ const cd = normalizeDomain(candidate.domain || candidate.website_url);
+ const ed = normalizeDomain(existing.domain || existing.website_url);
+ const cn = candidate.normalized_name || normalizeOrgName(candidate.display_name || candidate.legal_name);
+ const en = existing.normalized_name || normalizeOrgName(existing.display_name || existing.legal_name);
+ const sim = nameSimilarity(cn, en);
+
+ if (cd && ed && cd === ed) {
+ reasons.push(`same domain (${cd})`);
+ if (sim >= 0.5 || cn === en) { reasons.push('name agrees'); return { verdict: 'exact', score: 100, reasons }; }
+ // Same domain but very different names → branch/brand of same org; human decides.
+ reasons.push('name differs — possible brand/branch');
+ return { verdict: 'uncertain', score: 80, reasons };
+ }
+ if (cn && en && cn === en) {
+ reasons.push('identical normalized name');
+ const cs = (candidate.state_presence || []); const es = (existing.state_presence || []);
+ const stateOverlap = cs.some((s) => es.includes(s));
+ if (cd && ed && cd !== ed) { reasons.push(`different domains (${cd} vs ${ed})`); return { verdict: 'uncertain', score: 70, reasons }; }
+ if (!cs.length || !es.length || stateOverlap) return { verdict: cd || ed ? 'uncertain' : 'exact', score: 88, reasons };
+ reasons.push('no state overlap — may be same-name different firm');
+ return { verdict: 'uncertain', score: 55, reasons };
+ }
+ if (sim >= 0.8) { reasons.push(`high name similarity (${sim.toFixed(2)})`); return { verdict: 'uncertain', score: Math.round(sim * 80), reasons }; }
+ return { verdict: 'distinct', score: Math.round(sim * 40), reasons };
+}
+
+/**
+ * Classify a candidate person against an existing row.
+ * Considers: normalized name, organization, email, LinkedIn URL, office, title.
+ */
+function classifyPersonMatch(candidate, existing) {
+ const reasons = [];
+ const ce = normalizeEmail(candidate.public_work_email);
+ const ee = normalizeEmail(existing.public_work_email);
+ if (ce && ee && ce === ee) { reasons.push(`same email (${ce})`); return { verdict: 'exact', score: 100, reasons }; }
+ const cl = normalizeLinkedInUrl(candidate.linkedin_url);
+ const el = normalizeLinkedInUrl(existing.linkedin_url);
+ if (cl && el && cl === el) { reasons.push('same LinkedIn URL'); return { verdict: 'exact', score: 98, reasons }; }
+
+ const cn = candidate.normalized_name || normalizePersonName(candidate.full_name);
+ const en = existing.normalized_name || normalizePersonName(existing.full_name);
+ const sameOrg = candidate.organization_id && existing.organization_id &&
+ String(candidate.organization_id) === String(existing.organization_id);
+ if (cn && en && cn === en) {
+ reasons.push('identical normalized name');
+ if (sameOrg) {
+ reasons.push('same organization');
+ const ct = String(candidate.exact_title || '').toLowerCase();
+ const et = String(existing.exact_title || '').toLowerCase();
+ if (!ct || !et || nameSimilarity(ct, et) >= 0.4) return { verdict: 'exact', score: 95, reasons };
+ reasons.push('title differs — possible role change (preserve history)');
+ return { verdict: 'uncertain', score: 75, reasons };
+ }
+ reasons.push('different organization — possible move or namesake');
+ return { verdict: 'uncertain', score: 60, reasons };
+ }
+ const sim = nameSimilarity(cn, en);
+ if (sim >= 0.75 && sameOrg) { reasons.push(`similar name at same org (${sim.toFixed(2)})`); return { verdict: 'uncertain', score: 65, reasons }; }
+ return { verdict: 'distinct', score: Math.round(sim * 40), reasons };
+}
+
+module.exports = { classifyOrgMatch, classifyPersonMatch, nameSimilarity };
diff --git a/src/pr/lib/geo.js b/src/pr/lib/geo.js
new file mode 100644
index 00000000..281e3a4c
--- /dev/null
+++ b/src/pr/lib/geo.js
@@ -0,0 +1,50 @@
+'use strict';
+// Geographic rollout definition: the MANDATORY research order.
+// Phase 1 = California (in the exact metro order below). Phase 2 = Arizona,
+// which stays locked until the California quality gate passes (pr_settings.arizona_unlocked).
+
+const CA_METROS = [
+ { key: 'la', label: 'Los Angeles County', counties: ['Los Angeles'] },
+ { key: 'oc', label: 'Orange County', counties: ['Orange'] },
+ { key: 'inland-empire', label: 'Inland Empire', counties: ['Riverside', 'San Bernardino'] },
+ { key: 'san-diego', label: 'San Diego County', counties: ['San Diego'] },
+ { key: 'bay-area', label: 'San Francisco Bay Area', counties: ['San Francisco', 'Alameda', 'Contra Costa', 'Santa Clara', 'San Mateo', 'Marin'] },
+ { key: 'sacramento', label: 'Sacramento region', counties: ['Sacramento', 'Placer', 'El Dorado', 'Yolo'] },
+ { key: 'central-coast', label: 'Ventura & Santa Barbara', counties: ['Ventura', 'Santa Barbara'] },
+ { key: 'central-valley',label: 'Central Valley', counties: ['Fresno', 'Kern', 'San Joaquin', 'Stanislaus'] },
+];
+
+const AZ_METROS = [
+ { key: 'phoenix', label: 'Phoenix metro / Maricopa County', counties: ['Maricopa'] },
+ { key: 'scottsdale', label: 'Scottsdale', counties: ['Maricopa'] },
+ { key: 'tempe', label: 'Tempe', counties: ['Maricopa'] },
+ { key: 'mesa', label: 'Mesa', counties: ['Maricopa'] },
+ { key: 'chandler', label: 'Chandler', counties: ['Maricopa'] },
+ { key: 'gilbert', label: 'Gilbert', counties: ['Maricopa'] },
+ { key: 'glendale-az',label: 'Glendale', counties: ['Maricopa'] },
+ { key: 'pinal', label: 'Pinal County', counties: ['Pinal'] },
+ { key: 'tucson', label: 'Tucson / Pima County', counties: ['Pima'] },
+ { key: 'flagstaff', label: 'Flagstaff / Coconino County', counties: ['Coconino'] },
+];
+
+const STATES = { CA: { label: 'California', metros: CA_METROS }, AZ: { label: 'Arizona', metros: AZ_METROS } };
+
+function metrosFor(state) { return (STATES[state] || {}).metros || []; }
+function metroLabel(state, key) {
+ const m = metrosFor(state).find((x) => x.key === key);
+ return m ? m.label : key;
+}
+function metroForCounty(state, county) {
+ const c = String(county || '').toLowerCase().replace(/\s+county$/i, '').trim();
+ const m = metrosFor(state).find((x) => x.counties.some((k) => k.toLowerCase() === c));
+ return m ? m.key : null;
+}
+/** Geographic priority: earlier metro in the mandated order = higher (100..~40). */
+function geoPriority(state, metroKey) {
+ const list = metrosFor(state);
+ const i = list.findIndex((m) => m.key === metroKey);
+ if (i < 0) return 30;
+ return Math.max(30, 100 - Math.round((i / Math.max(list.length - 1, 1)) * 60));
+}
+
+module.exports = { CA_METROS, AZ_METROS, STATES, metrosFor, metroLabel, metroForCounty, geoPriority };
diff --git a/src/pr/lib/normalize.js b/src/pr/lib/normalize.js
new file mode 100644
index 00000000..8633c8e4
--- /dev/null
+++ b/src/pr/lib/normalize.js
@@ -0,0 +1,87 @@
+'use strict';
+// Normalization utilities — the foundation for dedup + matching.
+
+const ORG_STOPWORDS = new Set([
+ 'inc', 'incorporated', 'llc', 'llp', 'lp', 'ltd', 'limited', 'corp', 'corporation',
+ 'co', 'company', 'companies', 'group', 'holdings', 'partners', 'the', 'and', '&',
+ 'pc', 'pllc', 'plc', 'na', 'n.a',
+]);
+
+/** Normalize an organization name for matching: lowercase, strip punctuation + entity suffixes. */
+function normalizeOrgName(name) {
+ return String(name || '')
+ .toLowerCase()
+ .replace(/&/g, ' and ')
+ .replace(/[^a-z0-9\s]/g, ' ')
+ .split(/\s+/)
+ .filter((w) => w && !ORG_STOPWORDS.has(w))
+ .join(' ')
+ .trim();
+}
+
+/** Normalize a person name: lowercase, strip punctuation/credentials, collapse whitespace. */
+function normalizePersonName(name) {
+ return String(name || '')
+ .toLowerCase()
+ .replace(/,?\s*(jr|sr|ii|iii|iv|esq|cpa|mba|ccim|sior|aia|pe|mai)\.?$/i, '')
+ .replace(/[^a-z\s'-]/g, ' ')
+ .replace(/\s+/g, ' ')
+ .trim();
+}
+
+/** Split a full name into first/middle/last (best-effort; UI allows correction). */
+function splitName(full) {
+ const parts = String(full || '').trim().split(/\s+/).filter(Boolean);
+ if (!parts.length) return { first_name: null, middle_name: null, last_name: null };
+ if (parts.length === 1) return { first_name: parts[0], middle_name: null, last_name: null };
+ if (parts.length === 2) return { first_name: parts[0], middle_name: null, last_name: parts[1] };
+ return { first_name: parts[0], middle_name: parts.slice(1, -1).join(' '), last_name: parts[parts.length - 1] };
+}
+
+/** Normalize a domain: strip scheme/www/path, lowercase, apex-ish (keeps subdomain if meaningful). */
+function normalizeDomain(input) {
+ let s = String(input || '').trim().toLowerCase();
+ if (!s) return null;
+ s = s.replace(/^[a-z]+:\/\//, '').replace(/^www\./, '');
+ s = s.split(/[/?#]/)[0].split('@').pop().split(':')[0];
+ if (!/^[a-z0-9.-]+\.[a-z]{2,}$/.test(s)) return null;
+ // Reduce common hosting subdomains to apex, keep others (e.g. commercial.bank.com is meaningful).
+ const parts = s.split('.');
+ if (parts.length > 2 && ['www2', 'web', 'home', 'site'].includes(parts[0])) return parts.slice(1).join('.');
+ return s;
+}
+
+/** Normalize an email; returns null if not a plausible address. */
+function normalizeEmail(email) {
+ const e = String(email || '').trim().toLowerCase();
+ return /^[^@\s]+@[^@\s]+\.[^@\s]{2,}$/.test(e) ? e : null;
+}
+
+/** Extract the domain part of an email. */
+function emailDomain(email) {
+ const e = normalizeEmail(email);
+ return e ? e.split('@')[1] : null;
+}
+
+/** Normalize a LinkedIn URL to its canonical public form; null if not a LinkedIn profile/company URL. */
+function normalizeLinkedInUrl(url) {
+ const s = String(url || '').trim();
+ const m = s.match(/linkedin\.com\/(in|company)\/([A-Za-z0-9\-_%.]+)/i);
+ if (!m) return null;
+ return `https://www.linkedin.com/${m[1].toLowerCase()}/${m[2].replace(/\/+$/, '')}`;
+}
+
+/** Clamp + trim free text (mirrors server.js clean()). */
+function clean(s, n) { return String(s == null ? '' : s).trim().slice(0, n || 500); }
+
+/** US phone: digits-only, keep last 10-11; formatted display left to the UI. */
+function normalizePhone(p) {
+ const d = String(p || '').replace(/\D/g, '');
+ if (d.length === 11 && d.startsWith('1')) return d.slice(1);
+ return d.length === 10 ? d : (d.length >= 7 ? d : null);
+}
+
+module.exports = {
+ normalizeOrgName, normalizePersonName, splitName, normalizeDomain,
+ normalizeEmail, emailDomain, normalizeLinkedInUrl, clean, normalizePhone,
+};
diff --git a/src/pr/lib/scoring.js b/src/pr/lib/scoring.js
new file mode 100644
index 00000000..a06ae51d
--- /dev/null
+++ b/src/pr/lib/scoring.js
@@ -0,0 +1,95 @@
+'use strict';
+// Transparent 0–100 scoring. Confidence and priority are SEPARATE values, and every
+// score is returned with its components so the UI can display the breakdown.
+// A high priority never conceals weak evidence — the UI shows both side-by-side.
+const { roleRelevance } = require('./taxonomy');
+const { geoPriority } = require('./geo');
+
+const SOURCE_TYPE_WEIGHT = {
+ org_website: 90, gov_registry: 95, sec_edgar: 95, press_release: 80,
+ association: 70, news_article: 65, search_api: 40, user_import: 55,
+ licensed_provider: 75, manual: 85,
+};
+
+/** Freshness 0..100 from the newest evidence timestamp. */
+function freshnessScore(lastVerifiedAt, now = Date.now()) {
+ if (!lastVerifiedAt) return 0;
+ const days = (now - new Date(lastVerifiedAt).getTime()) / 864e5;
+ if (days <= 30) return 100;
+ if (days <= 90) return 85;
+ if (days <= 180) return 60;
+ if (days <= 365) return 35;
+ return 15;
+}
+
+/** Source confidence 0..100 from the entity's evidence rows (source_type + primary flag + count). */
+function sourceConfidence(evidenceRows) {
+ const rows = Array.isArray(evidenceRows) ? evidenceRows : [];
+ if (!rows.length) return 0;
+ let best = 0;
+ for (const r of rows) {
+ let w = SOURCE_TYPE_WEIGHT[r.source_type] != null ? SOURCE_TYPE_WEIGHT[r.source_type] : 30;
+ if (r.is_primary_source) w = Math.min(100, w + 5);
+ if (w > best) best = w;
+ }
+ // Corroboration bonus: multiple independent sources.
+ const uniqueSources = new Set(rows.map((r) => r.source_id || r.url)).size;
+ return Math.min(100, best + Math.min(10, (uniqueSources - 1) * 5));
+}
+
+/** Organization confidence: evidence + verified core fields. */
+function orgConfidence(org, evidenceRows) {
+ const src = sourceConfidence(evidenceRows);
+ const fresh = freshnessScore(org.last_verified_at);
+ let fields = 0;
+ if (org.website_url) fields += 25;
+ if (org.organization_type && org.organization_type !== 'unknown') fields += 25;
+ if ((org.metros || []).length || (org.counties || []).length) fields += 25;
+ if (org.press_page_url || org.newsroom_url || org.general_press_email || org.contact_form_url) fields += 25;
+ const score = Math.round(src * 0.5 + fields * 0.3 + fresh * 0.2);
+ return { score: Math.min(100, score), components: { source_confidence: src, field_completeness: fields, freshness: fresh } };
+}
+
+/** Person confidence: evidence for org+title, contact method quality, freshness. */
+function personConfidence(person, evidenceRows) {
+ const src = sourceConfidence(evidenceRows);
+ const fresh = freshnessScore(person.last_verified_at);
+ let contact = 0;
+ const ev = String(person.email_verification_status || 'unverified');
+ if (person.public_work_email && ['corroborated', 'company_site_verified', 'press_release_verified', 'manually_verified'].includes(ev)) contact = 100;
+ else if (person.public_work_email && ev === 'inferred') contact = 20; // stored, flagged, never sendable
+ else if (person.public_business_phone) contact = 50;
+ const li = String(person.linkedin_status || 'none');
+ const liScore = li === 'manually_verified' ? 100 : li === 'found_corroborated' ? 80 : li === 'found_uncorroborated' ? 35 : 0;
+ const score = Math.round(src * 0.45 + contact * 0.25 + liScore * 0.1 + fresh * 0.2);
+ return { score: Math.min(100, score), components: { source_confidence: src, contact_method: contact, linkedin: liScore, freshness: fresh } };
+}
+
+/**
+ * Outreach priority (suggested weighting from the spec):
+ * 25 active CRE relevance in target state · 20 role relevance · 15 source confidence ·
+ * 15 freshness · 10 direct verified contact method · 10 recent activity · 5 geographic priority
+ */
+function outreachPriority({ org, person, evidenceRows, recentActivity = false }) {
+ const c = {};
+ const states = (org && org.state_presence) || [];
+ c.state_relevance = states.includes('CA') || states.includes('AZ') ? 100 : states.length ? 40 : 0;
+ c.role_relevance = person ? roleRelevance(person.department, person.normalized_role) : 0;
+ c.source_confidence = sourceConfidence(evidenceRows);
+ c.freshness = freshnessScore((person && person.last_verified_at) || (org && org.last_verified_at));
+ const ev = String((person && person.email_verification_status) || 'unverified');
+ c.contact_method = person && person.public_work_email && ['corroborated', 'company_site_verified', 'press_release_verified', 'manually_verified'].includes(ev)
+ ? 100 : (org && (org.general_press_email || org.contact_form_url)) ? 60 : 0;
+ c.recent_activity = recentActivity ? 100 : 0;
+ const metro = (person && person.metro) || ((org && org.metros) || [])[0];
+ const state = states.includes('CA') ? 'CA' : states.includes('AZ') ? 'AZ' : null;
+ c.geographic = state && metro ? geoPriority(state, metro) : 30;
+
+ const score = Math.round(
+ c.state_relevance * 0.25 + c.role_relevance * 0.20 + c.source_confidence * 0.15 +
+ c.freshness * 0.15 + c.contact_method * 0.10 + c.recent_activity * 0.10 + c.geographic * 0.05
+ );
+ return { score: Math.min(100, score), components: c };
+}
+
+module.exports = { freshnessScore, sourceConfidence, orgConfidence, personConfidence, outreachPriority, SOURCE_TYPE_WEIGHT };
diff --git a/src/pr/lib/status.js b/src/pr/lib/status.js
new file mode 100644
index 00000000..5c6b30ac
--- /dev/null
+++ b/src/pr/lib/status.js
@@ -0,0 +1,60 @@
+'use strict';
+// Status machines. Transitions outside these maps are rejected by the services,
+// so a record can't silently jump (e.g.) discovered → contacted without review.
+
+const ORG_TRANSITIONS = {
+ discovered: ['needs_review', 'duplicate', 'archived'],
+ needs_review: ['verified', 'incomplete', 'duplicate', 'archived', 'discovered'],
+ verified: ['ready_for_outreach', 'needs_review', 'incomplete', 'duplicate', 'archived'],
+ incomplete: ['needs_review', 'archived', 'duplicate'],
+ duplicate: ['archived', 'needs_review'],
+ archived: ['needs_review'],
+ ready_for_outreach: ['contacted', 'needs_review', 'archived'],
+ contacted: ['active_relationship', 'ready_for_outreach', 'archived'],
+ active_relationship: ['contacted', 'archived'],
+};
+
+const PERSON_TRANSITIONS = {
+ discovered: ['needs_verification', 'duplicate', 'suppressed'],
+ needs_verification: ['verified', 'wrong_person', 'left_company', 'duplicate', 'suppressed', 'discovered'],
+ verified: ['ready_for_outreach', 'stale', 'wrong_person', 'left_company', 'duplicate', 'suppressed', 'needs_verification'],
+ stale: ['needs_verification', 'verified', 'left_company', 'suppressed'],
+ wrong_person: ['needs_verification', 'suppressed'],
+ left_company: ['needs_verification', 'suppressed'],
+ duplicate: ['needs_verification'],
+ suppressed: [], // permanent unless suppression row expires (handled separately)
+ ready_for_outreach: ['contacted', 'verified', 'stale', 'suppressed'],
+ contacted: ['replied', 'ready_for_outreach', 'stale', 'suppressed'],
+ replied: ['contacted', 'verified', 'suppressed'],
+};
+
+const OUTREACH_TRANSITIONS = {
+ not_started: ['draft_generated'],
+ draft_generated: ['needs_review', 'closed'],
+ needs_review: ['approved', 'draft_generated', 'closed'],
+ approved: ['provider_draft_created', 'needs_review', 'closed'],
+ provider_draft_created: ['sent', 'approved', 'closed'],
+ sent: ['replied', 'follow_up_due', 'bounced', 'opted_out', 'closed'],
+ replied: ['interested', 'referred', 'not_now', 'declined', 'opted_out', 'follow_up_due', 'closed'],
+ follow_up_due: ['sent', 'replied', 'closed', 'opted_out'],
+ interested: ['closed', 'follow_up_due'],
+ referred: ['closed', 'follow_up_due'],
+ not_now: ['follow_up_due', 'closed'],
+ declined: ['closed'],
+ bounced: ['closed'],
+ opted_out: ['closed'],
+ closed: [],
+};
+
+function canTransition(map, from, to) {
+ if (from === to) return true;
+ const allowed = map[from];
+ return Array.isArray(allowed) && allowed.includes(to);
+}
+
+module.exports = {
+ ORG_TRANSITIONS, PERSON_TRANSITIONS, OUTREACH_TRANSITIONS,
+ canOrgTransition: (f, t) => canTransition(ORG_TRANSITIONS, f, t),
+ canPersonTransition: (f, t) => canTransition(PERSON_TRANSITIONS, f, t),
+ canOutreachTransition: (f, t) => canTransition(OUTREACH_TRANSITIONS, f, t),
+};
diff --git a/src/pr/lib/taxonomy.js b/src/pr/lib/taxonomy.js
new file mode 100644
index 00000000..f19324c1
--- /dev/null
+++ b/src/pr/lib/taxonomy.js
@@ -0,0 +1,150 @@
+'use strict';
+// Organization taxonomy (the 35 required categories), asset-class tags, and the
+// contact-role dictionary (normalized role + department + seniority inference).
+
+const ORG_TYPES = [
+ ['brokerage', 'Commercial brokerage / advisory'],
+ ['developer_owner', 'Owner / operator / investor / developer / sponsor'],
+ ['reit', 'REIT (public or private)'],
+ ['asset_property_mgmt', 'Asset / property management'],
+ ['facility_mgmt', 'Facility management'],
+ ['bank', 'Bank / commercial bank'],
+ ['credit_union', 'Credit union (CRE-active)'],
+ ['commercial_mortgage_bank','Commercial mortgage bank'],
+ ['debt_equity_broker', 'Debt & equity broker'],
+ ['construction_lender', 'Construction lender'],
+ ['bridge_lender', 'Bridge lender'],
+ ['private_lender', 'Private lender / debt fund'],
+ ['cmbs_structured', 'CMBS / structured finance'],
+ ['life_co_lender', 'Life-company / institutional RE lender'],
+ ['title_insurer', 'Title insurance company'],
+ ['title_agency', 'Title agency'],
+ ['escrow_company', 'Independent escrow company'],
+ ['commercial_escrow_dept', 'Commercial escrow department'],
+ ['exchange_1031', '1031 exchange / qualified intermediary'],
+ ['law_firm', 'Real estate law firm'],
+ ['accounting_consulting', 'Accounting / tax / valuation / consulting'],
+ ['appraiser', 'Commercial appraiser'],
+ ['insurance_risk', 'Insurance / risk management / surety'],
+ ['architecture_firm', 'Architecture / design firm'],
+ ['engineering_firm', 'Engineering firm'],
+ ['general_contractor', 'General contractor'],
+ ['construction_mgmt', 'Construction / project management'],
+ ['economic_development', 'Economic-development agency'],
+ ['municipal_re', 'Municipal real estate / planning dept'],
+ ['association', 'CRE association / chapter'],
+ ['proptech', 'Proptech / CRE data / software / services'],
+ ['pr_agency', 'PR agency (CRE-specialized)'],
+ ['marketing_agency', 'Marketing / branding / content / comms firm serving CRE'],
+ ['public_affairs', 'Public affairs / government relations (CRE clients)'],
+ ['trade_publication', 'Trade publication / event organization'],
+];
+const ORG_TYPE_KEYS = ORG_TYPES.map(([k]) => k);
+const ORG_TYPE_LABEL = Object.fromEntries(ORG_TYPES);
+
+const ASSET_CLASSES = [
+ 'office', 'industrial', 'logistics', 'retail', 'multifamily', 'hospitality',
+ 'medical_office', 'life_science', 'senior_housing', 'student_housing',
+ 'self_storage', 'data_center', 'land', 'mixed_use', 'affordable_housing',
+ 'manufactured_housing', 'special_purpose',
+];
+
+// ── Contact roles: exact-title patterns → normalized role + department. ──────
+// Order matters: first match wins; more specific patterns first.
+const ROLE_RULES = [
+ // Communications & PR
+ [/chief communications officer|cco\b/i, 'chief_communications_officer', 'communications'],
+ [/head of communications/i, 'head_of_communications', 'communications'],
+ [/(vp|vice president)[^a-z]*(of )?communications/i, 'vp_communications', 'communications'],
+ [/director[^a-z]*(of )?communications|communications director/i, 'director_communications', 'communications'],
+ [/corporate communications/i, 'corporate_communications', 'communications'],
+ [/media relations/i, 'media_relations', 'communications'],
+ [/press officer/i, 'press_officer', 'communications'],
+ [/public relations|(^|[^a-z])pr(\b|\s)(manager|director|specialist|lead)?/i, 'public_relations', 'communications'],
+ [/communications manager/i, 'communications_manager', 'communications'],
+ [/communications specialist/i, 'communications_specialist', 'communications'],
+ [/public affairs/i, 'public_affairs', 'communications'],
+ [/external affairs/i, 'external_affairs', 'communications'],
+ [/community relations/i, 'community_relations', 'communications'],
+ // Marketing & content
+ [/chief marketing officer|cmo\b/i, 'chief_marketing_officer', 'marketing'],
+ [/(vp|vice president)[^a-z]*(of )?marketing/i, 'vp_marketing', 'marketing'],
+ [/regional marketing director/i, 'regional_marketing_director', 'marketing'],
+ [/marketing director|director[^a-z]*(of )?marketing/i, 'marketing_director', 'marketing'],
+ [/brand director/i, 'brand_director', 'marketing'],
+ [/content director/i, 'content_director', 'marketing'],
+ [/digital marketing/i, 'digital_marketing', 'marketing'],
+ [/social media/i, 'social_media', 'marketing'],
+ [/events director|director of events/i, 'events_director', 'marketing'],
+ [/research|thought leadership/i, 'research_thought_leadership', 'marketing'],
+ [/marketing/i, 'marketing_generalist', 'marketing'],
+ // Banking & lending
+ [/head of commercial real estate/i, 'head_of_cre', 'lending'],
+ [/head of commercial banking/i, 'head_of_commercial_banking', 'lending'],
+ [/commercial real estate lending|cre lending/i, 'cre_lending', 'lending'],
+ [/cre originations|originations/i, 'cre_originations', 'lending'],
+ [/debt (and|&) structured finance/i, 'debt_structured_finance', 'lending'],
+ [/capital markets/i, 'capital_markets', 'lending'],
+ [/loan production office manager/i, 'lpo_manager', 'lending'],
+ [/regional lending manager/i, 'regional_lending_manager', 'lending'],
+ [/relationship manager/i, 'relationship_manager', 'lending'],
+ // Title & escrow
+ [/national commercial services/i, 'national_commercial_services', 'title_escrow'],
+ [/commercial services manager/i, 'commercial_services_manager', 'title_escrow'],
+ [/commercial escrow officer/i, 'commercial_escrow_officer', 'title_escrow'],
+ [/escrow manager/i, 'escrow_manager', 'title_escrow'],
+ [/title operations manager/i, 'title_operations_manager', 'title_escrow'],
+ [/county manager/i, 'county_manager', 'title_escrow'],
+ [/sales executive/i, 'sales_executive', 'title_escrow'],
+ [/business development officer|business development/i, 'business_development', 'bd'],
+ // PR agency roles
+ [/cre practice lead|real estate practice lead/i, 'practice_lead', 'agency'],
+ [/account director/i, 'account_director', 'agency'],
+ [/account supervisor/i, 'account_supervisor', 'agency'],
+ [/media relations director/i, 'media_relations_director', 'agency'],
+ [/senior account executive/i, 'senior_account_executive', 'agency'],
+ // Commercial leadership (checked last so "VP of Marketing" doesn't land here)
+ [/chief executive officer|ceo\b/i, 'ceo', 'leadership'],
+ [/regional president/i, 'regional_president', 'leadership'],
+ [/president/i, 'president', 'leadership'],
+ [/managing director/i, 'managing_director', 'leadership'],
+ [/market leader/i, 'market_leader', 'leadership'],
+ [/office leader/i, 'office_leader', 'leadership'],
+ [/founder/i, 'founder', 'leadership'],
+ [/principal/i, 'principal', 'leadership'],
+ [/partner/i, 'partner', 'leadership'],
+];
+
+const SENIORITY_RULES = [
+ [/chief|^c[emc]o\b|president|founder/i, 'c_suite'],
+ [/executive vice president|evp|senior vice president|svp/i, 'evp_svp'],
+ [/vice president|vp\b/i, 'vp'],
+ [/director|head of/i, 'director'],
+ [/manager|lead\b/i, 'manager'],
+ [/principal|partner/i, 'principal'],
+ [/specialist|coordinator|associate|executive\b/i, 'specialist'],
+];
+
+/** Map an exact public title → { normalized_role, department, seniority }. Keeps the exact title separately. */
+function classifyTitle(exactTitle) {
+ const t = String(exactTitle || '').trim();
+ if (!t) return { normalized_role: null, department: null, seniority: null };
+ let normalized_role = 'other', department = 'other';
+ for (const [re, role, dept] of ROLE_RULES) {
+ if (re.test(t)) { normalized_role = role; department = dept; break; }
+ }
+ let seniority = 'other';
+ for (const [re, s] of SENIORITY_RULES) { if (re.test(t)) { seniority = s; break; } }
+ return { normalized_role, department, seniority };
+}
+
+// Role relevance for outreach-priority scoring (0..100): comms/PR first, then marketing,
+// then BD/agency, then leadership, then lending/title contacts.
+const DEPT_RELEVANCE = { communications: 100, agency: 92, marketing: 85, bd: 72, leadership: 65, lending: 60, title_escrow: 60, other: 30 };
+function roleRelevance(department, normalized_role) {
+ let base = DEPT_RELEVANCE[department] != null ? DEPT_RELEVANCE[department] : 30;
+ if (/press_officer|media_relations|public_relations|director_communications|head_of_communications|chief_communications/.test(normalized_role || '')) base = 100;
+ return base;
+}
+
+module.exports = { ORG_TYPES, ORG_TYPE_KEYS, ORG_TYPE_LABEL, ASSET_CLASSES, classifyTitle, roleRelevance };
diff --git a/src/pr/services/audit.js b/src/pr/services/audit.js
new file mode 100644
index 00000000..973fd16f
--- /dev/null
+++ b/src/pr/services/audit.js
@@ -0,0 +1,41 @@
+'use strict';
+// Audit log + activity stream. Every send, edit, approval, merge, import, and status
+// change flows through here. Accepts an optional client so callers inside a
+// transaction keep the audit row atomic with the change.
+const db = require('../db');
+
+async function log({ actor, action, entity_type, entity_id, before, after, detail }, client) {
+ const q = `INSERT INTO pr_audit_log (actor, action, entity_type, entity_id, before, after, detail)
+ VALUES ($1,$2,$3,$4,$5,$6,$7) RETURNING id`;
+ const params = [actor || 'system', action, entity_type || null, entity_id || null,
+ before ? JSON.stringify(before) : null, after ? JSON.stringify(after) : null,
+ JSON.stringify(detail || {})];
+ const r = client ? await client.query(q, params) : await db.query(q, params);
+ return r.rows[0].id;
+}
+
+async function activity({ entity_type, entity_id, activity: act, detail, actor }, client) {
+ const q = `INSERT INTO pr_activities (entity_type, entity_id, activity, detail, actor)
+ VALUES ($1,$2,$3,$4,$5) RETURNING id`;
+ const params = [entity_type, entity_id || null, act, JSON.stringify(detail || {}), actor || 'system'];
+ const r = client ? await client.query(q, params) : await db.query(q, params);
+ return r.rows[0].id;
+}
+
+async function recent({ entity_type, entity_id, limit = 100 }) {
+ if (entity_type && entity_id) {
+ return db.rows(`SELECT * FROM pr_activities WHERE entity_type=$1 AND entity_id=$2 ORDER BY created_at DESC LIMIT $3`,
+ [entity_type, entity_id, limit]);
+ }
+ return db.rows(`SELECT * FROM pr_activities ORDER BY created_at DESC LIMIT $1`, [limit]);
+}
+
+async function auditTrail({ entity_type, entity_id, limit = 200 }) {
+ if (entity_type && entity_id) {
+ return db.rows(`SELECT * FROM pr_audit_log WHERE entity_type=$1 AND entity_id=$2 ORDER BY at DESC LIMIT $3`,
+ [entity_type, entity_id, limit]);
+ }
+ return db.rows(`SELECT * FROM pr_audit_log ORDER BY at DESC LIMIT $1`, [limit]);
+}
+
+module.exports = { log, activity, recent, auditTrail };
diff --git a/src/pr/services/campaigns.js b/src/pr/services/campaigns.js
new file mode 100644
index 00000000..b54cd48d
--- /dev/null
+++ b/src/pr/services/campaigns.js
@@ -0,0 +1,80 @@
+'use strict';
+// Campaigns: target definitions + transactional enrollment (suppressed contacts are skipped).
+const db = require('../db');
+const audit = require('./audit');
+const suppression = require('./suppression');
+
+async function create(data, actor) {
+ const row = (await db.query(
+ `INSERT INTO pr_campaigns (name, state, metros, organization_types, asset_classes, status, base_template_id, owner_user)
+ VALUES ($1,$2,$3,$4,$5,$6,$7,$8) RETURNING *`,
+ [data.name, data.state || null, data.metros || [], data.organization_types || [],
+ data.asset_classes || [], data.status || 'draft', data.base_template_id || null, actor || 'admin'])).rows[0];
+ await audit.log({ actor, action: 'campaign.create', entity_type: 'campaign', entity_id: row.id, after: { name: row.name } });
+ return row;
+}
+
+async function update(id, patch, actor) {
+ const allowed = ['name', 'state', 'metros', 'organization_types', 'asset_classes', 'status', 'base_template_id'];
+ const sets = []; const params = [];
+ for (const k of allowed) { if (patch[k] !== undefined) { params.push(patch[k]); sets.push(`${k}=$${params.length}`); } }
+ if (!sets.length) return db.one('SELECT * FROM pr_campaigns WHERE id=$1', [id]);
+ params.push(id);
+ const row = (await db.query(`UPDATE pr_campaigns SET ${sets.join(', ')} WHERE id=$${params.length} RETURNING *`, params)).rows[0];
+ await audit.log({ actor, action: 'campaign.update', entity_type: 'campaign', entity_id: id, after: patch });
+ return row;
+}
+
+async function list() {
+ return db.rows(
+ `SELECT c.*,
+ (SELECT count(*) FROM pr_campaign_members m WHERE m.campaign_id=c.id)::int AS member_count,
+ (SELECT count(*) FROM pr_outreach_messages om WHERE om.campaign_id=c.id AND om.direction='outbound' AND om.status='sent')::int AS sent_count,
+ (SELECT count(*) FROM pr_outreach_messages om WHERE om.campaign_id=c.id AND om.direction='inbound')::int AS reply_count
+ FROM pr_campaigns c ORDER BY c.created_at DESC`, []);
+}
+
+async function get(id) {
+ const c = await db.one('SELECT * FROM pr_campaigns WHERE id=$1', [id]);
+ if (!c) return null;
+ c.members = await db.rows(
+ `SELECT m.person_id, m.added_at, p.full_name, p.exact_title, p.email_verification_status,
+ p.lifecycle_status, o.display_name AS organization_name,
+ (SELECT status FROM pr_outreach_messages om WHERE om.campaign_id=$1 AND om.person_id=m.person_id
+ AND om.direction='outbound' ORDER BY om.updated_at DESC LIMIT 1) AS outreach_status
+ FROM pr_campaign_members m
+ JOIN pr_people p ON p.id=m.person_id
+ LEFT JOIN pr_organizations o ON o.id=p.organization_id
+ WHERE m.campaign_id=$1 ORDER BY m.added_at DESC`, [id]);
+ return c;
+}
+
+/** Transactional enrollment; suppressed or ineligible-status people are skipped with reasons. */
+async function enroll(campaignId, personIds, actor) {
+ return db.tx(async (client) => {
+ const added = []; const skipped = [];
+ for (const pid of personIds) {
+ const p = (await client.query('SELECT id, full_name, public_work_email, lifecycle_status FROM pr_people WHERE id=$1', [pid])).rows[0];
+ if (!p) { skipped.push({ person_id: pid, reason: 'not found' }); continue; }
+ if (['suppressed', 'duplicate', 'wrong_person', 'left_company'].includes(p.lifecycle_status)) {
+ skipped.push({ person_id: pid, reason: 'lifecycle ' + p.lifecycle_status }); continue;
+ }
+ const sup = await suppression.isSuppressed({ email: p.public_work_email, person_id: pid });
+ if (sup.suppressed) { skipped.push({ person_id: pid, reason: 'suppressed: ' + sup.reason }); continue; }
+ const r = await client.query(
+ `INSERT INTO pr_campaign_members (campaign_id, person_id, added_by) VALUES ($1,$2,$3)
+ ON CONFLICT DO NOTHING RETURNING person_id`, [campaignId, pid, actor || 'admin']);
+ if (r.rows.length) added.push(pid); else skipped.push({ person_id: pid, reason: 'already enrolled' });
+ }
+ await audit.log({ actor, action: 'campaign.enroll', entity_type: 'campaign', entity_id: campaignId, detail: { added, skipped } }, client);
+ return { added, skipped };
+ });
+}
+
+async function removeMember(campaignId, personId, actor) {
+ await db.query('DELETE FROM pr_campaign_members WHERE campaign_id=$1 AND person_id=$2', [campaignId, personId]);
+ await audit.log({ actor, action: 'campaign.remove_member', entity_type: 'campaign', entity_id: campaignId, detail: { person_id: personId } });
+ return { ok: true };
+}
+
+module.exports = { create, update, list, get, enroll, removeMember };
diff --git a/src/pr/services/email-providers.js b/src/pr/services/email-providers.js
new file mode 100644
index 00000000..e5881eb8
--- /dev/null
+++ b/src/pr/services/email-providers.js
@@ -0,0 +1,149 @@
+'use strict';
+// Abstract email-provider interface + Gmail adapter (REST, OAuth2 refresh token) + a
+// mock provider for local dev/tests. The interface is deliberately small so another
+// provider (O365, SMTP+IMAP) can be added without touching outreach/reply code.
+//
+// interface Provider {
+// name, configured():bool, missingConfig():string[]
+// createDraft({to, subject, html, text, threadId?}) -> {draft_id, message_id, thread_id}
+// send({draft_id}) -> {message_id, thread_id} // EXPLICIT admin action only
+// listThreads({query, maxResults}) -> [{thread_id, snippet}]
+// getThread(threadId) -> {thread_id, messages:[{message_id, headers, text, html, internalDate}]}
+// }
+//
+// Gmail env (see docs/CRE_PR_EMAIL_SETUP.md):
+// PR_GMAIL_CLIENT_ID / PR_GMAIL_CLIENT_SECRET / PR_GMAIL_REFRESH_TOKEN / PR_GMAIL_USER
+
+const GMAIL_API = 'https://gmail.googleapis.com/gmail/v1/users/me';
+const TOKEN_URL = 'https://oauth2.googleapis.com/token';
+
+function b64url(s) { return Buffer.from(s).toString('base64').replace(/\+/g, '-').replace(/\//g, '_').replace(/=+$/, ''); }
+
+function buildMime({ from, to, subject, text, html }) {
+ const boundary = 'b' + Date.now().toString(36);
+ return [
+ `From: ${from}`, `To: ${to}`, `Subject: ${subject}`, 'MIME-Version: 1.0',
+ `Content-Type: multipart/alternative; boundary="${boundary}"`, '',
+ `--${boundary}`, 'Content-Type: text/plain; charset="UTF-8"', '', text || '', '',
+ `--${boundary}`, 'Content-Type: text/html; charset="UTF-8"', '', html || `<pre>${text || ''}</pre>`, '',
+ `--${boundary}--`, '',
+ ].join('\r\n');
+}
+
+class GmailProvider {
+ constructor() {
+ this.name = 'gmail';
+ this._token = null; this._tokenExp = 0;
+ }
+ missingConfig() {
+ return ['PR_GMAIL_CLIENT_ID', 'PR_GMAIL_CLIENT_SECRET', 'PR_GMAIL_REFRESH_TOKEN', 'PR_GMAIL_USER']
+ .filter((k) => !process.env[k]);
+ }
+ configured() { return this.missingConfig().length === 0; }
+ async _accessToken() {
+ if (this._token && Date.now() < this._tokenExp - 60000) return this._token;
+ const body = new URLSearchParams({
+ client_id: process.env.PR_GMAIL_CLIENT_ID,
+ client_secret: process.env.PR_GMAIL_CLIENT_SECRET,
+ refresh_token: process.env.PR_GMAIL_REFRESH_TOKEN,
+ grant_type: 'refresh_token',
+ });
+ const r = await fetch(TOKEN_URL, { method: 'POST', headers: { 'Content-Type': 'application/x-www-form-urlencoded' }, body, signal: AbortSignal.timeout(15000) });
+ if (!r.ok) throw new Error('gmail token refresh failed: ' + r.status + ' ' + (await r.text()).slice(0, 200));
+ const j = await r.json();
+ this._token = j.access_token; this._tokenExp = Date.now() + (j.expires_in || 3600) * 1000;
+ return this._token;
+ }
+ async _call(pathname, opts = {}) {
+ const token = await this._accessToken();
+ const r = await fetch(GMAIL_API + pathname, {
+ ...opts,
+ headers: { Authorization: 'Bearer ' + token, 'Content-Type': 'application/json', ...(opts.headers || {}) },
+ signal: AbortSignal.timeout(20000),
+ });
+ if (!r.ok) throw new Error(`gmail ${pathname} → ${r.status}: ${(await r.text()).slice(0, 300)}`);
+ return r.json();
+ }
+ async createDraft({ to, subject, html, text, threadId }) {
+ const raw = b64url(buildMime({ from: process.env.PR_GMAIL_USER, to, subject, text, html }));
+ const body = { message: { raw, ...(threadId ? { threadId } : {}) } };
+ const j = await this._call('/drafts', { method: 'POST', body: JSON.stringify(body) });
+ return { draft_id: j.id, message_id: j.message && j.message.id, thread_id: j.message && j.message.threadId };
+ }
+ async send({ draft_id }) {
+ const j = await this._call('/drafts/send', { method: 'POST', body: JSON.stringify({ id: draft_id }) });
+ return { message_id: j.id, thread_id: j.threadId };
+ }
+ async listThreads({ query, maxResults = 25 }) {
+ const j = await this._call(`/threads?q=${encodeURIComponent(query || '')}&maxResults=${maxResults}`);
+ return (j.threads || []).map((t) => ({ thread_id: t.id, snippet: t.snippet }));
+ }
+ async getThread(threadId) {
+ const j = await this._call(`/threads/${threadId}?format=full`);
+ const messages = (j.messages || []).map((m) => {
+ const headers = {};
+ ((m.payload || {}).headers || []).forEach((h) => { headers[h.name.toLowerCase()] = h.value; });
+ const parts = flattenParts(m.payload);
+ const text = decodePart(parts.find((p) => p.mimeType === 'text/plain'));
+ const html = decodePart(parts.find((p) => p.mimeType === 'text/html'));
+ return { message_id: m.id, thread_id: m.threadId, headers, text, html, internalDate: Number(m.internalDate) || null, labelIds: m.labelIds || [] };
+ });
+ return { thread_id: threadId, messages };
+ }
+}
+function flattenParts(payload, acc = []) {
+ if (!payload) return acc;
+ if (payload.body && payload.body.data) acc.push(payload);
+ (payload.parts || []).forEach((p) => flattenParts(p, acc));
+ return acc;
+}
+function decodePart(part) {
+ if (!part || !part.body || !part.body.data) return null;
+ return Buffer.from(part.body.data.replace(/-/g, '+').replace(/_/g, '/'), 'base64').toString('utf8');
+}
+
+// Mock provider: used automatically when Gmail is unconfigured, and by tests.
+// Drafts live in memory; send() records but delivers nothing.
+class MockProvider {
+ constructor() { this.name = 'mock'; this.drafts = new Map(); this.sent = []; this.threads = new Map(); this._n = 0; }
+ configured() { return true; }
+ missingConfig() { return []; }
+ async createDraft({ to, subject, html, text, threadId }) {
+ const id = 'mockd-' + (++this._n);
+ const thread = threadId || 'mockt-' + this._n;
+ this.drafts.set(id, { to, subject, html, text, thread_id: thread });
+ return { draft_id: id, message_id: 'mockm-' + this._n, thread_id: thread };
+ }
+ async send({ draft_id }) {
+ const d = this.drafts.get(draft_id);
+ if (!d) throw new Error('mock draft not found');
+ this.sent.push({ draft_id, ...d, at: new Date().toISOString() });
+ return { message_id: 'mocksent-' + draft_id, thread_id: d.thread_id };
+ }
+ async listThreads() { return [...this.threads.keys()].map((t) => ({ thread_id: t, snippet: '' })); }
+ async getThread(threadId) { return { thread_id: threadId, messages: this.threads.get(threadId) || [] }; }
+}
+
+let _provider = null;
+function getProvider() {
+ if (_provider) return _provider;
+ const gmail = new GmailProvider();
+ _provider = gmail.configured() ? gmail : new MockProvider();
+ return _provider;
+}
+function setProvider(p) { _provider = p; } // tests inject the mock explicitly
+
+/** Status for the settings page: which provider is active and what's missing. */
+function providerStatus() {
+ const gmail = new GmailProvider();
+ const active = getProvider();
+ return {
+ active: active.name,
+ gmail_configured: gmail.configured(),
+ gmail_missing_env: gmail.missingConfig(),
+ note: gmail.configured() ? 'Gmail connected via OAuth refresh token.' :
+ 'Gmail not configured — using the mock provider (drafts stored locally, nothing is sent). See docs/CRE_PR_EMAIL_SETUP.md.',
+ };
+}
+
+module.exports = { getProvider, setProvider, providerStatus, GmailProvider, MockProvider };
diff --git a/src/pr/services/importexport.js b/src/pr/services/importexport.js
new file mode 100644
index 00000000..ef38c48e
--- /dev/null
+++ b/src/pr/services/importexport.js
@@ -0,0 +1,222 @@
+'use strict';
+// Imports (CSV/JSON/LinkedIn URLs) with column-mapping, preview, validation, dry run,
+// duplicate detection, an import report, and a reversible batch id. Exports always
+// include verification + source + last-verified fields so unverified data can't be
+// mistaken for verified.
+const db = require('../db');
+const audit = require('./audit');
+const organizations = require('./organizations');
+const people = require('./people');
+const { normalizeLinkedInUrl } = require('../lib/normalize');
+
+// ── Tiny CSV parser (quoted fields, embedded commas/newlines). No dependency. ─
+function parseCSV(text) {
+ const rows = []; let row = []; let cell = ''; let inQ = false;
+ const s = String(text || '');
+ for (let i = 0; i < s.length; i++) {
+ const c = s[i];
+ if (inQ) {
+ if (c === '"') { if (s[i + 1] === '"') { cell += '"'; i++; } else inQ = false; }
+ else cell += c;
+ } else if (c === '"') inQ = true;
+ else if (c === ',') { row.push(cell); cell = ''; }
+ else if (c === '\n' || c === '\r') {
+ if (c === '\r' && s[i + 1] === '\n') i++;
+ row.push(cell); cell = '';
+ if (row.length > 1 || row[0] !== '') rows.push(row);
+ row = [];
+ } else cell += c;
+ }
+ if (cell !== '' || row.length) { row.push(cell); rows.push(row); }
+ return rows;
+}
+
+function toCSV(rows, headers) {
+ const escCell = (v) => { const s = String(v == null ? '' : v); return /[",\n\r]/.test(s) ? '"' + s.replace(/"/g, '""') + '"' : s; };
+ return [headers.map(escCell).join(','), ...rows.map((r) => headers.map((h) => escCell(r[h])).join(','))].join('\n');
+}
+
+/** Preview: parse + apply column map + validate, WITHOUT writing. */
+function preview({ csv, columnMap, kind }) {
+ const grid = parseCSV(csv);
+ if (!grid.length) return { headers: [], rows: [], issues: ['empty file'] };
+ const headers = grid[0].map((h) => h.trim());
+ const body = grid.slice(1);
+ const mapped = body.map((r) => {
+ const rec = {};
+ for (const [target, srcCol] of Object.entries(columnMap || {})) {
+ const idx = headers.indexOf(srcCol);
+ rec[target] = idx >= 0 ? (r[idx] || '').trim() : '';
+ }
+ return rec;
+ });
+ const issues = [];
+ mapped.forEach((rec, i) => {
+ if (kind === 'org_csv' && !rec.display_name) issues.push(`row ${i + 2}: missing display_name`);
+ if (kind === 'people_csv' && !rec.full_name) issues.push(`row ${i + 2}: missing full_name`);
+ if (kind === 'linkedin_urls' && !normalizeLinkedInUrl(rec.linkedin_url)) issues.push(`row ${i + 2}: not a LinkedIn URL`);
+ });
+ return { headers, rows: mapped.slice(0, 50), total_rows: mapped.length, issues: issues.slice(0, 100), all: mapped };
+}
+
+/**
+ * Run an import. dry_run=true reports what WOULD happen. Every created row carries the
+ * batch id, so reverse() can archive exactly what one import created.
+ */
+async function runImport({ kind, csv, json, columnMap, dry_run = false, filename, source_note, actor }) {
+ let records;
+ if (kind === 'json') {
+ records = Array.isArray(json) ? json : [];
+ } else {
+ const p = preview({ csv, columnMap, kind });
+ records = p.all || [];
+ }
+ const batch = (await db.query(
+ `INSERT INTO pr_import_batches (kind, filename, column_map, dry_run, row_count, created_by)
+ VALUES ($1,$2,$3,$4,$5,$6) RETURNING *`,
+ [kind, filename || null, JSON.stringify(columnMap || {}), dry_run, records.length, actor || 'admin'])).rows[0];
+
+ const report = { created: [], updated: [], duplicates: [], skipped: [], errors: [] };
+ const evidenceBase = {
+ source: {
+ source_type: 'user_import',
+ source_name: `Import batch ${batch.id}${filename ? ' (' + filename + ')' : ''}`,
+ license_or_usage_note: source_note || 'User-provided import; user asserts the right to use this data.',
+ adapter: 'import-' + kind,
+ },
+ };
+
+ for (const rec of records) {
+ try {
+ if (kind === 'org_csv' || (kind === 'json' && rec._type === 'organization')) {
+ if (!rec.display_name) { report.skipped.push({ rec, reason: 'missing display_name' }); continue; }
+ if (dry_run) {
+ const matches = await organizations.findMatches(rec);
+ if (matches.find((m) => m.verdict === 'exact')) report.duplicates.push({ name: rec.display_name, match: matches[0].org.id });
+ else report.created.push({ name: rec.display_name });
+ continue;
+ }
+ const r = await organizations.create({
+ ...rec,
+ state_presence: rec.state ? [rec.state] : (rec.state_presence || []),
+ metros: rec.metro ? [rec.metro] : (rec.metros || []),
+ counties: rec.county ? [rec.county] : (rec.counties || []),
+ asset_classes: typeof rec.asset_classes === 'string' ? rec.asset_classes.split(/[;|]/).map((s) => s.trim()).filter(Boolean) : (rec.asset_classes || []),
+ import_batch_id: batch.id,
+ }, { evidence: { ...evidenceBase, fields: { organization: 55 } }, actor });
+ (r.created ? report.created : report.duplicates).push({ name: rec.display_name, id: r.org.id, matched: r.matched });
+ } else if (kind === 'people_csv' || kind === 'contact_list' || (kind === 'json' && rec._type === 'person')) {
+ if (!rec.full_name) { report.skipped.push({ rec, reason: 'missing full_name' }); continue; }
+ let orgId = rec.organization_id || null;
+ if (!orgId && rec.organization_name) {
+ const m = await organizations.findMatches({ display_name: rec.organization_name, website_url: rec.organization_website });
+ if (m.length && m[0].verdict === 'exact') orgId = m[0].org.id;
+ else if (!dry_run) {
+ const created = await organizations.create({ display_name: rec.organization_name, website_url: rec.organization_website, organization_type: rec.organization_type || 'brokerage', state_presence: rec.state ? [rec.state] : [], import_batch_id: batch.id },
+ { evidence: { ...evidenceBase, fields: { organization: 50 } }, actor });
+ orgId = created.org.id;
+ }
+ }
+ if (dry_run) {
+ const matches = await people.findMatches(rec);
+ if (matches.find((m) => m.verdict === 'exact')) report.duplicates.push({ name: rec.full_name });
+ else report.created.push({ name: rec.full_name });
+ continue;
+ }
+ const r = await people.create({ ...rec, organization_id: orgId, import_batch_id: batch.id },
+ { evidence: { ...evidenceBase, fields: { organization: 50, exact_title: rec.exact_title ? 50 : 0 } }, actor });
+ (r.created ? report.created : report.duplicates).push({ name: rec.full_name, id: r.person.id, matched: r.matched });
+ } else if (kind === 'linkedin_urls') {
+ const url = normalizeLinkedInUrl(rec.linkedin_url);
+ if (!url) { report.skipped.push({ rec, reason: 'invalid LinkedIn URL' }); continue; }
+ if (dry_run) { report.created.push({ linkedin_url: url }); continue; }
+ if (rec.person_id) {
+ await people.update(rec.person_id, { linkedin_url: url, linkedin_status: 'found_uncorroborated' }, actor);
+ report.updated.push({ person_id: rec.person_id, linkedin_url: url });
+ } else if (rec.full_name) {
+ const r = await people.create({ full_name: rec.full_name, linkedin_url: url, linkedin_status: 'found_uncorroborated', import_batch_id: batch.id },
+ { evidence: { ...evidenceBase, fields: { linkedin_url: 40 } }, actor });
+ (r.created ? report.created : report.duplicates).push({ name: rec.full_name, id: r.person.id });
+ } else report.skipped.push({ rec, reason: 'need person_id or full_name' });
+ } else {
+ report.skipped.push({ rec, reason: 'unknown kind ' + kind });
+ }
+ } catch (e) { report.errors.push({ rec: rec.display_name || rec.full_name || null, error: e.message }); }
+ }
+
+ await db.query(
+ `UPDATE pr_import_batches SET created_count=$2, updated_count=$3, duplicate_count=$4, skipped_count=$5, report=$6 WHERE id=$1`,
+ [batch.id, report.created.length, report.updated.length, report.duplicates.length,
+ report.skipped.length + report.errors.length,
+ JSON.stringify({ ...report, created: report.created.slice(0, 200), errors: report.errors.slice(0, 100) })]);
+ await audit.log({ actor, action: 'import.run', entity_type: 'import', entity_id: batch.id,
+ detail: { kind, dry_run, created: report.created.length, duplicates: report.duplicates.length, errors: report.errors.length } });
+ return { batch_id: batch.id, dry_run, report };
+}
+
+/** Reverse a batch: archive (not delete) everything that batch created. */
+async function reverse(batchId, actor) {
+ return db.tx(async (client) => {
+ const b = (await client.query('SELECT * FROM pr_import_batches WHERE id=$1 FOR UPDATE', [batchId])).rows[0];
+ if (!b) throw new Error('batch not found');
+ if (b.reversed_at) throw new Error('batch already reversed');
+ const orgs = await client.query(`UPDATE pr_organizations SET lifecycle_status='archived' WHERE import_batch_id=$1 AND lifecycle_status NOT IN ('contacted','active_relationship') RETURNING id`, [batchId]);
+ const ppl = await client.query(`UPDATE pr_people SET lifecycle_status='suppressed', suppression_reason='import reversed', outreach_eligible=false WHERE import_batch_id=$1 AND lifecycle_status NOT IN ('contacted','replied') RETURNING id`, [batchId]);
+ await client.query('UPDATE pr_import_batches SET reversed_at=now() WHERE id=$1', [batchId]);
+ await audit.log({ actor, action: 'import.reverse', entity_type: 'import', entity_id: batchId, detail: { orgs: orgs.rowCount, people: ppl.rowCount } }, client);
+ return { ok: true, archived_orgs: orgs.rowCount, archived_people: ppl.rowCount };
+ });
+}
+
+async function listBatches() {
+ return db.rows('SELECT * FROM pr_import_batches ORDER BY created_at DESC LIMIT 100', []);
+}
+
+// ── Exports ──────────────────────────────────────────────────────────────────
+const ORG_EXPORT_COLS = ['id', 'display_name', 'legal_name', 'domain', 'website_url', 'organization_type', 'organization_subtype',
+ 'asset_classes', 'state_presence', 'metros', 'counties', 'internal_pr_status', 'general_press_email', 'press_page_url',
+ 'newsroom_url', 'contact_form_url', 'priority_score', 'confidence_score', 'verification_status', 'lifecycle_status',
+ 'last_verified_at', 'source_urls', 'created_at'];
+const PEOPLE_EXPORT_COLS = ['id', 'full_name', 'exact_title', 'normalized_role', 'department', 'organization_name', 'state', 'metro',
+ 'public_work_email', 'email_verification_status', 'public_business_phone', 'linkedin_url', 'linkedin_status',
+ 'contact_priority_score', 'confidence_score', 'verification_status', 'lifecycle_status', 'last_verified_at', 'source_urls', 'created_at'];
+
+async function exportOrganizations(filters = {}) {
+ const { rows } = await organizations.list({ ...filters, limit: 5000 });
+ for (const r of rows) {
+ const src = await db.rows(`SELECT DISTINCT s.url FROM pr_field_evidence e JOIN pr_sources s ON s.id=e.source_id WHERE e.entity_type='organization' AND e.entity_id=$1 AND s.url IS NOT NULL LIMIT 5`, [r.id]);
+ r.source_urls = src.map((x) => x.url).join(' | ');
+ ['asset_classes', 'state_presence', 'metros', 'counties'].forEach((k) => { r[k] = (r[k] || []).join(';'); });
+ }
+ return toCSV(rows, ORG_EXPORT_COLS);
+}
+
+async function exportPeople(filters = {}) {
+ const { rows } = await people.list({ ...filters, limit: 5000 });
+ for (const r of rows) {
+ const src = await db.rows(`SELECT DISTINCT s.url FROM pr_field_evidence e JOIN pr_sources s ON s.id=e.source_id WHERE e.entity_type='person' AND e.entity_id=$1 AND s.url IS NOT NULL LIMIT 5`, [r.id]);
+ r.source_urls = src.map((x) => x.url).join(' | ');
+ }
+ return toCSV(rows, PEOPLE_EXPORT_COLS);
+}
+
+async function exportSuppression() {
+ const { rows } = await require('./suppression').list({ limit: 10000 });
+ return toCSV(rows, ['id', 'email', 'domain', 'reason', 'source', 'permanent', 'created_at']);
+}
+
+async function exportOutreachHistory(filters = {}) {
+ const { rows } = await require('./outreach').list({ ...filters, limit: 5000 });
+ return toCSV(rows, ['id', 'direction', 'full_name', 'organization_name', 'subject', 'status', 'drafted_at', 'sent_at', 'replied_at']);
+}
+
+async function exportSourceEvidence(entityType) {
+ const rows = await db.rows(
+ `SELECT e.entity_type, e.entity_id, e.field_name, e.field_value, e.confidence,
+ s.source_type, s.source_name, s.url, s.published_at, s.retrieved_at, s.is_primary_source, s.license_or_usage_note
+ FROM pr_field_evidence e JOIN pr_sources s ON s.id=e.source_id
+ ${entityType ? 'WHERE e.entity_type=$1' : ''} ORDER BY e.created_at DESC LIMIT 20000`, entityType ? [entityType] : []);
+ return toCSV(rows, ['entity_type', 'entity_id', 'field_name', 'field_value', 'confidence', 'source_type', 'source_name', 'url', 'published_at', 'retrieved_at', 'is_primary_source', 'license_or_usage_note']);
+}
+
+module.exports = { parseCSV, toCSV, preview, runImport, reverse, listBatches, exportOrganizations, exportPeople, exportSuppression, exportOutreachHistory, exportSourceEvidence };
diff --git a/src/pr/services/letters.js b/src/pr/services/letters.js
new file mode 100644
index 00000000..2d6abb5f
--- /dev/null
+++ b/src/pr/services/letters.js
@@ -0,0 +1,166 @@
+'use strict';
+// Letter templates: immutable versions, locked blocks, merge-field rendering,
+// vertical/org snippets, HTML + plain-text output, version compare.
+const db = require('../db');
+const audit = require('./audit');
+const settings = require('./settings');
+
+async function baseTemplate() {
+ const t = await db.one('SELECT * FROM pr_letter_templates WHERE is_base=true ORDER BY id LIMIT 1', []);
+ if (!t) return null;
+ t.current = await db.one('SELECT * FROM pr_letter_versions WHERE template_id=$1 ORDER BY version DESC LIMIT 1', [t.id]);
+ return t;
+}
+
+async function listTemplates() {
+ const ts = await db.rows('SELECT * FROM pr_letter_templates ORDER BY is_base DESC, id', []);
+ for (const t of ts) {
+ t.current = await db.one('SELECT version, created_at, change_note FROM pr_letter_versions WHERE template_id=$1 ORDER BY version DESC LIMIT 1', [t.id]);
+ t.version_count = Number((await db.one('SELECT count(*)::int AS n FROM pr_letter_versions WHERE template_id=$1', [t.id])).n);
+ }
+ return ts;
+}
+
+async function versions(templateId) {
+ return db.rows('SELECT id, version, subject_template, created_by, change_note, created_at FROM pr_letter_versions WHERE template_id=$1 ORDER BY version DESC', [templateId]);
+}
+async function getVersion(versionId) {
+ return db.one('SELECT * FROM pr_letter_versions WHERE id=$1', [versionId]);
+}
+
+/**
+ * Save a new version. Locked blocks in the previous version may NOT be modified or
+ * removed unless `unlock` explicitly lists their keys (an audited human action).
+ */
+async function saveVersion(templateId, { subject_template, blocks, change_note, unlock = [], actor }) {
+ const prev = await db.one('SELECT * FROM pr_letter_versions WHERE template_id=$1 ORDER BY version DESC LIMIT 1', [templateId]);
+ const newBlocks = Array.isArray(blocks) ? blocks : [];
+ if (prev) {
+ const prevBlocks = Array.isArray(prev.blocks) ? prev.blocks : [];
+ for (const pb of prevBlocks) {
+ if (!pb.locked || unlock.includes(pb.key)) continue;
+ const nb = newBlocks.find((b) => b.key === pb.key);
+ if (!nb) throw new Error(`locked block "${pb.key}" cannot be removed (unlock it explicitly first)`);
+ if (nb.content !== pb.content) throw new Error(`locked block "${pb.key}" cannot be modified (unlock it explicitly first)`);
+ }
+ }
+ const version = prev ? prev.version + 1 : 1;
+ const merge_fields = extractMergeFields(newBlocks, subject_template);
+ const row = (await db.query(
+ `INSERT INTO pr_letter_versions (template_id, version, subject_template, blocks, merge_fields, created_by, change_note)
+ VALUES ($1,$2,$3,$4,$5,$6,$7) RETURNING *`,
+ [templateId, version, subject_template || (prev && prev.subject_template) || '', JSON.stringify(newBlocks), merge_fields, actor || 'admin', change_note || null])).rows[0];
+ await audit.log({ actor, action: 'letter.version_saved', entity_type: 'letter_template', entity_id: templateId, detail: { version, change_note, unlocked: unlock } });
+ return row;
+}
+
+function extractMergeFields(blocks, subject) {
+ const set = new Set();
+ const scan = (s) => { for (const m of String(s || '').matchAll(/\{\{\s*([a-z0-9_]+)\s*\}\}/gi)) set.add(m[1]); };
+ scan(subject);
+ (blocks || []).forEach((b) => scan(b.content));
+ return [...set];
+}
+
+/** Diff two versions: block-level added/removed/changed + subject change. */
+function compareVersions(a, b) {
+ const A = Object.fromEntries((a.blocks || []).map((x) => [x.key, x]));
+ const B = Object.fromEntries((b.blocks || []).map((x) => [x.key, x]));
+ const added = Object.keys(B).filter((k) => !A[k]);
+ const removed = Object.keys(A).filter((k) => !B[k]);
+ const changed = Object.keys(B).filter((k) => A[k] && A[k].content !== B[k].content);
+ return { from: a.version, to: b.version, added, removed, changed, subject_changed: a.subject_template !== b.subject_template };
+}
+
+// ── Blocks (vertical / org-specific / generic snippets) ──────────────────────
+async function listBlocks({ scope, org_type, organization_id } = {}) {
+ const where = ['enabled = true']; const params = [];
+ if (scope) { params.push(scope); where.push(`scope=$${params.length}`); }
+ if (org_type) { params.push(org_type); where.push(`(org_type=$${params.length} OR org_type IS NULL)`); }
+ if (organization_id) { params.push(organization_id); where.push(`(organization_id=$${params.length} OR organization_id IS NULL)`); }
+ return db.rows(`SELECT * FROM pr_letter_blocks WHERE ${where.join(' AND ')} ORDER BY scope, org_type NULLS LAST, id`, params);
+}
+async function saveBlock(b, actor) {
+ if (b.id) {
+ const row = (await db.query(
+ `UPDATE pr_letter_blocks SET label=$2, content=$3, scope=$4, org_type=$5, organization_id=$6,
+ internal_sources=$7, enabled=$8 WHERE id=$1 RETURNING *`,
+ [b.id, b.label, b.content, b.scope || 'generic', b.org_type || null, b.organization_id || null,
+ JSON.stringify(b.internal_sources || []), b.enabled !== false])).rows[0];
+ await audit.log({ actor, action: 'letter.block_updated', entity_type: 'letter_block', entity_id: b.id });
+ return row;
+ }
+ const row = (await db.query(
+ `INSERT INTO pr_letter_blocks (key, label, scope, org_type, organization_id, content, internal_sources, enabled)
+ VALUES ($1,$2,$3,$4,$5,$6,$7,$8) RETURNING *`,
+ [b.key || 'blk-' + Date.now().toString(36), b.label, b.scope || 'generic', b.org_type || null,
+ b.organization_id || null, b.content, JSON.stringify(b.internal_sources || []), b.enabled !== false])).rows[0];
+ await audit.log({ actor, action: 'letter.block_created', entity_type: 'letter_block', entity_id: row.id });
+ return row;
+}
+
+// ── Rendering ────────────────────────────────────────────────────────────────
+const esc = (s) => String(s == null ? '' : s).replace(/[<>&"]/g, (c) => ({ '<': '<', '>': '>', '&': '&', '"': '"' }[c]));
+
+/**
+ * Build the merge context for a person+org from settings + records.
+ * Unfilled fields stay as visible {{placeholders}} so a letter with missing
+ * required settings is obviously not send-ready.
+ */
+async function mergeContext({ person, org }) {
+ const s = async (k) => { const v = await settings.get(k, ''); return typeof v === 'string' ? v : (v == null ? '' : String(v)); };
+ return {
+ first_name: (person && (person.first_name || (person.full_name || '').split(' ')[0])) || '',
+ full_name: (person && person.full_name) || '',
+ organization_name: (org && org.display_name) || '',
+ market_name: marketLabel(org),
+ approved_submission_method: await s('approved_submission_method'),
+ sender_name: await s('sender_name'),
+ sender_title: await s('sender_title'),
+ sender_contact_information: await s('sender_contact_information'),
+ sender_postal_address: await s('sender_postal_address'),
+ unsubscribe_text: await s('unsubscribe_text'),
+ };
+}
+
+function marketLabel(org) {
+ const { metroLabel } = require('../lib/geo');
+ if (!org) return '';
+ const state = (org.state_presence || [])[0];
+ const metro = (org.metros || [])[0];
+ if (state && metro) return metroLabel(state, metro);
+ if (state === 'CA') return 'California';
+ if (state === 'AZ') return 'Arizona';
+ return org.headquarters || '';
+}
+
+/**
+ * Render blocks + context → { subject, text, html, missing:[unfilled placeholders], citations }.
+ * Internal citations (per-block internal_sources) are returned separately — they are
+ * kept on the admin record and NEVER included in the outgoing message body.
+ */
+function render({ subject_template, blocks, context, orgBlock }) {
+ const citations = [];
+ const fill = (s) => String(s || '').replace(/\{\{\s*([a-z0-9_]+)\s*\}\}/gi, (m, key) => {
+ if (key === 'organization_specific_block') return orgBlock ? orgBlock.trim() : '';
+ const v = context[key];
+ return v == null || v === '' ? m : String(v); // leave unfilled placeholders visible
+ });
+ const ordered = [...(blocks || [])].sort((a, b) => (a.order || 0) - (b.order || 0));
+ const paras = [];
+ for (const b of ordered) {
+ const t = fill(b.content).trim();
+ if (t) paras.push(t);
+ (b.internal_sources || []).forEach((sid) => citations.push({ block: b.key, source_id: sid }));
+ }
+ const text = paras.join('\n\n');
+ const subject = fill(subject_template || '');
+ const missing = [...new Set([...(subject + '\n' + text).matchAll(/\{\{\s*([a-z0-9_]+)\s*\}\}/gi)].map((m) => m[1]))];
+ const html = `<!doctype html><html><body style="margin:0;padding:24px;background:#f5f7fa">
+<div style="max-width:640px;margin:0 auto;background:#fff;border:1px solid #e3e8ee;border-radius:8px;padding:32px;font-family:-apple-system,BlinkMacSystemFont,'Segoe UI',Roboto,Helvetica,Arial,sans-serif;font-size:15px;line-height:1.65;color:#1a2733">
+${paras.map((p) => `<p style="margin:0 0 16px">${esc(p).replace(/\n/g, '<br>')}</p>`).join('\n')}
+</div></body></html>`;
+ return { subject, text, html, missing, citations };
+}
+
+module.exports = { baseTemplate, listTemplates, versions, getVersion, saveVersion, compareVersions, listBlocks, saveBlock, mergeContext, render, extractMergeFields };
diff --git a/src/pr/services/organizations.js b/src/pr/services/organizations.js
new file mode 100644
index 00000000..c332665f
--- /dev/null
+++ b/src/pr/services/organizations.js
@@ -0,0 +1,240 @@
+'use strict';
+// Organization service: upsert-with-dedupe, list/filter/search, review actions,
+// transactional merge, scoring refresh.
+const db = require('../db');
+const { normalizeOrgName, normalizeDomain, clean } = require('../lib/normalize');
+const { classifyOrgMatch } = require('../lib/dedupe');
+const { orgConfidence, outreachPriority } = require('../lib/scoring');
+const { canOrgTransition } = require('../lib/status');
+const { ORG_TYPE_KEYS } = require('../lib/taxonomy');
+const sources = require('./sources');
+const audit = require('./audit');
+
+/** Find likely duplicates of a candidate. Returns [{org, verdict, score, reasons}] sorted by score. */
+async function findMatches(candidate) {
+ const norm = normalizeOrgName(candidate.display_name || candidate.legal_name);
+ const domain = normalizeDomain(candidate.domain || candidate.website_url);
+ const params = []; const ors = [];
+ if (domain) { params.push(domain); ors.push(`domain = $${params.length}`); }
+ if (norm) {
+ params.push(norm); ors.push(`normalized_name = $${params.length}`);
+ const first = norm.split(' ')[0];
+ if (first && first.length >= 4) { params.push(first + '%'); ors.push(`normalized_name LIKE $${params.length}`); }
+ }
+ if (!ors.length) return [];
+ const rows = await db.rows(
+ `SELECT * FROM pr_organizations WHERE (${ors.join(' OR ')}) AND lifecycle_status <> 'duplicate' LIMIT 25`, params);
+ return rows
+ .map((org) => ({ org, ...classifyOrgMatch({ ...candidate, normalized_name: norm, domain }, org) }))
+ .filter((m) => m.verdict !== 'distinct')
+ .sort((a, b) => b.score - a.score);
+}
+
+/**
+ * Create an organization (or return the exact-match existing one) with source evidence.
+ * `evidence` = { source: {…pr_sources fields…}, fields: {field_name: confidence} }.
+ * Uncertain matches are NOT auto-merged: the new row is created flagged needs_review
+ * with the candidate ids recorded, so an admin decides.
+ */
+async function create(data, { evidence, actor } = {}) {
+ const display_name = clean(data.display_name || data.legal_name, 300);
+ if (!display_name) throw new Error('display_name required');
+ const organization_type = ORG_TYPE_KEYS.includes(data.organization_type) ? data.organization_type : 'brokerage';
+ const normalized_name = normalizeOrgName(display_name);
+ const domain = normalizeDomain(data.domain || data.website_url);
+ const matches = await findMatches({ ...data, display_name });
+
+ const exact = matches.find((m) => m.verdict === 'exact');
+ if (exact) {
+ // update-not-duplicate: attach any new evidence to the existing row
+ if (evidence && evidence.source) await attachEvidence(exact.org.id, evidence, null);
+ return { org: exact.org, created: false, matched: 'exact', matches };
+ }
+
+ const uncertain = matches.filter((m) => m.verdict === 'uncertain');
+ const row = await db.tx(async (client) => {
+ const r = await client.query(
+ `INSERT INTO pr_organizations
+ (legal_name, display_name, normalized_name, domain, website_url, linkedin_company_url,
+ organization_type, organization_subtype, asset_classes, headquarters, state_presence,
+ metros, counties, office_addresses, parent_organization_id, internal_pr_status,
+ press_page_url, newsroom_url, general_press_email, general_press_phone, contact_form_url,
+ verification_status, lifecycle_status, notes, tags, import_batch_id)
+ VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23,$24,$25,$26)
+ RETURNING *`,
+ [clean(data.legal_name, 300) || null, display_name, normalized_name, domain,
+ clean(data.website_url, 500) || null, clean(data.linkedin_company_url, 500) || null,
+ organization_type, clean(data.organization_subtype, 120) || null,
+ data.asset_classes || [], clean(data.headquarters, 300) || null,
+ data.state_presence || [], data.metros || [], data.counties || [],
+ JSON.stringify(data.office_addresses || []), data.parent_organization_id || null,
+ data.internal_pr_status || 'unknown',
+ clean(data.press_page_url, 500) || null, clean(data.newsroom_url, 500) || null,
+ clean(data.general_press_email, 300) || null, clean(data.general_press_phone, 60) || null,
+ clean(data.contact_form_url, 500) || null,
+ data.verification_status || 'unverified',
+ uncertain.length ? 'needs_review' : (data.lifecycle_status || 'discovered'),
+ clean(data.notes, 4000) || null, data.tags || [], data.import_batch_id || null]);
+ const org = r.rows[0];
+ if (evidence && evidence.source) {
+ const src = await sources.findOrCreateSource(evidence.source, client);
+ for (const [field, conf] of Object.entries(evidence.fields || { organization: 60 })) {
+ await sources.addEvidence({ source_id: src.id, entity_type: 'organization', entity_id: org.id, field_name: field, field_value: org[field] != null ? org[field] : null, confidence: conf }, client);
+ }
+ }
+ await audit.log({ actor, action: 'org.create', entity_type: 'organization', entity_id: org.id, after: { display_name, organization_type }, detail: { uncertain_matches: uncertain.map((m) => m.org.id) } }, client);
+ await audit.activity({ entity_type: 'organization', entity_id: org.id, activity: 'research', detail: { event: 'created', uncertain_matches: uncertain.map((m) => ({ id: m.org.id, score: m.score, reasons: m.reasons })) }, actor }, client);
+ return org;
+ });
+ await refreshScores(row.id);
+ return { org: await get(row.id), created: true, matched: uncertain.length ? 'uncertain' : 'none', matches: uncertain };
+}
+
+async function attachEvidence(orgId, evidence, actor) {
+ return db.tx(async (client) => {
+ const src = await sources.findOrCreateSource(evidence.source, client);
+ const out = [];
+ for (const [field, conf] of Object.entries(evidence.fields || {})) {
+ out.push(await sources.addEvidence({ source_id: src.id, entity_type: 'organization', entity_id: orgId, field_name: field, field_value: evidence.values ? evidence.values[field] : null, confidence: conf }, client));
+ }
+ await audit.log({ actor, action: 'org.evidence_added', entity_type: 'organization', entity_id: orgId, detail: { source: src.url || src.source_name, fields: Object.keys(evidence.fields || {}) } }, client);
+ return { source: src, evidence: out };
+ });
+}
+
+async function get(id) {
+ const org = await db.one('SELECT * FROM pr_organizations WHERE id=$1', [id]);
+ if (!org) return null;
+ org.evidence = await sources.evidenceFor('organization', id);
+ org.people = await db.rows(
+ `SELECT id, full_name, exact_title, normalized_role, department, lifecycle_status,
+ email_verification_status, public_work_email, linkedin_status, contact_priority_score, confidence_score
+ FROM pr_people WHERE organization_id=$1 AND lifecycle_status <> 'duplicate' ORDER BY contact_priority_score DESC`, [id]);
+ org.relationships = await db.rows(
+ `SELECT r.*, so.display_name AS source_name_org, tg.display_name AS target_name_org
+ FROM pr_org_relationships r
+ JOIN pr_organizations so ON so.id=r.source_organization_id
+ JOIN pr_organizations tg ON tg.id=r.target_organization_id
+ WHERE r.source_organization_id=$1 OR r.target_organization_id=$1`, [id]);
+ org.recent_activity = await audit.recent({ entity_type: 'organization', entity_id: id, limit: 30 });
+ org.outreach = await db.rows(
+ `SELECT id, person_id, campaign_id, subject, status, drafted_at, sent_at, replied_at
+ FROM pr_outreach_messages WHERE organization_id=$1 ORDER BY created_at DESC LIMIT 50`, [id]);
+ return org;
+}
+
+async function list(f = {}) {
+ const where = []; const params = [];
+ const add = (sql, v) => { params.push(v); where.push(sql.replace('?', '$' + params.length)); };
+ if (f.q) {
+ params.push('%' + String(f.q).toLowerCase() + '%');
+ const n = params.length;
+ where.push(`(lower(display_name) LIKE $${n} OR lower(coalesce(domain,'')) LIKE $${n} OR normalized_name LIKE $${n} OR lower(coalesce(legal_name,'')) LIKE $${n})`);
+ }
+ if (f.state) add(`? = ANY(state_presence)`, f.state);
+ if (f.metro) add(`? = ANY(metros)`, f.metro);
+ if (f.county) add(`? = ANY(counties)`, f.county);
+ if (f.type) add(`organization_type = ?`, f.type);
+ if (f.asset_class) add(`? = ANY(asset_classes)`, f.asset_class);
+ if (f.status) add(`lifecycle_status = ?::pr_org_status`, f.status);
+ if (f.min_confidence) add(`confidence_score >= ?`, Number(f.min_confidence));
+ if (f.min_priority) add(`priority_score >= ?`, Number(f.min_priority));
+ if (f.verified_before) add(`(last_verified_at IS NULL OR last_verified_at < ?)`, f.verified_before);
+ if (f.no_pr_contact === '1' || f.no_pr_contact === true) {
+ where.push(`NOT EXISTS (SELECT 1 FROM pr_people p WHERE p.organization_id = pr_organizations.id
+ AND p.department IN ('communications','agency') AND p.lifecycle_status NOT IN ('duplicate','suppressed'))`);
+ }
+ if (!f.include_duplicates) where.push(`lifecycle_status <> 'duplicate'`);
+ const sortable = { name: 'display_name', type: 'organization_type', status: 'lifecycle_status', confidence: 'confidence_score', priority: 'priority_score', verified: 'last_verified_at', created: 'created_at', updated: 'updated_at' };
+ const sort = sortable[f.sort] || 'priority_score';
+ const dir = f.dir === 'asc' ? 'ASC' : 'DESC';
+ const limit = Math.min(Number(f.limit) || 50, 500);
+ const offset = Number(f.offset) || 0;
+ const whereSql = where.length ? 'WHERE ' + where.join(' AND ') : '';
+ const total = await db.one(`SELECT count(*)::int AS n FROM pr_organizations ${whereSql}`, params);
+ params.push(limit, offset);
+ const rows = await db.rows(
+ `SELECT id, display_name, legal_name, domain, website_url, organization_type, organization_subtype,
+ asset_classes, state_presence, metros, counties, internal_pr_status, general_press_email,
+ press_page_url, newsroom_url, contact_form_url, priority_score, confidence_score, score_components,
+ verification_status, lifecycle_status, last_verified_at, tags, created_at, updated_at,
+ (SELECT count(*) FROM pr_people p WHERE p.organization_id=pr_organizations.id AND p.lifecycle_status NOT IN ('duplicate','suppressed'))::int AS people_count,
+ (SELECT count(*) FROM pr_field_evidence e WHERE e.entity_type='organization' AND e.entity_id=pr_organizations.id)::int AS evidence_count
+ FROM pr_organizations ${whereSql}
+ ORDER BY ${sort} ${dir} NULLS LAST LIMIT $${params.length - 1} OFFSET $${params.length}`, params);
+ return { total: total.n, rows };
+}
+
+async function update(id, patch, actor) {
+ const before = await db.one('SELECT * FROM pr_organizations WHERE id=$1', [id]);
+ if (!before) throw new Error('not found');
+ if (patch.lifecycle_status && !canOrgTransition(before.lifecycle_status, patch.lifecycle_status)) {
+ throw new Error(`invalid status transition ${before.lifecycle_status} → ${patch.lifecycle_status}`);
+ }
+ const allowed = ['legal_name', 'display_name', 'domain', 'website_url', 'linkedin_company_url', 'organization_type',
+ 'organization_subtype', 'asset_classes', 'headquarters', 'state_presence', 'metros', 'counties',
+ 'office_addresses', 'parent_organization_id', 'internal_pr_status', 'press_page_url', 'newsroom_url',
+ 'general_press_email', 'general_press_phone', 'contact_form_url', 'verification_status',
+ 'lifecycle_status', 'notes', 'tags', 'last_verified_at'];
+ const sets = []; const params = [];
+ for (const k of allowed) {
+ if (patch[k] === undefined) continue;
+ params.push(k === 'office_addresses' ? JSON.stringify(patch[k]) : patch[k]);
+ sets.push(`${k} = $${params.length}${k === 'lifecycle_status' ? '::pr_org_status' : k === 'verification_status' ? '::pr_verification' : ''}`);
+ }
+ if (patch.display_name) { params.push(normalizeOrgName(patch.display_name)); sets.push(`normalized_name = $${params.length}`); }
+ if (patch.domain || patch.website_url) { params.push(normalizeDomain(patch.domain || patch.website_url)); sets.push(`domain = $${params.length}`); }
+ if (!sets.length) return before;
+ params.push(id);
+ const row = (await db.query(`UPDATE pr_organizations SET ${sets.join(', ')} WHERE id=$${params.length} RETURNING *`, params)).rows[0];
+ await audit.log({ actor, action: 'org.update', entity_type: 'organization', entity_id: id, before: pick(before, Object.keys(patch)), after: pick(row, Object.keys(patch)) });
+ await refreshScores(id);
+ return row;
+}
+
+/** Transactional merge: loser's people/evidence/relationships/messages move to winner; loser marked duplicate. */
+async function merge(winnerId, loserId, actor) {
+ if (String(winnerId) === String(loserId)) throw new Error('cannot merge an organization into itself');
+ return db.tx(async (client) => {
+ const w = (await client.query('SELECT * FROM pr_organizations WHERE id=$1 FOR UPDATE', [winnerId])).rows[0];
+ const l = (await client.query('SELECT * FROM pr_organizations WHERE id=$1 FOR UPDATE', [loserId])).rows[0];
+ if (!w || !l) throw new Error('organization not found');
+ await client.query('UPDATE pr_people SET organization_id=$1 WHERE organization_id=$2', [winnerId, loserId]);
+ await client.query(`UPDATE pr_field_evidence SET entity_id=$1 WHERE entity_type='organization' AND entity_id=$2`, [winnerId, loserId]);
+ await client.query('UPDATE pr_outreach_messages SET organization_id=$1 WHERE organization_id=$2', [winnerId, loserId]);
+ await client.query('UPDATE pr_org_relationships SET source_organization_id=$1 WHERE source_organization_id=$2 AND target_organization_id<>$1', [winnerId, loserId]);
+ await client.query('UPDATE pr_org_relationships SET target_organization_id=$1 WHERE target_organization_id=$2 AND source_organization_id<>$1', [winnerId, loserId]);
+ await client.query('DELETE FROM pr_org_relationships WHERE source_organization_id=target_organization_id');
+ // fill gaps on the winner from the loser (never overwrite winner values)
+ const fill = ['legal_name', 'domain', 'website_url', 'linkedin_company_url', 'headquarters', 'press_page_url', 'newsroom_url', 'general_press_email', 'general_press_phone', 'contact_form_url'];
+ for (const k of fill) {
+ if (!w[k] && l[k]) await client.query(`UPDATE pr_organizations SET ${k}=$1 WHERE id=$2`, [l[k], winnerId]);
+ }
+ await client.query(`UPDATE pr_organizations SET
+ state_presence = (SELECT array_agg(DISTINCT x) FROM unnest(state_presence || $2::text[]) x),
+ metros = (SELECT array_agg(DISTINCT x) FROM unnest(metros || $3::text[]) x),
+ counties = (SELECT array_agg(DISTINCT x) FROM unnest(counties || $4::text[]) x),
+ asset_classes = (SELECT array_agg(DISTINCT x) FROM unnest(asset_classes || $5::text[]) x)
+ WHERE id=$1`, [winnerId, l.state_presence, l.metros, l.counties, l.asset_classes]);
+ await client.query(`UPDATE pr_organizations SET lifecycle_status='duplicate', duplicate_of_id=$1 WHERE id=$2`, [winnerId, loserId]);
+ await audit.log({ actor, action: 'org.merge', entity_type: 'organization', entity_id: winnerId, detail: { merged_from: loserId } }, client);
+ await audit.activity({ entity_type: 'organization', entity_id: winnerId, activity: 'duplicate_resolved', detail: { merged_from: loserId }, actor }, client);
+ return { ok: true, winner: winnerId, merged: loserId };
+ }).then(async (r) => { await refreshScores(winnerId); return r; });
+}
+
+/** Recompute + persist confidence/priority with components. */
+async function refreshScores(id) {
+ const org = await db.one('SELECT * FROM pr_organizations WHERE id=$1', [id]);
+ if (!org) return null;
+ const ev = await sources.evidenceFor('organization', id);
+ const conf = orgConfidence(org, ev.map((e) => ({ source_type: e.source_type, is_primary_source: e.is_primary_source, source_id: e.source_id })));
+ const pri = outreachPriority({ org, evidenceRows: ev.map((e) => ({ source_type: e.source_type, is_primary_source: e.is_primary_source, source_id: e.source_id })) });
+ await db.query(`UPDATE pr_organizations SET confidence_score=$1, priority_score=$2, score_components=$3 WHERE id=$4`,
+ [conf.score, pri.score, JSON.stringify({ confidence: conf.components, priority: pri.components }), id]);
+ return { confidence: conf, priority: pri };
+}
+
+function pick(obj, keys) { const o = {}; for (const k of keys) if (obj && obj[k] !== undefined) o[k] = obj[k]; return o; }
+
+module.exports = { create, get, list, update, merge, findMatches, attachEvidence, refreshScores };
diff --git a/src/pr/services/outreach.js b/src/pr/services/outreach.js
new file mode 100644
index 00000000..086a5134
--- /dev/null
+++ b/src/pr/services/outreach.js
@@ -0,0 +1,215 @@
+'use strict';
+// Outreach lifecycle: grounded draft generation → review → approval → provider draft →
+// EXPLICIT send → thread tracking. Hard rules enforced here, not in the UI:
+// - no draft for a suppressed contact
+// - readiness requires: org has ≥1 credible source, person's org+title have evidence,
+// and a verified email / official contact form / manually-approved method
+// - a provider draft is NEVER auto-sent; send() is a separate explicit call
+// - internal citations are stored on the record and stripped from outgoing content
+const db = require('../db');
+const audit = require('./audit');
+const letters = require('./letters');
+const suppression = require('./suppression');
+const providers = require('./email-providers');
+const { canOutreachTransition } = require('../lib/status');
+const { SENDABLE_EMAIL_STATUSES } = require('./people');
+
+/** Outreach-readiness check (Stage 7 gates). Returns { ready, reasons[] }. */
+async function readiness(personId) {
+ const p = await db.one('SELECT * FROM pr_people WHERE id=$1', [personId]);
+ if (!p) return { ready: false, reasons: ['person not found'] };
+ const reasons = [];
+ const org = p.organization_id ? await db.one('SELECT * FROM pr_organizations WHERE id=$1', [p.organization_id]) : null;
+ if (!org) reasons.push('person has no organization');
+ if (org) {
+ const orgEv = await db.one(`SELECT count(*)::int AS n FROM pr_field_evidence WHERE entity_type='organization' AND entity_id=$1`, [org.id]);
+ if (!Number(orgEv.n)) reasons.push('organization has no credible source');
+ }
+ const pEv = await db.rows(`SELECT field_name FROM pr_field_evidence WHERE entity_type='person' AND entity_id=$1`, [personId]);
+ const evFields = new Set(pEv.map((r) => r.field_name));
+ if (!evFields.has('organization') && !evFields.has('organization_id')) reasons.push("no source for the person's current organization");
+ if (!evFields.has('exact_title') && !evFields.has('title')) reasons.push("no source for the person's title");
+ const emailOk = p.public_work_email && SENDABLE_EMAIL_STATUSES.includes(p.email_verification_status);
+ const formOk = !!(org && org.contact_form_url);
+ const manualOk = p.outreach_eligible === true; // admin explicitly approved a method
+ if (!emailOk && !formOk && !manualOk) reasons.push('no verified email, official contact form, or manually approved outreach method');
+ const sup = await suppression.isSuppressed({ email: p.public_work_email, person_id: personId });
+ if (sup.suppressed) reasons.push('contact is suppressed (' + sup.reason + ')');
+ if (['suppressed', 'wrong_person', 'left_company', 'duplicate'].includes(p.lifecycle_status)) reasons.push('person lifecycle status is ' + p.lifecycle_status);
+ return { ready: reasons.length === 0, reasons, person: p, org };
+}
+
+/**
+ * Generate a grounded letter draft for a person (optionally within a campaign).
+ * Composition: base template version + vertical block for the org type + any
+ * org-specific snippet + extra blocks the admin chose.
+ */
+async function generateDraft({ person_id, campaign_id, extra_block_ids = [], actor }) {
+ const check = await readiness(person_id);
+ if (!check.person) throw new Error('person not found');
+ const sup = await suppression.isSuppressed({ email: check.person.public_work_email, person_id });
+ if (sup.suppressed) throw new Error('cannot draft to a suppressed contact (' + sup.reason + ')');
+
+ const tpl = await letters.baseTemplate();
+ if (!tpl || !tpl.current) throw new Error('no base letter template found');
+ const org = check.org;
+ const vertBlocks = org ? await letters.listBlocks({ scope: 'vertical', org_type: org.organization_type }) : [];
+ const orgBlocks = org ? await letters.listBlocks({ scope: 'organization', organization_id: org.id }) : [];
+ const chosen = [...orgBlocks, ...vertBlocks].filter((b) => org && (b.organization_id === org.id || b.org_type === org.organization_type));
+ const extra = extra_block_ids.length
+ ? await db.rows(`SELECT * FROM pr_letter_blocks WHERE id = ANY($1::bigint[])`, [extra_block_ids]) : [];
+ const orgBlockText = [...chosen, ...extra].map((b) => b.content).join('\n\n');
+ const context = await letters.mergeContext({ person: check.person, org });
+ const rendered = letters.render({
+ subject_template: tpl.current.subject_template,
+ blocks: tpl.current.blocks,
+ context,
+ orgBlock: orgBlockText,
+ });
+ const citations = [
+ ...rendered.citations,
+ ...[...chosen, ...extra].flatMap((b) => (b.internal_sources || []).map((sid) => ({ block: b.key, source_id: sid }))),
+ ];
+
+ return db.tx(async (client) => {
+ const r = await client.query(
+ `INSERT INTO pr_outreach_messages
+ (person_id, organization_id, campaign_id, direction, subject, rendered_html, rendered_text,
+ draft_blocks, draft_version_id, internal_citations, status, drafted_at, owner_user)
+ VALUES ($1,$2,$3,'outbound',$4,$5,$6,$7,$8,$9,'draft_generated', now(), $10) RETURNING *`,
+ [person_id, org ? org.id : null, campaign_id || null, rendered.subject, rendered.html, rendered.text,
+ JSON.stringify(tpl.current.blocks), tpl.current.id, JSON.stringify(citations), actor || 'admin']);
+ const msg = r.rows[0];
+ await audit.log({ actor, action: 'outreach.draft_generated', entity_type: 'message', entity_id: msg.id,
+ detail: { person_id, campaign_id, missing_merge_fields: rendered.missing, ready: check.ready, readiness_reasons: check.reasons } }, client);
+ await audit.activity({ entity_type: 'person', entity_id: person_id, activity: 'draft_created', detail: { message_id: msg.id }, actor }, client);
+ return { message: msg, missing_merge_fields: rendered.missing, readiness: { ready: check.ready, reasons: check.reasons } };
+ });
+}
+
+async function get(id) {
+ const m = await db.one(
+ `SELECT m.*, p.full_name, p.public_work_email, p.email_verification_status,
+ o.display_name AS organization_name
+ FROM pr_outreach_messages m
+ LEFT JOIN pr_people p ON p.id=m.person_id
+ LEFT JOIN pr_organizations o ON o.id=m.organization_id
+ WHERE m.id=$1`, [id]);
+ return m;
+}
+
+async function list(f = {}) {
+ const where = []; const params = [];
+ const add = (sql, v) => { params.push(v); where.push(sql.replace('?', '$' + params.length)); };
+ if (f.status) add('m.status = ?::pr_outreach_status', f.status);
+ if (f.campaign_id) add('m.campaign_id = ?', Number(f.campaign_id));
+ if (f.person_id) add('m.person_id = ?', Number(f.person_id));
+ if (f.direction) add('m.direction = ?', f.direction);
+ const limit = Math.min(Number(f.limit) || 50, 500);
+ const offset = Number(f.offset) || 0;
+ const whereSql = where.length ? 'WHERE ' + where.join(' AND ') : '';
+ const total = await db.one(`SELECT count(*)::int AS n FROM pr_outreach_messages m ${whereSql}`, params);
+ params.push(limit, offset);
+ const rows = await db.rows(
+ `SELECT m.id, m.person_id, m.organization_id, m.campaign_id, m.direction, m.subject, m.status,
+ m.provider, m.provider_thread_id, m.drafted_at, m.approved_at, m.sent_at, m.replied_at, m.follow_up_at,
+ p.full_name, p.public_work_email, o.display_name AS organization_name, m.created_at
+ FROM pr_outreach_messages m
+ LEFT JOIN pr_people p ON p.id=m.person_id
+ LEFT JOIN pr_organizations o ON o.id=m.organization_id
+ ${whereSql} ORDER BY m.updated_at DESC LIMIT $${params.length - 1} OFFSET $${params.length}`, params);
+ return { total: total.n, rows };
+}
+
+/** Edit a draft's rendered content (before approval). */
+async function updateDraft(id, { subject, rendered_text, rendered_html, follow_up_at, actor }) {
+ const m = await db.one('SELECT * FROM pr_outreach_messages WHERE id=$1', [id]);
+ if (!m) throw new Error('not found');
+ if (!['draft_generated', 'needs_review'].includes(m.status)) throw new Error('only unapproved drafts can be edited');
+ const row = (await db.query(
+ `UPDATE pr_outreach_messages SET subject=COALESCE($2,subject), rendered_text=COALESCE($3,rendered_text),
+ rendered_html=COALESCE($4,rendered_html), follow_up_at=COALESCE($5,follow_up_at), status='needs_review'
+ WHERE id=$1 RETURNING *`, [id, subject || null, rendered_text || null, rendered_html || null, follow_up_at || null])).rows[0];
+ await audit.log({ actor, action: 'outreach.draft_edited', entity_type: 'message', entity_id: id });
+ return row;
+}
+
+async function transition(id, to, actor, detail) {
+ return db.tx(async (client) => {
+ const m = (await client.query('SELECT * FROM pr_outreach_messages WHERE id=$1 FOR UPDATE', [id])).rows[0];
+ if (!m) throw new Error('not found');
+ if (!canOutreachTransition(m.status, to)) throw new Error(`invalid outreach transition ${m.status} → ${to}`);
+ const stamps = {
+ approved: 'approved_at', sent: 'sent_at', replied: 'replied_at',
+ bounced: 'bounced_at', opted_out: 'opted_out_at',
+ };
+ const stamp = stamps[to] ? `, ${stamps[to]} = now()` : '';
+ const by = to === 'approved' ? ', approved_by = $3' : '';
+ const params = [id, to]; if (to === 'approved') params.push(actor || 'admin');
+ const row = (await client.query(`UPDATE pr_outreach_messages SET status=$2::pr_outreach_status${stamp}${by} WHERE id=$1 RETURNING *`, params)).rows[0];
+ await audit.log({ actor, action: 'outreach.status', entity_type: 'message', entity_id: id, before: { status: m.status }, after: { status: to }, detail }, client);
+ return row;
+ });
+}
+
+async function approve(id, actor) {
+ const m = await db.one('SELECT * FROM pr_outreach_messages WHERE id=$1', [id]);
+ if (!m) throw new Error('not found');
+ const check = await readiness(m.person_id);
+ if (!check.ready) throw new Error('not outreach-ready: ' + check.reasons.join('; '));
+ if (m.status === 'draft_generated') await transition(id, 'needs_review', actor);
+ const row = await transition(id, 'approved', actor);
+ await audit.activity({ entity_type: 'person', entity_id: m.person_id, activity: 'approved', detail: { message_id: id }, actor });
+ return row;
+}
+
+/** Create the email-provider draft (Gmail draft or mock). Does NOT send. */
+async function createProviderDraft(id, actor) {
+ const m = await get(id);
+ if (!m) throw new Error('not found');
+ if (m.status !== 'approved') throw new Error('message must be approved first');
+ const check = await readiness(m.person_id);
+ if (!check.ready) throw new Error('not outreach-ready: ' + check.reasons.join('; '));
+ const to = m.public_work_email;
+ if (!to) throw new Error('person has no public work email; use the contact form / manual method instead');
+ // Strip internal citations: outgoing content is the rendered text/html only (citations
+ // were never merged into the rendered body — they live on internal_citations).
+ const provider = providers.getProvider();
+ const d = await provider.createDraft({ to, subject: m.subject, html: m.rendered_html, text: m.rendered_text });
+ const row = (await db.query(
+ `UPDATE pr_outreach_messages SET provider=$2, provider_message_id=$3, provider_thread_id=$4,
+ status='provider_draft_created' WHERE id=$1 RETURNING *`,
+ [id, provider.name, d.draft_id, d.thread_id || null])).rows[0];
+ await audit.log({ actor, action: 'outreach.provider_draft_created', entity_type: 'message', entity_id: id, detail: { provider: provider.name, draft_id: d.draft_id } });
+ return { message: row, provider: provider.name, draft_id: d.draft_id };
+}
+
+/**
+ * EXPLICIT send. Requires confirm:true from the admin click. Re-checks suppression at
+ * send time. Research-generated messages are never auto-sent — this is the only path.
+ */
+async function send(id, { confirm, actor }) {
+ if (confirm !== true) throw new Error('send requires explicit confirmation (confirm:true)');
+ const m = await get(id);
+ if (!m) throw new Error('not found');
+ if (m.status !== 'provider_draft_created') throw new Error('create the provider draft first');
+ const sup = await suppression.isSuppressed({ email: m.public_work_email, person_id: m.person_id });
+ if (sup.suppressed) throw new Error('contact became suppressed (' + sup.reason + '); send blocked');
+ const provider = providers.getProvider();
+ if (provider.name !== m.provider) throw new Error(`provider mismatch (draft on ${m.provider}, active ${provider.name})`);
+ const r = await provider.send({ draft_id: m.provider_message_id });
+ const row = await db.tx(async (client) => {
+ const upd = (await client.query(
+ `UPDATE pr_outreach_messages SET status='sent', sent_at=now(),
+ provider_message_id=$2, provider_thread_id=COALESCE($3, provider_thread_id) WHERE id=$1 RETURNING *`,
+ [id, r.message_id, r.thread_id || null])).rows[0];
+ await client.query(`UPDATE pr_people SET lifecycle_status='contacted' WHERE id=$1 AND lifecycle_status IN ('ready_for_outreach','verified')`, [m.person_id]);
+ if (m.organization_id) await client.query(`UPDATE pr_organizations SET lifecycle_status='contacted' WHERE id=$1 AND lifecycle_status='ready_for_outreach'`, [m.organization_id]);
+ await audit.log({ actor, action: 'outreach.sent', entity_type: 'message', entity_id: id, detail: { provider: provider.name, message_id: r.message_id } }, client);
+ await audit.activity({ entity_type: 'person', entity_id: m.person_id, activity: 'sent', detail: { message_id: id }, actor }, client);
+ return upd;
+ });
+ return row;
+}
+
+module.exports = { readiness, generateDraft, get, list, updateDraft, transition, approve, createProviderDraft, send };
diff --git a/src/pr/services/people.js b/src/pr/services/people.js
new file mode 100644
index 00000000..2a6b006c
--- /dev/null
+++ b/src/pr/services/people.js
@@ -0,0 +1,259 @@
+'use strict';
+// People service. Guarantees:
+// - exact public title AND normalized role are both stored
+// - inferred emails can be stored but NEVER become outreach_eligible (structural)
+// - manual corrections preserve the originally-extracted value (original_values)
+// - role changes append to role_history rather than silently overwriting
+const db = require('../db');
+const { normalizePersonName, splitName, normalizeEmail, normalizeLinkedInUrl, normalizePhone, clean } = require('../lib/normalize');
+const { classifyTitle } = require('../lib/taxonomy');
+const { classifyPersonMatch } = require('../lib/dedupe');
+const { personConfidence, outreachPriority } = require('../lib/scoring');
+const { canPersonTransition } = require('../lib/status');
+const sources = require('./sources');
+const audit = require('./audit');
+
+const SENDABLE_EMAIL_STATUSES = ['corroborated', 'company_site_verified', 'press_release_verified', 'manually_verified'];
+
+async function findMatches(candidate) {
+ const norm = normalizePersonName(candidate.full_name);
+ const email = normalizeEmail(candidate.public_work_email);
+ const li = normalizeLinkedInUrl(candidate.linkedin_url);
+ const params = []; const ors = [];
+ if (email) { params.push(email); ors.push(`public_work_email = $${params.length}`); }
+ if (li) { params.push(li); ors.push(`linkedin_url = $${params.length}`); }
+ if (norm) { params.push(norm); ors.push(`normalized_name = $${params.length}`); }
+ if (!ors.length) return [];
+ const rows = await db.rows(`SELECT * FROM pr_people WHERE (${ors.join(' OR ')}) AND lifecycle_status <> 'duplicate' LIMIT 25`, params);
+ return rows
+ .map((p) => ({ person: p, ...classifyPersonMatch({ ...candidate, normalized_name: norm, public_work_email: email, linkedin_url: li }, p) }))
+ .filter((m) => m.verdict !== 'distinct')
+ .sort((a, b) => b.score - a.score);
+}
+
+async function create(data, { evidence, actor } = {}) {
+ const full_name = clean(data.full_name, 200);
+ if (!full_name) throw new Error('full_name required');
+ const names = splitName(full_name);
+ const cls = classifyTitle(data.exact_title);
+ const email = normalizeEmail(data.public_work_email);
+ const li = normalizeLinkedInUrl(data.linkedin_url);
+ // Guard: an email may only enter as 'inferred' or 'unverified' unless evidence says otherwise;
+ // callers can pass a verified status ONLY alongside evidence.
+ let emailStatus = data.email_verification_status || (email ? 'unverified' : 'unverified');
+ if (SENDABLE_EMAIL_STATUSES.includes(emailStatus) && !(evidence && evidence.source)) emailStatus = 'unverified';
+
+ const matches = await findMatches({ ...data, full_name });
+ const exact = matches.find((m) => m.verdict === 'exact');
+ if (exact) {
+ if (evidence && evidence.source) await attachEvidence(exact.person.id, evidence, actor);
+ return { person: exact.person, created: false, matched: 'exact', matches };
+ }
+ const uncertain = matches.filter((m) => m.verdict === 'uncertain');
+
+ const row = await db.tx(async (client) => {
+ const r = await client.query(
+ `INSERT INTO pr_people
+ (organization_id, first_name, middle_name, last_name, full_name, normalized_name,
+ exact_title, normalized_role, department, seniority, office_location, state, metro,
+ public_work_email, email_verification_status, public_business_phone,
+ linkedin_url, linkedin_status, linkedin_indexed_title, linkedin_indexed_snippet,
+ biography_url, verification_status, lifecycle_status, notes, tags, import_batch_id)
+ VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23,$24,$25,$26)
+ RETURNING *`,
+ [data.organization_id || null, names.first_name, names.middle_name, names.last_name,
+ full_name, normalizePersonName(full_name),
+ clean(data.exact_title, 300) || null, cls.normalized_role, cls.department, cls.seniority,
+ clean(data.office_location, 200) || null, clean(data.state, 2) || null, clean(data.metro, 40) || null,
+ email, emailStatus, normalizePhone(data.public_business_phone),
+ li, data.linkedin_status || (li ? 'found_uncorroborated' : 'none'),
+ clean(data.linkedin_indexed_title, 300) || null, clean(data.linkedin_indexed_snippet, 500) || null,
+ clean(data.biography_url, 500) || null,
+ data.verification_status || 'unverified',
+ uncertain.length ? 'needs_verification' : (data.lifecycle_status || 'discovered'),
+ clean(data.notes, 4000) || null, data.tags || [], data.import_batch_id || null]);
+ const person = r.rows[0];
+ if (evidence && evidence.source) {
+ const src = await sources.findOrCreateSource(evidence.source, client);
+ for (const [field, conf] of Object.entries(evidence.fields || { organization: 60, exact_title: 60 })) {
+ await sources.addEvidence({ source_id: src.id, entity_type: 'person', entity_id: person.id, field_name: field, field_value: person[field] != null ? person[field] : null, confidence: conf }, client);
+ }
+ }
+ await audit.log({ actor, action: 'person.create', entity_type: 'person', entity_id: person.id, after: { full_name, exact_title: person.exact_title, organization_id: person.organization_id }, detail: { uncertain_matches: uncertain.map((m) => m.person.id) } }, client);
+ await audit.activity({ entity_type: 'person', entity_id: person.id, activity: 'research', detail: { event: 'created' }, actor }, client);
+ return person;
+ });
+ await refreshScores(row.id);
+ return { person: await get(row.id), created: true, matched: uncertain.length ? 'uncertain' : 'none', matches: uncertain };
+}
+
+async function attachEvidence(personId, evidence, actor) {
+ return db.tx(async (client) => {
+ const src = await sources.findOrCreateSource(evidence.source, client);
+ const out = [];
+ for (const [field, conf] of Object.entries(evidence.fields || {})) {
+ out.push(await sources.addEvidence({ source_id: src.id, entity_type: 'person', entity_id: personId, field_name: field, field_value: evidence.values ? evidence.values[field] : null, confidence: conf }, client));
+ }
+ await audit.log({ actor, action: 'person.evidence_added', entity_type: 'person', entity_id: personId, detail: { source: src.url || src.source_name, fields: Object.keys(evidence.fields || {}) } }, client);
+ return { source: src, evidence: out };
+ });
+}
+
+async function get(id) {
+ const p = await db.one(
+ `SELECT p.*, o.display_name AS organization_name, o.domain AS organization_domain
+ FROM pr_people p LEFT JOIN pr_organizations o ON o.id = p.organization_id WHERE p.id=$1`, [id]);
+ if (!p) return null;
+ p.evidence = await sources.evidenceFor('person', id);
+ p.recent_activity = await audit.recent({ entity_type: 'person', entity_id: id, limit: 30 });
+ p.outreach = await db.rows(`SELECT id, campaign_id, subject, status, drafted_at, sent_at, replied_at FROM pr_outreach_messages WHERE person_id=$1 ORDER BY created_at DESC LIMIT 50`, [id]);
+ p.suppressed = !!(await db.one(`SELECT 1 FROM pr_suppression WHERE (email=$1 AND email IS NOT NULL) OR person_id=$2 LIMIT 1`, [p.public_work_email, id]));
+ return p;
+}
+
+async function list(f = {}) {
+ const where = []; const params = [];
+ const add = (sql, v) => { params.push(v); where.push(sql.replace('?', '$' + params.length)); };
+ if (f.q) {
+ params.push('%' + String(f.q).toLowerCase() + '%');
+ const n = params.length;
+ where.push(`(lower(p.full_name) LIKE $${n} OR lower(coalesce(p.exact_title,'')) LIKE $${n}
+ OR lower(coalesce(p.public_work_email,'')) LIKE $${n} OR lower(coalesce(p.linkedin_url,'')) LIKE $${n}
+ OR lower(coalesce(o.display_name,'')) LIKE $${n})`);
+ }
+ if (f.organization_id) add(`p.organization_id = ?`, Number(f.organization_id));
+ if (f.role) add(`p.normalized_role = ?`, f.role);
+ if (f.department) add(`p.department = ?`, f.department);
+ if (f.metro) add(`p.metro = ?`, f.metro);
+ if (f.state) add(`p.state = ?`, f.state);
+ if (f.status) add(`p.lifecycle_status = ?::pr_person_status`, f.status);
+ if (f.email_status) add(`p.email_verification_status = ?::pr_verification`, f.email_status);
+ if (f.linkedin_status) add(`p.linkedin_status = ?::pr_linkedin_status`, f.linkedin_status);
+ if (f.outreach_eligible != null && f.outreach_eligible !== '') add(`p.outreach_eligible = ?`, f.outreach_eligible === '1' || f.outreach_eligible === true);
+ if (!f.include_duplicates) where.push(`p.lifecycle_status <> 'duplicate'`);
+ const sortable = { name: 'p.full_name', title: 'p.exact_title', role: 'p.normalized_role', status: 'p.lifecycle_status', confidence: 'p.confidence_score', priority: 'p.contact_priority_score', verified: 'p.last_verified_at', created: 'p.created_at', org: 'o.display_name' };
+ const sort = sortable[f.sort] || 'p.contact_priority_score';
+ const dir = f.dir === 'asc' ? 'ASC' : 'DESC';
+ const limit = Math.min(Number(f.limit) || 50, 500);
+ const offset = Number(f.offset) || 0;
+ const whereSql = where.length ? 'WHERE ' + where.join(' AND ') : '';
+ const base = `FROM pr_people p LEFT JOIN pr_organizations o ON o.id = p.organization_id ${whereSql}`;
+ const total = await db.one(`SELECT count(*)::int AS n ${base}`, params);
+ params.push(limit, offset);
+ const rows = await db.rows(
+ `SELECT p.id, p.full_name, p.exact_title, p.normalized_role, p.department, p.seniority,
+ p.state, p.metro, p.public_work_email, p.email_verification_status, p.public_business_phone,
+ p.linkedin_url, p.linkedin_status, p.contact_priority_score, p.confidence_score, p.score_components,
+ p.verification_status, p.lifecycle_status, p.last_verified_at, p.outreach_eligible,
+ p.organization_id, o.display_name AS organization_name, p.created_at, p.updated_at
+ ${base} ORDER BY ${sort} ${dir} NULLS LAST LIMIT $${params.length - 1} OFFSET $${params.length}`, params);
+ return { total: total.n, rows };
+}
+
+/**
+ * Update. Manual corrections preserve the original extracted value in original_values
+ * and title changes append the prior role to role_history.
+ */
+async function update(id, patch, actor) {
+ const before = await db.one('SELECT * FROM pr_people WHERE id=$1', [id]);
+ if (!before) throw new Error('not found');
+ if (patch.lifecycle_status && !canPersonTransition(before.lifecycle_status, patch.lifecycle_status)) {
+ throw new Error(`invalid status transition ${before.lifecycle_status} → ${patch.lifecycle_status}`);
+ }
+ // outreach_eligible may never be set while the email is inferred/unverified
+ if (patch.outreach_eligible === true) {
+ const status = patch.email_verification_status || before.email_verification_status;
+ const willHaveEmail = patch.public_work_email !== undefined ? normalizeEmail(patch.public_work_email) : before.public_work_email;
+ const hasForm = patch.approved_alternate_method === true; // explicit admin-approved non-email method
+ if (!hasForm && !(willHaveEmail && SENDABLE_EMAIL_STATUSES.includes(status))) {
+ throw new Error('outreach_eligible requires a verified business email or an explicitly approved alternate method');
+ }
+ }
+ const originals = { ...(before.original_values || {}) };
+ const history = Array.isArray(before.role_history) ? [...before.role_history] : [];
+ if (patch.exact_title !== undefined && patch.exact_title !== before.exact_title && before.exact_title) {
+ history.push({ exact_title: before.exact_title, normalized_role: before.normalized_role, department: before.department, recorded_at: new Date().toISOString(), replaced_by: actor || 'admin' });
+ }
+ const trackOriginal = ['full_name', 'exact_title', 'public_work_email', 'public_business_phone', 'linkedin_url'];
+ for (const k of trackOriginal) {
+ if (patch[k] !== undefined && patch[k] !== before[k] && originals[k] === undefined) originals[k] = before[k];
+ }
+
+ const sets = []; const params = [];
+ const push = (col, val, cast) => { params.push(val); sets.push(`${col} = $${params.length}${cast || ''}`); };
+ if (patch.full_name !== undefined) {
+ const fn = clean(patch.full_name, 200); const names = splitName(fn);
+ push('full_name', fn); push('normalized_name', normalizePersonName(fn));
+ push('first_name', names.first_name); push('middle_name', names.middle_name); push('last_name', names.last_name);
+ }
+ if (patch.exact_title !== undefined) {
+ const cls = classifyTitle(patch.exact_title);
+ push('exact_title', clean(patch.exact_title, 300) || null);
+ push('normalized_role', cls.normalized_role); push('department', cls.department); push('seniority', cls.seniority);
+ }
+ if (patch.public_work_email !== undefined) push('public_work_email', normalizeEmail(patch.public_work_email));
+ if (patch.email_verification_status !== undefined) push('email_verification_status', patch.email_verification_status, '::pr_verification');
+ if (patch.public_business_phone !== undefined) push('public_business_phone', normalizePhone(patch.public_business_phone));
+ if (patch.linkedin_url !== undefined) push('linkedin_url', normalizeLinkedInUrl(patch.linkedin_url));
+ if (patch.linkedin_status !== undefined) push('linkedin_status', patch.linkedin_status, '::pr_linkedin_status');
+ if (patch.verification_status !== undefined) push('verification_status', patch.verification_status, '::pr_verification');
+ if (patch.lifecycle_status !== undefined) push('lifecycle_status', patch.lifecycle_status, '::pr_person_status');
+ if (patch.organization_id !== undefined) push('organization_id', patch.organization_id);
+ if (patch.office_location !== undefined) push('office_location', clean(patch.office_location, 200) || null);
+ if (patch.state !== undefined) push('state', clean(patch.state, 2) || null);
+ if (patch.metro !== undefined) push('metro', clean(patch.metro, 40) || null);
+ if (patch.biography_url !== undefined) push('biography_url', clean(patch.biography_url, 500) || null);
+ if (patch.notes !== undefined) push('notes', clean(patch.notes, 4000) || null);
+ if (patch.tags !== undefined) push('tags', patch.tags);
+ if (patch.outreach_eligible !== undefined) push('outreach_eligible', !!patch.outreach_eligible);
+ if (patch.suppression_reason !== undefined) push('suppression_reason', clean(patch.suppression_reason, 300) || null);
+ if (patch.last_verified_at !== undefined) push('last_verified_at', patch.last_verified_at);
+ push('original_values', JSON.stringify(originals));
+ push('role_history', JSON.stringify(history));
+ params.push(id);
+ const row = (await db.query(`UPDATE pr_people SET ${sets.join(', ')} WHERE id=$${params.length} RETURNING *`, params)).rows[0];
+ await audit.log({ actor, action: 'person.update', entity_type: 'person', entity_id: id, before: pick(before, Object.keys(patch)), after: pick(row, Object.keys(patch)) });
+ if (patch.exact_title !== undefined || patch.public_work_email !== undefined || patch.linkedin_url !== undefined) {
+ await audit.activity({ entity_type: 'person', entity_id: id, activity: 'manual_verification', detail: { fields: Object.keys(patch) }, actor });
+ }
+ await refreshScores(id);
+ return row;
+}
+
+/** Transactional person merge (winner keeps; loser marked duplicate; evidence/messages move). */
+async function merge(winnerId, loserId, actor) {
+ if (String(winnerId) === String(loserId)) throw new Error('cannot merge a person into themselves');
+ return db.tx(async (client) => {
+ const w = (await client.query('SELECT * FROM pr_people WHERE id=$1 FOR UPDATE', [winnerId])).rows[0];
+ const l = (await client.query('SELECT * FROM pr_people WHERE id=$1 FOR UPDATE', [loserId])).rows[0];
+ if (!w || !l) throw new Error('person not found');
+ await client.query(`UPDATE pr_field_evidence SET entity_id=$1 WHERE entity_type='person' AND entity_id=$2`, [winnerId, loserId]);
+ await client.query('UPDATE pr_outreach_messages SET person_id=$1 WHERE person_id=$2', [winnerId, loserId]);
+ await client.query(`UPDATE pr_campaign_members SET person_id=$1 WHERE person_id=$2
+ AND NOT EXISTS (SELECT 1 FROM pr_campaign_members cm WHERE cm.campaign_id=pr_campaign_members.campaign_id AND cm.person_id=$1)`, [winnerId, loserId]);
+ await client.query('DELETE FROM pr_campaign_members WHERE person_id=$1', [loserId]);
+ const fill = ['public_work_email', 'public_business_phone', 'linkedin_url', 'biography_url', 'exact_title', 'metro', 'state', 'office_location'];
+ for (const k of fill) if (!w[k] && l[k]) await client.query(`UPDATE pr_people SET ${k}=$1 WHERE id=$2`, [l[k], winnerId]);
+ await client.query(`UPDATE pr_people SET lifecycle_status='duplicate', duplicate_of_id=$1 WHERE id=$2`, [winnerId, loserId]);
+ await audit.log({ actor, action: 'person.merge', entity_type: 'person', entity_id: winnerId, detail: { merged_from: loserId } }, client);
+ await audit.activity({ entity_type: 'person', entity_id: winnerId, activity: 'duplicate_resolved', detail: { merged_from: loserId }, actor }, client);
+ return { ok: true, winner: winnerId, merged: loserId };
+ }).then(async (r) => { await refreshScores(winnerId); return r; });
+}
+
+async function refreshScores(id) {
+ const p = await db.one('SELECT * FROM pr_people WHERE id=$1', [id]);
+ if (!p) return null;
+ const org = p.organization_id ? await db.one('SELECT * FROM pr_organizations WHERE id=$1', [p.organization_id]) : null;
+ const ev = await sources.evidenceFor('person', id);
+ const evRows = ev.map((e) => ({ source_type: e.source_type, is_primary_source: e.is_primary_source, source_id: e.source_id }));
+ const conf = personConfidence(p, evRows);
+ const pri = outreachPriority({ org, person: p, evidenceRows: evRows });
+ await db.query(`UPDATE pr_people SET confidence_score=$1, contact_priority_score=$2, score_components=$3 WHERE id=$4`,
+ [conf.score, pri.score, JSON.stringify({ confidence: conf.components, priority: pri.components }), id]);
+ return { confidence: conf, priority: pri };
+}
+
+function pick(obj, keys) { const o = {}; for (const k of keys) if (obj && obj[k] !== undefined) o[k] = obj[k]; return o; }
+
+module.exports = { create, get, list, update, merge, findMatches, attachEvidence, refreshScores, SENDABLE_EMAIL_STATUSES };
diff --git a/src/pr/services/relationships.js b/src/pr/services/relationships.js
new file mode 100644
index 00000000..79b4e204
--- /dev/null
+++ b/src/pr/services/relationships.js
@@ -0,0 +1,63 @@
+'use strict';
+// Organization relationships (agency↔client, parent↔subsidiary, affiliate, vendor).
+// A relationship requires a supporting source; current_status defaults to 'unknown'
+// so an old PR relationship is never implied to be current.
+const db = require('../db');
+const sources = require('./sources');
+const audit = require('./audit');
+
+const TYPES = ['agency_client', 'parent_subsidiary', 'affiliate', 'vendor'];
+
+async function create({ source_organization_id, target_organization_id, relationship_type, date_started, date_ended, current_status, confidence, notes, source, actor }) {
+ if (!TYPES.includes(relationship_type)) throw new Error('relationship_type must be one of ' + TYPES.join(', '));
+ if (String(source_organization_id) === String(target_organization_id)) throw new Error('relationship must join two different organizations');
+ return db.tx(async (client) => {
+ let source_id = null;
+ if (source) { const s = await sources.findOrCreateSource(source, client); source_id = s.id; }
+ const status = date_ended ? 'historical' : (current_status || 'unknown');
+ const r = await client.query(
+ `INSERT INTO pr_org_relationships
+ (source_organization_id, target_organization_id, relationship_type, date_started, date_ended,
+ current_status, source_id, confidence, notes)
+ VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9)
+ ON CONFLICT (source_organization_id, target_organization_id, relationship_type)
+ DO UPDATE SET date_started=COALESCE(EXCLUDED.date_started, pr_org_relationships.date_started),
+ date_ended=COALESCE(EXCLUDED.date_ended, pr_org_relationships.date_ended),
+ current_status=EXCLUDED.current_status,
+ source_id=COALESCE(EXCLUDED.source_id, pr_org_relationships.source_id),
+ confidence=GREATEST(EXCLUDED.confidence, pr_org_relationships.confidence),
+ notes=COALESCE(EXCLUDED.notes, pr_org_relationships.notes)
+ RETURNING *`,
+ [source_organization_id, target_organization_id, relationship_type,
+ date_started || null, date_ended || null, status, source_id, confidence || 50, notes || null]);
+ await audit.log({ actor, action: 'relationship.create', entity_type: 'relationship', entity_id: r.rows[0].id,
+ detail: { source_organization_id, target_organization_id, relationship_type, current_status: status } }, client);
+ return r.rows[0];
+ });
+}
+
+async function list({ organization_id, type, limit = 200 }) {
+ const where = []; const params = [];
+ if (organization_id) { params.push(organization_id); where.push(`(r.source_organization_id=$${params.length} OR r.target_organization_id=$${params.length})`); }
+ if (type) { params.push(type); where.push(`r.relationship_type=$${params.length}`); }
+ params.push(limit);
+ return db.rows(
+ `SELECT r.*, so.display_name AS source_org_name, tg.display_name AS target_org_name,
+ s.url AS source_url, s.source_name AS evidence_source
+ FROM pr_org_relationships r
+ JOIN pr_organizations so ON so.id=r.source_organization_id
+ JOIN pr_organizations tg ON tg.id=r.target_organization_id
+ LEFT JOIN pr_sources s ON s.id=r.source_id
+ ${where.length ? 'WHERE ' + where.join(' AND ') : ''}
+ ORDER BY r.updated_at DESC LIMIT $${params.length}`, params);
+}
+
+async function remove(id, actor) {
+ const before = await db.one('SELECT * FROM pr_org_relationships WHERE id=$1', [id]);
+ if (!before) return { ok: false, error: 'not found' };
+ await db.query('DELETE FROM pr_org_relationships WHERE id=$1', [id]);
+ await audit.log({ actor, action: 'relationship.delete', entity_type: 'relationship', entity_id: id, before });
+ return { ok: true };
+}
+
+module.exports = { create, list, remove, TYPES };
diff --git a/src/pr/services/replies.js b/src/pr/services/replies.js
new file mode 100644
index 00000000..7b2bff39
--- /dev/null
+++ b/src/pr/services/replies.js
@@ -0,0 +1,184 @@
+'use strict';
+// Inbound handling: thread sync, reply association, referrals, bounces, opt-outs.
+// Originals are preserved verbatim (raw headers + body); opt-outs suppress immediately.
+const db = require('../db');
+const audit = require('./audit');
+const suppression = require('./suppression');
+const providers = require('./email-providers');
+const outreach = require('./outreach');
+const people = require('./people');
+const { normalizeEmail } = require('../lib/normalize');
+
+/**
+ * Sync provider threads for all sent messages that have a provider_thread_id.
+ * New inbound messages are stored as direction='inbound' rows linked to the same
+ * person/org/campaign, associated by thread id (primary), then headers/recipient.
+ */
+async function syncThreads({ actor = 'system:sync-email-threads', limit = 50 } = {}) {
+ const provider = providers.getProvider();
+ const sent = await db.rows(
+ `SELECT DISTINCT ON (provider_thread_id) id, person_id, organization_id, campaign_id, provider_thread_id, provider
+ FROM pr_outreach_messages
+ WHERE direction='outbound' AND provider_thread_id IS NOT NULL AND status IN ('sent','replied','follow_up_due')
+ ORDER BY provider_thread_id, sent_at DESC NULLS LAST LIMIT $1`, [limit]);
+ let newReplies = 0; const errors = [];
+ for (const m of sent) {
+ try {
+ const thread = await provider.getThread(m.provider_thread_id);
+ for (const msg of thread.messages) {
+ const exists = await db.one('SELECT id FROM pr_outreach_messages WHERE provider_message_id=$1', [msg.message_id]);
+ if (exists) continue;
+ const fromHdr = (msg.headers && (msg.headers.from || '')) || '';
+ const ourAddr = (process.env.PR_GMAIL_USER || '').toLowerCase();
+ const isInbound = ourAddr ? !fromHdr.toLowerCase().includes(ourAddr) : true;
+ if (!isInbound) continue;
+ await associateInbound({
+ provider: provider.name, provider_message_id: msg.message_id, provider_thread_id: m.provider_thread_id,
+ person_id: m.person_id, organization_id: m.organization_id, campaign_id: m.campaign_id,
+ subject: msg.headers && msg.headers.subject, text: msg.text, html: msg.html, headers: msg.headers, actor,
+ });
+ newReplies++;
+ }
+ } catch (e) { errors.push({ thread: m.provider_thread_id, error: e.message }); }
+ }
+ return { threads_checked: sent.length, new_replies: newReplies, errors };
+}
+
+/** Store an inbound message + flip the parent outbound to 'replied' + detect bounce/opt-out. */
+async function associateInbound({ provider, provider_message_id, provider_thread_id, person_id, organization_id, campaign_id, subject, text, html, headers, actor }) {
+ return db.tx(async (client) => {
+ const r = await client.query(
+ `INSERT INTO pr_outreach_messages
+ (person_id, organization_id, campaign_id, direction, provider, provider_message_id, provider_thread_id,
+ subject, rendered_text, rendered_html, body_preserved, raw_headers, status, replied_at)
+ VALUES ($1,$2,$3,'inbound',$4,$5,$6,$7,$8,$9,$10,$11,'replied', now()) RETURNING *`,
+ [person_id, organization_id, campaign_id, provider, provider_message_id, provider_thread_id,
+ subject || null, text || null, html || null, text || html || null, JSON.stringify(headers || {})]);
+ const inbound = r.rows[0];
+ // Flip the latest outbound in this thread to replied (if legal).
+ const out = (await client.query(
+ `SELECT id, status FROM pr_outreach_messages
+ WHERE provider_thread_id=$1 AND direction='outbound' ORDER BY sent_at DESC NULLS LAST LIMIT 1`, [provider_thread_id])).rows[0];
+ if (out && ['sent', 'follow_up_due'].includes(out.status)) {
+ await client.query(`UPDATE pr_outreach_messages SET status='replied', replied_at=now() WHERE id=$1`, [out.id]);
+ }
+ if (person_id) {
+ await client.query(`UPDATE pr_people SET lifecycle_status='replied' WHERE id=$1 AND lifecycle_status='contacted'`, [person_id]);
+ await audit.activity({ entity_type: 'person', entity_id: person_id, activity: 'reply_received', detail: { message_id: inbound.id }, actor }, client);
+ }
+ await audit.log({ actor, action: 'reply.associated', entity_type: 'message', entity_id: inbound.id, detail: { thread: provider_thread_id } }, client);
+
+ // Bounce detection (mailer-daemon) and opt-out detection (unsubscribe language).
+ const blob = ((subject || '') + ' ' + (text || '')).toLowerCase();
+ const fromHdr = ((headers || {}).from || '').toLowerCase();
+ if (/mailer-daemon|postmaster@|delivery status notification|undeliverable|address not found/.test(fromHdr + ' ' + blob)) {
+ if (out) await client.query(`UPDATE pr_outreach_messages SET status='bounced', bounced_at=now() WHERE id=$1 AND status IN ('sent','replied','follow_up_due')`, [out.id]);
+ if (person_id) {
+ const p = (await client.query('SELECT public_work_email FROM pr_people WHERE id=$1', [person_id])).rows[0];
+ await suppression.add({ email: p && p.public_work_email, person_id, reason: 'bounce', source: 'reply', permanent: false, actor }, client);
+ await audit.activity({ entity_type: 'person', entity_id: person_id, activity: 'bounced', detail: { message_id: inbound.id }, actor }, client);
+ }
+ } else if (/\bunsubscribe\b|remove me|opt.?out|stop (contacting|emailing)|do not (contact|email)/.test(blob)) {
+ if (out) await client.query(`UPDATE pr_outreach_messages SET status='opted_out', opted_out_at=now() WHERE id=$1 AND status IN ('sent','replied','follow_up_due')`, [out.id]);
+ if (person_id) {
+ const p = (await client.query('SELECT public_work_email FROM pr_people WHERE id=$1', [person_id])).rows[0];
+ await suppression.add({ email: p && p.public_work_email, person_id, reason: 'opt_out', source: 'reply', permanent: true, actor }, client);
+ }
+ }
+ return inbound;
+ });
+}
+
+/** Full thread for the inbox viewer (outbound + inbound, oldest first). */
+async function thread(threadId) {
+ return db.rows(
+ `SELECT m.*, p.full_name, o.display_name AS organization_name
+ FROM pr_outreach_messages m
+ LEFT JOIN pr_people p ON p.id=m.person_id
+ LEFT JOIN pr_organizations o ON o.id=m.organization_id
+ WHERE m.provider_thread_id=$1 ORDER BY coalesce(m.sent_at, m.replied_at, m.created_at)`, [threadId]);
+}
+
+/** Inbox: threads needing action (replies newest first) + follow-ups due. */
+async function inbox({ limit = 50 } = {}) {
+ const replies = await db.rows(
+ `SELECT m.id, m.provider_thread_id, m.subject, m.replied_at, m.status, m.person_id, m.organization_id,
+ p.full_name, o.display_name AS organization_name,
+ left(coalesce(m.rendered_text, ''), 200) AS snippet
+ FROM pr_outreach_messages m
+ LEFT JOIN pr_people p ON p.id=m.person_id
+ LEFT JOIN pr_organizations o ON o.id=m.organization_id
+ WHERE m.direction='inbound' ORDER BY m.replied_at DESC NULLS LAST LIMIT $1`, [limit]);
+ const followups = await db.rows(
+ `SELECT m.id, m.provider_thread_id, m.subject, m.follow_up_at, m.status, p.full_name, o.display_name AS organization_name
+ FROM pr_outreach_messages m
+ LEFT JOIN pr_people p ON p.id=m.person_id
+ LEFT JOIN pr_organizations o ON o.id=m.organization_id
+ WHERE m.direction='outbound' AND m.follow_up_at IS NOT NULL AND m.follow_up_at <= now()
+ AND m.status IN ('sent','follow_up_due','replied','not_now')
+ ORDER BY m.follow_up_at LIMIT $1`, [limit]);
+ return { replies, followups };
+}
+
+/**
+ * Referral: the current contact pointed us to someone else. Creates the referred
+ * person (unverified, evidenced by the reply itself), marks the message referred,
+ * and opens a follow-up task.
+ */
+async function processReferral({ message_id, referred_name, referred_title, referred_email, actor }) {
+ const m = await outreach.get(message_id);
+ if (!m) throw new Error('message not found');
+ const created = await people.create({
+ full_name: referred_name,
+ exact_title: referred_title,
+ public_work_email: referred_email,
+ email_verification_status: referred_email ? 'corroborated' : 'unverified', // named directly by a colleague in-thread
+ organization_id: m.organization_id,
+ lifecycle_status: 'needs_verification',
+ }, {
+ evidence: {
+ source: {
+ source_type: 'manual', source_name: 'Email referral (thread ' + (m.provider_thread_id || m.id) + ')',
+ short_excerpt: `Referred by ${m.full_name || 'contact'} in reply to "${m.subject}"`,
+ is_primary_source: false, adapter: 'reply-referral',
+ },
+ fields: { organization: 70, exact_title: referred_title ? 55 : 0, public_work_email: referred_email ? 75 : 0 },
+ },
+ actor,
+ });
+ await outreach.transition(message_id, 'referred', actor, { referred_person_id: created.person.id });
+ await db.query(
+ `INSERT INTO pr_tasks (title, detail, entity_type, entity_id, due_at, created_by)
+ VALUES ($1,$2,'person',$3, now() + interval '2 days', $4)`,
+ [`Follow up with referred contact: ${referred_name}`,
+ `Referred from message #${message_id} (${m.subject}). Verify title/email, then draft outreach.`,
+ created.person.id, actor || 'admin']);
+ await audit.activity({ entity_type: 'person', entity_id: created.person.id, activity: 'reassigned', detail: { referred_from_message: message_id }, actor });
+ return { referred_person: created.person, message_id };
+}
+
+/** Compose a reply in-thread → provider draft (explicit send stays separate). */
+async function composeReply({ thread_id, body_text, body_html, subject, actor }) {
+ const msgs = await thread(thread_id);
+ if (!msgs.length) throw new Error('thread not found');
+ const root = msgs.find((x) => x.direction === 'outbound') || msgs[0];
+ const to = root.public_work_email || (await db.one('SELECT public_work_email FROM pr_people WHERE id=$1', [root.person_id]) || {}).public_work_email;
+ if (!to) throw new Error('no recipient email on this thread');
+ const sup = await suppression.isSuppressed({ email: to, person_id: root.person_id });
+ if (sup.suppressed) throw new Error('contact is suppressed (' + sup.reason + '); reply blocked');
+ const provider = providers.getProvider();
+ const subj = subject || (root.subject && root.subject.startsWith('Re:') ? root.subject : 'Re: ' + (root.subject || ''));
+ const d = await provider.createDraft({ to, subject: subj, text: body_text, html: body_html || undefined, threadId: thread_id });
+ const row = (await db.query(
+ `INSERT INTO pr_outreach_messages
+ (person_id, organization_id, campaign_id, direction, provider, provider_message_id, provider_thread_id,
+ subject, rendered_text, rendered_html, status, drafted_at, owner_user)
+ VALUES ($1,$2,$3,'outbound',$4,$5,$6,$7,$8,$9,'provider_draft_created', now(), $10) RETURNING *`,
+ [root.person_id, root.organization_id, root.campaign_id, provider.name, d.draft_id, thread_id,
+ subj, body_text || null, body_html || null, actor || 'admin'])).rows[0];
+ // Reply drafts skip generate→approve (human-authored), but sending still requires the explicit send call.
+ await audit.log({ actor, action: 'outreach.reply_draft_created', entity_type: 'message', entity_id: row.id, detail: { thread_id } });
+ return { message: row, draft_id: d.draft_id, provider: provider.name };
+}
+
+module.exports = { syncThreads, associateInbound, thread, inbox, processReferral, composeReply };
diff --git a/src/pr/services/runs.js b/src/pr/services/runs.js
new file mode 100644
index 00000000..3bacdcd0
--- /dev/null
+++ b/src/pr/services/runs.js
@@ -0,0 +1,72 @@
+'use strict';
+// Research runs: the resumable unit of discovery work. A run belongs to a
+// state/metro/category/adapter, carries a checkpoint (cursor), and can be
+// paused/resumed/cancelled from the admin.
+const db = require('../db');
+const audit = require('./audit');
+
+async function create({ state, metro, category, query, adapter, actor }) {
+ const row = (await db.query(
+ `INSERT INTO pr_research_runs (state, metro, category, query, adapter, status)
+ VALUES ($1,$2,$3,$4,$5,'queued') RETURNING *`,
+ [state, metro || null, category || null, query || null, adapter])).rows[0];
+ await audit.log({ actor, action: 'run.create', entity_type: 'run', entity_id: row.id, after: { state, metro, category, adapter } });
+ return row;
+}
+
+async function get(id) { return db.one('SELECT * FROM pr_research_runs WHERE id=$1', [id]); }
+
+async function list({ state, status, limit = 100 } = {}) {
+ const where = []; const params = [];
+ if (state) { params.push(state); where.push(`state=$${params.length}`); }
+ if (status) { params.push(status); where.push(`status=$${params.length}::pr_run_status`); }
+ params.push(limit);
+ return db.rows(
+ `SELECT * FROM pr_research_runs ${where.length ? 'WHERE ' + where.join(' AND ') : ''}
+ ORDER BY created_at DESC LIMIT $${params.length}`, params);
+}
+
+/** Progress update from inside a job — counters are cumulative, checkpoint replaces. */
+async function progress(id, { pages, candidates, created, updated, duplicates, error, checkpoint }) {
+ const sets = ['updated_at = now()']; const params = [id];
+ const bump = (col, v) => { if (v) { params.push(v); sets.push(`${col} = ${col} + $${params.length}`); } };
+ bump('pages_checked', pages); bump('candidates_found', candidates);
+ bump('records_created', created); bump('records_updated', updated); bump('duplicates_detected', duplicates);
+ if (checkpoint !== undefined) { params.push(JSON.stringify(checkpoint)); sets.push(`checkpoint = $${params.length}`); }
+ if (error) { params.push(JSON.stringify({ at: new Date().toISOString(), error: String(error).slice(0, 500) })); sets.push(`errors = errors || $${params.length}::jsonb`); }
+ return (await db.query(`UPDATE pr_research_runs SET ${sets.join(', ')} WHERE id=$1 RETURNING *`, params)).rows[0];
+}
+
+async function setStatus(id, status, actor) {
+ const stamps = status === 'running' ? ', started_at = COALESCE(started_at, now())'
+ : ['completed', 'failed', 'cancelled'].includes(status) ? ', completed_at = now()' : '';
+ const row = (await db.query(
+ `UPDATE pr_research_runs SET status=$2::pr_run_status${stamps} WHERE id=$1 RETURNING *`, [id, status])).rows[0];
+ await audit.log({ actor: actor || 'system', action: 'run.status', entity_type: 'run', entity_id: id, after: { status } });
+ return row;
+}
+
+/** Pause / resume from the admin — pausing also pauses the run's queued jobs. */
+async function pause(id, actor) {
+ await db.query(`UPDATE pr_jobs SET status='paused' WHERE run_id=$1 AND status IN ('queued','running')`, [id]);
+ return setStatus(id, 'paused', actor);
+}
+async function resume(id, actor) {
+ await db.query(`UPDATE pr_jobs SET status='queued' WHERE run_id=$1 AND status='paused'`, [id]);
+ return setStatus(id, 'queued', actor);
+}
+
+/** Health rollup for the dashboard. */
+async function health() {
+ return db.one(
+ `SELECT count(*) FILTER (WHERE status='running')::int AS running,
+ count(*) FILTER (WHERE status='queued')::int AS queued,
+ count(*) FILTER (WHERE status='paused')::int AS paused,
+ count(*) FILTER (WHERE status='failed')::int AS failed,
+ count(*) FILTER (WHERE status='completed')::int AS completed,
+ (SELECT count(*) FROM pr_jobs WHERE status='failed')::int AS failed_jobs,
+ (SELECT count(*) FROM pr_jobs WHERE status='queued')::int AS queued_jobs
+ FROM pr_research_runs`, []);
+}
+
+module.exports = { create, get, list, progress, setStatus, pause, resume, health };
diff --git a/src/pr/services/settings.js b/src/pr/services/settings.js
new file mode 100644
index 00000000..1fff050b
--- /dev/null
+++ b/src/pr/services/settings.js
@@ -0,0 +1,113 @@
+'use strict';
+// Settings + the California quality gate + the Arizona unlock.
+const db = require('../db');
+const audit = require('./audit');
+
+async function get(key, fallback = null) {
+ const row = await db.one('SELECT value FROM pr_settings WHERE key=$1', [key]);
+ return row ? row.value : fallback;
+}
+async function getAll() {
+ const rows = await db.rows('SELECT key, value, updated_at, updated_by FROM pr_settings ORDER BY key', []);
+ return Object.fromEntries(rows.map((r) => [r.key, { value: r.value, updated_at: r.updated_at, updated_by: r.updated_by }]));
+}
+async function set(key, value, actor) {
+ const before = await get(key);
+ await db.query(
+ `INSERT INTO pr_settings (key, value, updated_by) VALUES ($1,$2,$3)
+ ON CONFLICT (key) DO UPDATE SET value=$2, updated_at=now(), updated_by=$3`,
+ [key, JSON.stringify(value), actor || 'admin']);
+ await audit.log({ actor, action: 'settings.set', entity_type: 'settings', detail: { key, before, after: value } });
+ return { key, value };
+}
+
+/**
+ * The CALIFORNIA QUALITY GATE (computed live from real data, never hand-set):
+ * 1. resumability — pipeline has completed at least one run with a checkpoint (or resumed one)
+ * 2. dedupe working — at least one duplicate detected+resolved OR zero unresolved dup candidates
+ * 3. ≥80% of REVIEWED CA orgs have name+website+category+metro/county+≥1 source
+ * 4. ≥70% of high-priority CA orgs have ≥1 relevant contact (where publicly available)
+ * 5. workflow surfaces exercised (review, letters, drafts, opt-out, audit) — tests cover these;
+ * here we check the audit log has entries for the key action families
+ * 6. no inferred/generated contact mislabeled verified (structural query)
+ * 7. dashboard shows remaining coverage gaps (always true once the dashboard is live — reported)
+ */
+async function californiaGate() {
+ const checks = {};
+
+ const runs = await db.one(
+ `SELECT count(*) FILTER (WHERE status='completed') AS completed,
+ count(*) FILTER (WHERE checkpoint <> '{}'::jsonb) AS checkpointed
+ FROM pr_research_runs WHERE state='CA'`, []);
+ checks.pipeline_resumable = Number(runs.completed) > 0 && Number(runs.checkpointed) > 0;
+
+ const dups = await db.one(
+ `SELECT count(*) FILTER (WHERE lifecycle_status='duplicate') AS resolved,
+ count(*) FILTER (WHERE duplicate_of_id IS NOT NULL) AS linked
+ FROM pr_organizations`, []);
+ checks.dedupe_working = Number(dups.resolved) > 0 || Number(dups.linked) > 0;
+
+ const orgQ = await db.one(
+ `WITH reviewed AS (
+ SELECT o.*, EXISTS (SELECT 1 FROM pr_field_evidence e WHERE e.entity_type='organization' AND e.entity_id=o.id) AS has_evidence
+ FROM pr_organizations o
+ WHERE 'CA' = ANY(o.state_presence)
+ AND o.lifecycle_status IN ('verified','ready_for_outreach','contacted','active_relationship'))
+ SELECT count(*) AS n,
+ count(*) FILTER (WHERE display_name IS NOT NULL AND website_url IS NOT NULL
+ AND organization_type IS NOT NULL
+ AND (cardinality(metros)>0 OR cardinality(counties)>0)
+ AND has_evidence) AS complete
+ FROM reviewed`, []);
+ const nReviewed = Number(orgQ.n), nComplete = Number(orgQ.complete);
+ checks.org_completeness = { reviewed: nReviewed, complete: nComplete, pct: nReviewed ? Math.round(nComplete / nReviewed * 100) : 0, pass: nReviewed > 0 && nComplete / nReviewed >= 0.8 };
+
+ const hi = await db.one(
+ `WITH hi AS (SELECT id FROM pr_organizations WHERE 'CA'=ANY(state_presence) AND priority_score >= 60
+ AND lifecycle_status NOT IN ('duplicate','archived'))
+ SELECT count(*) AS n,
+ count(*) FILTER (WHERE EXISTS (
+ SELECT 1 FROM pr_people p WHERE p.organization_id = hi.id
+ AND p.lifecycle_status NOT IN ('duplicate','suppressed','wrong_person','left_company')
+ AND p.department IN ('communications','marketing','leadership','bd','agency'))) AS with_contact
+ FROM hi`, []);
+ const nHi = Number(hi.n), nWith = Number(hi.with_contact);
+ checks.contact_coverage = { high_priority: nHi, with_contact: nWith, pct: nHi ? Math.round(nWith / nHi * 100) : 0, pass: nHi > 0 && nWith / nHi >= 0.7 };
+
+ const flows = await db.one(
+ `SELECT count(*) FILTER (WHERE action LIKE 'review.%') AS review,
+ count(*) FILTER (WHERE action LIKE 'letter.%') AS letters,
+ count(*) FILTER (WHERE action LIKE 'outreach.%') AS outreach,
+ count(*) FILTER (WHERE action LIKE 'suppression.%') AS suppression
+ FROM pr_audit_log`, []);
+ checks.workflows_exercised = ['review', 'letters', 'outreach', 'suppression'].every((k) => Number(flows[k]) > 0);
+
+ const misl = await db.one(
+ `SELECT count(*) AS n FROM pr_people
+ WHERE email_verification_status='inferred' AND outreach_eligible = true`, []);
+ checks.no_mislabeled_inferred = Number(misl.n) === 0;
+
+ checks.dashboard_gaps_visible = true; // /api/pr/dashboard computes coverage gaps live
+
+ const passed = checks.pipeline_resumable && checks.dedupe_working &&
+ checks.org_completeness.pass && checks.contact_coverage.pass &&
+ checks.workflows_exercised && checks.no_mislabeled_inferred;
+ const result = { passed, checked_at: new Date().toISOString(), checks };
+ await set('california_gate', result, 'system:gate');
+ return result;
+}
+
+/** Arizona is unlocked ONLY by a passing CA gate or an explicit, audit-logged admin override. */
+async function unlockArizona({ actor, override = false }) {
+ const gate = await californiaGate();
+ if (!gate.passed && !override) {
+ return { ok: false, unlocked: false, gate, error: 'California quality gate has not passed; pass it or use an explicit override.' };
+ }
+ await set('arizona_unlocked', true, actor);
+ await audit.log({ actor, action: override ? 'gate.arizona_override_unlock' : 'gate.arizona_unlock', detail: { gate: gate.passed } });
+ return { ok: true, unlocked: true, override: !gate.passed, gate };
+}
+
+async function arizonaUnlocked() { return (await get('arizona_unlocked', false)) === true; }
+
+module.exports = { get, getAll, set, californiaGate, unlockArizona, arizonaUnlocked };
diff --git a/src/pr/services/sources.js b/src/pr/services/sources.js
new file mode 100644
index 00000000..58c5628a
--- /dev/null
+++ b/src/pr/services/sources.js
@@ -0,0 +1,71 @@
+'use strict';
+// Source records + field-level evidence. Every important fact must point at a row here.
+// We store the URL, a LIMITED evidentiary excerpt, metadata, and a content hash —
+// never complete copies of copyrighted pages.
+const crypto = require('crypto');
+const db = require('../db');
+
+const EXCERPT_MAX = 600; // hard cap on stored excerpt length
+
+async function createSource(s, client) {
+ const q = `INSERT INTO pr_sources
+ (source_type, source_name, url, page_title, publisher, published_at, retrieved_at,
+ short_excerpt, content_hash, license_or_usage_note, is_primary_source, adapter)
+ VALUES ($1,$2,$3,$4,$5,$6,COALESCE($7, now()),$8,$9,$10,$11,$12) RETURNING *`;
+ const excerpt = s.short_excerpt ? String(s.short_excerpt).slice(0, EXCERPT_MAX) : null;
+ const hash = s.content_hash || (excerpt ? crypto.createHash('sha256').update(excerpt).digest('hex').slice(0, 32) : null);
+ const params = [
+ s.source_type, s.source_name, s.url || null, s.page_title || null, s.publisher || null,
+ s.published_at || null, s.retrieved_at || null, excerpt, hash,
+ s.license_or_usage_note || null, !!s.is_primary_source, s.adapter || null,
+ ];
+ const r = client ? await client.query(q, params) : await db.query(q, params);
+ return r.rows[0];
+}
+
+/** Reuse an existing source row for the same URL retrieved recently, else create. */
+async function findOrCreateSource(s, client) {
+ if (s.url) {
+ const q = `SELECT * FROM pr_sources WHERE url=$1 AND source_type=$2
+ AND retrieved_at > now() - interval '7 days' ORDER BY retrieved_at DESC LIMIT 1`;
+ const r = client ? await client.query(q, [s.url, s.source_type]) : await db.query(q, [s.url, s.source_type]);
+ if (r.rows[0]) return r.rows[0];
+ }
+ return createSource(s, client);
+}
+
+/** Attach evidence: source S supports field F=V on entity E at confidence C. */
+async function addEvidence({ source_id, entity_type, entity_id, field_name, field_value, confidence }, client) {
+ const q = `INSERT INTO pr_field_evidence (source_id, entity_type, entity_id, field_name, field_value, confidence)
+ VALUES ($1,$2,$3,$4,$5,$6) RETURNING *`;
+ const params = [source_id, entity_type, entity_id, field_name,
+ field_value == null ? null : String(field_value).slice(0, 1000), confidence || 50];
+ const r = client ? await client.query(q, params) : await db.query(q, params);
+ return r.rows[0];
+}
+
+/** All evidence rows (joined with source metadata) for an entity. */
+async function evidenceFor(entity_type, entity_id) {
+ return db.rows(
+ `SELECT e.id AS evidence_id, e.field_name, e.field_value, e.confidence, e.created_at,
+ s.id AS source_id, s.source_type, s.source_name, s.url, s.page_title, s.publisher,
+ s.published_at, s.retrieved_at, s.short_excerpt, s.is_primary_source, s.license_or_usage_note, s.status
+ FROM pr_field_evidence e JOIN pr_sources s ON s.id = e.source_id
+ WHERE e.entity_type=$1 AND e.entity_id=$2
+ ORDER BY e.created_at DESC`, [entity_type, entity_id]);
+}
+
+async function listSources({ q, type, limit = 100, offset = 0 }) {
+ const where = []; const params = [];
+ if (q) { params.push('%' + q.toLowerCase() + '%'); where.push(`(lower(source_name) LIKE $${params.length} OR lower(url) LIKE $${params.length})`); }
+ if (type) { params.push(type); where.push(`source_type = $${params.length}`); }
+ params.push(limit, offset);
+ const rows = await db.rows(
+ `SELECT s.*, (SELECT count(*) FROM pr_field_evidence e WHERE e.source_id=s.id)::int AS evidence_count
+ FROM pr_sources s ${where.length ? 'WHERE ' + where.join(' AND ') : ''}
+ ORDER BY s.retrieved_at DESC LIMIT $${params.length - 1} OFFSET $${params.length}`, params);
+ const total = await db.one(`SELECT count(*)::int AS n FROM pr_sources ${where.length ? 'WHERE ' + where.join(' AND ') : ''}`, params.slice(0, params.length - 2));
+ return { total: total.n, rows };
+}
+
+module.exports = { createSource, findOrCreateSource, addEvidence, evidenceFor, listSources, EXCERPT_MAX };
diff --git a/src/pr/services/suppression.js b/src/pr/services/suppression.js
new file mode 100644
index 00000000..042b02fa
--- /dev/null
+++ b/src/pr/services/suppression.js
@@ -0,0 +1,61 @@
+'use strict';
+// Suppression list: opt-outs, bounces, do-not-contact. Checked before ANY outreach.
+const db = require('../db');
+const { normalizeEmail, emailDomain } = require('../lib/normalize');
+const audit = require('./audit');
+
+async function add({ email, domain, person_id, organization_id, reason, source, permanent = true, expires_at, actor }, client) {
+ const e = normalizeEmail(email);
+ const run = async (c) => {
+ const r = await c.query(
+ `INSERT INTO pr_suppression (email, domain, person_id, organization_id, reason, source, permanent, expires_at, created_by)
+ VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9)
+ ON CONFLICT (email) WHERE email IS NOT NULL DO UPDATE SET reason=EXCLUDED.reason, permanent=EXCLUDED.permanent
+ RETURNING *`,
+ [e, domain || null, person_id || null, organization_id || null, reason || 'manual', source || 'admin', permanent, expires_at || null, actor || 'admin']);
+ if (person_id) {
+ await c.query(`UPDATE pr_people SET lifecycle_status='suppressed', outreach_eligible=false, suppression_reason=$2 WHERE id=$1`, [person_id, reason || 'manual']);
+ } else if (e) {
+ await c.query(`UPDATE pr_people SET lifecycle_status='suppressed', outreach_eligible=false, suppression_reason=$2 WHERE public_work_email=$1`, [e, reason || 'manual']);
+ }
+ await audit.log({ actor, action: 'suppression.add', entity_type: 'suppression', entity_id: r.rows[0].id, detail: { email: e, domain, reason } }, c);
+ if (person_id) await audit.activity({ entity_type: 'person', entity_id: person_id, activity: 'opted_out', detail: { reason }, actor }, c);
+ return r.rows[0];
+ };
+ return client ? run(client) : db.tx(run);
+}
+
+/** Is this email/person suppressed right now? (expired temporary rows don't count) */
+async function isSuppressed({ email, person_id }) {
+ const e = normalizeEmail(email);
+ const d = e ? emailDomain(e) : null;
+ const row = await db.one(
+ `SELECT id, reason, permanent FROM pr_suppression
+ WHERE (permanent = true OR expires_at IS NULL OR expires_at > now())
+ AND ( (email IS NOT NULL AND email = $1)
+ OR (domain IS NOT NULL AND domain = $2)
+ OR (person_id IS NOT NULL AND person_id = $3) )
+ LIMIT 1`, [e, d, person_id || null]);
+ return row ? { suppressed: true, reason: row.reason, permanent: row.permanent } : { suppressed: false };
+}
+
+async function list({ limit = 200, offset = 0 } = {}) {
+ const total = await db.one('SELECT count(*)::int AS n FROM pr_suppression', []);
+ const rows = await db.rows(
+ `SELECT s.*, p.full_name FROM pr_suppression s LEFT JOIN pr_people p ON p.id=s.person_id
+ ORDER BY s.created_at DESC LIMIT $1 OFFSET $2`, [limit, offset]);
+ return { total: total.n, rows };
+}
+
+async function remove(id, actor) {
+ const before = await db.one('SELECT * FROM pr_suppression WHERE id=$1', [id]);
+ if (!before) return { ok: false, error: 'not found' };
+ if (before.permanent && before.reason === 'opt_out') {
+ return { ok: false, error: 'permanent opt-outs cannot be removed (CAN-SPAM); add a note instead' };
+ }
+ await db.query('DELETE FROM pr_suppression WHERE id=$1', [id]);
+ await audit.log({ actor, action: 'suppression.remove', entity_type: 'suppression', entity_id: id, before });
+ return { ok: true };
+}
+
+module.exports = { add, isSuppressed, list, remove };
diff --git a/src/pr/services/tasks.js b/src/pr/services/tasks.js
new file mode 100644
index 00000000..876bba4f
--- /dev/null
+++ b/src/pr/services/tasks.js
@@ -0,0 +1,34 @@
+'use strict';
+// Admin tasks (follow-ups, verification chores). Small by design.
+const db = require('../db');
+const audit = require('./audit');
+
+async function create({ title, detail, entity_type, entity_id, due_at, assignee, actor }) {
+ const row = (await db.query(
+ `INSERT INTO pr_tasks (title, detail, entity_type, entity_id, due_at, assignee, created_by)
+ VALUES ($1,$2,$3,$4,$5,$6,$7) RETURNING *`,
+ [title, detail || null, entity_type || null, entity_id || null, due_at || null, assignee || null, actor || 'admin'])).rows[0];
+ await audit.log({ actor, action: 'task.create', entity_type: 'task', entity_id: row.id, after: { title } });
+ return row;
+}
+
+async function list({ status = 'open', limit = 200 } = {}) {
+ return db.rows(
+ `SELECT t.*,
+ CASE t.entity_type WHEN 'person' THEN (SELECT full_name FROM pr_people WHERE id=t.entity_id)
+ WHEN 'organization' THEN (SELECT display_name FROM pr_organizations WHERE id=t.entity_id)
+ ELSE NULL END AS entity_name
+ FROM pr_tasks t WHERE ($1 = 'all' OR t.status = $1)
+ ORDER BY t.due_at NULLS LAST, t.created_at DESC LIMIT $2`, [status, limit]);
+}
+
+async function setStatus(id, status, actor) {
+ if (!['open', 'done', 'dismissed'].includes(status)) throw new Error('bad status');
+ const row = (await db.query(
+ `UPDATE pr_tasks SET status=$2, completed_at = CASE WHEN $2='done' THEN now() ELSE completed_at END WHERE id=$1 RETURNING *`,
+ [id, status])).rows[0];
+ await audit.log({ actor, action: 'task.status', entity_type: 'task', entity_id: id, after: { status } });
+ return row;
+}
+
+module.exports = { create, list, setStatus };
diff --git a/src/pr/worker.js b/src/pr/worker.js
new file mode 100644
index 00000000..e438d6fc
--- /dev/null
+++ b/src/pr/worker.js
@@ -0,0 +1,43 @@
+'use strict';
+// PR research worker — polls the pr_jobs queue and runs handlers. Run with:
+// npm run pr:worker
+// Safe to run alongside the web app (FOR UPDATE SKIP LOCKED claims) and safe to
+// Ctrl-C: in-flight jobs checkpoint, and an interrupted 'running' job is re-queued
+// on next boot (stale-claim recovery below).
+const db = require('./db');
+const jobs = require('./jobs');
+
+const POLL_MS = Number(process.env.PR_WORKER_POLL_MS || 3000);
+let stopping = false;
+
+async function recoverStaleClaims() {
+ // A worker crash leaves jobs 'running' forever; anything running >15 min is re-queued.
+ const r = await db.query(
+ `UPDATE pr_jobs SET status='queued', last_error='requeued after stale claim'
+ WHERE status='running' AND started_at < now() - interval '15 minutes' RETURNING id`);
+ if (r.rows.length) console.log(`[pr-worker] requeued ${r.rows.length} stale job(s)`);
+}
+
+async function loop() {
+ console.log('[pr-worker] starting (poll ' + POLL_MS + 'ms, db ' + db.DB_NAME + ')');
+ const h = await db.health();
+ if (!h.ok) { console.error('[pr-worker] database unavailable: ' + (h.reason || '')); process.exit(1); }
+ await db.runMigrations({ log: (m) => console.log('[pr-worker] ' + m) });
+ await recoverStaleClaims();
+ while (!stopping) {
+ let job = null;
+ try { job = await jobs.claim(); } catch (e) { console.error('[pr-worker] claim error: ' + e.message); }
+ if (!job) { await new Promise((r) => setTimeout(r, POLL_MS)); continue; }
+ const t0 = Date.now();
+ console.log(`[pr-worker] #${job.id} ${job.job_type} attempt ${job.attempts} …`);
+ const res = await jobs.runJob(job);
+ console.log(`[pr-worker] #${job.id} ${res.ok ? 'done' : 'FAILED: ' + res.error}${res.retried ? ` (retry in ${res.delaySec}s)` : ''} (${Date.now() - t0}ms)`);
+ }
+ console.log('[pr-worker] stopped');
+ process.exit(0);
+}
+
+process.on('SIGINT', () => { stopping = true; console.log('\n[pr-worker] stopping after current job …'); });
+process.on('SIGTERM', () => { stopping = true; });
+
+loop().catch((e) => { console.error('[pr-worker] fatal: ' + e.message); process.exit(1); });
← 459558da pr-intelligence: plan doc, isolated pg layer (rentv_pr), sch
·
back to Rentv 2026
·
auto-save: 2026-07-30T11:17:19 (4 files) — package.json serv 7a3217d0 →