feat(nostr): flag events whose NIP-52 publish didn't land

Inventory reaches clients only through the republished calendar event,
and until now a publish that failed or was skipped left no durable
trace — only a log line, if that. Twice the drift was caught by a human
reading a wrong number on a public page (#35 on aio-demo, #51 on cfaun,
where an event's relay copy sat 14 days behind the DB).

Adds `events.nostr_publish_pending`, set before every attempt and
cleared only on a confirmed success. Ordering it that way is what makes
"the attempt was never made" — no signer resolved, no NostrClient, the
process died mid-flight — as discoverable as "the attempt raised". Both
shapes have now been observed in production; only the second one was
ever visible.

`set_ticket_paid` raises the flag inside its own update so the counters
and "the relay doesn't know about them yet" commit atomically, and the
sale path pays no extra write.

`publish_or_delete_nostr_event` now returns a bool so callers can
branch. The flag, not the return value, is the durable record — the
existing call sites stay correct ignoring it.

Publish failures move from WARNING to ERROR: the published ticket count
has stopped tracking reality, which is not routine journal noise.

Refs #35

migrations_fork.py

@@ -127,3 +127,32 @@ async def m002_ticket_payment_hash(db):
         "UPDATE events.ticket SET payment_hash = id "
         "WHERE payment_hash IS NULL OR payment_hash = ''"
     )
+
+
+async def m003_event_nostr_publish_pending(db):
+    """
+    Add `events.nostr_publish_pending` — the marker that makes NIP-52
+    publish drift queryable instead of invisible.
+
+    Inventory reaches clients only through the republished calendar
+    event. When that publish doesn't land, the relay keeps serving the
+    counts it last saw and nothing anywhere records the divergence; it
+    has twice been caught only by a human reading a public page
+    (aiolabs/events#35, #51).
+
+    The flag is set before each publish attempt and cleared only on a
+    confirmed success, so it covers *both* observed failure shapes:
+    an attempt that raised (a signer outage) and an attempt that was
+    never made at all (no signer resolved, no NostrClient). A periodic
+    sweep republishes whatever is still marked.
+
+    Existing rows default to FALSE rather than TRUE: on upgrade we have
+    no evidence they're stale, and marking the whole table pending would
+    stampede the signer with a full-table republish on first boot.
+    `/republish-all` is the deliberate way to force that.
+    """
+    await _alter_add_column_safe(
+        db,
+        "ALTER TABLE events.events "
+        "ADD COLUMN nostr_publish_pending BOOLEAN NOT NULL DEFAULT FALSE",
+    )

models.py

@@ -125,6 +125,10 @@ class Event(BaseModel):
     status: str = "approved"
     nostr_event_id: str | None = None
     nostr_event_created_at: int | None = None
+    # Set before every publish attempt, cleared on confirmed success.
+    # True means the relay's copy may be behind this row — see
+    # migrations_fork.m003 and the sweep in __init__.events_start.
+    nostr_publish_pending: bool = False
 
     @validator("categories", pre=True)
     def parse_categories(cls, v):

nostr_hooks.py

@@ -12,7 +12,7 @@ from .models import Event
 from .nostr_publisher import publish_event_to_nostr
 
 
-async def publish_or_delete_nostr_event(event: Event, *, delete: bool = False) -> None:
+async def publish_or_delete_nostr_event(event: Event, *, delete: bool = False) -> bool:
     """Publish or delete the NIP-52 calendar event for `event`.
 
     Resolves a `NostrSigner` for the wallet owner — backend-agnostic
@@ -22,7 +22,22 @@ async def publish_or_delete_nostr_event(event: Event, *, delete: bool = False) -
     `await signer.sign_event(...)` for signing. Failures are logged
     and swallowed so a Nostr outage doesn't break the HTTP flow that
     triggered the publish.
+
+    Returns True when the event was signed and handed to the client,
+    False on any skip or failure. Callers are free to ignore it — the
+    `nostr_publish_pending` flag is the durable record, and the sweep
+    retries from that rather than from a return value.
     """
+    # Mark before attempting, clear only on confirmed success. Doing it
+    # in this order is what makes "the attempt was never made" — no
+    # signer, no NostrClient, process died mid-flight — as visible as
+    # "the attempt raised". Cheap guard so a re-publish of an already
+    # pending row doesn't write twice; `set_ticket_paid` sets the flag
+    # inside its own update so the sale path adds no extra write.
+    if not event.nostr_publish_pending:
+        event.nostr_publish_pending = True
+        await update_event(event)
+
     try:
         from lnbits.core.signers import resolve_for_wallet
 
@@ -45,14 +60,22 @@ async def publish_or_delete_nostr_event(event: Event, *, delete: bool = False) -
                 f"[EVENTS] No signer for wallet {event.wallet}, skipping "
                 f"NIP-52 {'delete' if delete else 'publish'} for event {event.id}"
             )
-            return
+            return False
 
         nostr_event = await publish_event_to_nostr(
             nostr_client, event, signer, delete=delete
         )
-        if nostr_event and not delete:
+        if nostr_event is None:
+            return False
+
+        event.nostr_publish_pending = False
+        if not delete:
             event.nostr_event_id = nostr_event.id
             event.nostr_event_created_at = nostr_event.created_at
         await update_event(event)
+        return True
     except Exception as exc:
-        logger.warning(f"[EVENTS] Nostr publish failed: {exc}")
+        # ERROR, not warning: the row stays flagged and its published
+        # counts stay behind until the sweep or a later edit succeeds.
+        logger.error(f"[EVENTS] Nostr publish failed for event {event.id}: {exc}")
+        return False

nostr_publisher.py

@@ -212,5 +212,8 @@ async def publish_event_to_nostr(
         return nostr_event
 
     except Exception as e:
-        logger.warning(f"[EVENTS] Failed to publish to Nostr: {e}")
+        # ERROR, not warning: this is the signer-outage shape of
+        # aiolabs/events#35 — the calendar event never reaches the relay
+        # and the published ticket counts stop tracking the DB.
+        logger.error(f"[EVENTS] Failed to publish event {event.id} to Nostr: {e}")
         return None

services.py

@@ -71,6 +71,10 @@ async def set_ticket_paid(ticket: Ticket) -> Ticket:
         assert event, "Couldn't get event from ticket being paid"
         event.sold += 1
         event.amount_tickets -= 1
+        # Flag inside this same write: the counters and "the relay does
+        # not know about them yet" land atomically, so a crash between
+        # here and the publish still leaves the drift discoverable.
+        event.nostr_publish_pending = True
         await update_event(event)
 
         # Republish the NIP-52 calendar event so connected clients see

tests/test_nostr_publish_pending.py

@@ -0,0 +1,171 @@
+"""The `nostr_publish_pending` marker and its lifecycle.
+
+Inventory reaches clients only through the republished NIP-52 calendar
+event. These tests pin the invariant that makes drift recoverable: the
+flag goes up before every attempt and comes down only on a confirmed
+success, so every shape of failure — raised, skipped, never attempted —
+leaves the row queryable by the sweep.
+"""
+
+from datetime import datetime, timezone
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+
+from .. import nostr_hooks, services
+from ..models import Event, Ticket
+
+
+def _event(**kwargs) -> Event:
+    defaults = {
+        "id": "evt",
+        "wallet": "w",
+        "name": "Test",
+        "info": "",
+        "closing_date": "2030-01-01",
+        "event_start_date": "2030-01-01",
+        "event_end_date": "2030-01-02",
+        "currency": "sat",
+        "price_per_ticket": 1000,
+        "amount_tickets": 10,
+        "time": datetime.now(timezone.utc),
+        "status": "approved",
+    }
+    defaults.update(kwargs)
+    return Event(**defaults)
+
+
+@pytest.fixture
+def saved(monkeypatch):
+    """Capture every update_event write so ordering can be asserted."""
+    writes: list[bool] = []
+
+    async def _update(event):
+        writes.append(event.nostr_publish_pending)
+        return event
+
+    monkeypatch.setattr(nostr_hooks, "update_event", _update)
+    return writes
+
+
+def _signer(monkeypatch, signer):
+    monkeypatch.setattr(
+        "lnbits.core.signers.resolve_for_wallet", AsyncMock(return_value=signer)
+    )
+
+
+def _publisher(monkeypatch, result):
+    monkeypatch.setattr(
+        nostr_hooks, "publish_event_to_nostr", AsyncMock(return_value=result)
+    )
+
+
+@pytest.mark.asyncio
+async def test_success_raises_then_clears_the_flag(monkeypatch, saved):
+    event = _event()
+    _signer(monkeypatch, SimpleNamespace(pubkey="pk"))
+    _publisher(monkeypatch, SimpleNamespace(id="nid", created_at=123))
+
+    assert await nostr_hooks.publish_or_delete_nostr_event(event) is True
+    # Flagged before the attempt, cleared after it — in that order.
+    assert saved == [True, False]
+    assert event.nostr_publish_pending is False
+    assert event.nostr_event_id == "nid"
+    assert event.nostr_event_created_at == 123
+
+
+@pytest.mark.asyncio
+async def test_missing_signer_leaves_the_flag_up(monkeypatch, saved):
+    event = _event()
+    _signer(monkeypatch, None)
+
+    assert await nostr_hooks.publish_or_delete_nostr_event(event) is False
+    assert saved == [True]
+    assert event.nostr_publish_pending is True
+
+
+@pytest.mark.asyncio
+async def test_publisher_returning_none_leaves_the_flag_up(monkeypatch, saved):
+    """The no-NostrClient shape: nothing raised, nothing published."""
+    event = _event()
+    _signer(monkeypatch, SimpleNamespace(pubkey="pk"))
+    _publisher(monkeypatch, None)
+
+    assert await nostr_hooks.publish_or_delete_nostr_event(event) is False
+    assert saved == [True]
+    assert event.nostr_publish_pending is True
+
+
+@pytest.mark.asyncio
+async def test_raised_publish_leaves_the_flag_up(monkeypatch, saved):
+    event = _event()
+    _signer(monkeypatch, SimpleNamespace(pubkey="pk"))
+    monkeypatch.setattr(
+        nostr_hooks,
+        "publish_event_to_nostr",
+        AsyncMock(side_effect=RuntimeError("signer timeout")),
+    )
+
+    assert await nostr_hooks.publish_or_delete_nostr_event(event) is False
+    assert saved == [True]
+    assert event.nostr_publish_pending is True
+
+
+@pytest.mark.asyncio
+async def test_already_pending_row_is_not_re_flagged(monkeypatch, saved):
+    """The sweep re-publishing a flagged row writes once, not twice."""
+    event = _event(nostr_publish_pending=True)
+    _signer(monkeypatch, SimpleNamespace(pubkey="pk"))
+    _publisher(monkeypatch, SimpleNamespace(id="nid", created_at=123))
+
+    assert await nostr_hooks.publish_or_delete_nostr_event(event) is True
+    assert saved == [False]
+
+
+@pytest.mark.asyncio
+async def test_delete_clears_the_flag_without_touching_the_coordinate(
+    monkeypatch, saved
+):
+    """A take-down must not overwrite the id/created_at of the event it
+    just deleted — the kind-5 has its own."""
+    event = _event(nostr_event_id="old", nostr_event_created_at=100)
+    _signer(monkeypatch, SimpleNamespace(pubkey="pk"))
+    _publisher(monkeypatch, SimpleNamespace(id="del", created_at=999))
+
+    assert await nostr_hooks.publish_or_delete_nostr_event(event, delete=True) is True
+    assert event.nostr_publish_pending is False
+    assert event.nostr_event_id == "old"
+    assert event.nostr_event_created_at == 100
+
+
+@pytest.mark.asyncio
+async def test_sale_flags_the_event_in_the_same_write(monkeypatch):
+    """`set_ticket_paid` must flag inside its own update, so the counters
+    and "the relay doesn't know yet" land atomically."""
+    event = _event(sold=4, amount_tickets=6)
+    seen: list[tuple[int, int, bool]] = []
+
+    async def _update_event(ev):
+        seen.append((ev.sold, ev.amount_tickets, ev.nostr_publish_pending))
+        return ev
+
+    monkeypatch.setattr(services, "update_ticket", AsyncMock())
+    monkeypatch.setattr(services, "get_event", AsyncMock(return_value=event))
+    monkeypatch.setattr(services, "update_event", _update_event)
+    monkeypatch.setattr(services, "publish_or_delete_nostr_event", AsyncMock())
+
+    ticket = Ticket(
+        id="t1",
+        wallet="w",
+        event="evt",
+        name="A",
+        email="a@example.com",
+        registered=False,
+        paid=False,
+        time=datetime.now(timezone.utc),
+        reg_timestamp=datetime.now(timezone.utc),
+    )
+    await services.set_ticket_paid(ticket)
+
+    assert seen == [(5, 5, True)]