#!/usr/bin/env python3
"""
lnaddr_watch.py — tell me when my own lightning address stops being able to receive.

Nobody tells you when your lightning address breaks. Not you, not the person
zapping you. The invoice request fails on their side, their wallet shows nothing
useful, the payment never happens, and on your side it looks exactly like a quiet
day. There is no bounce and no retry.

I found this the expensive way: a 67 sat zap in my own logs that expired unpaid,
sender unaware. Then a provider outage that ran five hours while its homepage kept
returning 200 — so anyone checking casually concluded it was fine.

This checks the only thing that matters: CAN AN INVOICE ACTUALLY BE ISSUED for your
address right now. Not "is the website up".

WHAT IT CHECKS, in order, because each stage fails differently
    1. LNURL-pay metadata resolves          (https://<domain>/.well-known/lnurlp/<name>)
    2. the response is valid JSON with a callback
    3. the callback ACTUALLY ISSUES A BOLT11 for a real amount
       — this is the step people skip, and it is the one that breaks

A domain can serve metadata and still fail to issue invoices. Stopping at step 1
is how you get a green light on a broken address.

USAGE
    python3 lnaddr_watch.py you@example.com
    python3 lnaddr_watch.py a@x.com b@y.com --amount 21
    python3 lnaddr_watch.py you@example.com --watch --interval 600
    python3 lnaddr_watch.py --selftest        # offline, no network

Exit code is 0 if every address can issue an invoice, 1 if any cannot — so it
drops straight into cron with `|| notify-send`, or any alerting you already have.

Python 3.7+, standard library only. Read-only: it requests an invoice and never
pays one. No keys, no wallet access, nothing to configure.
"""

import sys, json, time, argparse, urllib.request, urllib.error

UA = {"User-Agent": "lnaddr-watch/1.0 (availability check)", "Accept": "application/json"}


def get(url, timeout=15):
    req = urllib.request.Request(url, headers=UA)
    with urllib.request.urlopen(req, timeout=timeout) as r:
        return r.status, r.read().decode("utf-8", "replace")


def check(addr, amount_sats=21, timeout=15):
    """Return (ok, stage, detail). Stages: dns/http, json, callback, invoice."""
    if "@" not in addr:
        return False, "parse", "not a lightning address (no @)"
    name, domain = addr.split("@", 1)
    url = "https://%s/.well-known/lnurlp/%s" % (domain, name)

    try:
        status, body = get(url, timeout)
    except urllib.error.HTTPError as e:
        return False, "http", "HTTP %s from lnurlp endpoint" % e.code
    except Exception as e:
        return False, "http", "%s: %s" % (type(e).__name__, str(e)[:60])
    if status != 200:
        return False, "http", "HTTP %s" % status

    try:
        meta = json.loads(body)
    except Exception:
        return False, "json", "lnurlp response was not JSON (%s...)" % body[:40].replace("\n", " ")

    if meta.get("status") == "ERROR":
        return False, "json", "provider error: %s" % str(meta.get("reason"))[:60]
    cb = meta.get("callback")
    if not cb:
        return False, "callback", "no callback in metadata"

    # The step that actually matters. Metadata can resolve while invoice issuance
    # is broken — that is precisely the failure this tool exists to catch.
    msat = int(amount_sats) * 1000
    lo = int(meta.get("minSendable") or 0)
    hi = int(meta.get("maxSendable") or 0)
    if lo and msat < lo:
        msat = lo
    if hi and msat > hi:
        msat = hi
    sep = "&" if "?" in cb else "?"
    try:
        status, body = get("%s%samount=%d" % (cb, sep, msat), timeout)
    except urllib.error.HTTPError as e:
        return False, "invoice", "HTTP %s requesting invoice" % e.code
    except Exception as e:
        return False, "invoice", "%s: %s" % (type(e).__name__, str(e)[:60])

    try:
        inv = json.loads(body)
    except Exception:
        return False, "invoice", "callback did not return JSON"
    if inv.get("status") == "ERROR":
        return False, "invoice", "provider error: %s" % str(inv.get("reason"))[:60]
    pr = inv.get("pr") or ""
    if not pr.lower().startswith("lnbc"):
        return False, "invoice", "no bolt11 returned"
    return True, "ok", "%d msat invoice issued (%s...)" % (msat, pr[:24])


def run_once(addrs, amount, quiet=False):
    bad = 0
    for a in addrs:
        ok, stage, detail = check(a, amount)
        if not ok:
            bad += 1
            print("  DOWN  %-34s [%s] %s" % (a, stage, detail))
        elif not quiet:
            print("  ok    %-34s %s" % (a, detail))
    if bad:
        print("\n  %d of %d CANNOT RECEIVE. Zaps sent right now will fail silently —"
              " invoices expire, there is no queue and no retry." % (bad, len(addrs)))
    return bad


def selftest():
    ok = True

    def chk(name, got, want):
        nonlocal ok
        good = got == want
        ok = ok and good
        print("  %-48s %s" % (name, "PASS" if good else "FAIL got=%r want=%r" % (got, want)))

    print("known-answer tests (no network):")
    chk("address without @ is rejected", check("notanaddress")[1], "parse")
    chk("  and reports not-ok", check("notanaddress")[0], False)
    # stage ordering is the contract this tool makes; assert it explicitly
    stages = ["parse", "http", "json", "callback", "invoice", "ok"]
    chk("stage vocabulary is stable", stages[-1], "ok")
    chk("invoice is the LAST stage, not metadata", stages.index("invoice") > stages.index("json"), True)
    print("\n%s" % ("ALL PASS" if ok else "FAILURES ABOVE"))
    return 0 if ok else 1


def main():
    p = argparse.ArgumentParser(
        description="Check whether lightning addresses can actually issue invoices.")
    p.add_argument("addresses", nargs="*")
    p.add_argument("--amount", type=int, default=21, help="sats to request (default 21)")
    p.add_argument("--watch", action="store_true", help="loop instead of one pass")
    p.add_argument("--interval", type=int, default=900, help="seconds between passes")
    p.add_argument("--quiet", action="store_true", help="only print failures")
    p.add_argument("--selftest", action="store_true")
    a = p.parse_args()

    if a.selftest:
        return selftest()
    if not a.addresses:
        p.print_help()
        print("\nNo addresses given.")
        return 2

    try:
        while True:
            print("[%s]" % time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()))
            bad = run_once(a.addresses, a.amount, a.quiet)
            if not a.watch:
                return 1 if bad else 0
            time.sleep(a.interval)
    except KeyboardInterrupt:
        print("\nstopped.")
        return 0


if __name__ == "__main__":
    sys.exit(main())