34 lines
1.7 KiB
Python
34 lines
1.7 KiB
Python
"""Shared claim and completion rules for the two durable email queues."""
|
|
|
|
import uuid
|
|
|
|
|
|
def queue_table(table: str) -> str:
|
|
if table not in {"email_recap_deliveries", "newsletter_deliveries"}:
|
|
raise ValueError("Unknown email queue")
|
|
return table
|
|
|
|
|
|
def claim(conn, table: str, now: float) -> dict | None:
|
|
table = queue_table(table)
|
|
conn.execute(f"""UPDATE {table} SET state='unknown', detail='Delivery interrupted after sending began; check the mail server.', updated_at=?
|
|
WHERE state='sending' AND lease_until<?""", (now, now))
|
|
conn.execute(f"""UPDATE {table} SET state=CASE WHEN attempts>=3 THEN 'failed' ELSE 'retry' END,
|
|
next_attempt_at=?, updated_at=?, detail='Email preparation interrupted.'
|
|
WHERE state='preparing' AND lease_until<?""", (now, now, now))
|
|
row = conn.execute(f"""SELECT * FROM {table} WHERE state IN ('queued', 'retry') AND next_attempt_at<=?
|
|
ORDER BY created_at, id LIMIT 1""", (now,)).fetchone()
|
|
if not row:
|
|
return None
|
|
claim_id = uuid.uuid4().hex
|
|
conn.execute(f"""UPDATE {table} SET state='preparing', claim=?, lease_until=?,
|
|
attempts=attempts+1, updated_at=? WHERE id=?""", (claim_id, now + 1800, now, row["id"]))
|
|
return dict(conn.execute(f"SELECT * FROM {table} WHERE id=?", (row["id"],)).fetchone())
|
|
|
|
|
|
def finish(conn, table: str, delivery: dict, state: str, detail: str, now: float, delay: int = 0):
|
|
table = queue_table(table)
|
|
conn.execute(f"""UPDATE {table} SET state=?, detail=?, updated_at=?, next_attempt_at=?, lease_until=NULL
|
|
WHERE id=? AND claim=? AND state IN ('preparing', 'sending')""",
|
|
(state, detail, now, now + delay, delivery["id"], delivery["claim"]))
|