mail-address-semantics/mas_modes/email_mode.py

"""The email side of the MAS mode contract, in both roles (CONTRACT.md section 3).

Takes normalized gateway events (POST /mas-events), mode statuses
(POST /mode-status) and, in the slave role, outside-master commands
(POST /master). The role is configured per resource type:

- master: the case lives in a `MasterStore` (the reference ticket service).
  Email negotiates by mail, hands off, records every status in the thread,
  owns the deadline (`tick()`) and resumes by mail when a handoff ends.
- slave: an outside master issues case ids, gets every inbound mail as an
  event, and decides what email sends and where mode requests go.

Every outbound mail goes through principle 10 (`mailer.refusal`), whoever
asked for it. Outbound contract messages are queued while the lock is
held and sent after it is released, so peers may call back at once.
"""

from __future__ import annotations

import json
import re
import sqlite3
import threading
import time
import uuid
from dataclasses import dataclass, field
from typing import Callable, Optional, Protocol, Union

from mas_core.generator import resource_local_part
from mas_core.parser import MasSyntaxError, parse_address
from mas_core.policy import resolve
from mas_core.statute import Statute

from . import message_errors
from .contract import TERMINAL, Idempotent, allowed, json_server, message, post, refuse
from .mailer import Mailer

NOT_ENABLED = {"InteractionModeNotEnabled", "ProfileNotEnabled"}

SCHEMA = """
CREATE TABLE IF NOT EXISTS handoffs (
  id TEXT PRIMARY KEY, role TEXT, resource_type TEXT, case_address TEXT, case_key TEXT,
  case_ref TEXT, mode TEXT, status TEXT, seq INTEGER DEFAULT 0, deadline REAL,
  needs TEXT, participant TEXT, endpoint TEXT, relay_seq INTEGER DEFAULT 0
);
CREATE TABLE IF NOT EXISTS intakes (
  ref TEXT PRIMARY KEY, resource_type TEXT, domain TEXT, message_ref TEXT,
  requested_mode TEXT, resource_id TEXT
);
CREATE TABLE IF NOT EXISTS cases (
  key TEXT PRIMARY KEY, address TEXT, resource_type TEXT, resource_id TEXT, policy TEXT
);
CREATE TABLE IF NOT EXISTS case_messages (
  ref TEXT PRIMARY KEY, case_key TEXT, message TEXT, disposition TEXT
);
"""


class MasterStore(Protocol):
    """Where email-mastered cases live (implemented by the reference ticket service)."""

    def handle(self, event: dict) -> tuple[int, dict]: ...
    def open_case(self, response: dict) -> Optional[str]: ...   # case ref, if open
    def case_address(self, case_ref: str) -> str: ...
    def record(self, case_ref: str, kind: str, text: str) -> None: ...
    def reply(self, case_ref: str, msg: dict, subject: str, body: str) -> bool: ...


@dataclass
class RoleConfig:
    role: str                                   # "master" | "slave"
    modes: dict = field(default_factory=dict)   # master: mode -> endpoint URL
    fields: tuple = ("phone", "time")           # details read from "Key: value" lines
    timeout: float = 3600.0                     # master: handoff deadline (seconds)
    master_url: Optional[str] = None            # slave: outside master's event URL
    autonomous_send: bool = False               # slave: default policy
    transfers: Union[str, dict] = "block"       # slave: "block" | {"webhook": URL}


def extract_details(text: str, fields) -> dict:
    """`Phone: +49 ...` style lines, for the fields a mode asks for."""
    out = {}
    for line in (text or "").splitlines():
        m = re.match(r"\s*([A-Za-z_-]+)\s*:\s*(.+?)\s*$", line)
        if m and m.group(1).lower() in fields and m.group(1).lower() not in out:
            out[m.group(1).lower()] = m.group(2)
    return out


def _resume_text(mode: str, status: str, info: dict) -> str:
    why = info.get("reason") or info.get("detail") or ""
    texts = {
        "completed": f"Your {mode} has taken place.",
        "failed": f"Your {mode} did not succeed{f' ({why})' if why else ''}.",
        "expired": f"Your {mode} request expired{f' ({why})' if why else ''}.",
        "refused": f"We cannot continue by {mode} ({why}).",
        "cancelled": f"Your {mode} was cancelled.",
        "timed-out": f"We could not arrange your {mode} in time.",
    }
    return texts[status] + " We continue by email: reply to this message."


class EmailMode:
    def __init__(self, roles: dict[str, RoleConfig], store: Optional[MasterStore] = None,
                 db_path: str = ":memory:", statutes: Optional[dict[str, Statute]] = None,
                 clock: Callable[[], float] = time.time, smtp=None, max_sent=None,
                 require_dmarc_pass: bool = True):
        if any(c.role == "master" for c in roles.values()) and store is None:
            raise ValueError("a master role needs a MasterStore")
        self.roles = roles
        self.store = store
        self.statutes = statutes or {}
        self.clock = clock
        self.db = sqlite3.connect(db_path, check_same_thread=False)
        self.db.row_factory = sqlite3.Row
        self.db.executescript(SCHEMA)
        self.mailer = Mailer(self.db, smtp, max_sent, require_dmarc_pass)
        self.lock = threading.RLock()
        self.base_url = ""            # set by serve(); status_url = base_url + /mode-status
        self._queue: list = []
        self._once = Idempotent()

    # -- HTTP entry points ---------------------------------------------------------

    def server(self, host="127.0.0.1", port=0):
        return json_server({"/mas-events": self.handle_event, "/mode-status": self.mode_status,
                            "/master": self.master_command,
                            "/tick": lambda body: (200, {"ok": True, "timed_out": self.tick()})},
                           host, port)

    def _run(self, fn, *args):
        with self.lock, self.db:
            out = fn(*args)
        self._flush()
        return out

    def _send(self, url, msg, on_result=None):
        self._queue.append((url, msg, on_result))

    def _flush(self):
        while True:
            with self.lock:
                if not self._queue:
                    return
                url, msg, on_result = self._queue.pop(0)
            status, body = post(url, msg)
            if on_result:
                with self.lock, self.db:
                    on_result(status, body)

    # -- inbound mail --------------------------------------------------------------

    def handle_event(self, event: dict) -> tuple[int, dict]:
        mas = event.get("mas", {})
        rtype = mas.get("resource_type") if mas.get("kind") == "resource" else mas.get("intake_for")
        cfg = self.roles.get(rtype) if mas.get("valid") and mas.get("namespace_confirmed") else None
        if cfg is None:
            return self.store.handle(event) if self.store else (404, {"status": "not-handled"})
        if cfg.role == "master":
            return self._run(self._master_event, event, rtype, cfg)
        return self._run(self._slave_event, event, rtype, cfg)

    def _case(self, address: str) -> dict:
        try:
            addr = parse_address(address)
        except MasSyntaxError:
            return {"address": address, "key": address.rsplit("@", 1)[0]}
        sem = resolve(addr, self.statutes.get(addr.domain))
        return {"address": address, "key": sem.equivalence_key}

    def _open_handoff(self, case_key):
        rows = self.db.execute("SELECT * FROM handoffs WHERE case_key=? ORDER BY rowid", (case_key,))
        return next((r for r in rows if r["status"] not in TERMINAL), None)

    def _master_event(self, event, rtype, cfg):
        code, resp = self.store.handle(event)
        ref = self.store.open_case(resp)
        if ref is None:
            return code, resp
        mas, msg = event["mas"], event.get("message", {})
        case = self._case(self.store.case_address(ref))
        h = self._open_handoff(case["key"])
        if h is not None:
            if h["status"] == "negotiating":
                details = extract_details(msg.get("text"), json.loads(h["needs"] or "[]"))
                if details:
                    self.store.record(ref, "mode-update", f"{h['mode']}: details {sorted(details)}")
                    self._send(h["endpoint"], message("handoff.update", case,
                                                      handoff_id=h["id"], details=details))
            return code, {**resp, "handoff_id": h["id"]}
        mode = mas.get("interaction")
        if not mode:
            return code, resp
        failures = {f["code"] for f in mas.get("failures", [])}
        hid = self._new_handoff("master", rtype, case, ref, mode, msg,
                                cfg.modes.get(mode), self.clock() + cfg.timeout)
        if failures & NOT_ENABLED:
            # The creation reply already says so; only an existing case gets a notice.
            self._status(hid, "refused", {"reason": "mode-not-enabled"},
                         notify=resp.get("status") != "created")
        elif mode not in cfg.modes:
            self._status(hid, "refused", {"reason": "mode-unavailable"})
        else:
            self._request(hid, case, mode, extract_details(msg.get("text"), cfg.fields), msg)
        return code, {**resp, "handoff_id": hid}

    def _new_handoff(self, role, rtype, case, ref, mode, msg, endpoint, deadline):
        hid = "h-" + uuid.uuid4().hex[:12]
        self.db.execute("INSERT INTO handoffs (id, role, resource_type, case_address, case_key,"
                        " case_ref, mode, status, deadline, participant, endpoint)"
                        " VALUES (?,?,?,?,?,?,?,?,?,?,?)",
                        (hid, role, rtype, case["address"], case["key"], ref, mode, None,
                         deadline, json.dumps(msg), endpoint))
        return hid

    def _request(self, hid, case, mode, details, msg):
        h = self._handoff(hid)
        self._status(hid, "requested", {})
        req = message("handoff.request", case, handoff_id=hid, mode=mode, details=details,
                      status_url=self.base_url + "/mode-status",
                      participant={"address": msg.get("from")} if msg.get("from") else None)

        def on_result(status, body):
            if status is None or status >= 400:
                self._status(hid, "refused", {"reason": "mode-unavailable",
                                              "detail": f"mode endpoint answered {status}"})
        self._send(h["endpoint"], req, on_result)

    def _handoff(self, hid):
        return self.db.execute("SELECT * FROM handoffs WHERE id=?", (hid,)).fetchone()

    # -- statuses ------------------------------------------------------------------

    def mode_status(self, msg: dict) -> tuple[int, dict]:
        return self._once(msg, lambda m: self._run(self._mode_status, m))

    def _mode_status(self, msg):
        errors = message_errors(msg)
        if errors or msg.get("type") != "handoff.status":
            return refuse("schema", "; ".join(errors) or "expected handoff.status", 400)
        h = self._handoff(msg["handoff_id"])
        if h is None or h["case_key"] != msg["case"]["key"]:
            return refuse("unknown-handoff", msg["handoff_id"], 404)
        if msg["seq"] <= h["seq"]:
            return 200, {"ok": True, "ignored": "stale"}
        if msg["status"] == "timed-out":
            return refuse("invalid-transition", "only the timeout owner records timed-out")
        if not allowed(h["status"], msg["status"]):
            return refuse("invalid-transition", f"{h['status']} -> {msg['status']}")
        self.db.execute("UPDATE handoffs SET seq=? WHERE id=?", (msg["seq"], h["id"]))
        info = {k: msg[k] for k in ("needs", "at", "outcome", "reason", "detail") if k in msg}
        self._status(h["id"], msg["status"], info)
        return 200, {"ok": True}

    def _status(self, hid, status, info, notify=True):
        """Record a status and react to it, per role."""
        h = self._handoff(hid)
        self.db.execute("UPDATE handoffs SET status=?, needs=? WHERE id=?",
                        (status, json.dumps(info.get("needs")) if "needs" in info else h["needs"], hid))
        case = {"address": h["case_address"], "key": h["case_key"]}
        participant = json.loads(h["participant"])
        if h["role"] == "master":
            self.store.record(h["case_ref"], "mode-status",
                              f"{h['mode']}: {status}" + "".join(f" {k}={v}" for k, v in info.items()))
            if status == "negotiating":
                self._ask(h, participant, info["needs"], lambda subj, body:
                          self.store.reply(h["case_ref"], participant, subj, body))
            elif status in TERMINAL and notify:
                self.store.reply(h["case_ref"], participant, participant.get("subject") or "Your request",
                                 _resume_text(h["mode"], status, info))
            return
        # slave: the master owns the flow; relay, and ask only when allowed to
        cfg = self.roles[h["resource_type"]]
        if status != "requested":
            self._send(cfg.master_url, message("handoff.status", case, handoff_id=hid,
                                               seq=self._relay_seq(hid), status=status, **info))
        if status == "negotiating" and self._policy(h["case_key"])["autonomous_send"]:
            self._ask(h, participant, info["needs"], lambda subj, body: self.mailer.reply(
                h["case_key"], h["case_address"], participant, subj, body) is None)

    def _relay_seq(self, hid):
        # Statuses relayed to the master are numbered by email, per handoff.
        self.db.execute("UPDATE handoffs SET relay_seq=relay_seq+1 WHERE id=?", (hid,))
        return self._handoff(hid)["relay_seq"]

    def _ask(self, h, participant, needs, send):
        lines = [f"To arrange the {h['mode']}, please reply to this message with:"]
        lines += [f"{n.capitalize()}: ..." for n in needs]
        send(participant.get("subject") or "Your request", "\n".join(lines))

    def tick(self) -> list[str]:
        """Master role: time out handoffs past their deadline and resume by email."""
        with self.lock, self.db:
            now = self.clock()
            rows = self.db.execute("SELECT * FROM handoffs WHERE role='master' AND deadline<=?",
                                   (now,)).fetchall()
            out = []
            for h in rows:
                if h["status"] in TERMINAL or h["status"] is None:
                    continue
                self._status(h["id"], "timed-out", {})
                if h["endpoint"]:
                    self._send(h["endpoint"], message(
                        "handoff.cancel", {"address": h["case_address"], "key": h["case_key"]},
                        handoff_id=h["id"], reason="timed-out"))
                out.append(h["id"])
        self._flush()
        return out

    # -- slave role ----------------------------------------------------------------

    def _store_message(self, case_key, msg, disposition):
        ref = "m-" + uuid.uuid4().hex[:12]
        self.db.execute("INSERT INTO case_messages VALUES (?,?,?,?)",
                        (ref, case_key, json.dumps(msg), disposition))
        return ref

    def _slave_event(self, event, rtype, cfg):
        mas, msg = event["mas"], event.get("message", {})
        if mas.get("kind") != "resource":
            iref = "i-" + uuid.uuid4().hex[:12]
            mref = self._store_message(None, msg, "intake")
            self.db.execute("INSERT INTO intakes VALUES (?,?,?,?,?,NULL)",
                            (iref, rtype, mas["domain"], mref, mas.get("interaction")))
            self._send(cfg.master_url, message("case.intake", intake_ref=iref, resource_type=rtype,
                                               event=event, requested_mode=mas.get("interaction")))
            return 202, {"status": "intake-reported", "intake_ref": iref}
        row = self.db.execute("SELECT * FROM cases WHERE key=?", (mas["equivalence_key"],)).fetchone()
        if row is None:
            self._store_message(None, msg, "quarantined-unknown")
            return 202, {"status": "unknown-resource", "action": "quarantined, no reply"}
        case = {"address": row["address"], "key": row["key"]}
        mref = self._store_message(row["key"], msg, "attached")
        self._send(cfg.master_url, message("case.message", case, event=event, message_ref=mref))
        out = {"status": "reported", "case": case, "message_ref": mref}
        h = self._open_handoff(row["key"])
        if h is not None and h["status"] == "negotiating" and h["endpoint"]:
            details = extract_details(msg.get("text"), json.loads(h["needs"] or "[]"))
            if details:
                self._send(h["endpoint"], message("handoff.update", case, handoff_id=h["id"],
                                                  details=details))
        elif h is None and mas.get("interaction"):
            out["handoff_id"] = self._transfer(rtype, cfg, case, mas["interaction"], msg)
        return 200, out

    def _policy(self, case_key) -> dict:
        row = self.db.execute("SELECT policy FROM cases WHERE key=?", (case_key,)).fetchone()
        return json.loads(row["policy"])

    def _transfer(self, rtype, cfg, case, mode, msg):
        """A !MODE request in the slave role: blocked, or sent to the primed webhook."""
        transfers = self._policy(case["key"])["transfers"]
        webhook = transfers.get("webhook") if isinstance(transfers, dict) else None
        hid = self._new_handoff("slave", rtype, case, case["key"], mode, msg, webhook, None)
        self._send(cfg.master_url, message("handoff.requested", case, handoff_id=hid, mode=mode,
                                           forwarded=webhook is not None))
        if webhook is None:
            self._status(hid, "requested", {})
            self._status(hid, "refused", {"reason": "blocked"})
        else:
            self._request(hid, case, mode, extract_details(msg.get("text"), cfg.fields), msg)
        return hid

    def master_command(self, msg: dict) -> tuple[int, dict]:
        return self._once(msg, lambda m: self._run(self._master_command, m))

    def _master_command(self, msg):
        errors = message_errors(msg)
        if errors:
            return refuse("schema", "; ".join(errors), 400)
        t = msg["type"]
        if t == "case.assign":
            return self._assign(msg)
        row = self.db.execute("SELECT * FROM cases WHERE key=?", (msg["case"]["key"],)).fetchone()
        if row is None:
            return refuse("unknown-case", msg["case"]["key"], 404)
        if t == "case.policy":
            pol = {**self._policy(row["key"]),
                   **{k: msg[k] for k in ("autonomous_send", "transfers") if k in msg}}
            self.db.execute("UPDATE cases SET policy=? WHERE key=?", (json.dumps(pol), row["key"]))
            return 200, {"ok": True, "policy": pol}
        if t == "mail.send":
            m = self.db.execute("SELECT * FROM case_messages WHERE ref=? AND case_key=?",
                                (msg["in_reply_to"], row["key"])).fetchone()
            if m is None:
                return refuse("refused", msg["in_reply_to"], reason="unknown-message")
            orig = json.loads(m["message"])
            why = self.mailer.reply(row["key"], row["address"], orig,
                                    msg.get("subject") or orig.get("subject") or "Your request",
                                    msg["body"])
            return (200, {"ok": True, "sent": True}) if why is None else refuse("refused", why, reason=why)
        if t == "handoff.cancel":
            h = self._handoff(msg["handoff_id"])
            if h is None or h["case_key"] != row["key"]:
                return refuse("unknown-handoff", msg["handoff_id"], 404)
            if h["endpoint"] and h["status"] not in TERMINAL:
                self._send(h["endpoint"], message("handoff.cancel", msg["case"], handoff_id=h["id"],
                                                  reason=msg.get("reason")))
            return 200, {"ok": True}
        return refuse("unsupported", t, 400)

    def _assign(self, msg):
        it = self.db.execute("SELECT * FROM intakes WHERE ref=?", (msg["intake_ref"],)).fetchone()
        if it is None:
            return refuse("unknown-intake", msg["intake_ref"], 404)
        if it["resource_id"] is not None:
            return refuse("already-assigned", it["resource_id"])
        try:
            address = f"{resource_local_part(it['resource_type'], msg['resource_id'])}@{it['domain']}"
        except MasSyntaxError as e:
            return refuse("invalid-resource-id", str(e), 400)
        case = self._case(address)
        if self.db.execute("SELECT 1 FROM cases WHERE key=?", (case["key"],)).fetchone():
            return refuse("case-exists", case["key"])
        cfg = self.roles[it["resource_type"]]
        pol = {"autonomous_send": cfg.autonomous_send, "transfers": cfg.transfers,
               **msg.get("policy", {})}
        self.db.execute("INSERT INTO cases VALUES (?,?,?,?,?)",
                        (case["key"], address, it["resource_type"], msg["resource_id"], json.dumps(pol)))
        self.db.execute("UPDATE intakes SET resource_id=? WHERE ref=?", (msg["resource_id"], it["ref"]))
        self.db.execute("UPDATE case_messages SET case_key=? WHERE ref=?", (case["key"], it["message_ref"]))
        intake = json.loads(self.db.execute("SELECT message FROM case_messages WHERE ref=?",
                                            (it["message_ref"],)).fetchone()["message"])
        out = {"ok": True, "case": case, "message_ref": it["message_ref"]}
        if pol["autonomous_send"]:
            why = self.mailer.reply(case["key"], address, intake, intake.get("subject") or "Your request",
                                    f"Your request is now case {msg['resource_id']}.\n"
                                    f"Reply to {address} to add to it.")
            out["acknowledged"] = why is None
        if it["requested_mode"]:
            out["handoff_id"] = self._transfer(it["resource_type"], cfg, case, it["requested_mode"], intake)
        return 200, out