Idempotency, unknown outcomes, and task recovery

"""Two local SQLite files model independent caller/provider commits.

Not a real remote service, production queue or end-to-end exactly-once proof.
"""
import json
import sqlite3
import tempfile
from pathlib import Path


def provider(path, operation, payload):
    with sqlite3.connect(path) as db:
        db.execute("CREATE TABLE IF NOT EXISTS effects (operation TEXT PRIMARY KEY, payload TEXT NOT NULL, receipt TEXT NOT NULL)")
        # Serializes the read/check/write in this local demonstration.
        db.execute("BEGIN IMMEDIATE")
        existing = db.execute("SELECT payload,receipt FROM effects WHERE operation=?", (operation,)).fetchone()
        if existing:
            if existing[0] != payload:
                raise ValueError("same operation with different payload")
            return existing[1]
        receipt = "receipt:" + operation
        db.execute("INSERT INTO effects VALUES(?,?,?)", (operation, payload, receipt))
        return receipt


def demo():
    with tempfile.TemporaryDirectory() as folder:
        remote, local = Path(folder) / "provider.db", Path(folder) / "caller.db"
        with sqlite3.connect(local) as db:
            db.execute("CREATE TABLE intents(operation TEXT PRIMARY KEY, payload TEXT, status TEXT, receipt TEXT)")
            db.execute("INSERT INTO intents VALUES('publish-1','report-v1','prepared',NULL)")
        first = provider(remote, "publish-1", "report-v1")
        # Simulated response loss: provider committed, caller did not get receipt.
        with sqlite3.connect(local) as db:
            before = db.execute("SELECT status FROM intents").fetchone()[0]
        second = provider(remote, "publish-1", "report-v1")
        with sqlite3.connect(local) as db:
            db.execute("UPDATE intents SET status='confirmed', receipt=?", (second,))
        with sqlite3.connect(remote) as db:
            effects = db.execute("SELECT COUNT(*) FROM effects").fetchone()[0]
        assert before == "prepared" and first == second and effects == 1
        return dict(caller_before_recovery=before, same_receipt=first == second,
                    provider_effects=effects, caller_after_recovery="confirmed")


if __name__ == "__main__":
    print(json.dumps(demo(), sort_keys=True))