feat: implement Melt SAGA pattern for atomic melt operations

cashu/core/base.py

@@ -285,6 +285,38 @@ def __str__(self):
         return self.name
 
 
+class MeltSagaState(Enum):
+    initial = "Initial"
+    setup_complete = "SetupComplete"
+    payment_attempted = "PaymentAttempted"
+    payment_confirmed = "PaymentConfirmed"
+
+    def __str__(self):
+        return self.name
+
+
+class Saga(BaseModel):
+    operation_id: str
+    state: MeltSagaState
+    data: str  # JSON encoded data
+    created_at: int
+
+    @classmethod
+    def from_row(cls, row: Row):
+        try:
+            created_at = int(row["created_at"]) if row["created_at"] else 0
+        except Exception:
+            created_at = (
+                int(row["created_at"].timestamp()) if row["created_at"] else 0
+            )
+        return cls(
+            operation_id=row["operation_id"],
+            state=MeltSagaState(row["state"]),
+            data=row["data"],
+            created_at=created_at,
+        )
+
+
 class MeltQuote(LedgerEvent):
     quote: str
     method: str

cashu/mint/crud.py

@@ -14,6 +14,7 @@
     MintQuote,
     Proof,
     Unit,
+    Saga,
 )
 from ..core.db import (
     Connection,
@@ -290,6 +291,41 @@ async def get_melt_quotes_by_checking_id(
         conn: Optional[Connection] = None,
     ) -> List[MeltQuote]: ...
 
+    @abstractmethod
+    async def store_saga_state(
+        self,
+        *,
+        db: Database,
+        saga: "Saga",
+        conn: Optional[Connection] = None,
+    ) -> None: ...
+
+    @abstractmethod
+    async def get_saga_state(
+        self,
+        *,
+        db: Database,
+        operation_id: str,
+        conn: Optional[Connection] = None,
+    ) -> Optional["Saga"]: ...
+
+    @abstractmethod
+    async def delete_saga_state(
+        self,
+        *,
+        db: Database,
+        operation_id: str,
+        conn: Optional[Connection] = None,
+    ) -> None: ...
+
+    @abstractmethod
+    async def get_incomplete_sagas(
+        self,
+        *,
+        db: Database,
+        conn: Optional[Connection] = None,
+    ) -> List["Saga"]: ...
+
     @abstractmethod
     async def get_melt_quote_by_request(
         self,
@@ -1054,3 +1090,70 @@ async def get_melt_quotes_by_checking_id(
             {"checking_id": checking_id},
         )
         return [MeltQuote.from_row(row) for row in results]  # type: ignore
+
+    async def store_saga_state(
+        self,
+        *,
+        db: Database,
+        saga: Saga,
+        conn: Optional[Connection] = None,
+    ) -> None:
+        await (conn or db).execute(
+            f"""
+            INSERT INTO {db.table_with_schema('saga_state')}
+            (operation_id, state, data, created_at)
+            VALUES (:operation_id, :state, :data, :created_at)
+            ON CONFLICT(operation_id) DO UPDATE SET
+            state = :state, data = :data
+            """,
+            {
+                "operation_id": saga.operation_id,
+                "state": saga.state.value,
+                "data": saga.data,
+                "created_at": db.to_timestamp(db.timestamp_now_str()),
+            },
+        )
+
+    async def get_saga_state(
+        self,
+        *,
+        db: Database,
+        operation_id: str,
+        conn: Optional[Connection] = None,
+    ) -> Optional[Saga]:
+        row = await (conn or db).fetchone(
+            f"""
+            SELECT * FROM {db.table_with_schema('saga_state')}
+            WHERE operation_id = :operation_id
+            """,
+            {"operation_id": operation_id},
+        )
+        return Saga.from_row(row) if row else None  # type: ignore
+
+    async def delete_saga_state(
+        self,
+        *,
+        db: Database,
+        operation_id: str,
+        conn: Optional[Connection] = None,
+    ) -> None:
+        await (conn or db).execute(
+            f"""
+            DELETE FROM {db.table_with_schema('saga_state')}
+            WHERE operation_id = :operation_id
+            """,
+            {"operation_id": operation_id},
+        )
+
+    async def get_incomplete_sagas(
+        self,
+        *,
+        db: Database,
+        conn: Optional[Connection] = None,
+    ) -> List[Saga]:
+        rows = await (conn or db).fetchall(
+            f"""
+            SELECT * FROM {db.table_with_schema('saga_state')}
+            """
+        )
+        return [Saga.from_row(r) for r in rows]  # type: ignore