Coverage for src/lexigram/notification/delivery/stores.py: 51%
102 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-26 07:17 +0800
1"""Delivery-state stores for :class:`~lexigram.notification.delivery.retry.RetryingMailer`.
3Two implementations of
4:class:`~lexigram.contracts.notification.delivery.DeliveryStoreProtocol`:
6- :class:`MemoryDeliveryStore` — in-process, tests/dev.
7- :class:`SqlDeliveryStore` — SQL-backed via ``DatabaseProviderProtocol``;
8 the table is created automatically on first use (mirrors
9 ``DatabaseInboxStore``).
10"""
12from __future__ import annotations
14from datetime import UTC, datetime, timedelta
15from typing import Any
16import uuid
18from lexigram.contracts.data.sql.database import DatabaseProviderProtocol
19from lexigram.logging import get_logger
21logger = get_logger(__name__)
23_DEFAULT_TABLE = "notification_delivery_state"
25_CREATE_TABLE_SQL = """
26CREATE TABLE IF NOT EXISTS {table} (
27 delivery_id TEXT PRIMARY KEY,
28 recipient TEXT NOT NULL DEFAULT '',
29 subject TEXT NOT NULL DEFAULT '',
30 body TEXT NOT NULL DEFAULT '',
31 status TEXT NOT NULL DEFAULT 'pending',
32 attempts INTEGER NOT NULL DEFAULT 0,
33 last_error TEXT NOT NULL DEFAULT '',
34 message JSONB NOT NULL DEFAULT '{{}}',
35 created_at TIMESTAMPTZ NOT NULL,
36 retry_after TIMESTAMPTZ
37);
38CREATE INDEX IF NOT EXISTS {table}_status_idx ON {table} (status, retry_after);
39"""
42class MemoryDeliveryStore:
43 """In-process :class:`DeliveryStoreProtocol` for tests/dev."""
45 def __init__(self) -> None:
46 """Initialise empty stores."""
47 self._state: dict[str, dict[str, Any]] = {}
48 self._attempts: dict[str, list[dict[str, Any]]] = {}
50 async def record_attempt(
51 self, delivery_id: str, recipient: str, subject: str, attempt_number: int
52 ) -> None:
53 """Record one attempt row.
55 Args:
56 delivery_id: Delivery operation identifier.
57 recipient: Comma-separated recipients.
58 subject: Message subject.
59 attempt_number: 1-based attempt counter.
60 """
61 self._attempts.setdefault(delivery_id, []).append(
62 {
63 "recipient": recipient,
64 "subject": subject,
65 "attempt_number": attempt_number,
66 "attempted_at": datetime.now(UTC),
67 }
68 )
70 async def create_pending(self, message: Any) -> str:
71 """Persist a pending delivery and return its id.
73 Args:
74 message: Object exposing ``to``, ``subject``, ``body``.
76 Returns:
77 Generated delivery identifier.
78 """
79 delivery_id = str(uuid.uuid4())
80 self._state[delivery_id] = {
81 "delivery_id": delivery_id,
82 "recipient": ",".join(message.to) if getattr(message, "to", None) else "",
83 "subject": message.subject,
84 "body": message.body,
85 "message": {"subject": message.subject, "body": message.body},
86 "status": "pending",
87 "attempts": 0,
88 "last_error": "",
89 "created_at": datetime.now(UTC),
90 "retry_after": None,
91 }
92 return delivery_id
94 async def mark_delivered(self, delivery_id: str) -> None:
95 """Mark a delivery delivered.
97 Args:
98 delivery_id: Identifier of the delivery operation.
99 """
100 entry = self._state.get(delivery_id)
101 if entry is not None:
102 entry["status"] = "delivered"
104 async def get_retry_count(self, delivery_id: str) -> int:
105 """Return the number of attempts made for a delivery.
107 Args:
108 delivery_id: Identifier of the delivery operation.
110 Returns:
111 Current attempt count; 0 when unknown.
112 """
113 entry = self._state.get(delivery_id)
114 return int(entry["attempts"]) if entry else 0
116 async def increment_retry(self, delivery_id: str) -> int:
117 """Increment and return the attempt counter.
119 Args:
120 delivery_id: Identifier of the delivery operation.
122 Returns:
123 New attempt count after incrementing.
124 """
125 entry = self._state.get(delivery_id)
126 if entry is None:
127 return 0
128 entry["attempts"] = int(entry["attempts"]) + 1
129 entry["status"] = "retrying"
130 return int(entry["attempts"])
132 async def schedule_retry(self, delivery_id: str, delay_seconds: float) -> None:
133 """Record when the next retry should run.
135 Args:
136 delivery_id: Identifier of the delivery operation.
137 delay_seconds: Seconds from now.
138 """
139 entry = self._state.get(delivery_id)
140 if entry is not None:
141 entry["retry_after"] = datetime.now(UTC) + timedelta(seconds=delay_seconds)
143 async def mark_failed(
144 self, delivery_id: str, reason: str = "", final: bool = True
145 ) -> None:
146 """Mark a delivery failed.
148 Args:
149 delivery_id: Identifier of the delivery operation.
150 reason: Human-readable failure description.
151 final: Whether retries are exhausted.
152 """
153 entry = self._state.get(delivery_id)
154 if entry is not None:
155 entry["status"] = "failed"
156 entry["last_error"] = reason
158 # -- Worker support (not part of the protocol) ------------------------
160 async def due_deliveries(self, limit: int = 50) -> list[dict[str, Any]]:
161 """Return deliveries whose backoff window has elapsed.
163 Args:
164 limit: Maximum rows to return.
166 Returns:
167 Raw state dicts for due, still-retryable deliveries.
168 """
169 now = datetime.now(UTC)
170 due = [
171 dict(entry)
172 for entry in self._state.values()
173 if entry["status"] in ("pending", "retrying")
174 and int(entry["attempts"]) < 5
175 and (entry["retry_after"] is None or entry["retry_after"] <= now)
176 ]
177 return sorted(due, key=lambda e: e["created_at"])[:limit]
180class SqlDeliveryStore:
181 """SQL-backed :class:`DeliveryStoreProtocol` implementation.
183 Args:
184 db: Database provider injected via DI.
185 table: Table name. Defaults to ``notification_delivery_state``.
186 """
188 def __init__(
189 self,
190 db: DatabaseProviderProtocol,
191 *,
192 table: str = _DEFAULT_TABLE,
193 ) -> None:
194 """Initialise with a database provider and table name.
196 Args:
197 db: Database provider instance.
198 table: Backing table name.
199 """
200 self._db = db
201 self._table = table
203 async def _ensure_table(self) -> None:
204 """Create the backing table/index if absent."""
205 sql = _CREATE_TABLE_SQL.format(table=self._table)
206 for statement in sql.split(";"):
207 if statement.strip():
208 await self._db.execute_query(statement)
210 @staticmethod
211 def _row_to_state(row: Any) -> dict[str, Any]:
212 """Normalise a driver row into a plain state dict."""
213 return {
214 "delivery_id": str(row["delivery_id"]),
215 "recipient": str(row.get("recipient") or ""),
216 "subject": str(row.get("subject") or ""),
217 "body": str(row.get("body") or ""),
218 "message": {
219 "subject": str(row.get("subject") or ""),
220 "body": str(row.get("body") or ""),
221 },
222 "status": str(row.get("status") or "pending"),
223 "attempts": int(row.get("attempts") or 0),
224 "last_error": str(row.get("last_error") or ""),
225 "created_at": row.get("created_at"),
226 "retry_after": row.get("retry_after"),
227 }
229 async def record_attempt(
230 self, delivery_id: str, recipient: str, subject: str, attempt_number: int
231 ) -> None:
232 """Record one attempt row.
234 Args:
235 delivery_id: Delivery operation identifier.
236 recipient: Comma-separated recipients.
237 subject: Message subject.
238 attempt_number: 1-based attempt counter.
239 """
240 await self._db.execute_insert(
241 f"{self._table}_attempts",
242 {
243 "delivery_id": delivery_id,
244 "recipient": recipient,
245 "subject": subject,
246 "attempt_number": attempt_number,
247 "attempted_at": datetime.now(UTC),
248 },
249 )
251 async def create_pending(self, message: Any) -> str:
252 """Persist a pending delivery row and return its id.
254 Args:
255 message: Object exposing ``to``, ``subject``, ``body``.
257 Returns:
258 Generated delivery identifier.
259 """
260 await self._ensure_table()
261 delivery_id = str(uuid.uuid4())
262 await self._db.execute_insert(
263 self._table,
264 {
265 "delivery_id": delivery_id,
266 "recipient": ",".join(message.to)
267 if getattr(message, "to", None)
268 else "",
269 "subject": message.subject,
270 "body": message.body,
271 "status": "pending",
272 "attempts": 0,
273 "last_error": "",
274 "message": {"subject": message.subject, "body": message.body},
275 "created_at": datetime.now(UTC),
276 },
277 )
278 return delivery_id
280 async def mark_delivered(self, delivery_id: str) -> None:
281 """Mark a delivery delivered.
283 Args:
284 delivery_id: Identifier of the delivery operation.
285 """
286 result = await self._db.execute_update(
287 self._table,
288 {"status": "delivered"},
289 "delivery_id = ?",
290 [delivery_id],
291 )
292 if result.affected_rows == 0:
293 logger.warning("delivery_mark_delivered_missing", delivery_id=delivery_id)
295 async def get_retry_count(self, delivery_id: str) -> int:
296 """Return the attempt counter for a delivery.
298 Args:
299 delivery_id: Identifier of the delivery operation.
301 Returns:
302 Current attempt count; 0 when unknown.
303 """
304 await self._ensure_table()
305 result = await self._db.execute_query(
306 f"SELECT attempts FROM {self._table} WHERE delivery_id = ?",
307 [delivery_id],
308 )
309 rows = getattr(result, "rows", [])
310 return int(rows[0]["attempts"]) if rows else 0
312 async def increment_retry(self, delivery_id: str) -> int:
313 """Increment and return the attempt counter.
315 Args:
316 delivery_id: Identifier of the delivery operation.
318 Returns:
319 New attempt count after incrementing.
320 """
321 current = await self.get_retry_count(delivery_id)
322 new_value = current + 1
323 await self._db.execute_update(
324 self._table,
325 {"attempts": new_value, "status": "retrying"},
326 "delivery_id = ?",
327 [delivery_id],
328 )
329 return new_value
331 async def schedule_retry(self, delivery_id: str, delay_seconds: float) -> None:
332 """Record when the next retry should run.
334 Args:
335 delivery_id: Identifier of the delivery operation.
336 delay_seconds: Seconds from now.
337 """
338 retry_after = datetime.now(UTC) + timedelta(seconds=delay_seconds)
339 await self._db.execute_update(
340 self._table,
341 {"retry_after": retry_after},
342 "delivery_id = ?",
343 [delivery_id],
344 )
346 async def mark_failed(
347 self, delivery_id: str, reason: str = "", final: bool = True
348 ) -> None:
349 """Mark a delivery failed.
351 Args:
352 delivery_id: Identifier of the delivery operation.
353 reason: Human-readable failure description.
354 final: Whether retries are exhausted.
355 """
356 await self._db.execute_update(
357 self._table,
358 {"status": "failed", "last_error": reason},
359 "delivery_id = ?",
360 [delivery_id],
361 )
363 async def due_deliveries(self, limit: int = 50) -> list[dict[str, Any]]:
364 """Fetch deliveries whose backoff window has elapsed.
366 Args:
367 limit: Maximum rows to return.
369 Returns:
370 Normalised state dicts including a ``message`` payload.
371 """
372 await self._ensure_table()
373 now = datetime.now(UTC)
374 result = await self._db.execute_query(
375 f"SELECT * FROM {self._table} "
376 "WHERE status IN ('pending', 'retrying') AND attempts < 5 "
377 "AND (retry_after IS NULL OR retry_after <= ?) "
378 "ORDER BY created_at ASC LIMIT ?",
379 [now, limit],
380 )
381 rows = getattr(result, "rows", [])
382 states = []
383 for row in rows:
384 state = self._row_to_state(row)
385 raw_message = state["message"]
386 if isinstance(raw_message, str):
387 from lexigram.serialization import loads
389 try:
390 state["message"] = loads(raw_message)
391 except (TypeError, ValueError):
392 state["message"] = {}
393 states.append(state)
394 return states
397__all__ = ["MemoryDeliveryStore", "SqlDeliveryStore"]