Lead to Opportunity on DODIL: a Discovery Warehouse, a Cost-Gated Qualify, and a Sequence Engine That's Just Rows

# db.py — INSERT ... ON CONFLICT DO UPDATE (idempotent keyed write); DO NOTHING for pure edge rows
def upsert(session, model, rows, key):
    keys = [key] if isinstance(key, str) else list(key)
    table = model.__table__
    # normalise to a uniform column set — a multi-row VALUES needs every row to name the
    # same columns; fill any a caller omitted with None.
    cols = {c for r in rows for c in r}
    rows = [{c: r.get(c) for c in cols} for r in rows]
    stmt = pg_insert(table).values(rows)
    update_cols = [c.name for c in table.columns if c.name not in keys and c.name in cols]
    if update_cols:
        stmt = stmt.on_conflict_do_update(
            index_elements=keys,
            set_={c: getattr(stmt.excluded, c) for c in update_cols},
        )
    else:
        stmt = stmt.on_conflict_do_nothing(index_elements=keys)
    session.execute(stmt)

# routes.py — CRUD over the discovery warehouse, keyed on the natural PK (domain)
@router.post("/organizations")
def upsert_organization(o: OrganizationIn, s: Session = Depends(db)):
    upsert(s, Organization, [o.model_dump()], key="domain")
    s.commit()
    return {"ok": True, "domain": o.domain}
 
 
@router.get("/organizations/{domain}")
def get_organization(domain: str, s: Session = Depends(db)):
    row = s.get(Organization, domain)
    if not row:
        raise HTTPException(404, "no such organization")
    return row.__dict__ | {"_sa_instance_state": None}

# routes.py — workflow op 1: qualify a BOUNDED batch on Ignite Models, write the verdict back.
# This is one of the suite's four role gates: it spends model money, so it demands the
# orgs:qualify pool permission (see ## Auth) on top of the gateway's login.
@router.post("/organizations/qualify")
def qualify_batch(q: QualifyIn,
                  user: dict = Depends(require_permission("orgs:qualify")),
                  s: Session = Depends(db)):
    token = _models_token()
    verdicts = []
    for domain in q.domains:
        org = s.get(Organization, domain)
        if not org:
            continue
        v = _classify(token, org, q.product_pitch)
        row = {c.name: getattr(org, c.name) for c in Organization.__table__.columns}
        row.update(buyer=v["buyer"], fit_score=float(v["fit_score"]), angle=v["angle"], verified=True)
        upsert(s, Organization, [row], key="domain")     # merged full row -> discovery fields intact
        verdicts.append({"domain": domain, "buyer": v["buyer"], "fit_score": float(v["fit_score"])})
    s.commit()                                            # commit before any read-back
    return {"qualified": len(verdicts), "verdicts": verdicts}

# routes.py — workflow op 2: lookalike expansion (VECTOR)
@router.post("/organizations/expand")
def expand_lookalikes(q: ExpandIn, s: Session = Depends(db)):
    rows = s.execute(
        select(OrgEmbedding.domain, OrgEmbedding.embedding.cosine_distance(q.embedding).label("d"))
        .order_by("d")
        .limit(q.top_k)
    ).all()
    return {"matches": [{"domain": d, "distance": float(dist)} for d, dist in rows]}

# routes.py — workflow op 3: convert a strong org into a lead + opportunity (+ conversions audit)
@router.post("/organizations/{domain}/convert")
def convert_organization(domain: str, c: ConvertIn, s: Session = Depends(db)):
    org = s.get(Organization, domain)
    if not org:
        raise HTTPException(404, "no such organization")
    if org.buyer != "strong":
        raise HTTPException(409, f"organization is buyer={org.buyer!r}, not 'strong' — not convertible")
    now = _now()
    lead_id, opp_id = f"lead:{domain}", f"opp:{domain}"
    upsert(s, Lead, [{ ... }], key="lead_id")           # status qualified, score = org.fit_score
    upsert(s, Opportunity, [{ ... }], key="opportunity_id")
    upsert(s, Conversion, [{ ... }], key="conversion_id")  # conv:<domain> -> lead + opp + contact
    s.commit()
    return {"ok": True, "domain": domain, "lead_id": lead_id, "opportunity_id": opp_id}

# auth.py — a header-trust reader, not a verifier (the gateway already checked the JWT)
def current_user(request: Request) -> dict: ...       # reads x-dodil-user (+ permissions off the JWT)
def require_permission(perm: str): ...                # Depends-factory: 403 without the permission

python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt
 
cp .env.example .env      # then set DODIL_TOKEN (your `dodil auth login` token)
# BUCKET defaults to "crm"; create it once (Step 1), then create the tables from the models:
python -c "import db, models; models.Base.metadata.create_all(db.engine)"
 
uvicorn routes:app --reload
# POST /organizations · GET /organizations/{domain} · POST /organizations/qualify
# POST /organizations/expand · POST /organizations/{domain}/convert

# seq-engine/server.py — IMAGE-mode Ignite app (HTTP server on $PORT), PRIVATE. Calls NO Models.
#   GET  /healthz  -> 200 {"status":"ready"}     (probe path; no auth, no DataK3)
#   POST /tick     -> body {"limit": N}          (walks one tick of the nurture flow)
# All reads/writes go over the DROP-IN POSTGRES WIRE — the SA access token is the pg password.
 
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 
from psycopg import errors as pg_errors
 
BUCKET    = os.environ["BUCKET"]
SA_ID     = os.environ["DODIL_SERVICE_ACCOUNT_ID"]   # the cli-… serviceAccountId, NOT the uuid
SA_SECRET = os.environ["DODIL_SERVICE_ACCOUNT_SECRET"]
PG_HOST   = os.environ.get("PG_HOST", "pg.uk-lon-1.dodil.io")
PG_PORT   = int(os.environ.get("PG_PORT", "5432"))
ID_URL    = "https://id.dodil.io/realms/dodil/protocol/openid-connect/token"
# an explicit User-Agent is REQUIRED — stdlib urllib's default "Python-urllib/x" is
# banned by Cloudflare at id.dodil.io (HTTP 403 "error code: 1010").
UA = "crm-sequence-engine/1.0"
 
 
def _now():
    return datetime.now(timezone.utc).isoformat()
 
 
def _http_post(url, data, headers, form=False):
    headers = {"User-Agent": UA, **headers}
    body = urllib.parse.urlencode(data).encode() if form else json.dumps(data).encode()
    headers["Content-Type"] = "application/x-www-form-urlencoded" if form else "application/json"
    req = urllib.request.Request(url, data=body, headers=headers, method="POST")
    with urllib.request.urlopen(req, timeout=90) as r:
        return json.loads(r.read().decode())
 
 
def _token():
    out = _http_post(ID_URL, {"grant_type": "client_credentials",
                              "client_id": SA_ID, "client_secret": SA_SECRET}, headers={}, form=True)
    return out["access_token"]
 
 
def _pg(token):
    # drop-in Postgres wire: DB name = bucket, user "token", password = the SA access token.
    return psycopg.connect(host=PG_HOST, port=PG_PORT, dbname=BUCKET, user="token",
                           password=token, sslmode="require", connect_timeout=20, autocommit=False)
 
 
def _retry(fn):
    # the pg engine is serializable — retry a write on a transient serialization/deadlock.
    for attempt in range(4):
        try:
            return fn()
        except (pg_errors.SerializationFailure, pg_errors.DeadlockDetected):
            if attempt == 3:
                raise
            time.sleep(0.4 * (attempt + 1))
 
 
def _subscribed(token, email):
    with _pg(token) as conn, conn.cursor() as cur:
        cur.execute("SELECT subscribed FROM contacts WHERE email = %s", (email,))
        row = cur.fetchone()
    return bool(row and row[0])
 
 
def _next_run(delay):
    return datetime.fromtimestamp(time.time() + int(delay or 0), timezone.utc).isoformat()
 
 
def _send_email(to, subject, body_html):
    # hand off to YOUR ESP (SES / SendGrid / Postmark …) — the one intentional stub.
    # Everything around it (the DataK3 reads/writes) is real.
    pass
 
 
def _tick(token, limit):
    now = _now()
    with _pg(token) as conn, conn.cursor() as cur:
        cur.execute("SELECT enrollment_id, flow_id, contact_email, opportunity_id, current_position "
                    "FROM flow_enrollments WHERE status = 'active' AND next_run_at <= %s "
                    "ORDER BY next_run_at LIMIT %s", (now, limit))
        due = cur.fetchall()
 
    sent = skipped = done = 0
    for enr_id, flow_id, email, opp_id, pos in due:
        with _pg(token) as conn, conn.cursor() as cur:
            cur.execute("SELECT step_id, kind, subject, body_html, delay_seconds "
                        "FROM flow_steps WHERE flow_id = %s AND position = %s", (flow_id, pos))
            step = cur.fetchone()
 
        if step is None:                                  # walked off the end -> done
            def _finish():
                with _pg(token) as c, c.cursor() as cur:
                    cur.execute("UPDATE flow_enrollments SET status = 'done', next_run_at = %s "
                                "WHERE enrollment_id = %s", (now, enr_id))
                    c.commit()
            _retry(_finish); done += 1
            continue
 
        step_id, kind, subject, body_html, delay = step
        gated  = kind == "email" and not _subscribed(token, email)   # honor unsubscribe
        status = "skipped:unsubscribed" if gated else "sent"
 
        # action_id / activity_id are deterministic per (enrollment, position) — a poll retry or shard
        # re-entry re-writes the SAME PK, so these must be ON CONFLICT upserts to be re-run safe. A bare
        # re-INSERT of a committed PK raises duplicate-key 23505; DuckDB pg-wire supports ON CONFLICT.
        def _log_action():
            with _pg(token) as c, c.cursor() as cur:
                cur.execute("INSERT INTO flow_actions "
                            "(action_id, enrollment_id, flow_id, step_id, contact_email, kind, status, detail, ts) "
                            "VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s) "
                            "ON CONFLICT (action_id) DO UPDATE SET enrollment_id = EXCLUDED.enrollment_id, "
                            "flow_id = EXCLUDED.flow_id, step_id = EXCLUDED.step_id, "
                            "contact_email = EXCLUDED.contact_email, kind = EXCLUDED.kind, "
                            "status = EXCLUDED.status, detail = EXCLUDED.detail, ts = EXCLUDED.ts",
                            (f"act:{enr_id}:{pos}", enr_id, flow_id, step_id, email, kind, status, subject, now))
                c.commit()
        _retry(_log_action)
 
        if not gated:
            def _log_activity():
                with _pg(token) as c, c.cursor() as cur:
                    cur.execute("INSERT INTO activities "
                                "(activity_id, opportunity_id, contact_email, kind, subject, body, direction, status, ts) "
                                "VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s) "
                                "ON CONFLICT (activity_id) DO UPDATE SET opportunity_id = EXCLUDED.opportunity_id, "
                                "contact_email = EXCLUDED.contact_email, kind = EXCLUDED.kind, "
                                "subject = EXCLUDED.subject, body = EXCLUDED.body, "
                                "direction = EXCLUDED.direction, status = EXCLUDED.status, ts = EXCLUDED.ts",
                                (f"eml:{enr_id}:{pos}", opp_id, email, "email", subject, body_html,
                                 "outbound", "sent", now))
                    c.commit()
            _retry(_log_activity)
            _send_email(email, subject, body_html)
            sent += 1
        else:
            skipped += 1
 
        # advance the existing enrollment row (UPDATE — like moving leads.status in the scorer)
        def _advance():
            with _pg(token) as c, c.cursor() as cur:
                cur.execute("UPDATE flow_enrollments SET current_position = %s, next_run_at = %s "
                            "WHERE enrollment_id = %s", (pos + 1, _next_run(delay), enr_id))
                c.commit()
        _retry(_advance)
 
    return {"ticked": len(due), "sent": sent, "skipped": skipped, "done": done}
 
 
class Handler(BaseHTTPRequestHandler):
    def _send(self, code, body):
        payload = json.dumps(body).encode()
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(payload)))
        self.end_headers()
        self.wfile.write(payload)
 
    def do_GET(self):
        if self.path == "/healthz":
            return self._send(200, {"status": "ready"})
        return self._send(404, {"error": "no_route", "path": self.path})
 
    def do_POST(self):
        if self.path != "/tick":
            return self._send(404, {"error": "no_route", "path": self.path})
        try:
            n = int(self.headers.get("Content-Length") or 0)
            body = json.loads(self.rfile.read(n) or b"{}")
            return self._send(200, _tick(_token(), int(body.get("limit", 200))))
        except urllib.error.HTTPError as e:
            return self._send(502, {"error": "upstream", "code": e.code,
                                    "body": e.read().decode(errors="replace")[:600]})
        except Exception as e:
            return self._send(500, {"error": type(e).__name__, "detail": str(e)[:600]})
 
    def log_message(self, *a):
        pass
 
 
if __name__ == "__main__":
    port = int(os.environ.get("PORT", "8080"))
    print(f"crm-sequence-engine serving on 0.0.0.0:{port} bucket={BUCKET}", flush=True)
    ThreadingHTTPServer(("0.0.0.0", port), Handler).serve_forever()

# seq-engine/Dockerfile
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY server.py .
ENV PORT=8080
EXPOSE 8080
CMD ["python", "server.py"]

# seq-engine/requirements.txt
psycopg[binary]==3.2.3

# tracking/server.py — IMAGE-mode Ignite app (HTTP server on $PORT), PUBLIC (no auth). Calls NO Models.
# The open/click/unsubscribe links real recipients hit — kept OFF the admin box, carrying the fewest
# secrets possible: DataK3 pg-wire access + the unsubscribe signing key (the ONLY secret beyond DataK3).
#   GET /healthz            -> 200 {"status":"ready"}   (probe; no token, no DataK3)
#   GET /track/open?a=&c=   -> writes an 'open' email_events row, returns a 1x1 gif
#   GET /track/click?a=&c=&u= -> writes a 'click' row, 302-redirects to u
#   GET /unsubscribe?e=&t=  -> verifies the HMAC, flips contacts.subscribed, logs it
 
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 
from psycopg import errors as pg_errors
 
BUCKET    = os.environ["BUCKET"]
SA_ID     = os.environ["DODIL_SERVICE_ACCOUNT_ID"]   # the cli-… serviceAccountId, NOT the uuid
SA_SECRET = os.environ["DODIL_SERVICE_ACCOUNT_SECRET"]
UNSUB_KEY = os.environ["UNSUB_SIGNING_KEY"].encode()  # the ONLY secret beyond DataK3 access
PG_HOST   = os.environ.get("PG_HOST", "pg.uk-lon-1.dodil.io")
PG_PORT   = int(os.environ.get("PG_PORT", "5432"))
ID_URL    = "https://id.dodil.io/realms/dodil/protocol/openid-connect/token"
UA = "crm-tracking/1.0"   # explicit UA — urllib's default is Cloudflare-banned at id.dodil.io (403)
GIF = (b"GIF89a\x01\x00\x01\x00\x80\x00\x00\xff\xff\xff\x00\x00\x00!\xf9\x04\x01\x00\x00"
       b"\x00\x00,\x00\x00\x00\x00\x01\x00\x01\x00\x00\x02\x02D\x01\x00;")   # 1x1 transparent pixel
 
 
def _now():
    return datetime.now(timezone.utc).isoformat()
 
 
def _http_post(url, data, headers, form=False):
    headers = {"User-Agent": UA, **headers}
    body = urllib.parse.urlencode(data).encode() if form else json.dumps(data).encode()
    headers["Content-Type"] = "application/x-www-form-urlencoded" if form else "application/json"
    req = urllib.request.Request(url, data=body, headers=headers, method="POST")
    with urllib.request.urlopen(req, timeout=90) as r:
        return json.loads(r.read().decode())
 
 
def _token():
    out = _http_post(ID_URL, {"grant_type": "client_credentials",
                              "client_id": SA_ID, "client_secret": SA_SECRET}, headers={}, form=True)
    return out["access_token"]
 
 
def _pg(token):
    # drop-in Postgres wire: DB name = bucket, user "token", password = the SA access token.
    return psycopg.connect(host=PG_HOST, port=PG_PORT, dbname=BUCKET, user="token",
                           password=token, sslmode="require", connect_timeout=20, autocommit=False)
 
 
def _retry(fn):
    for attempt in range(4):
        try:
            return fn()
        except (pg_errors.SerializationFailure, pg_errors.DeadlockDetected):
            if attempt == 3:
                raise
            time.sleep(0.4 * (attempt + 1))
 
 
def _sign(email):
    return hmac.new(UNSUB_KEY, email.encode(), hashlib.sha256).hexdigest()
 
 
def _event(ev_type, action_id, email, detail):
    now = _now()
    def _w():
        with _pg(_token()) as c, c.cursor() as cur:
            # event_id carries the timestamp -> a fresh PK every event (append-only), so this first-write
            # INSERT is correct. (Re-writing the SAME key would instead need ON CONFLICT DO UPDATE — a bare
            # re-INSERT of a committed PK raises duplicate-key 23505.)
            cur.execute("INSERT INTO email_events (event_id, action_id, contact_email, type, detail, ts) "
                        "VALUES (%s,%s,%s,%s,%s,%s)",
                        (f"ev:{ev_type}:{action_id or email}:{now}", action_id, email, ev_type, detail, now))
            c.commit()
    _retry(_w)
 
 
def _unsubscribe(email):
    now = _now()
    def _w():
        with _pg(_token()) as c, c.cursor() as cur:
            cur.execute("UPDATE contacts SET subscribed = false, updated_at = %s WHERE email = %s",
                        (now, email))
            cur.execute("INSERT INTO activities (activity_id, contact_email, kind, direction, status, ts) "
                        "VALUES (%s,%s,%s,%s,%s,%s)",
                        (f"unsub:{email}:{now}", email, "unsubscribe", "inbound", "done", now))
            c.commit()
    _retry(_w)
 
 
class Handler(BaseHTTPRequestHandler):
    def _json(self, code, body):
        payload = json.dumps(body).encode()
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(payload)))
        self.end_headers()
        self.wfile.write(payload)
 
    def _raw(self, code, ctype, payload, extra=None):
        self.send_response(code)
        self.send_header("Content-Type", ctype)
        self.send_header("Content-Length", str(len(payload)))
        for k, v in (extra or {}).items():
            self.send_header(k, v)
        self.end_headers()
        self.wfile.write(payload)
 
    def do_GET(self):
        u = urllib.parse.urlparse(self.path)
        q = urllib.parse.parse_qs(u.query)
        one = lambda k: (q.get(k) or [""])[0]
        try:
            if u.path == "/healthz":
                return self._json(200, {"status": "ready"})          # no token, no DataK3
            if u.path == "/track/open":
                _event("open", one("a"), one("c"), None)
                return self._raw(200, "image/gif", GIF)              # the tracking pixel
            if u.path == "/track/click":
                _event("click", one("a"), one("c"), one("u"))
                return self._raw(302, "text/plain", b"", {"Location": one("u") or "/"})
            if u.path == "/unsubscribe":
                email, tok = one("e"), one("t")
                if not hmac.compare_digest(tok, _sign(email)):       # the ONLY secret this app carries
                    return self._json(403, {"status": "error", "reason": "bad_signature"})
                _unsubscribe(email)
                return self._raw(200, "text/html; charset=utf-8", b"<h1>You are unsubscribed.</h1>")
            return self._json(404, {"error": "no_route", "path": u.path})
        except urllib.error.HTTPError as e:
            return self._json(502, {"error": "upstream", "code": e.code})
        except Exception as e:
            return self._json(500, {"error": type(e).__name__, "detail": str(e)[:600]})
 
    def log_message(self, *a):
        pass
 
 
if __name__ == "__main__":
    port = int(os.environ.get("PORT", "8080"))
    print(f"crm-tracking serving on 0.0.0.0:{port} bucket={BUCKET}", flush=True)
    ThreadingHTTPServer(("0.0.0.0", port), Handler).serve_forever()

# tracking/Dockerfile
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY server.py .
ENV PORT=8080
EXPOSE 8080
CMD ["python", "server.py"]

# tracking/requirements.txt
psycopg[binary]==3.2.3

# 1. warehouse + flow tables exist (15 incl. crm_node/crm_edge)
dodil data table list -b "$BUCKET"
#  organizations, org_embeddings, flows, flow_steps, flow_enrollments, flow_actions, email_events,
#  conversions, leads, contacts, accounts, opportunities, activities, crm_node, crm_edge
 
# 2. the qualify gate returns a valid buyer-tier JSON verdict
dodil ignite models chat kimi-k2.6 \
  --system 'Return ONLY JSON: buyer (strong|possible|weak), fit_score (0-1), angle (<=12 words).' \
  --message 'org: Harborlytics | harborlytics.io | CDP stitching Postgres, Pinecone, and a warehouse.'
#  {"buyer":"strong","fit_score":0.92,"angle":"Replace stitched Postgres/Pinecone/warehouse with unified backend"}
 
# 3. strong orgs converted to leads + a conversions audit row
dodil data sql -b "$BUCKET" "SELECT
  (SELECT count(*) FROM organizations WHERE buyer='strong') AS strong_orgs,
  (SELECT count(*) FROM leads WHERE status='qualified')     AS qualified_leads,
  (SELECT count(*) FROM conversions)                        AS conversions"
#  strong_orgs 3 | qualified_leads 3 | conversions 1
 
# 4. the enrollment is staged at position 1, due now
dodil data sql -b "$BUCKET" \
  "SELECT current_position, status FROM flow_enrollments WHERE contact_email='[email protected]'"
#  current_position 1 | status active
 
# 5. the converted org is in the account graph (works_at reaches the contact)
dodil data pg -b "$BUCKET" "SELECT c.name FROM graph_khop('crm_graph',1,1,'in') g
  JOIN crm_node c ON c.id=g.node AND c.kind='contact'"
#  Harborlytics Data Team
 
# 6. lookalike KNN ranks the data-infra buyers ahead of the weak DTC store
dodil data vsearch -b "$BUCKET" -t org_embeddings --column embedding \
  --text "fragmented data stack stitching SQL, a vector store, and a graph database" \
  --model jina-embeddings-v4 --metric cosine --top-k 4
#  harborlytics.io 0.33 | nimbusgraph.dev 0.39 | greyparrot.ai 0.47 | brightmugs.com 0.56

Build the discover → qualify → nurture → convert front door of a CRM on DataK3 (one bucket = SQL + vector +
graph). Confirm each step. Assume crm/core masters exist (or stub leads/contacts/accounts/opportunities/
activities first).
 
1. Create a DataK3 bucket `crm`. Discovery warehouse: organizations (key domain) + org_embeddings
   (key domain, VECTOR(2048)). Upsert ~4 discovered companies, buyer=unqualified.
2. Qualify a bounded batch (25) on kimi-k2.6 — system prompt: "Qualify a company as a sales lead for a
   unified data backend (SQL+vector+graph in one bucket). Return ONLY JSON: buyer (strong|possible|weak),
   fit_score (0-1), angle (<=12 words)." Merge buyer/fit_score/angle back onto each organizations row.
3. Embed each description with jina-embeddings-v4 into org_embeddings (one vector per upsert). Vsearch a
   strong-buyer profile for lookalikes.
4. Convert buyer='strong' orgs into leads (status qualified). Fully convert the top one: account + contact +
   opportunity + a conversions row. Project account/contact nodes + a works_at edge into crm_node/crm_edge,
   then CREATE GRAPH crm_graph; traverse graph_khop('crm_graph',<acct>,1,'in') to confirm.
5. Sequence tables: flows, flow_steps, flow_enrollments, flow_actions. Author an `onboarding` flow (3 email
   steps, delays 0 / 3d / 7d) and enroll the converted contact at position 1, due now.
6. Deploy Ignite `crm-sequence-engine` (own SA: k3.editor + ignite.app-developer): a tick loop over due
   enrollments that sends the current step (skip if contacts.subscribed is false) and advances position.
7. Create email_events; deploy a SECOND public Ignite app `crm-tracking` (own SA: k3.editor only + the
   unsubscribe signing key) serving /healthz, /track/open, /track/click, /unsubscribe.