"""The use cases — contacts, what was said, and what happens next.
Each is declared once and served as an API route AND an MCP tool.
WHO: every signed-in user may READ contacts and the activity record — CRM-02 exists so "the next person who
picks up the account knows what was said", which a private log defeats. (Whether rows are owned at all is
the spec's customer_varies question for CRM-01: this brick answers "shared", stated here.) Writing needs
activity:contact:write / activity:activity:write; changing someone else's contact or task needs to be its
owner or hold activity:contact:manage / activity:activity:manage (0.3.0: a permission, not the `manager` role
name; roles.yaml: the default roles). Tasks are listed per person (CRM-12).
NEIGHBOURS, asked through their routes:
business-partner a contact works at a business partner that must exist; a company's timeline or contact list is
for a partner that exists.
pipeline an activity on a deal is reported with touch_opportunity, so a deal being worked is not
flagged as gone quiet (CRM-16). When an activity names a company but no deal, and the rep has
exactly ONE open deal there, it is attached to that deal (CRM-02: "it appears on the contact,
the account and the open deal"); with several, it says so rather than guess. In process (tier
1) the pipeline answers as the rep; across apps (tier 2) it answers as a service and sees no
deals — the on-behalf-of loss erp_platform.seam documents — so nothing is guessed.
0.2: a deal NAMED by the caller — the timeline of opportunity O-1, a call logged onto it — is
looked up AS THE CALLER (seam caller=): a deal they may not see is 404, as it is in the pipeline.
A rep could otherwise read another rep's deal through its activities (the CRM audit, H5).
"""
import time
from contextlib import contextmanager
from datetime import date, datetime, timezone
from typing import Literal, Optional
from uuid import uuid4
from erp_platform import auth
from erp_platform.bricks import Brick, Refused
from erp_platform.repository import Conflict, retry_on_conflict
from erp_platform.seam import Seam
from erp_platform.types import BigId, Email, IsoDate, IsoTime, Limit, Name, Note, text
from brick_activity.domain.activity import (PAST, Activity, ActivityIncomplete, ActivityRuleBroken, Contact,
normalise_email, task_board)
CONTACTS_WRITE, ACTIVITY_WRITE = "activity:contact:write", "activity:activity:write"
# 0.3.0: a brick decides by PERMISSION, never by role name (erp-platform 0.9.2). Changing a contact or a task someone
# else owns was the `manager` ROLE; it is now these — held by whichever of the customer's roles should (roles.yaml).
CONTACTS_MANAGE, TASKS_MANAGE = "activity:contact:manage", "activity:activity:manage"
Kind = Literal["NOTE", "CALL", "MEETING", "EMAIL", "TASK"]
Phone, Ref = text(64), text(64)
ACT = Brick("activity", version="0.3.1", ids_as_strings=True, instructions=(
"Contacts (people at business partners) and the record of what was said and what happens next. "
"Log a NOTE, CALL, MEETING or EMAIL that happened, or a TASK with a due date; a task is completed, "
"cancelled or reopened, never deleted. Activities show on the contact, the company and the deal, newest "
"first. A person's open tasks are grouped overdue, today and this week. bp_id is a 64-bit id, sent and "
"returned as a string of digits. A contact and a task carry a `version`: pass the one you read back with a "
"change; a 409 means someone changed it since."))
# 0.3.0: the process is declared (process.yaml → GET /process/bpmn), as erp-platform ≥ 0.8 requires.
ACT.declare_process("process.yaml")
PARTNERS = Seam("business-partner", "BUSINESS_PARTNER_BASE_URL", roles=["service"], permissions=[])
PIPELINE = Seam("pipeline", "PIPELINE_BASE_URL", roles=["service"], permissions=["pipeline:opportunity:touch"])
@contextmanager
def _repository_default():
from erp_platform import db
from brick_activity.infrastructure.tables import ActivityRepository
with db.engine.begin() as conn:
yield ActivityRepository(conn)
repository = _repository_default
def _now() -> datetime:
return datetime.now(timezone.utc)
@contextmanager
def _rules():
try:
yield
except ActivityIncomplete as e:
raise Refused(422, str(e)) from None
except ActivityRuleBroken as e:
raise Refused(409, str(e)) from None
def _may_change(caller: dict, owner: str, what: str) -> None:
manage = CONTACTS_MANAGE if what == "contact" else TASKS_MANAGE
if owner != caller.get("email") and not auth.has_permission(caller, manage):
raise Refused(403, f"this {what} belongs to {owner}; only they or a manager can change it")
def _contact_view(c: Contact) -> dict:
return {"contact_id": c.contact_id, "bp_id": c.bp_id, "full_name": c.full_name, "email": c.email,
"title": c.title, "phone": c.phone, "owner": c.owner, "subscribed": c.subscribed,
"status": c.status, "version": c.version}
def _activity_view(a: Activity, today: Optional[str] = None) -> dict:
return {"activity_id": a.activity_id, "kind": a.kind, "subject": a.subject, "note": a.note,
"owner": a.owner, "bp_id": a.bp_id, "contact_id": a.contact_id,
"opportunity_id": a.opportunity_id, "direction": a.direction, "occurred_at": a.occurred_at,
"due_on": a.due_on, "status": a.status, "done_at": a.done_at,
"overdue": a.is_overdue(today or date.today().isoformat()), "version": a.version}
def _get_contact(repo, contact_id: str) -> Contact:
c = repo.contact(contact_id)
if c is None:
raise Refused(404, f"no contact {contact_id}")
return c
def _get_activity(repo, activity_id: str) -> Activity:
a = repo.activity(activity_id)
if a is None:
raise Refused(404, f"no activity {activity_id}")
return a
def _at(x, version: Optional[int], table: str, key: dict):
"""The contact or task, if it is still at the version the caller loaded; a Conflict (409) if not."""
if version is not None and x.version != int(version):
raise Conflict(table, key, int(version), x.version)
return x
def _changing(version: Optional[int], fn):
"""With the caller's version, a race is their 409; with none, the change is re-run on the latest record."""
return fn() if version is not None else retry_on_conflict(fn)
def _deal_visible(caller: dict, opportunity_id: str) -> None:
"""A deal the caller names must be one they may see — asked of the pipeline AS THE CALLER. 404 otherwise,
the pipeline's own answer. Not installed: the id is taken as given (there is nothing to ask)."""
if not PIPELINE.installed:
return
r = PIPELINE.get(f"/opportunities/{opportunity_id}", caller=caller)
if r.status == 404:
raise Refused(404, f"no opportunity {opportunity_id}")
if not r.ok:
raise Refused(503, f"cannot check deal {opportunity_id} right now ({r.reason}); try again")
def _partner_known(caller: dict, bp_id: int) -> None:
"""A company the caller names must exist (partners are readable by everyone signed in)."""
if PARTNERS.installed:
r = PARTNERS.get(f"/partners/{int(bp_id)}", caller=caller)
if r.status == 404:
raise Refused(404, f"no partner {bp_id}")
def _partner_exists(caller: dict, bp_id: int) -> None:
if PARTNERS.installed:
r = PARTNERS.get(f"/partners/{int(bp_id)}", actor=caller.get("email", ""))
if r.status == 404:
raise Refused(422, f"there is no business partner {bp_id} — add the company first")
def _the_open_deal(caller: dict, bp_id: int) -> tuple:
"""(opportunity_id, note): the rep's single open deal at this company, or None and why not."""
if not PIPELINE.installed:
return None, ""
r = PIPELINE.get(f"/partners/{int(bp_id)}/group-pipeline", caller=caller) # the caller's deals only
if not r.ok:
return None, f"not attached to a deal: the pipeline did not answer ({r.reason})"
here = [o["opportunity_id"] for o in r.data.get("opportunities", []) if int(o["bp_id"]) == int(bp_id)]
if len(here) == 1:
return here[0], ""
if len(here) > 1:
return None, f"not attached to a deal: {len(here)} open deals at this company — name one"
return None, ""
def _report_to_pipeline(caller: dict, a: Activity, at: str) -> str:
"""'' when reported or nothing to report; a note when the deal could not be told."""
if not a.opportunity_id or not PIPELINE.installed:
return ""
r = PIPELINE.post(f"/opportunities/{a.opportunity_id}/touch",
{"at": at, "what": f"{a.kind.lower()}: {a.subject}"}, actor=caller.get("email", ""))
return "" if r.ok else f"the deal was not told about it ({r.reason})"
# ── contacts ───────────────────────────────────────────────────────────────────────────────
@ACT.use_case(kind="command", route="POST /contacts", permission=CONTACTS_WRITE, idempotent=True)
def add_contact(caller: dict, bp_id: BigId, full_name: Name, email: Email, title: Name = "",
phone: Phone = "") -> dict:
"""Add a person at a company (a business partner), owned by the caller. One email is one contact: adding
an email that already exists returns that contact if it is at the same company, and is refused if it
belongs to someone at another."""
_partner_exists(caller, bp_id)
try:
return _add_contact(caller, bp_id, full_name, email, title, phone)
except Exception as e: # noqa: BLE001 — re-raised unless the race
from brick_activity.infrastructure.tables import lost_email_claim
if not lost_email_claim(e):
raise
# Lost the race for this email to a concurrent add (its lookup saw nothing; its claim failed at the INSERT
# or at COMMIT). Answer as if it had come second: that contact, or the same refusal (H12: never a 500).
# The winner's claim can be visible a moment before its contact row is (a claim that won at the INSERT, its
# transaction still committing), so the contact is read again for up to ~2 s before giving up with a 409.
for attempt in range(6):
with repository() as repo, _rules():
existing = repo.contact_by_email(normalise_email(email))
if existing is not None:
return _same_company(existing, bp_id)
time.sleep(0.1 * (attempt + 1))
raise Refused(409, f"{normalise_email(email)} is being added by another request; try again")
def _same_company(existing: Contact, bp_id: int) -> dict:
if existing.bp_id != int(bp_id):
raise Refused(409, f"{existing.email} is already {existing.full_name} at partner {existing.bp_id}")
return _contact_view(existing)
def _add_contact(caller, bp_id, full_name, email, title, phone) -> dict:
with repository() as repo, _rules():
existing = repo.contact_by_email(normalise_email(email))
if existing is not None:
return _same_company(existing, bp_id)
c = Contact.new(f"C-{uuid4().hex[:10].upper()}", bp_id, full_name, email, caller["email"],
title=title, phone=phone)
repo.claim_email(c.email, c.contact_id) # first: a duplicate fails before anything is written
repo.save_contact(c)
return _contact_view(c)
@ACT.use_case(kind="command", route="PATCH /contacts/{contact_id}", permission=CONTACTS_WRITE, idempotent=True,
if_match="version")
def edit_contact(caller: dict, contact_id: Ref, full_name: Optional[Name] = None, title: Optional[Name] = None,
phone: Optional[Phone] = None, email: Optional[Email] = None, version: Optional[int] = None) -> dict:
"""Change a contact's name, title, phone or email. Fields left out are unchanged. A new email must not
belong to anyone else. `version` (or If-Match): the one you read; a 409 means it changed since."""
def run():
with repository() as repo, _rules():
c = _at(_get_contact(repo, contact_id), version, "activity_contacts", {"contact_id": contact_id})
_may_change(caller, c.owner, "contact")
c.describe(full_name=full_name, title=title, phone=phone)
if email is not None and normalise_email(email) != c.email:
new = normalise_email(email)
repo.claim_email(new, c.contact_id)
repo.release_email(c.email, c.contact_id)
c.email = new
repo.save_contact(c)
return _contact_view(c)
return _changing(version, run)
@ACT.use_case(kind="command", route="PUT /contacts/{contact_id}/subscription", permission=CONTACTS_WRITE,
idempotent=True, if_match="version")
def set_subscription(caller: dict, contact_id: Ref, subscribed: bool, version: Optional[int] = None) -> dict:
"""Record whether the contact consents to marketing email. One-to-one sales email does not need it. A
contact who has left the company cannot be subscribed."""
def run():
with repository() as repo, _rules():
c = _at(_get_contact(repo, contact_id), version, "activity_contacts", {"contact_id": contact_id})
if subscribed and c.status == "left":
raise Refused(409, f"{c.full_name} has left the company; their old address cannot be subscribed")
if c.subscribed == bool(subscribed):
return _contact_view(c)
c.subscribed = bool(subscribed)
repo.save_contact(c)
return _contact_view(c)
return _changing(version, run)
@ACT.use_case(kind="command", route="POST /contacts/{contact_id}/left", permission=CONTACTS_WRITE,
idempotent=True, if_match="version")
def contact_left(caller: dict, contact_id: Ref, version: Optional[int] = None) -> dict:
"""The person has left the company. They stay — every activity that mentions them points here — but
are unsubscribed and marked as gone."""
def run():
with repository() as repo, _rules():
c = _at(_get_contact(repo, contact_id), version, "activity_contacts", {"contact_id": contact_id})
_may_change(caller, c.owner, "contact")
if c.status == "left" and not c.subscribed:
return _contact_view(c)
c.left_company()
repo.save_contact(c)
return _contact_view(c)
return _changing(version, run)
@ACT.use_case(kind="query", route="GET /contacts")
def find_contacts(caller: dict, bp_id: Optional[BigId] = None, email: Optional[Email] = None) -> dict:
"""The people at a company, or the contact with an email. A company that does not exist is 404."""
if not email and bp_id is not None:
_partner_known(caller, bp_id)
with repository() as repo, _rules():
if email:
c = repo.contact_by_email(normalise_email(email))
return {"contacts": [_contact_view(c)] if c else []}
if bp_id is None:
raise Refused(422, "name a company (bp_id) or an email")
return {"contacts": [_contact_view(c) for c in repo.contacts_at(bp_id)]}
@ACT.use_case(kind="query", route="GET /contacts/{contact_id}", etag="version")
def get_contact(caller: dict, contact_id: Ref) -> dict:
"""One contact, with its `version` (send it back with a change)."""
with repository() as repo:
return _contact_view(_get_contact(repo, contact_id))
# ── activities ─────────────────────────────────────────────────────────────────────────────
@ACT.use_case(kind="command", route="POST /activities", permission=ACTIVITY_WRITE)
def log_activity(caller: dict, kind: Kind, subject: Name, contact_id: Optional[Ref] = None,
bp_id: Optional[BigId] = None, opportunity_id: Optional[Ref] = None, note: Note = "",
direction: Optional[Literal["inbound", "outbound"]] = None, occurred_at: Optional[IsoTime] = None,
due_on: Optional[IsoDate] = None, assignee: Optional[Email] = None) -> dict:
"""CRM-02: log a call, email, meeting or note that happened — or a TASK due on a date — in one step.
Name the contact and the company comes from them; name the company alone for a note about it. If no deal
is named and the caller has exactly one open deal there, the activity goes on that deal too; a deal you
name must be one you can see. A task can be assigned to someone else with `assignee`."""
if opportunity_id:
_deal_visible(caller, opportunity_id)
with repository() as repo, _rules():
if contact_id:
c = _get_contact(repo, contact_id)
if bp_id is not None and int(bp_id) != c.bp_id:
raise Refused(409, f"{c.full_name} works at partner {c.bp_id}, not {bp_id}")
bp_id = c.bp_id
if bp_id is None:
raise Refused(422, "an activity has to be about someone or some company — name a contact or a bp_id")
notes = []
if opportunity_id is None:
opportunity_id, why = _the_open_deal(caller, bp_id)
notes.append(why)
owner = (assignee or caller["email"]).strip().lower() if kind == "TASK" else caller["email"]
a = Activity.log(f"A-{uuid4().hex[:12].upper()}", kind, subject, owner, bp_id, contact_id=contact_id,
opportunity_id=opportunity_id, note=note, direction=direction,
occurred_at=occurred_at, due_on=due_on, now=_now())
repo.save_activity(a)
if a.kind in PAST:
notes.append(_report_to_pipeline(caller, a, a.occurred_at))
return {**_activity_view(a), "notes": [n for n in notes if n]}
@ACT.use_case(kind="query", route="GET /timeline")
def timeline(caller: dict, contact_id: Optional[Ref] = None, bp_id: Optional[BigId] = None,
opportunity_id: Optional[Ref] = None, limit: Limit = 100) -> dict:
"""What was said and done, newest first — for a contact, a company (including everything logged against
the people there), or a deal. A deal's timeline is for those who may see the deal (404 otherwise, as in
the pipeline); a contact or company that does not exist is 404."""
if not (contact_id or bp_id is not None or opportunity_id):
raise Refused(422, "name a contact, a company (bp_id) or a deal (opportunity_id)")
if opportunity_id:
_deal_visible(caller, opportunity_id)
if bp_id is not None:
_partner_known(caller, bp_id)
with repository() as repo:
if contact_id:
_get_contact(repo, contact_id)
acts = repo.timeline(contact_id=contact_id, bp_id=bp_id, opportunity_id=opportunity_id,
limit=min(limit, 500))
return {"activities": [_activity_view(a) for a in acts]}
@ACT.use_case(kind="query", route="GET /tasks/mine")
def my_tasks(caller: dict, today: Optional[IsoDate] = None) -> dict:
"""CRM-12: the caller's open tasks as overdue, today and this week (the next seven days), with counts."""
day = (today or date.today().isoformat())[:10]
with repository() as repo:
board = task_board(repo.tasks_of(caller["email"]), day)
return {"today": day, "counts": {k: len(v) for k, v in board.items()},
**{k: [_activity_view(a, day) for a in v] for k, v in board.items()}}
def _task(repo, caller: dict, activity_id: str, version: Optional[int]) -> Activity:
a = _at(_get_activity(repo, activity_id), version, "activity_log", {"activity_id": activity_id})
_may_change(caller, a.owner, "task")
return a
@ACT.use_case(kind="command", route="POST /tasks/{activity_id}/complete", permission=ACTIVITY_WRITE,
idempotent=True, if_match="version")
def complete_task(caller: dict, activity_id: Ref, note: Optional[Note] = None, version: Optional[int] = None) -> dict:
"""Mark a task done, optionally adding what came of it. Its text is kept. Completing it again changes
nothing. A task on a deal tells the deal it was worked. `version` (or If-Match): the one you read."""
def run():
with repository() as repo, _rules():
a = _task(repo, caller, activity_id, version)
changed = a.complete(_now(), note)
if changed:
repo.save_activity(a)
return a, changed
a, changed = _changing(version, run)
notes = [_report_to_pipeline(caller, a, a.done_at)] if changed else []
return {**_activity_view(a), "notes": [n for n in notes if n]}
@ACT.use_case(kind="command", route="POST /tasks/{activity_id}/cancel", permission=ACTIVITY_WRITE,
idempotent=True, if_match="version")
def cancel_task(caller: dict, activity_id: Ref, version: Optional[int] = None) -> dict:
"""Cancel a task that will not be done. It stays on the record."""
def run():
with repository() as repo, _rules():
a = _task(repo, caller, activity_id, version)
if a.cancel():
repo.save_activity(a)
return _activity_view(a)
return _changing(version, run)
@ACT.use_case(kind="command", route="POST /tasks/{activity_id}/reopen", permission=ACTIVITY_WRITE,
idempotent=True, if_match="version")
def reopen_task(caller: dict, activity_id: Ref, version: Optional[int] = None) -> dict:
"""Open a done or cancelled task again."""
def run():
with repository() as repo, _rules():
a = _task(repo, caller, activity_id, version)
if a.reopen():
repo.save_activity(a)
return _activity_view(a)
return _changing(version, run)
@ACT.use_case(kind="command", route="PUT /tasks/{activity_id}/due", permission=ACTIVITY_WRITE, idempotent=True,
if_match="version")
def reschedule_task(caller: dict, activity_id: Ref, due_on: IsoDate, version: Optional[int] = None) -> dict:
"""Move an open task's due date."""
def run():
with repository() as repo, _rules():
a = _task(repo, caller, activity_id, version)
a.reschedule(due_on)
repo.save_activity(a)
return _activity_view(a)
return _changing(version, run)