from __future__ import annotations

import hashlib
import json
import threading
from datetime import datetime, timedelta, timezone

import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

from yovision_brain.integration.machine_identity.token import KeyRecord, Registry, Signer, Verifier
from yovision_brain.integration.sense_control.connector import ConnectorResponse, SourceConfigEndpoint, StatusSender
from yovision_brain.integration.sense_control.consumer import SourceConfigConsumer, SourceConfigError
from yovision_brain.integration.sense_control.replay import SQLiteReplayStore
from yovision_brain.integration.sense_control.status import RuntimeStatusPublisher


def source_document(revision: int = 1, *, state: str = "active", effective: int = 0) -> bytes:
    now=datetime(2026,8,31,tzinfo=timezone.utc)
    document={"schema_version":"yovision.source-config/v1","config_id":"gate-primary","revision":revision,"published_at":now.isoformat().replace("+00:00","Z"),"effective_at":(now+timedelta(seconds=effective)).isoformat().replace("+00:00","Z"),"site":{"id":"site-east"},"logical_device":{"id":"camera-1"},"profile":{"id":"main","width":1920,"height":1080,"encoding":"H264","frame_rate":25},"media":{"ref":"media:site-east/camera-1/main","transport":"rtsp"},"rule_set":{"version":f"rules-{revision}","state":state,"profile_binding":{"profile_id":"main","width":1920,"height":1080},"areas":[{"id":"danger","version":1,"kind":"danger_area","enabled":True,"points":[{"x":.1,"y":.1},{"x":.8,"y":.1},{"x":.5,"y":.8}]}],"directional_lines":[]}}
    digest=hashlib.sha256(json.dumps(document,separators=(",",":"),sort_keys=True).encode()).hexdigest();document["integrity"]={"algorithm":"sha256","value":digest}
    return json.dumps(document,separators=(",",":"),sort_keys=True).encode()


def with_extension(body: bytes, extensions: object) -> bytes:
    document=json.loads(body);document["extensions"]=extensions;unsigned=dict(document);unsigned.pop("integrity");document["integrity"]["value"]=hashlib.sha256(json.dumps(unsigned,separators=(",",":"),sort_keys=True).encode()).hexdigest();return json.dumps(document,separators=(",",":"),sort_keys=True).encode()


def identity(tmp_path, now: int):
    private=Ed25519PrivateKey.generate();signer=Signer("yv:sense:east","sense-key-01",private,clock=lambda:now)
    registry=Registry([KeyRecord("yv:sense:east","sense-key-01",private.public_key(),"yovision-brain",frozenset({"source-config:write"}))])
    replay=SQLiteReplayStore(tmp_path/"replay.sqlite")
    return signer,Verifier(registry,replay,clock=lambda:now)


def test_authenticated_apply_is_idempotent_and_replay_survives_restart(tmp_path):
    now=int(datetime(2026,8,31,tzinfo=timezone.utc).timestamp());signer,verifier=identity(tmp_path,now);body=source_document();consumer=SourceConfigConsumer(tmp_path/"state.sqlite",clock=lambda:now);endpoint=SourceConfigEndpoint(consumer,verifier)
    token=signer.mint("yovision-brain",("source-config:write",),"POST","/machine/v1/source-config",body)
    result=endpoint.receive("Bearer "+token,body,"corr-request-0001");assert result.state=="applied" and result.config.rules is not None
    with pytest.raises(ValueError,match="machine_token_replayed"):endpoint.receive("Bearer "+token,body,"corr-request-0001")
    restarted=SourceConfigEndpoint(SourceConfigConsumer(tmp_path/"state.sqlite",clock=lambda:now),Verifier(verifier._registry,SQLiteReplayStore(tmp_path/"replay.sqlite"),clock=lambda:now))
    with pytest.raises(ValueError,match="machine_token_replayed"):restarted.receive("Bearer "+token,body,"corr-request-0002")
    new_token=signer.mint("yovision-brain",("source-config:write",),"POST","/machine/v1/source-config",body);assert restarted.receive("Bearer "+new_token,body,"corr-request-0003").state=="idempotent"


def test_atomic_replay_accepts_once_under_concurrency(tmp_path):
    store=SQLiteReplayStore(tmp_path/"atomic.sqlite");results=[]
    threads=[threading.Thread(target=lambda:results.append(store.consume("yv:sense:east","token-id",200,100))) for _ in range(12)]
    for thread in threads:thread.start()
    for thread in threads:thread.join()
    assert results.count(True)==1


def test_stale_unknown_profile_and_recalibration_are_safe(tmp_path):
    now=int(datetime(2026,8,31,tzinfo=timezone.utc).timestamp());consumer=SourceConfigConsumer(tmp_path/"state.sqlite",clock=lambda:now)
    assert consumer.apply(source_document(2)).config.rules is not None
    with pytest.raises(SourceConfigError,match="STALE_REVISION"):consumer.apply(source_document(1))
    invalid=json.loads(source_document(3));invalid["profile"]["width"]=1280;invalid["integrity"]["value"]="0"*64
    with pytest.raises(SourceConfigError,match="CONFIG_INVALID"):consumer.apply(json.dumps(invalid).encode())
    unknown=json.loads(source_document(3));unknown["schema_version"]="yovision.source-config/v2"
    with pytest.raises(SourceConfigError,match="UNSUPPORTED_SCHEMA_VERSION"):consumer.apply(json.dumps(unknown).encode())
    safe=consumer.apply(source_document(3,state="recalibration_required"));assert safe.config.rule_state=="recalibration_required" and safe.config.rules is None


def test_future_effective_snapshot_activates_atomically_after_restart(tmp_path):
    base=int(datetime(2026,8,31,tzinfo=timezone.utc).timestamp());current=[base];path=tmp_path/"state.sqlite";consumer=SourceConfigConsumer(path,clock=lambda:current[0])
    assert consumer.apply(source_document(1)).state=="applied";scheduled=consumer.apply(source_document(2,effective=60));assert scheduled.state=="scheduled" and scheduled.config.revision==1
    current[0]+=61;restarted=SourceConfigConsumer(path,clock=lambda:current[0]);activated=restarted.activate_due();assert activated[0].revision==2 and restarted.get_active("gate-primary").revision==2


def test_unknown_valid_extension_namespace_is_ignored(tmp_path):
    now=int(datetime(2026,8,31,tzinfo=timezone.utc).timestamp());consumer=SourceConfigConsumer(tmp_path/"state.sqlite",clock=lambda:now)
    result=consumer.apply(with_extension(source_document(),{"vendor.example":{"feature":"safe"}}));assert result.state=="applied" and result.config.revision==1


@pytest.mark.parametrize("extensions", [[], {"1invalid":{}}, {"vendor_ok":{}}, {"vendor.example":"not-an-object"}])
def test_invalid_extensions_are_rejected(tmp_path, extensions):
    now=int(datetime(2026,8,31,tzinfo=timezone.utc).timestamp());consumer=SourceConfigConsumer(tmp_path/"state.sqlite",clock=lambda:now)
    with pytest.raises(SourceConfigError,match="CONFIG_INVALID"):consumer.apply(with_extension(source_document(),extensions))


def test_status_sequence_restart_retry_timeout_and_disable(tmp_path):
    path=tmp_path/"status.sqlite";publisher=RuntimeStatusPublisher(path,"brain-east-01");health={"overall":"healthy","error_codes":[],"metrics":{"load_percent":1.0,"queue_depth":0,"latency_ms":2.0}}
    one=json.loads(publisher.build(runtime_state="running",runtime_version="1.0.0",started_at=datetime.now(timezone.utc),model_ref="people-detection",model_version="1",configurations=[],health=health,inputs=[]));two=json.loads(RuntimeStatusPublisher(path,"brain-east-01").build(runtime_state="running",runtime_version="1.0.0",started_at=None,model_ref="people-detection",model_version="1",configurations=[],health=health,inputs=[]));assert (one["sequence"],two["sequence"])==(0,1)
    private=Ed25519PrivateKey.generate();signer=Signer("yv:brain:east","brain-key-01",private,clock=lambda:1_787_000_000);attempts=[]
    def send(_auth,_body,_corr,timeout):attempts.append(timeout);raise TimeoutError
    sender=StatusSender(signer,send,max_attempts=3,sleeper=lambda _:None)
    with pytest.raises(RuntimeError,match="STATUS_DELIVERY_EXHAUSTED"):sender.publish(b"{}","corr-request-0001")
    assert attempts==[5.0,5.0,5.0]
    disabled=StatusSender(signer,lambda *_:ConnectorResponse(204,b"","corr-request-0001"),enabled=False)
    with pytest.raises(RuntimeError,match="CONNECTOR_DISABLED"):disabled.publish(b"{}","corr-request-0001")


@pytest.mark.parametrize("request_id", ["short", "0123456789abcde\n", "0123456789abcde!", "a"*129])
def test_connector_rejects_unsafe_request_ids(tmp_path, request_id):
    now=int(datetime(2026,8,31,tzinfo=timezone.utc).timestamp());signer,verifier=identity(tmp_path,now);body=source_document();endpoint=SourceConfigEndpoint(SourceConfigConsumer(tmp_path/"state.sqlite",clock=lambda:now),verifier)
    token=signer.mint("yovision-brain",("source-config:write",),"POST","/machine/v1/source-config",body)
    with pytest.raises(ValueError,match="INVALID_CORRELATION_ID"):endpoint.receive("Bearer "+token,body,request_id)
    status_signer=Signer("yv:brain:east","brain-key-01",Ed25519PrivateKey.generate(),clock=lambda:now)
    sender=StatusSender(status_signer,lambda *_:ConnectorResponse(204,b"",request_id))
    with pytest.raises(ValueError,match="INVALID_CORRELATION_ID"):sender.publish(b"{}",request_id)