mail-address-semantics/examples/ticket_service/ticket_service.py

#!/usr/bin/env python3
"""Reference ticket service for the MAS v0.1 experiment.

Consumes normalized MAS events from `mas-gateway` (POST /mas-events) and
proves the canonical lifecycle:

  A create     ticket!callback@DOMAIN      -> new ticket, reply from ticket=ID@DOMAIN
               (any address the statute declares with "intake_for": "ticket")
  B correlate  ticket=ID[+facet]@DOMAIN     -> message attached to ID; subject ignored
  C closed     ticket=ID@DOMAIN (closed)    -> archived, ticket not mutated, one notice
  D unknown    ticket=NOPE@DOMAIN           -> quarantined, no reply (no backscatter)

Ticket ids match case-insensitively (`ticket=gs-id@` finds `GS-ID`), because
forwarders and mailing lists lower-case local-parts in practice.

Resource addresses are correlation identifiers, not credentials: knowing
one lets you send mail to the ticket, nothing more. Admin endpoints here are
unauthenticated and meant for local experiments only.
"""

from __future__ import annotations

import argparse
import json
import secrets
import sqlite3
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Callable, Optional

from mas_core.generator import resource_local_part
from mas_modes.mailer import Mailer, parse_duration, parse_size

RESOURCE_TYPE = "ticket"
INTAKE_TARGET = "ticket"
_ALPHABET = "0123456789abcdefghjkmnpqrstvwxyz"  # Crockford base32, no i l o u


def default_ticket_id() -> str:
    """Non-guessable-preferred, per the great-service statute.

    Lower case, so the address a customer sees is the one forwarders and
    mailing lists deliver (spec 6.2, SCP-7).
    """
    return "gs-" + "".join(secrets.choice(_ALPHABET) for _ in range(10))


SCHEMA = """
CREATE TABLE IF NOT EXISTS tickets (
  id TEXT PRIMARY KEY, domain TEXT NOT NULL, status TEXT NOT NULL,
  requester TEXT, subject TEXT, requested_interaction TEXT, created_seq INTEGER
);
CREATE TABLE IF NOT EXISTS messages (
  seq INTEGER PRIMARY KEY AUTOINCREMENT, ticket_id TEXT, disposition TEXT NOT NULL,
  recipient TEXT, facet TEXT, sender TEXT, subject TEXT, message_id TEXT, body TEXT
);
"""
# Before the shared Mailer (MAS-WP-0012) replies lived in this table. It is
# read once at start so mail sent before keeps counting against the cap.
LEGACY_OUTBOX = "outbox"


class TicketService:
    def __init__(self, db_path: str = ":memory:", id_factory: Callable[[], str] = default_ticket_id,
                 smtp: Optional[tuple[str, int]] = None, closed_notice: bool = True,
                 max_sent: Optional[int] = None, require_dmarc_pass: bool = False,
                 mailer: Optional[Mailer] = None):
        """Replies go through `mailer`, the deployment's one outbound path (principle
        10, one lifetime cap). Without one, the service makes its own from smtp,
        max_sent and require_dmarc_pass."""
        self.db = sqlite3.connect(db_path, check_same_thread=False)
        self.db.row_factory = sqlite3.Row
        self.db.executescript(SCHEMA)
        self.lock = threading.Lock()
        self.new_id = id_factory
        self.closed_notice = closed_notice
        self.mailer = mailer or Mailer(smtp=smtp, max_sent=max_sent,
                                       require_dmarc_pass=require_dmarc_pass)
        self._import_legacy_outbox()

    def _import_legacy_outbox(self):
        exists = self.db.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name=?",
                                 (LEGACY_OUTBOX,)).fetchone()
        if exists:
            rows = [{**dict(r), "case_ref": r["ticket_id"]}
                    for r in self.db.execute(f"SELECT * FROM {LEGACY_OUTBOX} ORDER BY seq")]
            self.mailer.import_legacy(f"ticket-service:{LEGACY_OUTBOX}", rows)

    # -- event handling -------------------------------------------------------------

    def handle(self, event: dict) -> tuple[int, dict]:
        mas, msg = event.get("mas", {}), event.get("message", {})
        with self.lock, self.db:
            if not mas.get("valid"):
                return 422, {"status": "not-mas", "action": "ignored"}
            if not mas.get("namespace_confirmed"):
                # No validated statute: we refuse to act on unconfirmed semantics.
                return 422, {"status": "namespace-unconfirmed", "action": "ignored",
                             "discovery": mas.get("discovery")}
            if mas.get("kind") == "resource" and mas.get("resource_type") == RESOURCE_TYPE:
                return self._to_resource(event, mas, msg)
            if mas.get("intake_for") == RESOURCE_TYPE:  # declared by the statute (SCP-3)
                return self._create(event, mas, msg)
            return 404, {"status": "not-handled", "action": "ignored"}

    def _create(self, event, mas, msg):
        interaction = mas.get("interaction")
        failures = {f["code"] for f in mas.get("failures", [])}
        honored = interaction if interaction and not failures else None
        tid = self.new_id()
        domain = mas["domain"]
        self.db.execute("INSERT INTO tickets VALUES (?,?,?,?,?,?,NULL)",
                        (tid, domain, "open", msg.get("from"), msg.get("subject"), honored))
        self._store(tid, "attached", event, mas, msg)
        address = f"{resource_local_part(RESOURCE_TYPE, tid)}@{domain}"
        lines = [f"Your request is now ticket {tid}.",
                 f"Reply to {address} to add to it; no reference in the subject is needed."]
        if honored:
            lines.append(f"You asked to continue via '{honored}'. We will try to honour that preference.")
        elif interaction:
            lines.append(f"The requested continuation mode '{interaction}' is not available "
                         f"({', '.join(sorted(failures))}); we will continue by email.")
        self._reply(tid, address, msg, msg.get("subject") or "Your request", "\n".join(lines))
        return 201, {"status": "created", "ticket_id": tid, "resource_address": address,
                     "requested_interaction": honored}

    def _to_resource(self, event, mas, msg):
        # Ids match case-insensitively: forwarders and list servers lower-case
        # local-parts (SCP-7). From here on use the id as issued.
        row = self.db.execute("SELECT * FROM tickets WHERE id=? COLLATE NOCASE AND domain=?",
                              (mas["resource_id"], mas["domain"])).fetchone()
        if row is None:
            self._store(None, "quarantined-unknown", event, mas, msg)
            return 202, {"status": "unknown-resource", "action": "quarantined, no reply"}
        tid = row["id"]
        if row["status"] == "closed":
            self._store(tid, "archived-after-close", event, mas, msg)
            notified = False
            if self.closed_notice and msg.get("from") and msg.get("from") == row["requester"]:
                address = f"{resource_local_part(RESOURCE_TYPE, tid)}@{row['domain']}"
                notified = self._reply(tid, address, msg, msg.get("subject") or "Ticket closed",
                                       f"Ticket {tid} is closed. Your message was archived and did "
                                       f"not reopen it. Write to {INTAKE_TARGET}@{row['domain']} "
                                       f"for a new request.")
            return 200, {"status": "closed", "ticket_id": tid, "action": "archived",
                         "mutated": False, "notified": notified}
        self._store(tid, "attached", event, mas, msg)
        return 200, {"status": "correlated", "ticket_id": tid, "facet": mas.get("facet")}

    def _store(self, tid, disposition, event, mas, msg):
        self.db.execute(
            "INSERT INTO messages (ticket_id, disposition, recipient, facet, sender, subject, message_id, body)"
            " VALUES (?,?,?,?,?,?,?,?)",
            (tid, disposition, event.get("recipient"), mas.get("facet"), msg.get("from"),
             msg.get("subject"), msg.get("message_id"), msg.get("text")))

    def _reply(self, tid, mail_from, msg, subject, body) -> bool:
        """True when the reply went out (or was recorded, without SMTP). False for
        automated or unauthenticated senders, and over the cap (then it is held)."""
        return self.mailer.reply(tid, mail_from, msg, subject, body) is None

    def send_cap_reached(self) -> bool:
        return self.mailer.cap_reached()

    # -- MasterStore (mas_modes.email_mode): email as master for tickets -------------

    def open_case(self, response: dict) -> Optional[str]:
        return response.get("ticket_id") if response.get("status") in ("created", "correlated") else None

    def case_address(self, tid: str) -> str:
        row = self.db.execute("SELECT domain FROM tickets WHERE id=?", (tid,)).fetchone()
        return f"{resource_local_part(RESOURCE_TYPE, tid)}@{row['domain']}"

    def record(self, tid: str, kind: str, text: str) -> None:
        """A mode status or update in the ticket thread."""
        with self.lock, self.db:
            self.db.execute("INSERT INTO messages (ticket_id, disposition, body) VALUES (?,?,?)",
                            (tid, kind, text))

    def reply(self, tid: str, msg: dict, subject: str, body: str) -> bool:
        """Mail the participant of `msg` from the ticket address, under principle 10."""
        with self.lock, self.db:
            return self._reply(tid, self.case_address(tid), msg, subject, body)

    # -- admin / inspection ----------------------------------------------------------

    def close(self, tid: str) -> bool:
        with self.lock, self.db:
            return self.db.execute("UPDATE tickets SET status='closed' WHERE id=? COLLATE NOCASE",
                                   (tid,)).rowcount == 1

    def ticket(self, tid: str) -> Optional[dict]:
        row = self.db.execute("SELECT * FROM tickets WHERE id=? COLLATE NOCASE", (tid,)).fetchone()
        if row is None:
            return None
        msgs = self.db.execute("SELECT * FROM messages WHERE ticket_id=? ORDER BY seq",
                               (row["id"],)).fetchall()
        return {**dict(row), "messages": [dict(m) for m in msgs]}

    def outbox(self) -> list[dict]:
        """This service's replies in the shared outbox (`ticket_id` = case ref)."""
        return [{**r, "ticket_id": r["case_ref"]} for r in self.mailer.outbox()]

    def quarantine(self) -> list[dict]:
        return [dict(r) for r in self.db.execute(
            "SELECT * FROM messages WHERE disposition='quarantined-unknown' ORDER BY seq")]


def make_server(service: TicketService, host="127.0.0.1", port=8025) -> ThreadingHTTPServer:
    class Handler(BaseHTTPRequestHandler):
        def do_POST(self):
            body = self.rfile.read(int(self.headers.get("Content-Length", "0")))
            if self.path == "/mas-events":
                try:
                    event = json.loads(body.decode())
                except (UnicodeDecodeError, json.JSONDecodeError):
                    return self._json(400, {"error": "invalid JSON"})
                return self._json(*service.handle(event))
            parts = self.path.strip("/").split("/")
            if len(parts) == 3 and parts[0] == "tickets" and parts[2] == "close":
                return self._json(200 if service.close(parts[1]) else 404, {"ticket_id": parts[1]})
            self._json(404, {"error": "not found"})

        def do_GET(self):
            parts = self.path.strip("/").split("/")
            if len(parts) == 2 and parts[0] == "tickets":
                t = service.ticket(parts[1])
                return self._json(200 if t else 404, t or {"error": "not found"})
            if parts == ["outbox"]:
                return self._json(200, service.outbox())
            self._json(404, {"error": "not found"})

        def _json(self, status, obj):
            out = json.dumps(obj).encode()
            self.send_response(status)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(out)))
            self.end_headers()
            self.wfile.write(out)

        def log_message(self, *a):
            pass

    return ThreadingHTTPServer((host, port), Handler)


if __name__ == "__main__":
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--db", default="tickets.sqlite3")
    ap.add_argument("--listen", default="127.0.0.1:8025")
    ap.add_argument("--smtp", metavar="HOST:PORT", help="send replies via this SMTP relay")
    ap.add_argument("--require-dmarc-pass", action="store_true",
                    help="auto-reply only when the event says the sender passed DMARC")
    ap.add_argument("--max-sent", type=int, metavar="N",
                    help="stop sending once N replies have been sent in total (kept in the outbox)")
    ap.add_argument("--outbox-db", metavar="FILE",
                    help="the shared outbox (default: DB with '-outbox' added before the suffix)")
    ap.add_argument("--email-held-duration-max", default="7d", metavar="DURATION",
                    help="purge mail held over the cap after this long (default 7d)")
    ap.add_argument("--emails-held-storage-max", default="100MB", metavar="SIZE",
                    help="hold no more mail once held mail would exceed this (default 100MB)")
    a = ap.parse_args()
    smtp = (a.smtp.rsplit(":", 1)[0], int(a.smtp.rsplit(":", 1)[1])) if a.smtp else None
    outbox_db = a.outbox_db or (a.db if a.db == ":memory:" else
                                str(Path(a.db).with_name(Path(a.db).stem + "-outbox" + Path(a.db).suffix)))
    mailer = Mailer(outbox_db, smtp, a.max_sent, a.require_dmarc_pass,
                    held_max_age=parse_duration(a.email_held_duration_max),
                    held_max_bytes=parse_size(a.emails_held_storage_max))
    h, p = a.listen.rsplit(":", 1)
    print(f"ticket service on http://{a.listen}/mas-events (outbox {outbox_db})")
    make_server(TicketService(a.db, mailer=mailer), h, int(p)).serve_forever()