diff --git a/.gitignore b/.gitignore index a2fbcfe..fbbe8fc 100644 --- a/.gitignore +++ b/.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 diff --git a/opencode_hawk/dashboard.py b/opencode_hawk/dashboard.py index c92907a..b5ed1f8 100644 --- a/opencode_hawk/dashboard.py +++ b/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: diff --git a/opencode_hawk/notify.py b/opencode_hawk/notify.py index fff36d7..0a6fa89 100644 --- a/opencode_hawk/notify.py +++ b/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 / is a 404), so - each live message (ntfy_channel_state: not yet deleted) gets its own - DELETE //, 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: diff --git a/opencode_hawk/web/app.js b/opencode_hawk/web/app.js index 2119c99..2e1947b 100644 --- a/opencode_hawk/web/app.js +++ b/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; } diff --git a/tests/selftest_dashboard.py b/tests/selftest_dashboard.py index afb6faa..dc8c934 100644 --- a/tests/selftest_dashboard.py +++ b/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() diff --git a/tests/selftest_notify.py b/tests/selftest_notify.py index 7659330..840973d 100644 --- a/tests/selftest_notify.py +++ b/tests/selftest_notify.py @@ -616,14 +616,6 @@ def boom_net(req, timeout=None): check("delete: single message ok", notify.delete_message(cfg, "abc123") is True and calls == [("DELETE", "https://ntfy.sh/t/abc123")]) - calls.clear() - r = notify.delete_all(cfg) - check("delete_all: enumerates then deletes each message id", - r == {"ok": True, "deleted": 2, "total": 2, - "ok_ids": ["ida", "idb"], "failed_ids": []} - and calls == [("GET", "https://ntfy.sh/t/json?poll=1&since=all"), - ("DELETE", "https://ntfy.sh/t/ida"), - ("DELETE", "https://ntfy.sh/t/idb")]) urllib.request.urlopen = boom_404 check("delete: HTTP 404 -> gone is success", notify.delete_message(cfg, "abc123") is True) @@ -633,25 +625,45 @@ def boom_net(req, timeout=None): finally: urllib.request.urlopen = orig - # partial failure mid-clear: only confirmed deletes are counted - def rec_partial(req, timeout=None): + # clear-all returns at once; the server deletes run in the background, a failing one does not stop + # the others, and only live messages inside ntfy's 12 h cache are deleted + import time as _tm + now_s = int(_tm.time()) + tmp_home = Path(tempfile.mkdtemp(prefix="ntfy-clear-")) + orig_home = hawk.home_dir + hawk.home_dir = lambda: tmp_home + deletes = [] + + def rec_bg(req, timeout=None): method = getattr(req, "method", None) or "GET" if method == "DELETE": + deletes.append(req.full_url) if req.full_url.endswith("/idb"): raise urllib.error.HTTPError("u", 500, "boom", {}, None) return FakeResp(json.dumps({"id": "del-x"}), 200) return FakeStreamResp([ - '{"event":"message","id":"ida","message":"A"}', - '{"event":"message","id":"idb","message":"B"}']) + '{"event":"message","id":"ida","message":"A","time":%d}' % (now_s - 60), + '{"event":"message","id":"idb","message":"B","time":%d}' % (now_s - 30), + '{"event":"message","id":"idc","message":"C","time":%d}' % (now_s - 20), + '{"event":"message","id":"old","message":"O","time":%d}' % (now_s - 13 * 3600)]) - urllib.request.urlopen = rec_partial + urllib.request.urlopen = rec_bg try: + t0 = _tm.time() rp = notify.delete_all(cfg) - check("delete_all: partial failure -> confirmed ids only", - rp == {"ok": True, "deleted": 1, "total": 2, "ok_ids": ["ida"], - "failed_ids": ["idb"]}) + quick = _tm.time() - t0 < 2.0 + th = notify._bg["thread"] + if th: + th.join(timeout=10) + check("delete_all: returns at once, hides all live, queues the young ones", + quick and rp == {"ok": True, "hidden": 4, "queued": 3} + and notify.hidden_ids() >= {"ida", "idb", "idc", "old"}) + check("delete_all: background deletes continue past a failure, skip expired", + sorted(deletes) == ["https://ntfy.sh/t/ida", "https://ntfy.sh/t/idb", "https://ntfy.sh/t/idc"]) finally: urllib.request.urlopen = orig + hawk.home_dir = orig_home + shutil.rmtree(tmp_home, ignore_errors=True) # 429 rate limit: waits and retries instead of failing the delete tries = []