A Write the Owner Makes

# delta/server.py (trimmed excerpt): the owner's gateway receives a call
REQUEST_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$")   # a write call's own name


def call_request_id(h: Handler) -> str:
    """The caller's own name for one write call, from ``X-Delta-Request-Id``."""
    request_id = h.headers.get("X-Delta-Request-Id") or ""
    if not REQUEST_ID.match(request_id):
        raise BadRequest("a call names itself in X-Delta-Request-Id: the owner keeps its outcome "
                         "by that id, which is what makes asking again safe")
    return request_id


def _require_write(h: Handler, share) -> None:
    if not share.can_write(h.caller, h.app or None):
        raise Forbidden(f"{h.caller} may not write {share.id}" + (f" for {h.app}" if h.app else ""))


def serve_call(h: Handler, tail: str) -> None:
    """Who may is the share's: the calling Mac (from the tailnet roster or its
    ticket) held to ``writers``, the app (``X-Delta-App``) held to ``apps``. The
    child gets the verified Mac, never a name the request carries. The answer is
    the child's JSON, kept nowhere by Delta."""
    share, op = _share_and_op(h, tail)
    _require_write(h, share)
    request_id = call_request_id(h)
    body = h.read_json()
    h.delta.count("calls")
    result = share.call(op, dict(h.query), body, caller=h.caller, app=h.app, request_id=request_id)
    h.send_json(200, result, {"X-Delta-Share": share.id, "X-Delta-Request-Id": request_id})

# tests/test_write_call.py (trimmed excerpt): the smallest owner Delta's suite declares
import json

from delta.errors import Conflict
from delta.shares import DataShare


class LedgerShare(DataShare):
    """An owner with one canonical writer: every call it carries out is one line."""

    def writable(self):
        return True

    def call(self, op, params, body, *, caller, app, request_id):
        done = self.repo_root / ("done-" + self.name) / (request_id + ".json")
        if done.exists():  # the same call again: its first answer, never a second line
            return dict(json.loads(done.read_text()), replayed=True)
        if op == "refuse":
            raise Conflict("the ledger refuses this entry")
        entry = {"op": op, "params": params, "body": body, "caller": caller, "app": app,
                 "request_id": request_id}
        with open(self.repo_root / ("ledger-" + self.name + ".jsonl"), "a") as handle:
            handle.write(json.dumps(entry) + "\n")
        answer = {"ok": True, "entry": entry}
        done.parent.mkdir(exist_ok=True)
        done.write_text(json.dumps(answer))
        return answer

# owner.py: a canonical writer that answers a repeated request id with its first answer
import json
import sqlite3


class Owner:
    """One writer, one ledger; every outcome kept by the caller's request id."""

    def __init__(self, path: str = ":memory:"):
        self.db = sqlite3.connect(path)
        self.db.execute("CREATE TABLE IF NOT EXISTS notes (name TEXT PRIMARY KEY, text TEXT, rev INTEGER)")
        self.db.execute("CREATE TABLE IF NOT EXISTS outcomes (request_id TEXT PRIMARY KEY, answer TEXT)")

    def call(self, op: str, body: dict, *, caller: str, request_id: str) -> dict:
        with self.db:  # one transaction: the write and its outcome land together, or neither
            row = self.db.execute("SELECT answer FROM outcomes WHERE request_id=?", (request_id,)).fetchone()
            if row:
                return dict(json.loads(row[0]), replayed=True)
            if op != "write":
                raise ValueError(f"no operation {op!r}")
            cur = self.db.execute("SELECT rev FROM notes WHERE name=?", (body["name"],)).fetchone()
            rev = cur[0] if cur else 0
            if body.get("expected_rev", rev) != rev:
                answer = {"ok": False, "error": "conflict", "rev": rev}
            else:
                self.db.execute("INSERT OR REPLACE INTO notes VALUES (?,?,?)", (body["name"], body["text"], rev + 1))
                answer = {"ok": True, "rev": rev + 1, "by": caller}
            self.db.execute("INSERT INTO outcomes VALUES (?,?)", (request_id, json.dumps(answer)))
            return answer


owner = Owner()
print(owner.call("write", {"name": "a", "text": "one", "expected_rev": 0}, caller="laptop", request_id="w-1"))
# The answer was lost on the way back; the caller asks again under the same id.
print(owner.call("write", {"name": "a", "text": "one", "expected_rev": 0}, caller="laptop", request_id="w-1"))
# A new request from a stale view is refused by the writer's own rule.
print(owner.call("write", {"name": "a", "text": "two", "expected_rev": 0}, caller="laptop", request_id="w-2"))
print(owner.db.execute("SELECT * FROM notes").fetchall())