#!/bin/sh
# 在临时 Docker 网络和一次性 MySQL 中验证首次初始化、控制台登录、任务执行、重试和定时任务。
set -eu

E2E_DOCKER=${DOCKER:-docker}
E2E_GO=${GO:-go}
E2E_IMAGE=${ORBIT_E2E_IMAGE:-orbit:e2e}
E2E_SUFFIX=$$
E2E_MYSQL="orbit-e2e-mysql-$E2E_SUFFIX"
E2E_ORBIT="orbit-e2e-server-$E2E_SUFFIX"
E2E_ORBIT_SECOND="orbit-e2e-login-peer-$E2E_SUFFIX"
E2E_DIR=$(mktemp -d "${TMPDIR:-/tmp}/orbit-console-e2e.XXXXXX")
E2E_PASSWORD_FILE="$E2E_DIR/console-password"
E2E_CREDENTIALS_FILE="$E2E_DIR/credentials.json"

cleanup() {
    "$E2E_DOCKER" stop "$E2E_ORBIT_SECOND" "$E2E_ORBIT" "$E2E_MYSQL" >/dev/null 2>&1 || true
    "$E2E_DOCKER" container rm "$E2E_ORBIT_SECOND" "$E2E_ORBIT" "$E2E_MYSQL" >/dev/null 2>&1 || true
    if [ -d "$E2E_DIR/archive-job" ]; then
        for e2e_artifact in "$E2E_DIR/archive-job"/.lock "$E2E_DIR/archive-job"/manifest.json "$E2E_DIR/archive-job"/batch-000001.json; do
            if [ -f "$e2e_artifact" ]; then unlink "$e2e_artifact"; fi
        done
        rmdir "$E2E_DIR/archive-job" >/dev/null 2>&1 || true
    fi
    if [ -d "$E2E_DIR/archive-volume" ]; then
        python3 - "$E2E_DIR/archive-volume" <<'PY_CLEAN'
import pathlib, shutil, sys
path = pathlib.Path(sys.argv[1])
# 仅清理由该测试 mktemp 目录下创建的独立归档卷。
if path.name == "archive-volume" and path.parent.name.startswith("orbit-console-e2e."):
    shutil.rmtree(path)
PY_CLEAN
    fi
    for e2e_artifact in "$E2E_DIR"/*; do
        if [ -f "$e2e_artifact" ]; then
            unlink "$e2e_artifact"
        fi
    done
    rmdir "$E2E_DIR" >/dev/null 2>&1 || true
}
trap cleanup EXIT HUP INT TERM

umask 077
E2E_DB_PASSWORD=$(openssl rand -hex 24)
E2E_CONSOLE_PASSWORD=$(openssl rand -hex 24)
E2E_APPLICATION="console-e2e-$E2E_SUFFIX"
E2E_ADMIN="orbit-admin-$E2E_SUFFIX"
printf '%s\n' "$E2E_CONSOLE_PASSWORD" > "$E2E_PASSWORD_FILE"

python3 - "$E2E_DIR/credential-keys.json" <<'PY_KEY'
import base64, json, pathlib, secrets, sys
path = pathlib.Path(sys.argv[1])
path.write_text(json.dumps({'active':'test-v1', 'keys':{'test-v1':base64.b64encode(secrets.token_bytes(32)).decode()}}))
path.chmod(0o600)
PY_KEY

printf '%s\n' "[1/8] build image"
"$E2E_DOCKER" build --tag "$E2E_IMAGE" . >/dev/null
"$E2E_DOCKER" run -d --name "$E2E_MYSQL" --network bridge \
    --env MYSQL_ROOT_PASSWORD="$E2E_DB_PASSWORD" \
    --env MYSQL_DATABASE=orbit_test mysql:8.4 >/dev/null

E2E_READY=0
for e2e_attempt in $(seq 1 60); do
    if "$E2E_DOCKER" exec "$E2E_MYSQL" sh -c 'MYSQL_PWD="$MYSQL_ROOT_PASSWORD" mysql -uroot --protocol=tcp -e "SELECT 1" >/dev/null 2>&1'; then
        E2E_READY=1
        break
    fi
    sleep 1
done
if [ "$E2E_READY" -ne 1 ]; then
    printf '%s\n' 'temporary MySQL failed to become ready' >&2
    "$E2E_DOCKER" logs --tail 40 "$E2E_MYSQL" >&2
    exit 1
fi

E2E_DSN="root:$E2E_DB_PASSWORD@tcp(mysql:3306)/orbit_test"
printf '%s\n' "[2/8] migrate empty database"
"$E2E_DOCKER" run --rm --network bridge --link "$E2E_MYSQL:mysql" \
    --env ORBIT_MYSQL_DSN="$E2E_DSN" \
    --entrypoint /usr/local/bin/orbit-migrate "$E2E_IMAGE" up >/dev/null

printf '%s\n' "[3/8] bootstrap resources and console administrator"
"$E2E_DOCKER" run --rm -i --network bridge --link "$E2E_MYSQL:mysql" \
    --env ORBIT_MYSQL_DSN="$E2E_DSN" \
    --volume "$E2E_DIR:/run/orbit" \
    --env ORBIT_CREDENTIAL_KEY_FILE=/run/orbit/credential-keys.json \
    --entrypoint /usr/local/bin/orbit-admin "$E2E_IMAGE" bootstrap \
    --application "$E2E_APPLICATION" \
    --namespace default \
    --environment test \
    --queue files \
    --worker-access \
    --schedule-access \
    --console-username "$E2E_ADMIN" \
    --console-display-name 'Orbit E2E 管理员' \
    --console-password-stdin \
    --output /run/orbit/credentials.json < "$E2E_PASSWORD_FILE" >/dev/null
"$E2E_DOCKER" run --rm --user 0:0 --volume "$E2E_DIR:/run/orbit" --entrypoint /bin/sh mysql:8.4 -c 'chown 65532:65532 /run/orbit/credential-keys.json'
E2E_QUEUE=$(python3 - "$E2E_CREDENTIALS_FILE" <<'PY'
import json
import sys

value = json.load(open(sys.argv[1], encoding="utf-8"))
queue = value.get("queue_id", "")
if not queue:
    raise SystemExit("bootstrap output has no queue_id")
print(queue)
PY
)

mkdir -m 700 "$E2E_DIR/archive-volume"
E2E_STORAGE_ID=$(openssl rand -hex 16)
printf '%s\n' "[4/8] start Orbit container"
"$E2E_DOCKER" run -d --name "$E2E_ORBIT" --network bridge --link "$E2E_MYSQL:mysql" \
    --env ORBIT_MYSQL_DSN="$E2E_DSN" \
    --env ORBIT_HTTP_ADDR=0.0.0.0:8080 \
    --env ORBIT_CONSOLE_QUEUE_ID="$E2E_QUEUE" \
    --env ORBIT_CONSOLE_LOGIN_FAILURE_LIMIT=3 \
    --env ORBIT_CONSOLE_LOGIN_LOCK_DURATION=10s \
    --env ORBIT_CONSOLE_LOGIN_RATE_WINDOW=2s \
    --env ORBIT_CONSOLE_APPLICATION_ADMIN="$E2E_ADMIN" \
    --env ORBIT_CREDENTIAL_KEY_FILE=/run/credential-keys.json \
    --volume "$E2E_DIR/credential-keys.json:/run/credential-keys.json:ro" \
    --env ORBIT_ARCHIVE_ROOT=/var/lib/orbit/archive \
    --env ORBIT_ARCHIVE_STORAGE_ID="$E2E_STORAGE_ID" \
    --env ORBIT_ARCHIVE_TICK_INTERVAL=1s \
    --volume "$E2E_DIR/archive-volume:/var/lib/orbit/archive" \
    --publish 127.0.0.1::8080 "$E2E_IMAGE" >/dev/null
E2E_PORT=
for e2e_attempt in $(seq 1 30); do
    E2E_PORT=$("$E2E_DOCKER" port "$E2E_ORBIT" 8080/tcp 2>/dev/null | sed -n 's/.*:\([0-9][0-9]*\)$/\1/p' | head -n 1)
    if [ -n "$E2E_PORT" ]; then
        break
    fi
    sleep 1
done
if [ -z "$E2E_PORT" ]; then
    printf '%s\n' 'Orbit container did not publish HTTP port' >&2
    "$E2E_DOCKER" logs --tail 60 "$E2E_ORBIT" >&2
    exit 1
fi
E2E_URL="http://127.0.0.1:$E2E_PORT"
E2E_READY=0
for e2e_attempt in $(seq 1 30); do
    if curl -fsS "$E2E_URL/health/ready" >/dev/null 2>&1; then
        E2E_READY=1
        break
    fi
    sleep 1
done
if [ "$E2E_READY" -ne 1 ]; then
    printf '%s\n' 'Orbit readiness check failed' >&2
    "$E2E_DOCKER" logs --tail 60 "$E2E_ORBIT" >&2
    exit 1
fi

ORBIT_E2E_URL="$E2E_URL" ORBIT_E2E_USER="$E2E_ADMIN" \
    ORBIT_E2E_PASSWORD_FILE="$E2E_PASSWORD_FILE" python3 ./scripts/test-console-applications.py

printf '%s\n' "[5/8] console login and protected-page check"
ORBIT_E2E_URL="$E2E_URL" ORBIT_E2E_QUEUE="$E2E_QUEUE" ORBIT_E2E_USER="$E2E_ADMIN" \
    ORBIT_E2E_PASSWORD_FILE="$E2E_PASSWORD_FILE" python3 - <<'PY'
import http.cookiejar
import json
import os
import pathlib
import urllib.error
import urllib.parse
import urllib.request

base = os.environ["ORBIT_E2E_URL"]
queue = os.environ["ORBIT_E2E_QUEUE"]
username = os.environ["ORBIT_E2E_USER"]
password = pathlib.Path(os.environ["ORBIT_E2E_PASSWORD_FILE"]).read_text(encoding="utf-8").strip()
jar = http.cookiejar.CookieJar()
client = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar))

def request(path, method="GET", body=None):
    request_body = None if body is None else json.dumps(body).encode("utf-8")
    request = urllib.request.Request(base + path, method=method, data=request_body)
    request.add_header("Accept", "application/json")
    if body is not None:
        request.add_header("Content-Type", "application/json")
    try:
        with client.open(request, timeout=8) as response:
            return response.status, response.read()
    except urllib.error.HTTPError as error:
        return error.code, error.read()

status, _ = request("/console/auth/session")
assert status == 200, status
status, _ = request("/console/auth/login", "POST", {"username": username, "password": password})
assert status == 200, status
status, body = request("/console/auth/session")
assert status == 200 and json.loads(body).get("authenticated") is True
status, body = request("/console/")
assert status == 200 and b'id="overview-task-list"' in body and b'id="tasks-filters"' in body and b'/console/console-core.js' in body, status
status, body = request("/console/config")
assert status == 200 and json.loads(body).get("default_queue_id") == queue
status, _ = request("/v1/tasks?" + urllib.parse.urlencode({"queue_id": queue, "limit": 20}))
assert status == 200, status
print("console_login=ok protected_page=ok")
PY

printf '%s\n' "[login] cross-node rejection, restart and recovery"
"$E2E_DOCKER" run -d --name "$E2E_ORBIT_SECOND" --network bridge --link "$E2E_MYSQL:mysql" \
    --env ORBIT_MYSQL_DSN="$E2E_DSN" --env ORBIT_HTTP_ADDR=0.0.0.0:8080 \
    --env ORBIT_CONSOLE_APPLICATION_ADMIN="$E2E_ADMIN" \
    --env ORBIT_CREDENTIAL_KEY_FILE=/run/credential-keys.json \
    --env ORBIT_CONSOLE_LOGIN_FAILURE_LIMIT=3 \
    --env ORBIT_CONSOLE_LOGIN_LOCK_DURATION=10s \
    --env ORBIT_CONSOLE_LOGIN_RATE_WINDOW=2s \
    --volume "$E2E_DIR/credential-keys.json:/run/credential-keys.json:ro" \
    --publish 127.0.0.1::8080 "$E2E_IMAGE" >/dev/null
E2E_SECOND_PORT=$("$E2E_DOCKER" port "$E2E_ORBIT_SECOND" 8080/tcp | sed -n 's/.*:\([0-9][0-9]*\)$/\1/p' | head -n 1)
ORBIT_E2E_URL="$E2E_URL" ORBIT_E2E_SECOND_URL="http://127.0.0.1:$E2E_SECOND_PORT" \
    ORBIT_E2E_SECOND_CONTAINER="$E2E_ORBIT_SECOND" ORBIT_E2E_USER="$E2E_ADMIN" \
    ORBIT_E2E_PASSWORD_FILE="$E2E_PASSWORD_FILE" DOCKER="$E2E_DOCKER" python3 scripts/test-console-login.py
"$E2E_DOCKER" stop "$E2E_ORBIT_SECOND" >/dev/null

printf '%s\n' "[6/8] WebSocket Worker task execution"
(cd examples && ORBIT_URL="$E2E_URL" ORBIT_CREDENTIALS_FILE="$E2E_CREDENTIALS_FILE" \
    "$E2E_GO" run ./worker) | tee "$E2E_DIR/worker.log"
grep -q 'worker_drained=true started=3 confirmed=3 running=0' "$E2E_DIR/worker.log"

printf '%s\n' "[7/8] retry and schedule execution"
(cd examples && ORBIT_URL="$E2E_URL" ORBIT_CREDENTIALS_FILE="$E2E_CREDENTIALS_FILE" \
    "$E2E_GO" run ./retry) | tee "$E2E_DIR/retry.log"
grep -q 'attempts=2' "$E2E_DIR/retry.log"
(cd examples && ORBIT_URL="$E2E_URL" ORBIT_CREDENTIALS_FILE="$E2E_CREDENTIALS_FILE" \
    "$E2E_GO" run ./schedules) | tee "$E2E_DIR/schedule.log"
grep -q 'source_kind=SCHEDULE' "$E2E_DIR/schedule.log"

printf '%s\n' "[8/8] final console data and logout check"
ORBIT_E2E_URL="$E2E_URL" ORBIT_E2E_QUEUE="$E2E_QUEUE" ORBIT_E2E_PASSWORD_FILE="$E2E_PASSWORD_FILE" \
    ORBIT_E2E_USER="$E2E_ADMIN" python3 - <<'PY'
import hashlib
import http.cookiejar
import json
import os
import pathlib
import urllib.error
import urllib.parse
import urllib.request

base = os.environ["ORBIT_E2E_URL"]
queue = os.environ["ORBIT_E2E_QUEUE"]
username = os.environ["ORBIT_E2E_USER"]
password = pathlib.Path(os.environ["ORBIT_E2E_PASSWORD_FILE"]).read_text(encoding="utf-8").strip()
jar = http.cookiejar.CookieJar()
client = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar))

def request(path, method="GET", body=None, headers=None):
    data = None if body is None else json.dumps(body).encode("utf-8")
    req = urllib.request.Request(base + path, method=method, data=data)
    req.add_header("Accept", "application/json")
    for name, value in (headers or {}).items():
        req.add_header(name, value)
    if body is not None:
        req.add_header("Content-Type", "application/json")
    try:
        with client.open(req, timeout=8) as response:
            return response.status, response.read()
    except urllib.error.HTTPError as error:
        return error.code, error.read()

status, _ = request("/console/auth/login", "POST", {"username": username, "password": password})
assert status == 200, status
status, body = request("/v1/tasks?" + urllib.parse.urlencode({"queue_id": queue, "limit": 100}))
assert status == 200, status
items = json.loads(body).get("data", {}).get("items", [])
assert any(item.get("state") == "SUCCEEDED" for item in items), "no succeeded task visible in console scope"
assert any(item.get("handler", "").startswith("text.normalize.runtime.") for item in items)
status, body = request("/console/retention")
assert status == 200, (status, body)
policy = json.loads(body)["data"]
assert policy["revision"] == "0" and policy["enabled"] is False
update = {key: policy[key] for key in ("enabled", "idempotency_days", "task_history_days", "workflow_history_days", "event_log_days", "delivery_history_days", "inbox_history_days", "audit_log_days")}
update.update(expected_revision="0", enabled=True, reason="console retention end-to-end validation")
headers = {"Idempotency-Key": "e2e-retention-save"}
status, body = request("/console/retention", "PUT", update, headers)
assert status == 200 and json.loads(body)["data"]["revision"] == "1", (status, body)
status, body = request("/console/retention", "PUT", update, headers)
assert status == 200 and json.loads(body)["data"]["revision"] == "1", (status, body)
status, _ = request("/console/retention", "PUT", update, {"Idempotency-Key": "e2e-retention-stale"})
assert status == 409, status
status, body = request("/console/retention/preview")
assert status == 200 and len(json.loads(body)["data"]["items"]) == 7, (status, body)
status, body = request("/console/retention/export", "POST", {"kind": "tasks", "expected_revision": "1", "limit": 20})
assert status == 200, (status, body)
snapshot = json.loads(body)["data"]
assert snapshot["format_version"] == 1
assert hashlib.sha256(snapshot["content"].encode("utf-8")).hexdigest() == snapshot["sha256"]
content = json.loads(snapshot["content"])
assert content["source_retained"] is True and content["root_count"] == 0
status, _ = request("/console/retention/export", "POST", {"kind": "tasks", "expected_revision": "0"})
assert status == 409, status
print("retention_policy=ok replay=ok stale_revision=409 preview_categories=7 empty_export_digest=ok")
status, body = request("/console/archive/plans")
assert status == 200, (status, body)
plans = json.loads(body)["data"]
assert len(plans["items"]) == 7 and plans["executor_configured"] is True
plan = dict(expected_revision="0", policy_revision="1", enabled=True, interval_seconds=3600, batch_size=20, max_bytes=1048576, reason="console scheduler e2e")
plan_headers = {"Idempotency-Key": "archive-plan-e2e"}
status, body = request("/console/archive/plans/tasks", "PUT", plan, {**plan_headers, "Origin": "https://different-origin.invalid"})
assert status == 403, (status, body)
status, body = request("/console/archive/plans/tasks", "PUT", plan, plan_headers)
assert status == 200 and json.loads(body)["data"]["revision"] == "1", (status, body)
status, body = request("/console/archive/plans/tasks", "PUT", plan, plan_headers)
assert status == 200 and json.loads(body)["data"]["revision"] == "1", (status, body)
import time
for attempt in range(30):
    status, body = request("/console/archive/jobs")
    assert status == 200, (status, body)
    jobs = json.loads(body)["data"]["items"]
    if jobs and jobs[0]["state"] in ("SUCCEEDED", "FAILED"):
        break
    time.sleep(1)
assert len(jobs) == 1 and jobs[0]["state"] == "SUCCEEDED", jobs
assert jobs[0]["batches"] == 1 and jobs[0]["roots"] == 0 and len(jobs[0]["last_sha256"]) == 64
assert "lease_id" not in jobs[0]
print("archive_scheduler=ok plan_replay=ok cross_origin=403 durable_job=SUCCEEDED")

status, _ = request("/console/auth/logout", "POST", {})
assert status == 200, status
status, _ = request("/v1/tasks?" + urllib.parse.urlencode({"queue_id": queue, "limit": 1}))
assert status == 401, status
status_retention, _ = request("/console/retention")
assert status_retention == 401, status_retention
status_archive, _ = request("/console/archive/jobs")
assert status_archive == 401, status_archive
print("console_task_data=ok succeeded_tasks=%d logout=ok unauthorized_after_logout=%d" % (sum(item.get("state") == "SUCCEEDED" for item in items), status))
PY

printf '%s\n' "[archive] durable job and offline verification"
E2E_NAMESPACE=$(python3 - "$E2E_CREDENTIALS_FILE" <<'PYCODE'
import json, sys
print(json.load(open(sys.argv[1], encoding="utf-8"))["namespace_id"])
PYCODE
)
"$E2E_DOCKER" run --rm --network bridge --link "$E2E_MYSQL:mysql" \
    --env ORBIT_MYSQL_DSN="$E2E_DSN" --volume "$E2E_DIR:/run/orbit" \
    --entrypoint /usr/local/bin/orbit-admin "$E2E_IMAGE" retention archive \
    --namespace "$E2E_NAMESPACE" --kind tasks --revision 1 --directory /run/orbit/archive-job > "$E2E_DIR/archive-result.json"
"$E2E_DOCKER" run --rm --network none --volume "$E2E_DIR:/run/orbit" \
    --entrypoint /usr/local/bin/orbit-admin "$E2E_IMAGE" retention verify-archive \
    --directory /run/orbit/archive-job > "$E2E_DIR/archive-verified.json"
python3 - "$E2E_DIR/archive-result.json" "$E2E_DIR/archive-verified.json" <<'PYCODE'
import json, sys
result, verified = [json.load(open(p, encoding="utf-8")) for p in sys.argv[1:]]
assert result == verified
assert result["completed"] is True and result["batches"] == 1 and result["roots"] == 0
print("durable_archive=ok offline_verification=ok roots=0 completed=true")
PYCODE

printf '%s\n' "console-e2e=passed image=$E2E_IMAGE url=$E2E_URL"