"""Real-process contract for Jetstream V2 planSnapshot operator controls."""

import json
import os
import urllib.error
import urllib.request


BASE = os.environ["STREAM_BASE_URL"].rstrip("/") + "/xrpc/network.bsky.jetstream.planSnapshot"
MODE = os.environ["STREAM_PLAN_CONFIG_MODE"]

# listener-first startup: xrpc paths 503 as "starting" until storage is
# published; wait out the gate before asserting plan responses
import time as _time

_status_url = os.environ["STREAM_BASE_URL"].rstrip("/") + "/status"
_deadline = _time.monotonic() + 15
while True:
    try:
        with urllib.request.urlopen(_status_url, timeout=1) as _r:
            if b"state            starting" not in _r.read():
                break
    except OSError:
        pass
    assert _time.monotonic() < _deadline, "startup gate never opened"
    _time.sleep(0.05)


def post(body):
    request = urllib.request.Request(
        BASE,
        data=json.dumps(body).encode(),
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    try:
        response = urllib.request.urlopen(request)
        return response.status, json.load(response)
    except urllib.error.HTTPError as error:
        return error.code, json.load(error)


if MODE == "limited":
    status, result = post({"collections": ["app.bsky.feed.post"]})
    assert status == 200
    assert result["sealedTipSeq"] == 12
    assert result["plannedThroughSeq"] == 1
    assert result["stats"]["entries"] == 1
    assert len(result["segments"]) == 1
    assert result["segments"][0]["mode"] == "blocks"
    assert result["segments"][0]["blocks"] == [{"first": 0, "last": 0}]
    for body in (
        {"dids": ["did:plc:a", "did:plc:b"]},
        {"collections": ["app.bsky.feed.post", "app.bsky.feed.like"]},
    ):
        status, error = post(body)
        assert status == 400 and error["error"] == "InvalidRequest"
elif MODE == "whole":
    status, result = post({"collections": ["app.bsky.feed.post"]})
    assert status == 200
    assert result["plannedThroughSeq"] == result["sealedTipSeq"] == 12
    assert result["stats"]["entries"] == 3
    assert len(result["segments"]) == 3
    assert all(segment["mode"] == "segment" for segment in result["segments"])
elif MODE == "disabled":
    for body in (
        {"dids": ["did:plc:planconfigaccount"]},
        {"collections": ["app.bsky.feed.post"]},
    ):
        status, error = post(body)
        assert status == 400 and error["error"] == "InvalidRequest"
else:
    raise AssertionError(f"unknown mode: {MODE}")

print(f"plan config {MODE}: PASS")