From 59559ad909b7726c014fcfa680030f0f43bedffb Mon Sep 17 00:00:00 2001 From: Mike Wichers Date: Fri, 4 Sep 2026 18:34:01 -0400 Subject: [PATCH] lead claims: outreach v1, decided by Mike Who picked a lead up and when, so nothing sits waiting unnoticed. No outcomes, no end states, nothing on join_leads changes: the lead stays the record of what the family said and the claim is only who has it. One row per claim in lead_claims; the current claim is the latest unreleased. Taking over is release then claim, never a silent overwrite: 409 names the holder. The holder or an owner may release. Every claim and release lands in the action log; the summary carries an unclaimed count. tests/smoke_admin.py 112 -> 122. --- app/admin_api.py | 38 ++++++++++++++++++ app/store.py | 93 ++++++++++++++++++++++++++++++++++++++++++++ tests/smoke_admin.py | 22 +++++++++++ 3 files changed, 153 insertions(+) diff --git a/app/admin_api.py b/app/admin_api.py index 45ac32c..909a0de 100644 --- a/app/admin_api.py +++ b/app/admin_api.py @@ -175,6 +175,44 @@ def get_one(request: Request, record_id: str, x_admin_token: str = Header(None)) return rec +@router.post("/leads/{record_id}/claim") +def claim_lead(request: Request, record_id: str, x_admin_token: str = Header(None)): + """Take a lead so it is not sitting unclaimed. Outreach v1 is exactly + this: who picked it up and when. No outcomes. 409 if someone else has + it - taking over is release, then claim, never a silent overwrite.""" + _auth(request, x_admin_token, "leads:read") + person = _person(request) + if not person: + raise HTTPException(403, "claiming needs a signed-in person") + try: + cur = store.claim_lead(record_id, person["id"], person["email"]) + except store.ClaimRejected as e: + raise _reject(e) + if not cur: + raise HTTPException(404, "no such lead") + _log(request, "lead.claimed", record_id) + return {"claim": {"email": cur["email"], "claimed_at": cur["claimed_at"]}} + + +@router.post("/leads/{record_id}/release") +def release_lead(request: Request, record_id: str, x_admin_token: str = Header(None)): + """Let a lead go. The holder may; so may an owner (people:manage), + which is how a lead gets reassigned when someone steps back.""" + _auth(request, x_admin_token, "leads:read") + person = _person(request) + if not person: + raise HTTPException(403, "releasing needs a signed-in person") + try: + cur = store.release_lead(record_id, person["id"], person["email"], + can_release_any=identity.can(person, "people:manage")) + except store.ClaimRejected as e: + raise _reject(e) + if not cur: + raise HTTPException(404, "no such lead") + _log(request, "lead.released", "%s (was %s)" % (record_id, cur["email"])) + return {"claim": None} + + @router.get("/mirrors/failed") def failed(request: Request, target: str = "google_sheet", x_admin_token: str = Header(None)): _auth(request, x_admin_token, "leads:read") diff --git a/app/store.py b/app/store.py index 0a7d283..f0138e2 100644 --- a/app/store.py +++ b/app/store.py @@ -74,6 +74,22 @@ CREATE TABLE IF NOT EXISTS join_leads ( CREATE INDEX IF NOT EXISTS idx_join_leads_submitted ON join_leads(submitted_at DESC); CREATE INDEX IF NOT EXISTS idx_join_leads_email ON join_leads(email); +-- Lead claims (outreach v1, decided by Mike 2026-09-04): who picked a lead +-- up and when, so nothing sits waiting unnoticed. No outcomes, no end +-- states - the lead stays the record of what the family said, this is only +-- who has it. One row per claim; the current claim is the latest with no +-- released_at. Nothing here touches join_leads. +CREATE TABLE IF NOT EXISTS lead_claims ( + id TEXT PRIMARY KEY, + lead_id TEXT NOT NULL REFERENCES join_leads(id), + person_id TEXT NOT NULL, + email TEXT NOT NULL, + claimed_at TEXT NOT NULL, + released_at TEXT, + released_by TEXT +); +CREATE INDEX IF NOT EXISTS idx_lead_claims_lead ON lead_claims(lead_id, released_at); + CREATE TABLE IF NOT EXISTS mirrors ( record_id TEXT NOT NULL, target TEXT NOT NULL, @@ -272,6 +288,11 @@ def get_lead(record_id): out["mirrors"] = [dict(m) for m in con.execute( "SELECT target, state, attempts, last_error, last_attempt_at" " FROM mirrors WHERE record_id=?", (record_id,))] + cur = current_claim(con, record_id) + out["claim"] = {"email": cur["email"], "claimed_at": cur["claimed_at"]} if cur else None + out["claim_history"] = [dict(c) for c in con.execute( + "SELECT email, claimed_at, released_at, released_by FROM lead_claims WHERE lead_id=?" + " ORDER BY claimed_at DESC", (record_id,))] return out finally: con.close() @@ -302,6 +323,13 @@ def list_leads(since=None, q=None, limit=100, offset=0): mir.setdefault(m["record_id"], {})[m["target"]] = m["state"] for r in rows: r["mirrors"] = mir.get(r["id"], {}) + claims = {} + for c in con.execute( + "SELECT lead_id, email, claimed_at FROM lead_claims WHERE released_at IS NULL" + " AND lead_id IN (%s)" % marks, ids): + claims[c["lead_id"]] = {"email": c["email"], "claimed_at": c["claimed_at"]} + for r in rows: + r["claim"] = claims.get(r["id"]) return rows finally: con.close() @@ -313,6 +341,8 @@ def summary(): try: return { "total": con.execute("SELECT COUNT(*) c FROM join_leads").fetchone()["c"], + "unclaimed": con.execute("SELECT count(*) FROM join_leads l WHERE NOT EXISTS" + " (SELECT 1 FROM lead_claims c WHERE c.lead_id=l.id AND c.released_at IS NULL)").fetchone()[0], "by_interest": {r["interested_in"]: r["c"] for r in con.execute( "SELECT interested_in, COUNT(*) c FROM join_leads GROUP BY interested_in")}, "by_heard_from": {r["heard_from"]: r["c"] for r in con.execute( @@ -567,6 +597,69 @@ def revoke_announcement(aid): # Rows are deactivated, never deleted. A unit that folds or moves keeps its # row with active=0, so "why did that pack disappear from the page" stays # answerable. Reactivation is an update setting active back to 1. + +# --------------------------------------------------------------------------- +# Lead claims - outreach v1. See the schema comment. +# --------------------------------------------------------------------------- + +class ClaimRejected(Rejected): + pass + + +def current_claim(con, lead_id): + r = con.execute("SELECT * FROM lead_claims WHERE lead_id=? AND released_at IS NULL" + " ORDER BY claimed_at DESC LIMIT 1", (lead_id,)).fetchone() + return dict(r) if r else None + + +def claim_lead(lead_id, person_id, email): + """Take a lead. If someone else holds it, 409 with who - taking it over + is a deliberate release-then-claim, never a silent overwrite.""" + con = connect() + try: + if not con.execute("SELECT 1 FROM join_leads WHERE id=?", (lead_id,)).fetchone(): + return None + cur = current_claim(con, lead_id) + if cur: + if cur["person_id"] == person_id: + raise ClaimRejected(409, "you already have this lead") + raise ClaimRejected(409, "%s has this lead; release it first" % cur["email"], {"claimed_by": cur["email"]}) + con.execute("INSERT INTO lead_claims (id, lead_id, person_id, email, claimed_at) VALUES (?,?,?,?,?)", + (str(uuid.uuid4()), lead_id, person_id, email, _now())) + con.commit() + return current_claim(con, lead_id) + finally: + con.close() + + +def release_lead(lead_id, person_id, email, can_release_any=False): + """Let a lead go. The holder may; so may someone with people:manage + (an owner reassigning), which is what can_release_any means.""" + con = connect() + try: + if not con.execute("SELECT 1 FROM join_leads WHERE id=?", (lead_id,)).fetchone(): + return None + cur = current_claim(con, lead_id) + if not cur: + raise ClaimRejected(409, "nobody has this lead") + if cur["person_id"] != person_id and not can_release_any: + raise ClaimRejected(403, "%s has this lead; only they or an owner can release it" % cur["email"]) + con.execute("UPDATE lead_claims SET released_at=?, released_by=? WHERE id=?", (_now(), email, cur["id"])) + con.commit() + return cur + finally: + con.close() + + +def unclaimed_count(): + con = connect() + try: + return con.execute("SELECT count(*) FROM join_leads l WHERE NOT EXISTS" + " (SELECT 1 FROM lead_claims c WHERE c.lead_id=l.id AND c.released_at IS NULL)").fetchone()[0] + finally: + con.close() + + # ---------------------------------------------------------------------------- diff --git a/tests/smoke_admin.py b/tests/smoke_admin.py index c07eb60..59cf698 100644 --- a/tests/smoke_admin.py +++ b/tests/smoke_admin.py @@ -301,6 +301,28 @@ check("kind prefix filter and limit", all(e["kind"].startswith("nearby.") for e check("history is admin and above, and not scopable on a key", "history:read" in I.CAPS["admin"] and "history:read" not in I.CAPS["leader"] and "history:read" in I.KEY_UNSCOPABLE) +print("\nlead claims") +lid = S.record_lead({"parent_name": "Claim Test", "email": "claim@example.test", "interested_in": "Pack 73"}) if hasattr(S, "record_lead") else None +if lid is None: + con = S.connect(); lid = "lead-claim-test" + con.execute("INSERT INTO join_leads (id, submitted_at, recorded_at, parent_name, email, payload) VALUES (?,?,?,?,?,?)", + (lid, I._now(), I._now(), "Claim Test", "claim@example.test", "{}")); con.commit(); con.close() +check("unclaimed count sees it", S.unclaimed_count() >= 1 and S.summary()["unclaimed"] >= 1) +c = S.claim_lead(lid, "p-a", "a@example.test") +check("claim recorded and listed on the lead", c["email"] == "a@example.test" and S.get_lead(lid)["claim"]["email"] == "a@example.test" + and [r for r in S.list_leads(limit=500) if r["id"] == lid][0]["claim"]["email"] == "a@example.test") +raises("claiming again yourself", 409, S.ClaimRejected, S.claim_lead, lid, "p-a", "a@example.test") +raises("someone else cannot take it over silently", 409, S.ClaimRejected, S.claim_lead, lid, "p-b", "b@example.test") +raises("someone else cannot release it", 403, S.ClaimRejected, S.release_lead, lid, "p-b", "b@example.test") +check("an owner can release it", S.release_lead(lid, "p-b", "b@example.test", can_release_any=True)["email"] == "a@example.test" + and S.get_lead(lid)["claim"] is None) +raises("release when nobody has it", 409, S.ClaimRejected, S.release_lead, lid, "p-a", "a@example.test") +S.claim_lead(lid, "p-b", "b@example.test"); S.release_lead(lid, "p-b", "b@example.test") +check("history keeps every claim", len(S.get_lead(lid)["claim_history"]) == 2 and all(h["released_at"] for h in S.get_lead(lid)["claim_history"])) +check("nothing on the lead itself changed", "status" not in S.get_lead(lid) and "claimed" not in S.get_lead(lid)) +check("unknown lead is None", S.claim_lead("nope", "p", "e") is None and S.release_lead("nope", "p", "e") is None) +con = S.connect(); con.execute("DELETE FROM lead_claims WHERE lead_id=?", (lid,)); con.execute("DELETE FROM join_leads WHERE id=?", (lid,)); con.commit(); con.close() + print("\napi docs registry") import admin_api as A reg = A.describe_routes()