import threading
import os
import sqlite3

import pytest

from src import live_snapshot_store
from src.live_snapshot_store import LiveSnapshotStore, RefreshLeaseLost
from src.state_encryption import PrivateStateEncryptionError


PRIVATE_KEY = b"l" * 32


def test_published_live_snapshot_is_encrypted_at_rest_and_survives_restart(tmp_path):
    path = tmp_path / "live.sqlite3"
    store = LiveSnapshotStore(path, clock=lambda: 100.0, encryption_key=PRIVATE_KEY)
    lease = store.try_acquire_refresh({"context"}, lease_seconds=5)
    canary = "private-live-title-canary"

    published = store.publish_refresh(
        lease,
        value={"context": {"title": canary}, "events": [], "notifications": []},
        created_at={section: 100.0 for section in live_snapshot_store.SECTIONS},
        failure_count={section: 0 for section in live_snapshot_store.SECTIONS},
        retry_at={section: None for section in live_snapshot_store.SECTIONS},
        changed_sections={"context"},
    )

    with sqlite3.connect(path) as connection:
        payload = connection.execute(
            "SELECT value_json FROM live_snapshot WHERE singleton = 1"
        ).fetchone()[0]
    assert payload.startswith("v1:")
    assert canary not in payload
    assert LiveSnapshotStore(path, clock=lambda: 100.0, encryption_key=PRIVATE_KEY).load().value == published.value


def test_live_snapshot_lazily_migrates_plaintext_without_changing_metadata(tmp_path):
    path = tmp_path / "live.sqlite3"
    store = LiveSnapshotStore(path, clock=lambda: 100.0, encryption_key=PRIVATE_KEY)
    legacy = {"context": {"title": "legacy"}, "events": [], "notifications": []}
    with sqlite3.connect(path) as connection:
        connection.execute(
            "UPDATE live_snapshot SET value_json = ?, revisions_json = ? WHERE singleton = 1",
            ('{"context":{"title":"legacy"},"events":[],"notifications":[]}',
             '{"context":4,"events":2,"notifications":1}'),
        )

    state = store.load()

    with sqlite3.connect(path) as connection:
        migrated = connection.execute(
            "SELECT value_json FROM live_snapshot WHERE singleton = 1"
        ).fetchone()[0]
    assert state.value == legacy
    assert state.revisions == {"context": 4, "events": 2, "notifications": 1}
    assert migrated.startswith("v1:")


def test_live_snapshot_authentication_failure_returns_no_private_content(tmp_path):
    path = tmp_path / "live.sqlite3"
    store = LiveSnapshotStore(path, clock=lambda: 100.0, encryption_key=PRIVATE_KEY)
    lease = store.try_acquire_refresh({"context"}, lease_seconds=5)
    metadata = {section: None for section in live_snapshot_store.SECTIONS}
    store.publish_refresh(
        lease,
        value={"context": {"title": "secret"}},
        created_at=metadata,
        failure_count={section: 0 for section in metadata},
        retry_at=metadata,
        changed_sections={"context"},
    )

    with pytest.raises(PrivateStateEncryptionError, match="private state could not be decrypted"):
        LiveSnapshotStore(path, encryption_key=b"x" * 32).load()


def test_metadata_load_does_not_retrieve_or_decode_snapshot_value(tmp_path):
    store = LiveSnapshotStore(tmp_path / "live.sqlite3", clock=lambda: 100.0)
    with sqlite3.connect(store.path) as connection:
        connection.execute(
            "UPDATE live_snapshot SET value_json = ? WHERE singleton = 1",
            ("not-json",),
        )

    metadata = store.load_metadata()

    assert metadata.created_at == {
        section: None for section in ("context", "events", "notifications")
    }
    assert metadata.revisions == {
        section: 0 for section in ("context", "events", "notifications")
    }
    with pytest.raises(PrivateStateEncryptionError):
        store.load()


def test_independent_stores_racing_for_refresh_have_one_lease_winner(tmp_path):
    path = tmp_path / "live.sqlite3"
    first = LiveSnapshotStore(path, clock=lambda: 100.0)
    second = LiveSnapshotStore(path, clock=lambda: 100.0)
    barrier = threading.Barrier(2)
    results = []

    def acquire(store):
        barrier.wait()
        results.append(store.try_acquire_refresh({"context", "events"}, lease_seconds=5))

    threads = [threading.Thread(target=acquire, args=(store,)) for store in (first, second)]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()

    assert sum(result is not None for result in results) == 1


def test_snapshot_and_shared_revisions_publish_atomically(tmp_path):
    path = tmp_path / "live.sqlite3"
    writer = LiveSnapshotStore(path, clock=lambda: 100.0)
    reader = LiveSnapshotStore(path, clock=lambda: 100.0)
    lease = writer.try_acquire_refresh(set(("context", "events", "notifications")), lease_seconds=5)
    assert lease is not None

    state = writer.publish_refresh(
        lease,
        value={
            "context": {"generation": 1},
            "events": [],
            "notifications": [],
            "sections": {section: "fresh" for section in ("context", "events", "notifications")},
        },
        created_at={section: 100.0 for section in ("context", "events", "notifications")},
        failure_count={section: 0 for section in ("context", "events", "notifications")},
        retry_at={section: None for section in ("context", "events", "notifications")},
        changed_sections=set(("context", "events", "notifications")),
    )

    observed = reader.load()
    assert observed.value == state.value
    assert observed.revisions == {"context": 1, "events": 1, "notifications": 1}
    assert observed.refreshing_sections == set()
    with pytest.raises(RefreshLeaseLost):
        writer.publish_refresh(
            lease,
            value={"context": {"generation": 2}},
            created_at=observed.created_at,
            failure_count=observed.failure_count,
            retry_at=observed.retry_at,
            changed_sections={"context"},
        )


def test_expired_refresh_lease_can_be_recovered(tmp_path):
    now = 100.0
    path = tmp_path / "live.sqlite3"
    abandoned = LiveSnapshotStore(path, clock=lambda: now)
    recovery = LiveSnapshotStore(path, clock=lambda: now)

    first = abandoned.try_acquire_refresh({"context"}, lease_seconds=5)
    assert first is not None
    assert recovery.try_acquire_refresh({"context"}, lease_seconds=5) is None
    now = 106.0
    second = recovery.try_acquire_refresh({"context"}, lease_seconds=5)

    assert second is not None
    assert second != first


def test_default_store_clock_uses_reboot_stable_wall_time(tmp_path, monkeypatch):
    epoch = 1_700_000_000.0
    monkeypatch.setattr(live_snapshot_store.time, "time", lambda: epoch)
    store = LiveSnapshotStore(tmp_path / "live.sqlite3")

    lease = store.try_acquire_refresh({"context"}, lease_seconds=5)

    assert lease is not None
    assert store.load().lease_expires_at == epoch + 5


def test_store_is_private_and_does_not_persist_upstream_token(tmp_path, monkeypatch):
    token = "gitea-super-secret-token-material"
    monkeypatch.setenv("GITEA_TOKEN", token)
    path = tmp_path / "private-state" / "live.sqlite3"

    LiveSnapshotStore(path)

    assert os.stat(path).st_mode & 0o777 == 0o600
    assert os.stat(path.parent).st_mode & 0o777 == 0o700
    assert token.encode() not in path.read_bytes()


def test_remove_notifications_filters_batch_with_one_revision_advance(tmp_path):
    store = LiveSnapshotStore(tmp_path / "live.sqlite3", clock=lambda: 100.0)
    lease = store.try_acquire_refresh({"notifications"}, lease_seconds=5)
    assert lease is not None
    metadata = {section: None for section in ("context", "events", "notifications")}
    published = store.publish_refresh(
        lease,
        value={
            "notifications": [{"id": 7}, {"id": 8}, {"id": 9}],
            "sections": {"notifications": "fresh"},
        },
        created_at=metadata,
        failure_count={section: 0 for section in metadata},
        retry_at=metadata,
        changed_sections={"notifications"},
    )

    removed = store.remove_notifications([7, 9, 7])

    assert removed.value["notifications"] == [{"id": 8}]
    assert removed.revisions["notifications"] == published.revisions["notifications"] + 1


def test_read_notification_filter_is_shared_with_future_publications(tmp_path):
    path = tmp_path / "live.sqlite3"
    first = LiveSnapshotStore(path, clock=lambda: 100.0)
    second = LiveSnapshotStore(path, clock=lambda: 100.0)
    lease = first.try_acquire_refresh({"notifications"}, lease_seconds=5)
    assert lease is not None
    metadata = {section: None for section in ("context", "events", "notifications")}
    first.publish_refresh(
        lease,
        value={"notifications": [{"id": 7}, {"id": 8}], "sections": {"notifications": "fresh"}},
        created_at=metadata,
        failure_count={section: 0 for section in metadata},
        retry_at=metadata,
        changed_sections={"notifications"},
    )

    removed = second.remove_notification(7)
    lease = first.try_acquire_refresh({"notifications"}, lease_seconds=5)
    assert lease is not None
    republished = first.publish_refresh(
        lease,
        value={"notifications": [{"id": 7}, {"id": 8}], "sections": {"notifications": "fresh"}},
        created_at=metadata,
        failure_count={section: 0 for section in metadata},
        retry_at=metadata,
        changed_sections={"notifications"},
    )

    assert removed.value["notifications"] == [{"id": 8}]
    assert republished.value["notifications"] == [{"id": 8}]