← back to Rentv 2026
PR intel: Gmail correspondence sync + per-contact timeline
0884d55a1810876566096e24e78c0ef3f492702a · 2026-08-06 09:12:32 -0700 · Steve
- services/gmail-sync.js: per-tenant sync — refresh-token → access token, list
messages, match participants to CRM contacts (email→person, domain→org),
record each as a timestamped pr_outreach_message (corresponded_at = real
msg date/time) + activity. Read-only (gmail.readonly), idempotent (unique
tenant+provider_message_id), incremental (pageToken cursor). Direction by
sender/recipient.
- migration 006: dedup unique index + person timeline index.
- routes: POST /tenant/gmail/sync (integrations cap), GET /people/:id/timeline.
- Verified: matching + recording + timeline on local; live run needs tenant creds.
Files touched
M src/pr/index.jsA src/pr/migrations/006_gmail_sync.sqlA src/pr/services/gmail-sync.js
Diff
commit 0884d55a1810876566096e24e78c0ef3f492702a
Author: Steve <steve@designerwallcoverings.com>
Date: Thu Aug 6 09:12:32 2026 -0700
PR intel: Gmail correspondence sync + per-contact timeline
- services/gmail-sync.js: per-tenant sync — refresh-token → access token, list
messages, match participants to CRM contacts (email→person, domain→org),
record each as a timestamped pr_outreach_message (corresponded_at = real
msg date/time) + activity. Read-only (gmail.readonly), idempotent (unique
tenant+provider_message_id), incremental (pageToken cursor). Direction by
sender/recipient.
- migration 006: dedup unique index + person timeline index.
- routes: POST /tenant/gmail/sync (integrations cap), GET /people/:id/timeline.
- Verified: matching + recording + timeline on local; live run needs tenant creds.
---
src/pr/index.js | 16 ++++++
src/pr/migrations/006_gmail_sync.sql | 9 ++++
src/pr/services/gmail-sync.js | 101 +++++++++++++++++++++++++++++++++++
3 files changed, 126 insertions(+)
diff --git a/src/pr/index.js b/src/pr/index.js
index e10326b7..15be1d95 100644
--- a/src/pr/index.js
+++ b/src/pr/index.js
@@ -136,6 +136,22 @@ module.exports = function mountPR(app, { adminOnly, sendPage }) {
res.json({ ok: false, connected: false, error: j.error_description || j.error || ('HTTP ' + r.status) });
} catch (e) { res.json({ ok: false, connected: false, error: e.message }); }
}));
+ // Sync Gmail → correspondence records (matches messages to contacts by email/domain).
+ const gmailSync = require('./services/gmail-sync');
+ app.post('/api/pr/tenant/gmail/sync', adminOnly, requireCap('integrations'), h(async (req, res) => {
+ const max = Math.min(Number((req.body || {}).max) || 300, 2000);
+ res.json(await gmailSync.syncTenant(req.prAuth.tenant.id, { max }));
+ }));
+ // Per-contact correspondence timeline (dates & times of every recorded message).
+ app.get('/api/pr/people/:id/timeline', adminOnly, requireCap('read'), h(async (req, res) => {
+ const rows = await db.rows(
+ `SELECT id, direction, provider, subject, corresponded_at, sent_at, replied_at, provider_thread_id
+ FROM pr_outreach_messages
+ WHERE tenant_id=$1 AND person_id=$2
+ ORDER BY COALESCE(corresponded_at, sent_at, created_at) DESC LIMIT 200`,
+ [req.prAuth.tenant.id, Number(req.params.id)]);
+ res.json({ rows });
+ }));
// ── Health & meta ──────────────────────────────────────────────────────────
app.get('/api/pr/health', adminOnly, async (_q, res) => res.json(await db.health()));
diff --git a/src/pr/migrations/006_gmail_sync.sql b/src/pr/migrations/006_gmail_sync.sql
new file mode 100644
index 00000000..6fc4b479
--- /dev/null
+++ b/src/pr/migrations/006_gmail_sync.sql
@@ -0,0 +1,9 @@
+-- Gmail correspondence sync support: idempotent message dedup + timeline lookup.
+-- One row per (tenant, gmail message) so re-runs never double-record a correspondence.
+CREATE UNIQUE INDEX IF NOT EXISTS pr_msg_tenant_provmsg_uidx
+ ON pr_outreach_messages (tenant_id, provider_message_id)
+ WHERE provider_message_id IS NOT NULL;
+
+-- per-person correspondence timeline (newest first) reads on person + corresponded_at
+CREATE INDEX IF NOT EXISTS pr_msg_person_when_idx
+ ON pr_outreach_messages (tenant_id, person_id, corresponded_at DESC);
diff --git a/src/pr/services/gmail-sync.js b/src/pr/services/gmail-sync.js
new file mode 100644
index 00000000..76e4747c
--- /dev/null
+++ b/src/pr/services/gmail-sync.js
@@ -0,0 +1,101 @@
+'use strict';
+// Per-tenant Gmail correspondence sync. Reads the tenant's saved OAuth creds, lists
+// recent messages, matches each message's participants to the tenant's CRM contacts
+// (by email → person, else sender/recipient domain → org), and records every match as a
+// timestamped pr_outreach_message (corresponded_at = the message's real date/time) plus
+// a pr_activity. READ-ONLY against Gmail (gmail.readonly). Idempotent (unique on
+// tenant+provider_message_id). Incremental via a per-tenant pageToken cursor.
+const db = require('../db');
+const audit = require('./audit');
+
+const TOKEN_URL = 'https://oauth2.googleapis.com/token';
+const GMAIL = 'https://gmail.googleapis.com/gmail/v1/users/me';
+const EMAIL_RE = /[a-z0-9._%+-]+@[a-z0-9.-]+\.[a-z]{2,}/ig;
+
+function addrs(s) { return [...new Set([...String(s || '').matchAll(EMAIL_RE)].map((m) => m[0].toLowerCase()))]; }
+function domainOf(e) { const i = String(e).indexOf('@'); return i < 0 ? null : e.slice(i + 1); }
+
+async function accessToken(g) {
+ const r = await fetch(TOKEN_URL, {
+ method: 'POST', headers: { 'Content-Type': 'application/x-www-form-urlencoded' },
+ body: new URLSearchParams({ client_id: g.client_id, client_secret: g.client_secret, refresh_token: g.refresh_token, grant_type: 'refresh_token' }),
+ signal: AbortSignal.timeout(15000),
+ });
+ const j = await r.json();
+ if (!r.ok || !j.access_token) throw new Error('gmail token: ' + (j.error_description || j.error || ('HTTP ' + r.status)));
+ return j.access_token;
+}
+async function gget(token, path) {
+ const r = await fetch(GMAIL + path, { headers: { Authorization: 'Bearer ' + token }, signal: AbortSignal.timeout(20000) });
+ if (!r.ok) throw new Error('gmail ' + r.status + ' ' + path.slice(0, 48));
+ return r.json();
+}
+
+/** Record ONE gmail message against a matched contact. Returns 1 if newly recorded. */
+async function recordMessage(tenantId, match, m, H) {
+ const when = m.internalDate ? new Date(Number(m.internalDate)).toISOString() : null;
+ const cols = match.person_id
+ ? { person_id: match.person_id, organization_id: null }
+ : { person_id: null, organization_id: match.organization_id };
+ const ins = await db.query(
+ `INSERT INTO pr_outreach_messages
+ (tenant_id, person_id, organization_id, direction, provider, provider_message_id, provider_thread_id, subject, corresponded_at, status)
+ VALUES ($1,$2,$3,$4,'gmail',$5,$6,$7,$8,'sent')
+ ON CONFLICT (tenant_id, provider_message_id) WHERE provider_message_id IS NOT NULL DO NOTHING
+ RETURNING id`,
+ [tenantId, cols.person_id, cols.organization_id, match.direction, m.id, m.threadId, String(H.subject || '').slice(0, 300), when]);
+ if (!ins.rows[0]) return 0;
+ await audit.activity({
+ entity_type: match.person_id ? 'person' : 'organization', entity_id: match.person_id || match.organization_id,
+ activity: match.direction === 'inbound' ? 'reply_received' : 'sent',
+ detail: { source: 'gmail', message_id: m.id, at: when, direction: match.direction, subject: String(H.subject || '').slice(0, 140) },
+ actor: 'system:gmail-sync',
+ });
+ return 1;
+}
+
+/** Match a message's participants to a CRM contact. person > org; direction by sender. */
+function matchContact(from, to, emap, dmap) {
+ for (const e of from) { if (emap.has(e)) return { person_id: emap.get(e), direction: 'inbound' }; }
+ for (const e of to) { if (emap.has(e)) return { person_id: emap.get(e), direction: 'outbound' }; }
+ for (const e of from) { const d = domainOf(e); if (d && dmap.has(d)) return { organization_id: dmap.get(d), direction: 'inbound' }; }
+ for (const e of to) { const d = domainOf(e); if (d && dmap.has(d)) return { organization_id: dmap.get(d), direction: 'outbound' }; }
+ return null;
+}
+
+async function syncTenant(tenantId, { max = 300 } = {}) {
+ const t = await db.one(`SELECT gmail_user, settings->'gmail' AS g FROM pr_tenants WHERE id=$1`, [tenantId]);
+ if (!t || !t.g || !t.g.refresh_token) return { skipped: 'no gmail credentials' };
+ const token = await accessToken(t.g);
+ // contact maps (tenant-scoped): email→person, domain→org
+ const people = await db.rows(`SELECT id, lower(public_work_email) em FROM pr_people WHERE tenant_id=$1 AND public_work_email IS NOT NULL AND lifecycle_status NOT IN ('duplicate','suppressed')`, [tenantId]);
+ const emap = new Map(people.map((p) => [p.em, p.id]));
+ const orgs = await db.rows(`SELECT id, domain FROM pr_organizations WHERE tenant_id=$1 AND domain IS NOT NULL AND lifecycle_status NOT IN ('duplicate','archived')`, [tenantId]);
+ const dmap = new Map(orgs.map((o) => [o.domain, o.id]));
+
+ let scanned = 0, created = 0, matched = 0;
+ let pageToken = (t.g._cursor && t.g._cursor.pageToken) || undefined;
+ while (scanned < max) {
+ const list = await gget(token, `/messages?maxResults=${Math.min(100, max - scanned)}${pageToken ? '&pageToken=' + pageToken : ''}`);
+ const ids = (list.messages || []).map((x) => x.id);
+ if (!ids.length) { pageToken = null; break; }
+ for (const id of ids) {
+ scanned++;
+ const m = await gget(token, `/messages/${id}?format=metadata&metadataHeaders=From&metadataHeaders=To&metadataHeaders=Cc&metadataHeaders=Subject`);
+ const H = Object.fromEntries((((m.payload || {}).headers) || []).map((h) => [h.name.toLowerCase(), h.value]));
+ const from = addrs(H.from), to = addrs(H.to).concat(addrs(H.cc));
+ const match = matchContact(from, to, emap, dmap);
+ if (!match) continue;
+ matched++;
+ created += await recordMessage(tenantId, match, m, H);
+ }
+ pageToken = list.nextPageToken;
+ if (!pageToken) break;
+ }
+ await db.query(`UPDATE pr_tenants SET settings = jsonb_set(settings, '{gmail,_cursor}', $2::jsonb, true) WHERE id=$1`,
+ [tenantId, JSON.stringify({ pageToken: pageToken || null, at: null })]);
+ await audit.log({ actor: 'system:gmail-sync', action: 'gmail.sync', entity_type: 'tenant', entity_id: tenantId, detail: { scanned, matched, created } });
+ return { scanned, matched, created };
+}
+
+module.exports = { syncTenant, matchContact, addrs, domainOf };
← 7ac0a3d5 News map: add From/To year-range filter (2015-present)
·
back to Rentv 2026
·
fix(deals): parseOccupancy counts 'N% pre-leased' (new-dev p 83953f44 →