"""The push bundle (moved verbatim from the pre-split web_stack_test.py;
see test/harness.py for the shared machinery and the runner)."""
from harness import *  # noqa: F401,F403 — the shared wire helpers


# =============================================================================
# Bundle: push — plan 11 task E: real Web Push over VAPID. VAPID
# persistence, the service worker, subscription CRUD + validation,
# notify-me prefs, the block/lifecycle dispatch matrix (global vs
# opted-in), 410 pruning, retry-next-tick delivery and protocol-level
# payload verification (RFC 8291 decryption + ES256 JWT) against the
# fake push service.
# =============================================================================

def run(ctx):
    auth, fake, fake_id, fake_token = ctx.auth, ctx.fake, ctx.fake_id, ctx.fake_token

    def subscription_count(endpoint):
        conn = db()
        try:
            return conn.execute(
                "select count(*) from push_subscriptions where endpoint = ?",
                (endpoint,)).fetchone()[0]
        finally:
            conn.close()

    print("== task E: service worker asset ==")
    status, sw = http("GET", "/sw.js", timeout=15)
    check(status == 200 and "addEventListener('push'" in sw
          and "notificationclick" in sw and "clients.openWindow" in sw,
          "the service worker serves at the origin root with push + click routing")

    print("== task E: VAPID keypair generation + persistence ==")
    config = json.load(open(os.path.join(WORK, "web-config.json")))
    vapid = config.get("vapid") or {}
    check(bool(vapid.get("public_key")) and bool(vapid.get("private_key")),
          "the VAPID keypair persists into web-config.json's vapid section")
    vapid_private_key = ec.derive_private_key(
        int.from_bytes(b64url_dec(vapid["private_key"]), "big"), ec.SECP256R1())
    derived_public = vapid_private_key.public_key().public_bytes(
        Encoding.X962, PublicFormat.UncompressedPoint)
    check(b64url_dec(vapid["public_key"]) == derived_public,
          "the persisted public key matches the private scalar (a valid P-256 pair)")
    dash = slot_public(auth, "/", "dashboard")
    check(dash.get("push_vapid_key") == vapid["public_key"],
          "the dashboard exposes the same VAPID key to subscribers")
    check(dash.get("push_subscribed") is False and bool(dash.get("push_subscribe")),
          "the bell starts unsubscribed and carries the subscribe action")

    print("== task E: subscription CRUD + auth validation ==")
    browser = ec.generate_private_key(ec.SECP256R1())
    browser_public = browser.public_key().public_bytes(
        Encoding.X962, PublicFormat.UncompressedPoint)
    browser_auth = os.urandom(16)
    subscribe_fields = {"p256dh": b64url(browser_public), "auth": b64url(browser_auth)}
    valid_endpoint = "http://127.0.0.1:1/x"
    check("absolute http(s)" in (action_notify(act(auth, dash["push_subscribe"],
        {"endpoint": "ftp://example.com/push", **subscribe_fields})) or ""),
        "a non-http endpoint is rejected")
    check("p256dh" in (action_notify(act(auth, dash["push_subscribe"],
        {"endpoint": valid_endpoint, "p256dh": b64url(os.urandom(64)),
         "auth": subscribe_fields["auth"]})) or ""),
        "a malformed p256dh point is rejected")
    check("auth" in (action_notify(act(auth, dash["push_subscribe"],
        {"endpoint": valid_endpoint, "p256dh": subscribe_fields["p256dh"],
         "auth": b64url(os.urandom(15))})) or ""),
        "a malformed auth secret is rejected")

    push = FakePushService()
    push.serve()
    endpoint_main = push.url("/main")
    act(auth, dash["push_subscribe"],
        {"endpoint": endpoint_main, **subscribe_fields})
    conn = db()
    rows = conn.execute("select user_id, p256dh, auth from push_subscriptions"
                        " where endpoint = ?", (endpoint_main,)).fetchall()
    admin_id = conn.execute("select id from spry_users where username = 'admin'"
                            ).fetchone()[0]
    conn.close()
    check(len(rows) == 1 and rows[0][0] == admin_id,
          "a valid subscription stores against the signed-in user")
    # Re-subscribing the same endpoint replaces (never duplicates).
    rotated_auth = os.urandom(16)
    act(auth, dash["push_subscribe"],
        {"endpoint": endpoint_main, "p256dh": subscribe_fields["p256dh"],
         "auth": b64url(rotated_auth)})
    conn = db()
    rows = conn.execute("select auth from push_subscriptions where endpoint = ?",
                        (endpoint_main,)).fetchall()
    conn.close()
    check(len(rows) == 1 and rows[0][0] == b64url(rotated_auth),
          "re-subscribing an endpoint replaces its keys in place")

    print("== task E: notify-me pref persistence ==")
    act(auth, slot_public(auth, "/tasks", "tasks")["create"],
        {"project_id": "1", "title": "Push pref task", "priority": "5"})
    pref_task = task_id_of("Push pref task")
    act(auth, task_state(auth, pref_task)["toggle_notify"], {})
    conn = db()
    check(conn.execute("select count(*) from task_notify_prefs where task_id = ?"
                       " and user_id = ?", (pref_task, admin_id)).fetchone()[0] == 1,
          "Notify me persists a task_notify_prefs row for the user")
    conn.close()
    check(task_state(auth, pref_task).get("notify_me") is True,
          "the task slot reflects notify_me after the toggle")
    act(auth, task_state(auth, pref_task)["toggle_notify"], {})
    conn = db()
    check(conn.execute("select count(*) from task_notify_prefs where task_id = ?",
                       (pref_task,)).fetchone()[0] == 0,
          "Don't notify me removes the pref row")
    conn.close()
    check(task_state(auth, pref_task).get("notify_me") is False,
          "the task slot reflects the opt-out")

    print("== task E: dispatch matrix — global blocks vs opted-in lifecycle ==")
    # Two "browsers" for admin (/main, /gone) and one for another user
    # (id 999, rows seeded directly — push targeting is per user).
    endpoint_gone = push.url("/gone")
    push.status_for["/gone"] = 410
    act(auth, dash["push_subscribe"],
        {"endpoint": endpoint_gone, **subscribe_fields})
    other_browser = ec.generate_private_key(ec.SECP256R1())
    other_public = other_browser.public_key().public_bytes(
        Encoding.X962, PublicFormat.UncompressedPoint)
    other_auth = os.urandom(16)
    endpoint_other = push.url("/other")
    conn = db()
    conn.execute("insert into push_subscriptions (user_id, endpoint, p256dh, auth,"
                 " created_at) values (999, ?, ?, ?, strftime('%s','now'))",
                 (endpoint_other, b64url(other_public), b64url(other_auth)))
    conn.commit()
    conn.close()

    # Task 1: admin opted in via the UI toggle; the other user is not.
    # Baselines are captured BEFORE the enqueue: the first-start push
    # fires at the allocation, the instant the dispatch lands.
    act(auth, slot_public(auth, "/tasks", "tasks")["create"],
        {"project_id": "1", "title": "Push lifecycle task", "priority": "5"})
    task1 = task_id_of("Push lifecycle task")
    act(auth, task_state(auth, task1)["toggle_notify"], {})
    main_before = len(push.hits("/main"))
    other_before = len(push.hits("/other"))
    gone_before = len(push.hits("/gone"))
    enqueue_task(auth, task1)
    check(wait_for(lambda: fake.running_task == task1 or False, timeout=30,
                   what="task 1 assignment") is not None,
          "task 1 dispatches to the fake daemon")
    task1_session = fake.session_id
    check(wait_for(lambda: len(push.hits("/main")) > main_before, timeout=15,
                   what="first-start push to the opted-in admin") is not None,
          "first start notifies the opted-in user's subscription")
    check(wait_for(lambda: len(push.hits("/gone")) > gone_before, timeout=15,
                   what="first-start push to the gone endpoint") is not None,
          "first start reaches every subscription of the opted-in user")
    check(quiet_window(quiet_seconds(),
                       lambda: len(push.hits("/other")) == other_before,
                       "no first-start push to the not-opted-in user"),
          "first start never reaches users without the notify pref")
    check(wait_task_done(auth, task1, timeout=60) is not None,
          "task 1 completes")
    _, done_payload = decrypt_webpush(push.hits("/main")[-1], browser,
                                      browser_public, rotated_auth)
    check(done_payload["title"] == "Push lifecycle task — completed"
          and done_payload["url"] == "/tasks/%d" % task1,
          "completion notifies the opted-in user (decrypted payload)")
    check(len(push.hits("/other")) == other_before,
          "completion stays with the opted-in user only")
    check(wait_for(lambda: subscription_count(endpoint_gone) == 0, timeout=15,
                   what="410 endpoint pruned") is not None,
          "a 410 Gone endpoint is pruned from push_subscriptions")

    # Block events go to EVERY push-subscribed user globally. The
    # poll-era /update raise is gone: a live (non-terminal) host task
    # with an open session on the fake carries the block through the
    # results push — task1 is done and terminal pushes are consumed
    # silently, so the host is its own fixture row.
    conn = db()
    conn.execute("insert into tasks (project_id, title, status, priority,"
                 " created_at) values (1, 'Push block host', 'working', 5,"
                 " strftime('%s','now'))")
    push_host = conn.execute("select max(id) from tasks").fetchone()[0]
    conn.execute("insert into sessions (task_id, saddle_id, started_at)"
                 " values (?, ?, strftime('%s','now'))", (push_host, fake_id))
    push_host_session = conn.execute(
        "select max(id) from sessions").fetchone()[0]
    conn.commit()
    conn.close()

    def raise_global_block(question):
        return saddle_api("POST", "/api/saddle/results", fake_token, {
            "session_id": push_host_session, "state": "blocked",
            "block": {"question": question, "type": "question"}})

    main_before = len(push.hits("/main"))
    other_before = len(push.hits("/other"))
    raise_global_block("Global question for everyone?")
    check(wait_for(lambda: len(push.hits("/main")) > main_before
                   and len(push.hits("/other")) > other_before, timeout=15,
                   what="global block push") is not None,
          "a raised block notifies every push-subscribed user globally")
    block_hit = push.hits("/other")[-1]
    _, payload = decrypt_webpush(block_hit, other_browser, other_public, other_auth)
    check(payload["title"] == "Push block host — question raised"
          and payload["body"] == "Global question for everyone?"
          and payload["url"] == "/tasks/%d" % push_host,
          "the block payload carries task title + kind, question and click URL (%s)"
          % payload)
    ciphertext_len = len(block_hit["body"]) - (21 + 65)
    envelope = block_hit["body"]
    record_size = int.from_bytes(envelope[16:20], "big")
    # rs = plaintext + delimiter (1) + tag (16) == ciphertext-with-tag len.
    check(record_size == ciphertext_len,
          "the RFC 8291 envelope record size covers payload + delimiter + tag")
    claims = verify_vapid(block_hit, vapid["public_key"], vapid["private_key"])
    check(claims.get("aud") == "http://127.0.0.1:%d" % push.port
          and claims.get("sub", "").startswith("mailto:")
          and claims.get("exp", 0) > time.time(),
          "the VAPID JWT verifies: aud = the endpoint origin, mailto sub, future exp")
    check(block_hit["headers"].get("ttl") == "60"
          and block_hit["headers"].get("content-type") == "application/octet-stream",
          "the push POST carries TTL and the encrypted content type")

    # Task 2: only the OTHER user opted in — start + failure reach them
    # alone. Baseline again before the enqueue (first start fires at
    # the allocation).
    fake.fail_next = True
    act(auth, slot_public(auth, "/tasks", "tasks")["create"],
        {"project_id": "1", "title": "Push failure task", "priority": "5"})
    task2 = task_id_of("Push failure task")
    conn = db()
    conn.execute("insert into task_notify_prefs (task_id, user_id) values (?, 999)",
                 (task2,))
    conn.commit()
    conn.close()
    main_before2 = len(push.hits("/main"))
    other_before2 = len(push.hits("/other"))
    enqueue_task(auth, task2)
    check(wait_for(lambda: fake.running_task == task2 or False, timeout=30,
                   what="task 2 assignment") is not None,
          "task 2 dispatches (scripted to fail)")
    check(wait_for(lambda: task_state(auth, task2)["status"] == "failed"
                   or False, timeout=30, what="task 2 failed") is not None,
          "task 2 fails")
    check(wait_for(lambda: len(push.hits("/other")) >= other_before2 + 2,
                   timeout=15, what="opted-in user's started+failed pushes"),
          "the opted-in user gets first start AND failure for the task")
    started_hit = [hit for hit in push.hits("/other")[other_before2:]][0]
    _, started_payload = decrypt_webpush(started_hit, other_browser,
                                         other_public, other_auth)
    check(started_payload["title"] == "Push failure task — started",
          "the first-start payload names task + kind (%s)" % started_payload)
    _, failure_payload = decrypt_webpush(push.hits("/other")[-1], other_browser,
                                         other_public, other_auth)
    check(failure_payload["title"] == "Push failure task — failed"
          and failure_payload["url"] == "/tasks/%d" % task2,
          "the failure payload names the task and links its page")
    check(len(push.hits("/main")) == main_before2,
          "an agent failure never reaches users without the pref")

    print("== task E: retry next tick (best-effort delivery) ==")
    endpoint_flaky = push.url("/flaky")
    act(auth, dash["push_subscribe"],
        {"endpoint": endpoint_flaky, **subscribe_fields})
    push.fail_next["/flaky"] = 1
    flaky_before = len(push.hits("/flaky"))
    raise_global_block("Flaky delivery question?")
    check(wait_for(lambda: len(push.hits("/flaky")) >= flaky_before + 2,
                   timeout=20, what="flaky endpoint retried") is not None,
          "a failed send retries on the next dispatcher tick and lands")

    print("== plan 12: goal notifications — blocks/success/failure always ==")
    # Goal blocks, completion approval and failure push to EVERY
    # subscribed user (the mirror on the goal-run task stays quiet).
    push_goal = None
    act(auth, slot_public(auth, "/goals", "goals")["create"], {
        "project_id": "1", "title": "Push goal", "brief": "Notifies everyone.",
        "branch_name": "goal/push"})
    conn = db()
    row = conn.execute("select id from goals where title = 'Push goal'").fetchone()
    conn.close()
    push_goal = int(row[0])
    fake.goal_script = [("plan", "# Push plan\n\n1. Ring the bells")]
    main_before = len(push.hits("/main"))
    other_before = len(push.hits("/other"))
    act(auth, goal_state(push_goal)["start"], {})
    check(wait_for(lambda: goal_state(push_goal)["status"] == "plan_review" or False,
                   timeout=60, what="push goal enters plan review") is not None,
          "the push goal reaches plan review")
    check(wait_for(lambda: len(push.hits("/main")) > main_before
                   and len(push.hits("/other")) > other_before, timeout=15,
                   what="goal plan-review push") is not None,
          "a goal plan review notifies every subscribed user globally")
    _, plan_payload = decrypt_webpush(push.hits("/other")[-1], other_browser,
                                      other_public, other_auth)
    check("Push goal" in plan_payload["title"] and "plan" in plan_payload["title"]
          and plan_payload["url"] == "/goals/%d" % push_goal,
          "the goal block payload carries the goal title + click URL (%s)"
          % plan_payload)
    main_before = len(push.hits("/main"))
    fake.goal_script = [("suspend",)]
    act(auth, goal_state(push_goal)["plan_pending_block"]["answer_action"],
        {"answer": "Plan approved. Proceed with implementation."})
    check(wait_for(lambda: goal_state(push_goal)["status"] == "active" or False,
                   timeout=30, what="push goal active") is not None,
          "the push goal activates")
    check(quiet_window(3, lambda: len(push.hits("/main")) == main_before,
                       "no push for an internal resume"),
          "the goal agent's own park/resume does not notify (internal)")

    # `saddle failure` in goal context fails the goal — global push.
    fake.goal_script = [("fail", "the harness collapsed")]
    other_before2 = len(push.hits("/other"))
    main_before2 = len(push.hits("/main"))
    conn = db()
    conn.execute("insert into tasks (project_id, goal_id, kind, title, status,"
                 " priority, completed_at, created_at) values (1, ?, 'task',"
                 " 'Warm', 'done', 5,"
                 " strftime('%s','now'),"
                 " strftime('%s','now','-30 seconds'))", (push_goal,))
    conn.commit()
    conn.close()
    check(wait_for(lambda: goal_state(push_goal)["status"] == "failed" or False,
                   timeout=60, what="goal failure lands") is not None,
          "the goal agent's failure fails the goal")
    check(wait_for(lambda: len(push.hits("/other")) > other_before2
                   and len(push.hits("/main")) > main_before2, timeout=15,
                   what="goal failure push") is not None,
          "a goal failure notifies every subscribed user globally")
    _, fail_payload = decrypt_webpush(push.hits("/other")[-1], other_browser,
                                      other_public, other_auth)
    check("Push goal" in fail_payload["title"] and "failed" in fail_payload["title"]
          and fail_payload["url"] == "/goals/%d" % push_goal,
          "the goal failure payload carries the goal title + click URL (%s)"
          % fail_payload)

    # Completion approval pushes "complete" globally (success always).
    act(auth, slot_public(auth, "/goals", "goals")["create"], {
        "project_id": "1", "title": "Push done goal", "brief": "Completes.",
        "branch_name": "goal/push-done"})
    conn = db()
    row = conn.execute("select id from goals where title = 'Push done goal'").fetchone()
    conn.close()
    done_goal = int(row[0])
    fake.goal_script = [("plan", "# Done plan")]
    act(auth, goal_state(done_goal)["start"], {})
    check(wait_for(lambda: goal_state(done_goal)["status"] == "plan_review" or False,
                   timeout=60, what="done goal plan review") is not None,
          "the done goal reaches plan review")
    fake.goal_script = [("done", "# Completion\n\nShipped.")]
    act(auth, goal_state(done_goal)["plan_pending_block"]["answer_action"],
        {"answer": "Plan approved. Proceed with implementation."})
    check(wait_for(lambda: goal_state(done_goal)["status"] == "completion_review"
                   or False, timeout=60, what="done goal completion review") is not None,
          "the approved done goal reaches completion review")
    main_before = len(push.hits("/main"))
    act(auth, goal_state(done_goal)["completion_pending_block"]["answer_action"],
        {"answer": "Review approved. Continue with the task."})
    check(wait_for(lambda: goal_state(done_goal)["status"] == "complete" or False,
                   timeout=30, what="done goal completes") is not None,
          "the done goal completes")
    check(wait_for(lambda: len(push.hits("/main")) > main_before, timeout=15,
                   what="goal success push") is not None,
          "a goal completion notifies every subscribed user globally")
    _, done_goal_payload = decrypt_webpush(push.hits("/main")[-1], browser,
                                           browser_public, rotated_auth)
    check("Push done goal" in done_goal_payload["title"]
          and "complete" in done_goal_payload["title"]
          and done_goal_payload["url"] == "/goals/%d" % done_goal,
          "the goal success payload carries the goal title + click URL (%s)"
          % done_goal_payload)

    print("== task E: unsubscribe removes the subscription ==")
    dash = slot_public(auth, "/", "dashboard")
    check(dash.get("push_subscribed") is True,
          "the bell shows subscribed while endpoints exist")
    act(auth, dash["push_unsubscribe"], {"endpoint": endpoint_main})
    check(subscription_count(endpoint_main) == 0,
          "unsubscribing removes the endpoint's row")