ntfy clear-all is instant; server deletes in the background

.gitignore

@@ -42,3 +42,6 @@ dist/
 
 # Single-instance locks (monitor, dashboard, llama-start)
 *.lock
+
+# Dashboard messages cleared but not yet expired on ntfy
+ntfy_hidden.json

opencode_hawk/dashboard.py

@@ -1061,7 +1061,9 @@ def ntfy_payload() -> dict:
         "topic": n["topic"],
         "server": n["server"],
         "log": notify.load_notify_log(),
-        "channel": _ntfy_channel_cache["v"],
+        # Cleared in the dashboard but possibly not yet deleted on the server: stays hidden (12 h).
+        "channel": [m for m in _ntfy_channel_cache["v"]
+                    if m.get("seq", m.get("id")) not in notify.hidden_ids()],
     }
 
 
@@ -1088,25 +1090,13 @@ def ntfy_delete(msg_id: str) -> dict:
 
 
 def ntfy_delete_all() -> dict:
-    """POST /api/ntfy/delete-all handler body: send a DELETE tombstone for
-    every message cached on the topic (enumerated from the server cache PLUS
-    local-log ids; expired/unknown ids 404 -> treated as gone), then drop
-    the matching local rows. Rows without an id (received replies) are
-    cleared too; ids whose request failed keep their local row.
-    NOTE (verified against ntfy.sh): a delete is a tombstone — it dismisses
-    the notification on connected devices but the message STAYS in ntfy's
-    pollable history until its cache TTL (12h) expires, so /api/ntfy/cache-
-    ids will keep listing ids that were "deleted"."""
+    """POST /api/ntfy/delete-all: instant, like the ntfy app's "clear all". The list (local log and channel
+    view) is cleared at once; the server deletes run in the background (notify.delete_all)."""
     from . import notify
     r = notify.delete_all(coordinator.load_config())
-    # Everything goes except rows whose live message failed to delete (they
-    # keep their delete button for a retry).
-    failed = set(r.get("failed_ids") or [])
-    notify.remove_log_entries(lambda e: e.get("id") not in failed)
-    _ntfy_channel_cache["v"] = [m for m in _ntfy_channel_cache["v"]
-                                if m.get("seq", m.get("id")) in failed]
-    return {"ok": True, "deleted": r.get("deleted", 0),
-            "total": r.get("total", 0)}
+    notify.remove_log_entries(lambda e: True)
+    _ntfy_channel_cache["v"] = []
+    return {"ok": True, "cleared": r.get("hidden", 0), "server_pending": r.get("queued", 0)}
 
 
 def ntfy_cache_ids() -> dict:

opencode_hawk/notify.py

@@ -26,6 +26,7 @@
 import time as _time
 import urllib.error
 import urllib.request
+import threading
 import uuid
 from urllib.parse import quote
 
@@ -487,17 +488,86 @@ def ntfy_channel_state(cfg, events=None) -> list:
     return out
 
 
+NTFY_CACHE_S = 12 * 3600  # ntfy.sh keeps a message this long; older ones are gone without a delete
+HIDDEN_FN = "ntfy_hidden.json"
+_bg = {"queue": [], "thread": None}
+_bg_lock = threading.Lock()
+
+
+def hidden_ids() -> set:
+    """Channel messages cleared in the dashboard whose server delete may still be pending: hidden from the
+    list until ntfy drops them (12 h)."""
+    data = _hawk.load_json(_hawk.home_dir() / HIDDEN_FN, {}) or {}
+    now = _time.time()
+    return {k for k, t in data.items() if now - float(t or 0) < NTFY_CACHE_S}
+
+
+def _hide(seqs) -> None:
+    path = _hawk.home_dir() / HIDDEN_FN
+    data = _hawk.load_json(path, {}) or {}
+    now = _time.time()
+    data = {k: t for k, t in data.items() if now - float(t or 0) < NTFY_CACHE_S}
+    for s_ in seqs:
+        data[str(s_)] = now
+    _hawk.save_json(path, data)
+
+
+def delete_once(cfg, msg_id: str):
+    """One DELETE without waiting: ("ok"|"gone"|"limited"|"error", seconds to wait before retrying)."""
+    n = _ntfy_cfg(cfg)
+    url = "%s/%s/%s" % (n["server"], n["topic"], quote(str(msg_id), safe=""))
+    try:
+        with urllib.request.urlopen(urllib.request.Request(url, method="DELETE"), timeout=15) as resp:
+            return ("ok" if 200 <= (resp.status or 200) < 300 else "error"), 0
+    except urllib.error.HTTPError as e:
+        if e.code == 429:
+            try:
+                wait = float(e.headers.get("Retry-After") or 5)
+            except (TypeError, ValueError):
+                wait = 5.0
+            return "limited", max(1.0, wait)
+        return ("gone" if e.code == 404 else "error"), 0
+    except Exception:
+        return "error", 0
+
+
+def _delete_worker(cfg) -> None:
+    """Server deletes in the background: as fast as ntfy allows, waiting only when it says so (429).
+    Nothing waits for this; whatever it does not finish, ntfy drops after 12 h by itself."""
+    done = failed = 0
+    while True:
+        with _bg_lock:
+            if not _bg["queue"]:
+                _bg["thread"] = None
+                break
+            mid, until = _bg["queue"][0]
+        if _time.time() > until:  # expired on the server by now: nothing to delete
+            with _bg_lock:
+                _bg["queue"].pop(0)
+            continue
+        state, wait = delete_once(cfg, mid)
+        if state == "limited":
+            _time.sleep(wait)
+            continue
+        with _bg_lock:
+            _bg["queue"].pop(0)
+        if state in ("ok", "gone"):
+            done += 1
+        else:
+            failed += 1
+        _time.sleep(0.1)
+    log("notify: background clear finished: %d deleted on the server, %d failed (ntfy drops those within 12 h)"
+        % (done, failed))
+
+
 def delete_all(cfg) -> dict:
-    """Delete every message an ntfy client still shows for the topic.
-
-    ntfy.sh has no topic-wide delete endpoint (DELETE /<topic> is a 404), so
-    each live message (ntfy_channel_state: not yet deleted) gets its own
-    DELETE /<topic>/<sequence_id>, paced at 0.1s. Local log ids are NOT
-    sent: ntfy answers 200 for unknown ids and appends a tombstone for each,
-    so they only bloat the topic. Returns {"ok", "deleted", "total",
-    "ok_ids"}; raises ValueError when no topic is configured and re-raises
-    any enumeration failure so callers can surface it.
-    """
+    """Clear every message an ntfy client still shows, without making anyone wait.
+
+    The ntfy app's own "clear all" only clears the phone, which is why it is instant. Deleting on the server
+    takes one DELETE per message (ntfy.sh has no topic-wide delete) and is rate-limited, so: every live message
+    is hidden from the dashboard at once, and the server deletes run in a background thread, only for messages
+    still inside ntfy's 12 h cache. Returns {"ok", "hidden", "queued"}; raises ValueError without a topic and
+    re-raises an enumeration failure."""
     n = _ntfy_cfg(cfg)
     if not n["topic"]:
         raise ValueError("ntfy not configured")
@@ -506,28 +576,19 @@ def delete_all(cfg) -> dict:
     except Exception as e:
         log("notify: ntfy cache enumeration failed: %s" % e)
         raise
-    # Only messages a client still shows; already-deleted ones would just
-    # pile more tombstones onto the topic.
-    seen = set()
-    ids = []
-    for m in live:
-        i = m["seq"]
-        if i not in seen:
-            seen.add(i)
-            ids.append(i)
-    deleted = 0
-    ok_ids = []
-    for i, mid in enumerate(ids):
-        if delete_message(cfg, mid):
-            deleted += 1
-            ok_ids.append(mid)
-        if i + 1 < len(ids):
-            _time.sleep(0.1)  # polite pacing between deletes (spec §3)
-    log("notify: ntfy clear-all deleted %d of %d id(s) "
-        "(live channel messages)" % (deleted, len(ids)))
-    return {"ok": True, "deleted": deleted, "total": len(ids),
-            "ok_ids": ok_ids,
-            "failed_ids": [i for i in ids if i not in set(ok_ids)]}
+    now = _time.time()
+    seqs = list(dict.fromkeys(m["seq"] for m in live))
+    _hide(seqs)
+    young = [(m["seq"], m["ts"] / 1000.0 + NTFY_CACHE_S) for m in live if now - m["ts"] / 1000.0 < NTFY_CACHE_S]
+    with _bg_lock:
+        queued = {q for q, _ in _bg["queue"]}
+        _bg["queue"].extend(item for item in dict(young).items() if item[0] not in queued)
+        start = _bg["thread"] is None and bool(_bg["queue"])
+        if start:
+            _bg["thread"] = threading.Thread(target=_delete_worker, args=(cfg,), daemon=True, name="ntfy-clear")
+            _bg["thread"].start()
+    log("notify: clear-all hid %d message(s); %d server delete(s) run in the background" % (len(seqs), len(young)))
+    return {"ok": True, "hidden": len(seqs), "queued": len(young)}
 
 
 def backfill_bodies(cfg) -> dict:

opencode_hawk/web/app.js

@@ -721,8 +721,6 @@ const NOTIFY_TAG_KINDS = {
 };
 // ntfy clear-all pacing: one DELETE per message (ntfy.sh rate-limits
 // bursts); failed ones get one slower retry pass.
-const NTFY_CLEAR_SPACING_MS = 2500;
-const NTFY_CLEAR_RETRY_SPACING_MS = 6000;
 
 let ntfyCache = { log: [] };
 
@@ -1015,50 +1013,21 @@ async function deleteNtfyMessage(id, btn) {
 }
 
 async function clearAllNtfy() {
-  let ids = (ntfyCache.channel || []).map((m) => String(m.seq || m.id));
-  const total = ids.length;
+  const total = (ntfyCache.channel || []).length;
   const topic = (ntfyCache && ntfyCache.topic) || "";
-  if (!confirm("Delete all " + total + " message(s) on ntfy channel '"
-      + topic + "'?\n\nThey disappear from the ntfy app on every device and "
-      + "from this list.")) return;
+  if (!confirm("Clear all " + total + " message(s) on ntfy channel '" + topic + "'?\n\n"
+      + "The list clears at once; the messages are deleted on the server in the background "
+      + "(they disappear from the ntfy app on every device).")) return;
   const btn = $("btn-clear-all");
   if (btn) btn.disabled = true;
-  let done = 0;
-  let failed = [];
   try {
-    // One delete per message (ntfy has no topic-wide delete); each row
-    // disappears as soon as its delete lands, so progress is visible.
-    // Failures get one slower retry pass.
-    for (let pass = 0; pass < 2 && ids.length; pass++) {
-      failed = [];
-      for (const id of ids) {
-        try {
-          await fetchJSON("/api/ntfy/delete", {
-            method: "POST",
-            headers: { "Content-Type": "application/json" },
-            body: JSON.stringify({ id }),
-          });
-          done++;
-          dropNtfyRow(id);
-        } catch (e) { failed.push(id); }
-        devMsg("deleting… " + done + " / " + total
-            + (failed.length ? " (" + failed.length + " failed)" : ""));
-        await sleep(pass ? NTFY_CLEAR_RETRY_SPACING_MS : NTFY_CLEAR_SPACING_MS);
-      }
-      ids = failed;
-    }
-    try {
-      await fetchJSON("/api/ntfy/prune-local", {
-        method: "POST",
-        headers: { "Content-Type": "application/json" },
-        body: JSON.stringify({ all: true }),
-      });
-    } catch (e) { /* not fatal */ }
+    const r = await fetchJSON("/api/ntfy/delete-all", { method: "POST" });
     ntfyCache.lastKey = "";
     await loadNtfy();
-    devMsg(failed.length
-        ? "deleted " + done + " of " + total + "; " + failed.length + " failed — try again"
-        : "deleted " + done + " message(s) — channel is empty");
+    devMsg("cleared " + (r.cleared || 0) + " message(s)"
+        + (r.server_pending ? "; " + r.server_pending + " server delete(s) running in the background" : ""));
+  } catch (e) {
+    devMsg("clear failed: " + e.message);
   } finally {
     if (btn) btn.disabled = false;
   }

tests/selftest_dashboard.py

@@ -1471,7 +1471,6 @@ def seed_part8(subagent_type):
     orig_hm = coordinator.home_dir
     orig_cfg = coordinator.load_config
     orig_del = _n.delete_message
-    orig_delall = _n.delete_all
     tmp2 = Path(tempfile.mkdtemp(prefix="ntfy-del-test-"))
     coordinator.home_dir = lambda: tmp2
     coordinator.load_config = lambda: {"notify": {"ntfy":
@@ -1502,49 +1501,40 @@ def seed_part8(subagent_type):
             check("ntfy_delete_engine_fail", False)
         except RuntimeError:
             check("ntfy_delete_engine_fail", True)
-        _n.delete_message = lambda cfg, mid: True
-        _n.delete_all = lambda cfg: {"ok": True, "deleted": 2, "total": 2,
-                                     "ok_ids": ["a1", "a2"]}
-        r2 = ntfy_delete_all()
-        check("ntfy_delete_all_local_sync",
-              r2 == {"ok": True, "deleted": 2, "total": 2}
-              and _n.load_notify_log() == [])
-        # partial remote wipe: only confirmed-deleted ids prune local rows,
-        # unconfirmed ids keep their local row (and their delete button)
-        _n.append_notify_log({"dir": "sent", "id": "a3", "title": "partial",
-                              "ts": 5, "topic": "t"})
-        _n.append_notify_log({"dir": "sent", "id": "a4", "title": "unconfirmed",
-                              "ts": 6, "topic": "t"})
-        _n.delete_all = lambda cfg: {"ok": True, "deleted": 1, "total": 2,
-                                     "ok_ids": ["a3"], "failed_ids": ["a4"]}
-        r3 = ntfy_delete_all()
-        check("ntfy_delete_all_partial_sync",
-              r3 == {"ok": True, "deleted": 1, "total": 2}
-              and [e.get("id") for e in _n.load_notify_log()] == ["a4"])
-        # clear-all deletes only messages still live on the channel (not the
-        # tombstoned ones, not local-only ids) and clears the local list.
-        _n.append_notify_log({"dir": "sent", "id": "exp1", "title": "expired",
-                              "ts": 6, "topic": "t"})
-        _n.append_notify_log({"dir": "received", "id": "rr1", "kind": "reply",
-                              "ts": 7})
-        _n.delete_all = orig_delall
-        _n.delete_message = lambda cfg, mid: True
+        # clear-all is instant (like the ntfy app): the list empties at once, the cleared messages stay
+        # hidden, and the server deletes run in the background for live messages younger than 12 h only;
+        # a 429 makes the worker wait and retry instead of blocking anyone.
+        import time as _tm
+        now_s = _tm.time()
         orig_rh = _n._ntfy_raw_history
+        orig_once = _n.delete_once
         _n._ntfy_raw_history = lambda cfg: [
-            {"event": "message", "id": "c1", "message": "x", "time": 1},
-            {"event": "message", "id": "c2", "message": "y", "time": 2},
-            {"event": "message_delete", "id": "t1", "sequence_id": "c2"}]
+            {"event": "message", "id": "c1", "message": "x", "time": now_s - 60},
+            {"event": "message", "id": "c2", "message": "y", "time": now_s - 30},
+            {"event": "message_delete", "id": "t1", "sequence_id": "c2"},
+            {"event": "message", "id": "c3", "message": "old", "time": now_s - 13 * 3600}]
+        calls = []
+        def _once(cfg, mid):
+            calls.append(mid)
+            return ("limited", 0.01) if calls.count(mid) == 1 else ("ok", 0)
+        _n.delete_once = _once
         try:
-            r4 = _n.delete_all(coordinator.load_config())
-            check("ntfy_clearall_live_only",
-                  r4["total"] == 1 and r4["ok_ids"] == ["c1"]
-                  and r4["failed_ids"] == [])
+            t0 = _tm.time()
             r5 = ntfy_delete_all()
-            check("ntfy_clearall_clears_list",
-                  r5 == {"ok": True, "deleted": 1, "total": 1}
+            instant = _tm.time() - t0 < 1.0
+            check("ntfy_clearall_instant",
+                  instant and r5 == {"ok": True, "cleared": 2, "server_pending": 1}
                   and _n.load_notify_log() == [])
+            th = _n._bg["thread"]
+            if th:
+                th.join(timeout=10)
+            check("ntfy_clearall_background_live_young_only", calls == ["c1", "c1"])
+            _ntfy_channel_cache.update(t=_tm.time(), v=_n.ntfy_channel_state(coordinator.load_config()))
+            check("ntfy_clearall_hidden_until_expired",
+                  [m["seq"] for m in ntfy_payload()["channel"]] == [])
         finally:
             _n._ntfy_raw_history = orig_rh
+            _n.delete_once = orig_once
         # legacy backfill: sent entries without body recover their full body
         # from the ntfy cache by publish id; dlt- fallbacks, already-complete
         # and received rows stay untouched.
@@ -1635,7 +1625,6 @@ def seed_part8(subagent_type):
         coordinator.home_dir = orig_hm
         coordinator.load_config = orig_cfg
         _n.delete_message = orig_del
-        _n.delete_all = orig_delall
         shutil.rmtree(tmp2, ignore_errors=True)
 
     print()