outbox.py

"""A local outbox: rows in SQLite that turn into emails when their time comes.

Used by send_later.py for messages you want to stay able to cancel or edit.
Nothing here is specific to one mail provider except the single call in
deliver().
"""

from __future__ import annotations

import os
import re
import sqlite3
import uuid
from datetime import datetime, timedelta, timezone

from infrai import infrai

DB_PATH = os.environ.get("OUTBOX_DB", "outbox.db")

SCHEMA = """
CREATE TABLE IF NOT EXISTS outbox (
    id          TEXT PRIMARY KEY,
    due_at      TEXT NOT NULL,
    recipient   TEXT NOT NULL,
    subject     TEXT NOT NULL,
    html        TEXT NOT NULL,
    state       TEXT NOT NULL DEFAULT 'pending',
    message_id  TEXT,
    attempts    INTEGER NOT NULL DEFAULT 0,
    last_error  TEXT,
    created_at  TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS outbox_due ON outbox (state, due_at);
"""

_OFFSET = re.compile(r"^(\d+)([mhd])$")
_UNITS = {"m": "minutes", "h": "hours", "d": "days"}


def now() -> datetime:
    return datetime.now(timezone.utc)


def parse_when(text, base=None) -> datetime:
    """'45m' / '6h' / '3d' counted from now, or an RFC 3339 timestamp.

    A timestamp without an offset is rejected: 09:00 means two different
    moments depending on where the customer lives.
    """
    base = base or now()
    match = _OFFSET.match(text.strip())
    if match:
        return base + timedelta(**{_UNITS[match.group(2)]: int(match.group(1))})
    stamp = datetime.fromisoformat(text.strip().replace("Z", "+00:00"))
    if stamp.tzinfo is None:
        raise ValueError("timestamps need an offset, e.g. 2026-09-01T09:00:00-07:00")
    return stamp.astimezone(timezone.utc)


def rfc3339(moment: datetime) -> str:
    return moment.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")


def connect(path=DB_PATH) -> sqlite3.Connection:
    conn = sqlite3.connect(path)
    conn.row_factory = sqlite3.Row
    conn.executescript(SCHEMA)
    return conn


def enqueue(conn, recipient, subject, html, due_at) -> str:
    row_id = uuid.uuid4().hex
    conn.execute(
        "INSERT INTO outbox (id, due_at, recipient, subject, html, created_at)"
        " VALUES (?, ?, ?, ?, ?, ?)",
        (row_id, rfc3339(due_at), recipient, subject, html, rfc3339(now())),
    )
    conn.commit()
    return row_id


def cancel(conn, row_id) -> bool:
    cursor = conn.execute(
        "UPDATE outbox SET state = 'canceled' WHERE id = ? AND state = 'pending'",
        (row_id,),
    )
    conn.commit()
    return cursor.rowcount == 1


def pending(conn):
    return conn.execute(
        "SELECT * FROM outbox WHERE state = 'pending' ORDER BY due_at"
    ).fetchall()


def due(conn, moment=None):
    return conn.execute(
        "SELECT * FROM outbox WHERE state = 'pending' AND due_at <= ? ORDER BY due_at",
        (rfc3339(moment or now()),),
    ).fetchall()


def deliver(conn, row) -> str:
    """Send one due row and record the message_id.

    The row id doubles as idempotency_key: if the process dies between the API
    call and the UPDATE below, the retry resolves to the same message rather
    than a second copy in someone's inbox.
    """
    payload = {
        "to": row["recipient"],
        "subject": row["subject"],
        "html": row["html"],
        "idempotency_key": row["id"],
    }
    sender = os.environ.get("INFRAI_EMAIL_FROM")
    if sender:
        payload["from"] = sender
    data = infrai.email.send(**payload)
    message_id = data.get("message_id", "")
    conn.execute(
        "UPDATE outbox SET state = 'sent', message_id = ?, attempts = attempts + 1"
        " WHERE id = ?",
        (message_id, row["id"]),
    )
    conn.commit()
    return message_id


def run_due(conn, moment=None):
    """One tick: send everything whose due_at has passed. Returns (id, result)."""
    results = []
    for row in due(conn, moment):
        try:
            results.append((row["id"], deliver(conn, row)))
        except Exception as exc:
            # Keep the tick alive; the row stays pending and is retried later.
            conn.execute(
                "UPDATE outbox SET attempts = attempts + 1, last_error = ? WHERE id = ?",
                (str(exc), row["id"]),
            )
            conn.commit()
            results.append((row["id"], "error: {0}".format(exc)))
    return results