API sync: remote data into contact attributes

// Pull customer data from a remote JSON API into Kommio contact attributes.
// Node 22, no dependencies:
//
//   KOMMIO_URL=https://support.example.com KOMMIO_API_KEY=<api key> \
//   SOURCE_URL=http://127.0.0.1:4700/customers \
//   node --experimental-strip-types sync.ts [--dry-run] [--every 900]
//
// Each run reads the source, finds each record's contact (by external id,
// else by email), and writes the mapped custom attributes through the
// Kommio API. It only writes what changed, so running it again is safe;
// it waits out 429 answers (Retry-After) and retries server errors.
import { pathToFileURL } from 'node:url';

/** One record of the remote API. Adapt to yours. */
export type SourceRecord = { customer_id?: string; email?: string; plan?: string; lifetime_value?: number };

/**
 * Contact attribute key → value from a record. The keys must exist in
 * Settings → Custom attributes (contact): here `ltv` (number) and `plan` (text).
 */
export const MAPPING: Record<string, (r: SourceRecord) => unknown> = {
  ltv: (r) => r.lifetime_value,
  plan: (r) => r.plan,
};

export type Contact = { id: number; name: string; company?: string; avatar_url?: string; emails: string[]; external_ids: { kind: string; value: string }[]; custom_attributes: Record<string, unknown> | null };
export type Outcome = { key: string; result: 'updated' | 'unchanged' | 'missing' | 'would-update' | 'failed'; contact?: number; changes?: Record<string, unknown>; error?: string };

export type Options = {
  kommioUrl: string; // the Kommio address; the API is under /api
  apiKey: string; // an API key (starts with kmk_) with contacts:read and contacts:write
  sourceUrl: string;
  dryRun?: boolean;
  fetch?: typeof fetch;
  sleep?: (ms: number) => Promise<void>;
  log?: (line: string) => void;
  maxRetries?: number;
};

const wait = (ms: number) => new Promise<void>((r) => setTimeout(r, ms));

/**
 * One API call with the key. 429: wait Retry-After (seconds) and try
 * again; 5xx or a network error: back off 1, 2, 4, 8, 16 seconds; other errors
 * are thrown with the API's message.
 */
export async function call<T>(o: Options, method: string, path: string, body?: unknown): Promise<T> {
  const f = o.fetch ?? fetch;
  const sleep = o.sleep ?? wait;
  const max = o.maxRetries ?? 5;
  for (let attempt = 0; ; attempt++) {
    let r: Response | null = null;
    try {
      r = await f(`${o.kommioUrl.replace(/\/$/, '')}/api${path}`, {
        method,
        headers: { Authorization: `Bearer ${o.apiKey}`, Accept: 'application/json', ...(body ? { 'Content-Type': 'application/json' } : {}) },
        body: body ? JSON.stringify(body) : undefined,
      });
    } catch (e) {
      if (attempt >= max) throw e;
    }
    if (r && r.ok) return (r.status === 204 ? undefined : await r.json()) as T;
    if (r && r.status !== 429 && r.status < 500) {
      const detail = await r.json().catch(() => ({})) as { detail?: string; title?: string };
      throw new Error(`${method} ${path}: ${r.status} ${detail.detail ?? detail.title ?? ''}`.trim());
    }
    if (attempt >= max) throw new Error(`${method} ${path}: gave up after ${attempt + 1} attempts (${r?.status ?? 'network error'})`);
    const retryAfter = Number(r?.headers.get('Retry-After'));
    const ms = r?.status === 429 && retryAfter > 0 ? retryAfter * 1000 : 1000 * 2 ** attempt;
    o.log?.(`${method} ${path}: ${r?.status ?? 'network error'}, retrying in ${ms / 1000}s`);
    await sleep(ms);
  }
}

/** The contact a record belongs to: an exact external id, else an exact email. */
export async function findContact(o: Options, r: SourceRecord): Promise<Contact | null> {
  const search = async (q: string) => (await call<{ items: Contact[] }>(o, 'GET', `/contacts?q=${encodeURIComponent(q)}&limit=50`)).items;
  if (r.customer_id) {
    const hit = (await search(r.customer_id)).find((c) => c.external_ids?.some((x) => x.kind === 'external' && x.value === r.customer_id));
    if (hit) return hit;
  }
  if (r.email) {
    const email = r.email.trim().toLowerCase();
    // ?q= matches parts of names and identities; keep only the exact address.
    return (await search(email)).find((c) => c.emails?.some((e) => e.toLowerCase() === email)) ?? null;
  }
  return null;
}

/** The attributes that differ from what the contact has (undefined values are skipped). */
export function changes(current: Record<string, unknown> | null, record: SourceRecord): Record<string, unknown> {
  const out: Record<string, unknown> = {};
  for (const [key, pick] of Object.entries(MAPPING)) {
    const v = pick(record);
    if (v !== undefined && JSON.stringify(current?.[key]) !== JSON.stringify(v)) out[key] = v;
  }
  return out;
}

/** One pass over the source. */
export async function syncOnce(o: Options): Promise<Outcome[]> {
  const f = o.fetch ?? fetch;
  const src = await f(o.sourceUrl, { headers: { Accept: 'application/json' } });
  if (!src.ok) throw new Error(`source answered ${src.status}`);
  const records = await src.json() as SourceRecord[];
  const out: Outcome[] = [];
  for (const r of records) {
    const key = r.customer_id ?? r.email ?? '?';
    try {
      const c = await findContact(o, r);
      if (!c) { out.push({ key, result: 'missing' }); continue; }
      const diff = changes(c.custom_attributes, r);
      if (Object.keys(diff).length === 0) { out.push({ key, result: 'unchanged', contact: c.id }); continue; }
      if (o.dryRun) { out.push({ key, result: 'would-update', contact: c.id, changes: diff }); continue; }
      // PUT replaces name, company, avatar and attributes: send them back as they are.
      await call(o, 'PUT', `/contacts/${c.id}`, { name: c.name, company: c.company, avatar_url: c.avatar_url, custom_attributes: { ...(c.custom_attributes ?? {}), ...diff } });
      out.push({ key, result: 'updated', contact: c.id, changes: diff });
    } catch (e) {
      out.push({ key, result: 'failed', error: (e as Error).message });
    }
  }
  return out;
}

function summary(rows: Outcome[]): string {
  const n = (k: Outcome['result']) => rows.filter((r) => r.result === k).length;
  return `updated ${n('updated')}, would update ${n('would-update')}, unchanged ${n('unchanged')}, no contact ${n('missing')}, failed ${n('failed')}`;
}

async function main() {
  const args = process.argv.slice(2);
  const every = args.includes('--every') ? Number(args[args.indexOf('--every') + 1]) : 0;
  const o: Options = {
    kommioUrl: process.env.KOMMIO_URL || 'http://localhost:8080',
    apiKey: process.env.KOMMIO_API_KEY || '',
    sourceUrl: process.env.SOURCE_URL || 'http://127.0.0.1:4700/customers',
    dryRun: args.includes('--dry-run'),
    log: (l) => console.log(l),
  };
  if (!o.apiKey) throw new Error('set KOMMIO_API_KEY (Settings → API keys, with contacts:read and contacts:write)');
  for (;;) {
    const rows = await syncOnce(o);
    for (const r of rows) console.log(`${r.result.padEnd(12)} ${r.key}${r.contact ? ` → contact ${r.contact}` : ''}${r.changes ? ` ${JSON.stringify(r.changes)}` : ''}${r.error ? ` ${r.error}` : ''}`);
    console.log(summary(rows));
    if (!every) { if (rows.some((r) => r.result === 'failed')) process.exitCode = 1; return; }
    await wait(every * 1000);
  }
}

if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) main().catch((e) => { console.error(e.message); process.exit(1); });

"""Pull customer data from a remote JSON API into Kommio contact attributes.
Python 3.9+, standard library only:

    KOMMIO_URL=https://support.example.com KOMMIO_API_KEY=<api key> \\
    SOURCE_URL=http://127.0.0.1:4700/customers \\
    python3 sync.py [--dry-run] [--every 900]

Each run reads the source, finds each record's contact (by external id,
else by email), and writes the mapped custom attributes through the Kommio
API. It only writes what changed, so running it again is safe; it waits out
429 answers (Retry-After) and retries server errors.
"""

import argparse
import json
import os
import sys
import time
import urllib.error
import urllib.parse
import urllib.request

# Contact attribute key -> value from a record. The keys must exist in
# Settings -> Custom attributes (contact): here `ltv` (number) and `plan` (text).
MAPPING = {
    "ltv": lambda r: r.get("lifetime_value"),
    "plan": lambda r: r.get("plan"),
}


class ApiError(Exception):
    pass


def urllib_transport(method, url, headers, body):
    """(status, headers, body bytes) of one HTTP request."""
    req = urllib.request.Request(url, data=body, method=method, headers=headers)
    try:
        with urllib.request.urlopen(req, timeout=30) as r:
            return r.status, dict(r.headers), r.read()
    except urllib.error.HTTPError as e:
        return e.code, dict(e.headers or {}), e.read()


class Sync:
    def __init__(self, kommio_url, api_key, source_url, dry_run=False, transport=urllib_transport, sleep=time.sleep, log=print, max_retries=5):
        self.api = kommio_url.rstrip("/") + "/api"
        self.key = api_key
        self.source = source_url
        self.dry_run = dry_run
        self.transport = transport
        self.sleep = sleep
        self.log = log
        self.max_retries = max_retries

    def call(self, method, path, body=None):
        """One API call with the key. 429: wait Retry-After (seconds) and try
        again; 5xx or a network error: back off 1, 2, 4, 8, 16 seconds; other
        errors raise ApiError with the API's message."""
        headers = {"Authorization": f"Bearer {self.key}", "Accept": "application/json"}
        data = None
        if body is not None:
            headers["Content-Type"] = "application/json"
            data = json.dumps(body).encode()
        for attempt in range(self.max_retries + 1):
            try:
                status, resp_headers, raw = self.transport(method, self.api + path, headers, data)
            except OSError as e:
                if attempt >= self.max_retries:
                    raise
                status, resp_headers, raw = None, {}, b""
            if status is not None and 200 <= status < 300:
                return json.loads(raw) if raw else None
            if status is not None and status != 429 and status < 500:
                try:
                    detail = json.loads(raw)
                except ValueError:
                    detail = {}
                raise ApiError(f"{method} {path}: {status} {detail.get('detail') or detail.get('title') or ''}".strip())
            if attempt >= self.max_retries:
                raise ApiError(f"{method} {path}: gave up after {attempt + 1} attempts ({status or 'network error'})")
            retry_after = {k.lower(): v for k, v in resp_headers.items()}.get("retry-after")
            delay = float(retry_after) if status == 429 and retry_after and retry_after.isdigit() else 2 ** attempt
            self.log(f"{method} {path}: {status or 'network error'}, retrying in {delay:g}s")
            self.sleep(delay)

    def search(self, q):
        return self.call("GET", "/contacts?" + urllib.parse.urlencode({"q": q, "limit": 50}))["items"]

    def find_contact(self, r):
        """The contact a record belongs to: an exact external id, else an exact email."""
        cid = r.get("customer_id")
        if cid:
            for c in self.search(cid):
                if any(x.get("kind") == "external" and x.get("value") == cid for x in c.get("external_ids") or []):
                    return c
        email = (r.get("email") or "").strip().lower()
        if email:
            # ?q= matches parts of names and identities; keep only the exact address.
            for c in self.search(email):
                if any(e.lower() == email for e in c.get("emails") or []):
                    return c
        return None

    def run_once(self):
        status, _, raw = self.transport("GET", self.source, {"Accept": "application/json"}, None)
        if status != 200:
            raise ApiError(f"source answered {status}")
        out = []
        for r in json.loads(raw):
            key = r.get("customer_id") or r.get("email") or "?"
            try:
                c = self.find_contact(r)
                if c is None:
                    out.append({"key": key, "result": "missing"})
                    continue
                diff = changes(c.get("custom_attributes"), r)
                if not diff:
                    out.append({"key": key, "result": "unchanged", "contact": c["id"]})
                    continue
                if self.dry_run:
                    out.append({"key": key, "result": "would-update", "contact": c["id"], "changes": diff})
                    continue
                # PUT replaces name, company, avatar and attributes: send them back as they are.
                attrs = dict(c.get("custom_attributes") or {}, **diff)
                self.call("PUT", f"/contacts/{c['id']}", {"name": c.get("name"), "company": c.get("company"), "avatar_url": c.get("avatar_url"), "custom_attributes": attrs})
                out.append({"key": key, "result": "updated", "contact": c["id"], "changes": diff})
            except (ApiError, OSError) as e:
                out.append({"key": key, "result": "failed", "error": str(e)})
        return out


def changes(current, record):
    """The attributes that differ from what the contact has (missing values are skipped)."""
    current = current or {}
    out = {}
    for key, pick in MAPPING.items():
        v = pick(record)
        if v is not None and current.get(key) != v:
            out[key] = v
    return out


def main():
    p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    p.add_argument("--dry-run", action="store_true", help="report what would change, write nothing")
    p.add_argument("--every", type=int, default=0, help="run again every N seconds (default: once)")
    a = p.parse_args()
    key = os.environ.get("KOMMIO_API_KEY", "")
    if not key:
        sys.exit("set KOMMIO_API_KEY (Settings -> API keys, with contacts:read and contacts:write)")
    s = Sync(os.environ.get("KOMMIO_URL", "http://localhost:8080"), key, os.environ.get("SOURCE_URL", "http://127.0.0.1:4700/customers"), dry_run=a.dry_run)
    while True:
        rows = s.run_once()
        for r in rows:
            extra = f" -> contact {r['contact']}" if r.get("contact") else ""
            extra += f" {json.dumps(r['changes'])}" if r.get("changes") else ""
            extra += f" {r['error']}" if r.get("error") else ""
            print(f"{r['result']:<12} {r['key']}{extra}")
        counts = {k: sum(1 for r in rows if r["result"] == k) for k in ("updated", "would-update", "unchanged", "missing", "failed")}
        print(f"updated {counts['updated']}, would update {counts['would-update']}, unchanged {counts['unchanged']}, no contact {counts['missing']}, failed {counts['failed']}")
        if not a.every:
            sys.exit(1 if counts["failed"] else 0)
        time.sleep(a.every)


if __name__ == "__main__":
    main()