Process mailbox events and attachments

pip install "fastapi[standard]"
export BIRD_MAILBOX_ID="mbx_..."
export BIRD_WEBHOOK_SECRET="whsec_..."

import os
import sqlite3
from contextlib import asynccontextmanager

from bird import Bird, WebhookVerificationError
from fastapi import FastAPI, HTTPException, Request, Response

DB = os.environ.get("MAILBOX_EVENTS_DB", "mailbox-events.sqlite3")
MAILBOX_ID = os.environ["BIRD_MAILBOX_ID"]
os.umask(0o077)


@asynccontextmanager
async def lifespan(app):
    with Bird(webhook_secret=os.environ["BIRD_WEBHOOK_SECRET"]) as bird:
        app.state.bird = bird
        yield


app = FastAPI(lifespan=lifespan)


def database():
    db = sqlite3.connect(DB, timeout=5)
    db.execute("""CREATE TABLE IF NOT EXISTS incoming (
        message_id TEXT PRIMARY KEY, thread_id TEXT NOT NULL,
        event_id TEXT NOT NULL, occurred_at TEXT NOT NULL,
        state TEXT NOT NULL DEFAULT 'pending')""")
    db.commit()
    return db


@app.post("/webhooks/mailboxes")
async def receive(request: Request):
    raw = bytearray()
    async for chunk in request.stream():
        raw.extend(chunk)
        if len(raw) > 2 * 1024 * 1024:
            raise HTTPException(413, "This receiver accepts bodies up to 2 MiB.")
    try:
        event = request.app.state.bird.webhooks.unwrap(bytes(raw), dict(request.headers)).root
    except WebhookVerificationError:
        raise HTTPException(400, "Invalid webhook signature or payload.")
    if event.type != "email_mailbox.message_received":
        return Response(status_code=204)
    data = event.data
    if data.mailbox_id != MAILBOX_ID:
        return Response(status_code=204)
    db = database()
    try:
        with db:
            db.execute("BEGIN IMMEDIATE")
            existing = db.execute("SELECT thread_id FROM incoming WHERE message_id=?", (data.message_id,)).fetchone()
            if existing and existing[0] != data.thread_id:
                raise HTTPException(409, "Message and thread identity disagree.")
            if not existing:
                if db.execute("SELECT count(*) FROM incoming").fetchone()[0] >= 1000:
                    raise HTTPException(503, "Demo queue is full; retain deduplication records while recovering.")
                db.execute("INSERT INTO incoming(message_id, thread_id, event_id, occurred_at) VALUES (?, ?, ?, ?)",
                           (data.message_id, data.thread_id, request.headers["webhook-id"], event.timestamp))
    finally:
        db.close()
    return Response(status_code=204)

import json
from pathlib import Path
from mailbox_app import run
from mailbox_receiver import database


def drain():
    db = database()
    try:
        pending = db.execute("SELECT message_id, thread_id FROM incoming WHERE state='pending' LIMIT 20").fetchall()
        for message_id, thread_id in pending:
            try:
                run("sync", thread_id)
                local = json.loads(Path("mailbox-state.json").read_text())
                job = local["jobs"].get(message_id)
                outcome = "done" if job and job["state"] == "accepted" else "needs_review"
                with db:
                    db.execute("UPDATE incoming SET state=? WHERE message_id=? AND state='pending'", (outcome, message_id))
            except Exception as error:
                print(json.dumps({"message": message_id, "result": "pending", "error": str(error)}))
    finally:
        db.close()


if __name__ == "__main__":
    drain()

python -c 'import sqlite3; db=sqlite3.connect("mailbox-events.sqlite3"); print(db.execute("SELECT message_id, thread_id, state FROM incoming").fetchall()); db.close()'

export BIRD_API_BASE="https://us1.platform.bird.com"
export THREAD_ID="thr_..."
export RECEIVED_MESSAGE_ID="rem_..."
export ATTACHMENT_ID="rea_..."
curl --fail --show-error --silent \
  "$BIRD_API_BASE/v1/email/threads/$THREAD_ID/messages/$RECEIVED_MESSAGE_ID/attachments/$ATTACHMENT_ID" \
  -H "Authorization: Bearer $BIRD_API_KEY" \
  --max-time 30 \
  --max-filesize 1048576 \
  --output attachment.bin