#!/usr/bin/env python3 """ End-to-end diagnostic for the push-notification chain, for ONE user. The chain has four independent links, and a break in any of them looks the same from the phone: nothing arrives. Worse, a flush reports `pushed=1` even when OneSignal accepted the call but matched no device, so "sent" is not proof of "delivered". This script walks the links in order and names the one that breaks. 1. Hydrogen : does the user exist? GET /admin/users/{hex} 2. OneSignal : is a device aliased to them? GET /apps/{id}/users/by/external_id/{hex} 3. Hydrogen : can a notification be queued? POST /admin/notifications 4. Hydrogen : does the flush push it? POST /admin/jobs/flush/notifications Steps 1-2 are READ-ONLY. Steps 3-4 write (they queue and send a real push), so they only run with --send. Config is read from the Hydrogen `.env` (APP_URL, ADMIN_API_TOKEN, and for step 2 ONESIGNAL_APP_ID / ONESIGNAL_REST_API_KEY); every value can be overridden by a real environment variable of the same name. Dependencies: pip install requests Usage: python notifications_doctor.py --env .envpp python notifications_doctor.py --env .envpp --send """ from __future__ import annotations import argparse import os import re import sys from pathlib import Path try: import requests except ImportError as exc: # pragma: no cover sys.stderr.write(f"Missing dependency: {exc}. Run: pip install requests\n") sys.exit(2) OK, KO, WARN, INFO = " OK ", " FAIL ", " WARN ", " .. " def load_env(env_path: Path) -> dict[str, str]: values: dict[str, str] = {} if not env_path.is_file(): return values for line in env_path.read_text(encoding="utf-8").splitlines(): line = line.strip() if line == "" or line.startswith("#"): continue if line.startswith("export "): line = line[len("export "):] if "=" not in line: continue key, _, val = line.partition("=") values[key.strip()] = val.strip().strip('"').strip("'") return values def say(tag: str, msg: str) -> None: print(f"[{tag}] {msg}", flush=True) def main() -> int: ap = argparse.ArgumentParser(description="Diagnose the push chain for one user.") ap.add_argument("user_hex", help="User id, hex 32 lowercase (no dashes).") ap.add_argument("--env", default=str(Path(__file__).resolve().parent.parent / ".env")) ap.add_argument("--send", action="store_true", help="Also queue a test notification and flush it (writes!).") args = ap.parse_args() env = load_env(Path(args.env)) def cfg(key: str, default: str = "") -> str: return os.environ.get(key, env.get(key, default)) user_hex = args.user_hex.strip().lower() if not re.fullmatch(r"[0-9a-f]{32}", user_hex): say(KO, f"'{args.user_hex}' is not a 32-char lowercase hex id " f"(strip the dashes from the UUID).") return 2 api = cfg("APP_URL", "").rstrip("/") token = cfg("ADMIN_API_TOKEN") os_app = cfg("ONESIGNAL_APP_ID") os_key = cfg("ONESIGNAL_REST_API_KEY") if not api or not token: say(KO, "APP_URL and ADMIN_API_TOKEN are required.") return 2 admin = {"Authorization": f"Bearer {token}", "Accept": "application/json"} print(f"\nuser : {user_hex}\ntarget : {api}\n") verdicts: list[str] = [] # -- 1. the user exists in Hydrogen ------------------------------------- r = requests.get(f"{api}/admin/users/{user_hex}", headers=admin, timeout=30) if r.status_code == 200: say(OK, "1/4 Hydrogen knows this user.") elif r.status_code == 404: say(KO, "1/4 Hydrogen does NOT know this user — wrong id, or wrong environment.") verdicts.append("The id does not exist in this database: check you are targeting " "the right environment, or take the id from GET /admin/users/search.") print() for v in verdicts: print(f"=> {v}") return 1 else: say(WARN, f"1/4 unexpected status {r.status_code}: {r.text[:200]}") # -- 2. a device is aliased to them in OneSignal ------------------------ if not os_app or not os_key: say(WARN, "2/4 skipped — ONESIGNAL_APP_ID / ONESIGNAL_REST_API_KEY missing from the env.") verdicts.append("Without the OneSignal credentials this script cannot tell whether a " "device is linked; the server needs them too, to push at all.") else: r = requests.get( f"https://onesignal.com/api/v1/apps/{os_app}/users/by/external_id/{user_hex}", headers={"Authorization": f"Basic {os_key}"}, timeout=30, ) body = r.text if r.status_code == 200 and '"identity"' in body and "errors" not in body: subs = body.count('"subscription_id"') say(OK, f"2/4 OneSignal has an alias for this user ({subs} subscription(s)).") if subs == 0: say(WARN, " ...but no subscription attached: the device is aliased yet not " "push-subscribed (permission revoked, or a web-only record).") verdicts.append("The alias exists but carries no push subscription — check the " "notification permission on the device.") else: say(KO, "2/4 OneSignal has NO device aliased to this user.") say(INFO, f" response: {body[:200]}") verdicts.append("THIS IS THE BREAK: the phone is not linked to this account. The app " "must call OneSignal.login(\"" + user_hex + "\") after sign-in " "(hex 32, lowercase, no dashes). Signing out and back in on this " "environment normally fixes it. Also confirm the app uses the SAME " "ONESIGNAL_APP_ID as the server, and points at this environment.") if not args.send: print() if verdicts: for v in verdicts: print(f"=> {v}") else: say(OK, "Read-only checks pass. Re-run with --send to queue and push a test.") return 1 if verdicts else 0 # -- 3. queue a notification ------------------------------------------- r = requests.post( f"{api}/admin/notifications", headers={**admin, "Content-Type": "application/json"}, json={"userIds": [user_hex], "translations": {"fr-FR": {"title": "Test push", "body": "Diagnostic Hydrogen"}}}, timeout=60, ) payload = r.json() if r.headers.get("content-type", "").startswith("application/json") else {} if r.status_code == 200 and payload.get("dispatched") == 1: say(OK, "3/4 notification queued.") elif payload.get("skipped"): say(KO, "3/4 skipped — this user disabled the in-app channel in their preferences.") verdicts.append("Re-enable the in-app notification preference for this user.") else: say(KO, f"3/4 could not queue: {r.status_code} {r.text[:300]}") verdicts.append("The queue write failed — the error above is the real cause.") # -- 4. flush ------------------------------------------------------------ r = requests.post(f"{api}/admin/jobs/flush/notifications", headers=admin, timeout=300) if r.status_code == 200: summary = r.json().get("summary", {}) counts = {k: v for k, v in summary.items() if k != "errors"} say(OK, f"4/4 flush: {counts}") for reason, count in (summary.get("errors") or {}).items(): say(KO, f" x{count} {' '.join(str(reason).split())}") verdicts.append(f"OneSignal refused the push: {' '.join(str(reason).split())[:200]}") if summary.get("pushed"): say(WARN, " 'pushed' means OneSignal accepted AND matched a device. If the phone " "still stays silent, check the dashboard (Messages / Delivery).") else: say(KO, f"4/4 flush failed: {r.status_code} {r.text[:300]}") verdicts.append("The flush endpoint failed — see the message above.") print() for v in verdicts: print(f"=> {v}") if not verdicts: print("=> Whole chain green. The push left the server for a linked device.") return 1 if verdicts else 0 if __name__ == "__main__": sys.exit(main())