"""
Тесты портативного артефакта стартовой истории.
"""

from dataclasses import replace
from datetime import datetime, timezone
import hashlib
import json
import lzma

import pytest

from generator import GeneratorState


def test_plain_json_roundtrip_keeps_existing_serialization(tmp_path):
    """Обычный JSON по-прежнему пишется без сжатия и читается обратно."""
    from clickstream_generator.startup_history_artifact import (
        load_startup_history_artifact,
        write_startup_history_artifact,
    )

    artifact = {"text": "мир", "nested": {"value": 42}}
    path = tmp_path / "artifact.json"

    write_startup_history_artifact(path, artifact)

    assert path.read_text(encoding="utf-8") == json.dumps(
        artifact,
        ensure_ascii=True,
        indent=2,
    )
    assert load_startup_history_artifact(path) == artifact


def test_xz_artifact_roundtrip(tmp_path):
    """XZ-артефакт потоково пишется и читается без изменения данных."""
    from clickstream_generator.startup_history_artifact import (
        load_startup_history_artifact,
        write_startup_history_artifact,
    )

    artifact = {"text": "мир", "nested": {"value": 42}}
    path = tmp_path / "artifact.json.xz"

    write_startup_history_artifact(path, artifact)

    assert load_startup_history_artifact(path) == artifact
    with lzma.open(path, "rt", encoding="utf-8") as input_file:
        assert input_file.read() == json.dumps(
            artifact,
            ensure_ascii=True,
            indent=2,
        )


def _state() -> GeneratorState:
    import random

    now = datetime(2026, 1, 1, 1, 0, tzinfo=timezone.utc)
    return GeneratorState(
        tick=1,
        rng_state=random.Random(42).getstate(),
        last_batch_id="startup-history-abc",
        last_timestamp=now,
        model_timestamp=now,
        wall_timestamp=now,
        model_time_speed=1,
        model_timezone="UTC",
        model_t0=datetime(2026, 1, 1, 0, 0, tzinfo=timezone.utc),
        gen_seed=42,
        population=[
            {
                "user_domain_id": "user-1",
                "seed_click_id": "seed-1",
                "active_click_id": None,
                "last_finished_at": None,
            }
        ],
    )


def _batch() -> dict[str, list[dict]]:
    return {
        "browser_events": [
            {
                "event_id": "event-1",
                "click_id": "click-1",
                "user_domain_id": "user-1",
                "event_timestamp": "2026-01-01 00:00:00.000000",
            }
        ],
        "location_events": [
            {
                "event_id": "event-1",
                "page_url_path": "/home",
            }
        ],
        "device_events": [
            {
                "click_id": "click-1",
                "user_domain_id": "user-1",
            }
        ],
        "geo_events": [
            {
                "click_id": "click-1",
                "country": "RU",
            }
        ],
    }


def test_incremental_counters_equal_full_recompute():
    """Накопление по частям даёт те же числа, что полный пересчёт мира."""
    from clickstream_generator.startup_history_artifact import ManifestCounters

    first_day = _batch()
    second_day = {
        "browser_events": [
            {
                "event_id": "event-2",
                "click_id": "click-1",
                "user_domain_id": "user-1",
                "event_timestamp": "2026-01-02 00:00:00.000000",
            },
            {
                "event_id": "event-3",
                "click_id": "click-2",
                "user_domain_id": "user-2",
                "event_timestamp": "2026-01-02 00:01:00.000000",
            },
        ],
        "location_events": [
            {"event_id": "event-2", "page_url_path": "/cart"},
            {"event_id": "event-3", "page_url_path": "/home"},
        ],
        "device_events": [
            {"click_id": "click-1", "user_domain_id": "user-1"},
            {"click_id": "click-2", "user_domain_id": "user-2"},
        ],
        "geo_events": [
            {"click_id": "click-1", "country": "RU"},
            {"click_id": "click-2", "country": "KZ"},
        ],
    }

    incremental = ManifestCounters()
    incremental.add_batch(first_day)
    incremental = ManifestCounters.from_state(
        incremental.to_state(),
        known_click_ids={"click-1"},
        known_user_ids={"user-1"},
    )
    incremental.add_batch(second_day)

    full = ManifestCounters()
    full.add_batch({
        topic: first_day[topic] + second_day[topic]
        for topic in first_day
    })

    assert incremental.to_manifest_topics() == full.to_manifest_topics()
    assert incremental.to_manifest_totals() == full.to_manifest_totals()


def test_exact_id_sets_are_split_into_bounded_hash_chain(monkeypatch):
    """Manifest ссылается на цепочку малых точных фрагментов, а не растёт сам."""
    from clickstream_generator import startup_history_artifact as module

    monkeypatch.setattr(module, "ID_SET_CHUNK_SIZE", 2)
    counters = module.ManifestCounters()
    batch = _batch()
    batch["browser_events"].extend([
        {
            "event_id": "event-2",
            "click_id": "click-2",
            "event_timestamp": "2026-01-01 00:01:00.000000",
        },
        {
            "event_id": "event-3",
            "click_id": "click-3",
            "event_timestamp": "2026-01-01 00:02:00.000000",
        },
    ])
    counters.add_batch(batch)

    state, chunks = counters.to_state_with_chunks()

    assert len(chunks) == 2
    assert state["totals"] == {"visits": 3, "users": 1}
    assert state["id_sets"] == {
        "storage": "manifest-topic-sha256-chain-v1",
        "latest_sha256": chunks[-1][0],
        "chunks": 2,
        "click_ids": 3,
        "user_ids": 1,
    }
    assert chunks[0][1]["previous_sha256"] is None
    assert chunks[1][1]["previous_sha256"] == chunks[0][0]
    assert "click-1" not in json.dumps(state)

    restored = module.ManifestCounters.from_state(
        state,
        known_click_ids={"click-3"},
        known_user_ids={"user-1"},
    )
    restored.add_batch({
        "browser_events": [
            {
                "event_id": "event-4",
                "click_id": "click-3",
                "event_timestamp": "2026-01-02 00:00:00.000000",
            },
            {
                "event_id": "event-5",
                "click_id": "click-4",
                "event_timestamp": "2026-01-02 00:01:00.000000",
            },
        ],
        "location_events": [],
        "device_events": [{"click_id": "click-4", "user_domain_id": "user-2"}],
        "geo_events": [],
    })
    next_state, next_chunks = restored.to_state_with_chunks()

    assert next_state["totals"] == {"visits": 4, "users": 2}
    assert len(next_chunks) == 1
    assert next_chunks[0][1]["previous_sha256"] == chunks[-1][0]
    assert next_state["id_sets"]["chunks"] == 3


def test_counter_chunk_has_explicit_kafka_size_and_hash_guards():
    """Большой или неверно адресованный фрагмент падает до отправки в Kafka."""
    from clickstream_generator.kafka_io import (
        MAX_COMPACT_MESSAGE_BYTES,
        KafkaStartupHistoryManifest,
    )

    manager = object.__new__(KafkaStartupHistoryManifest)
    with pytest.raises(ValueError, match="безопасный предел"):
        manager._assert_message_size(
            {"payload": "x" * MAX_COMPACT_MESSAGE_BYTES},
            "фрагмент",
        )
    with pytest.raises(ValueError, match="хеш.*не совпадает"):
        manager.save_counter_chunk("0" * 64, {"version": "1.0"})

    assert manager.COUNTER_TOPIC != manager.MANIFEST_TOPIC


def test_artifact_roundtrip_keeps_events_state_and_manifest(base_config):
    """Артефакт хранит события, state и manifest одним проверяемым набором."""
    from clickstream_generator.startup_history_artifact import (
        StartupHistoryArtifactBuilder,
        build_manifest,
        load_startup_history_artifact,
        validate_startup_history_artifact,
        write_startup_history_artifact,
    )

    state = _state()
    builder = StartupHistoryArtifactBuilder()
    builder.add_batch(_batch())
    manifest = build_manifest(
        config=replace(
            base_config,
            model_t0=state.model_t0,
            model_t_end=state.model_timestamp,
            model_time_speed=1,
            model_timezone="UTC",
            seed=42,
        ),
        counters=builder.counters,
        state=state,
    )

    artifact = builder.to_artifact(manifest=manifest, state=state)
    (
        loaded_state,
        loaded_manifest,
        loaded_topics,
        loaded_raw_topics,
    ) = validate_startup_history_artifact(
        artifact,
        expected_config=replace(
            base_config,
            model_t0=state.model_t0,
            model_t_end=state.model_timestamp,
            model_time_speed=1,
            model_timezone="UTC",
            seed=42,
        ),
    )

    assert artifact["artifact_version"] == "1.0"
    assert loaded_state.last_batch_id == state.last_batch_id
    assert loaded_manifest["state"]["last_batch_id"] == state.last_batch_id
    assert loaded_topics["browser_events"][0]["event_id"] == "event-1"
    assert loaded_raw_topics["browser_events"][0]["value_json"].encode("utf-8") == (
        b'{"event_id": "event-1", "click_id": "click-1", '
        b'"user_domain_id": "user-1", '
        b'"event_timestamp": "2026-01-01 00:00:00.000000"}'
    )

    path = base_config.data_dir.parent / "tmp-startup-history-artifact.json"
    try:
        write_startup_history_artifact(path, artifact)
        loaded = load_startup_history_artifact(path)
        assert loaded["manifest"]["totals"]["events"] == 1
    finally:
        path.unlink(missing_ok=True)


def test_legacy_artifact_checksum_remains_valid(base_config):
    """Эталонный артефакт со старой суммой не требует пересборки."""
    from clickstream_generator.startup_history_artifact import (
        StartupHistoryArtifactBuilder,
        build_manifest,
        validate_startup_history_artifact,
    )

    state = _state()
    config = replace(
        base_config,
        model_t0=state.model_t0,
        model_t_end=state.model_timestamp,
        model_time_speed=1,
        model_timezone="UTC",
        seed=42,
    )
    builder = StartupHistoryArtifactBuilder()
    builder.add_batch(_batch())
    manifest = build_manifest(config, builder.counters, state)
    for topic, events in builder.topics.items():
        checksum = hashlib.sha256()
        for event in events:
            checksum.update(
                json.dumps(event, sort_keys=True, ensure_ascii=True).encode("utf-8")
            )
        manifest["topics"][topic]["checksum_sha256"] = checksum.hexdigest()

    artifact = builder.to_artifact(manifest=manifest, state=state)

    _, loaded_manifest, _, _ = validate_startup_history_artifact(
        artifact,
        expected_config=config,
    )
    assert loaded_manifest["topics"] == artifact["manifest"]["topics"]


def test_manifest_and_artifact_show_launch_profile(base_config):
    """Manifest и файл артефакта показывают выбранный профиль запуска."""
    from clickstream_generator.startup_history_artifact import (
        StartupHistoryArtifactBuilder,
        build_manifest,
        load_startup_history_artifact,
        write_startup_history_artifact,
    )

    state = _state()
    builder = StartupHistoryArtifactBuilder()
    builder.add_batch(_batch())
    manifest = build_manifest(
        config=replace(
            base_config,
            model_t0=state.model_t0,
            model_t_end=state.model_timestamp,
            model_time_speed=1,
            model_timezone="UTC",
            seed=42,
            launch_profile="daily-wave",
        ),
        counters=builder.counters,
        state=state,
    )
    artifact = builder.to_artifact(manifest=manifest, state=state)

    path = base_config.data_dir.parent / "tmp-startup-history-profile.json"
    try:
        write_startup_history_artifact(path, artifact)
        loaded = load_startup_history_artifact(path)
        assert manifest["launch_profile"] == "daily-wave"
        assert loaded["manifest"]["launch_profile"] == "daily-wave"
    finally:
        path.unlink(missing_ok=True)


def test_manifest_records_initial_boundary_chain(base_config):
    """Новый manifest явно хранит начало и правую границу истории."""
    from clickstream_generator.startup_history_artifact import (
        StartupHistoryArtifactBuilder,
        build_manifest,
        manifest_boundaries,
    )

    state = _state()
    builder = StartupHistoryArtifactBuilder()
    builder.add_batch(_batch())
    manifest = build_manifest(
        config=replace(
            base_config,
            model_t0=state.model_t0,
            model_t_end=state.model_timestamp,
            model_time_speed=1,
            model_timezone="UTC",
            seed=42,
        ),
        counters=builder.counters,
        state=state,
    )

    assert manifest["boundaries"] == [
        "2026-01-01T00:00:00+00:00",
        "2026-01-01T01:00:00+00:00",
    ]
    assert manifest_boundaries(manifest) == manifest["boundaries"]


def test_legacy_manifest_without_boundaries_gets_endpoint_chain():
    """Старый manifest без boundaries читается как пара [T0, T_end]."""
    from clickstream_generator.startup_history_artifact import manifest_boundaries

    manifest = {
        "model_t0": "2026-01-01T00:00:00+00:00",
        "model_t_end": "2026-01-03T00:00:00+00:00",
    }

    assert manifest_boundaries(manifest) == [
        "2026-01-01T00:00:00+00:00",
        "2026-01-03T00:00:00+00:00",
    ]


def test_manifest_rejects_boundary_chain_with_wrong_endpoint():
    """Цепочка не может расходиться с текущим model_t_end."""
    from clickstream_generator.startup_history_artifact import manifest_boundaries

    manifest = {
        "model_t0": "2026-01-01T00:00:00+00:00",
        "model_t_end": "2026-01-03T00:00:00+00:00",
        "boundaries": [
            "2026-01-01T00:00:00+00:00",
            "2026-01-02T00:00:00+00:00",
        ],
    }

    with pytest.raises(ValueError, match="model_t_end"):
        manifest_boundaries(manifest)


def test_build_manifest_rejects_explicit_empty_boundaries(base_config):
    """Явно пустая цепочка не подменяется границами по умолчанию."""
    from clickstream_generator.startup_history_artifact import (
        StartupHistoryArtifactBuilder,
        build_manifest,
    )

    state = _state()
    builder = StartupHistoryArtifactBuilder()
    builder.add_batch(_batch())
    config = replace(
        base_config,
        model_t0=state.model_t0,
        model_t_end=state.model_timestamp,
        model_time_speed=1,
        model_timezone="UTC",
        seed=42,
    )

    with pytest.raises(ValueError, match="T0 и T_end"):
        build_manifest(config, builder.counters, state, boundaries=[])


def test_artifact_validation_rejects_mismatched_manifest(base_config):
    """Валидация отвергает артефакт, где manifest не совпадает с событиями."""
    from clickstream_generator.startup_history_artifact import (
        StartupHistoryArtifactBuilder,
        build_manifest,
        validate_startup_history_artifact,
    )

    state = _state()
    config = replace(
        base_config,
        model_t0=state.model_t0,
        model_t_end=state.model_timestamp,
        model_time_speed=1,
        model_timezone="UTC",
        seed=42,
    )
    builder = StartupHistoryArtifactBuilder()
    builder.add_batch(_batch())
    manifest = build_manifest(config=config, counters=builder.counters, state=state)
    artifact = builder.to_artifact(manifest=manifest, state=state)
    artifact["manifest"]["gen_seed"] = 43

    with pytest.raises(ValueError, match="GEN_SEED"):
        validate_startup_history_artifact(artifact, expected_config=config)


def test_manifest_clickhouse_stats_detect_mismatch():
    """Сверка ClickHouse с manifest ловит расхождение контрольных чисел."""
    from clickstream_generator.startup_history_artifact import (
        compare_clickhouse_stats_to_manifest,
    )

    manifest = {
        "totals": {
            "events": 10,
            "visits": 4,
            "users": 3,
            "min_event_timestamp": "2026-01-01 00:00:00.000000",
            "max_event_timestamp": "2026-01-01 00:10:00.000000",
        }
    }
    stats = {
        "events": "9",
        "visits": "4",
        "users": "3",
        "min_event_timestamp": "2026-01-01 00:00:00.000000",
        "max_event_timestamp": "2026-01-01 00:10:00.000000",
    }

    assert compare_clickhouse_stats_to_manifest(manifest, stats) == ["events"]


def test_raw_topic_value_can_keep_original_json_bytes(base_config):
    """Raw value проверяется как JSON, но импортируется без пересборки строки."""
    from clickstream_generator.startup_history_artifact import (
        StartupHistoryArtifactBuilder,
        build_manifest,
        import_startup_history_artifact,
    )

    state = _state()
    config = replace(
        base_config,
        model_t0=state.model_t0,
        model_t_end=state.model_timestamp,
        model_time_speed=1,
        model_timezone="UTC",
        seed=42,
    )
    builder = StartupHistoryArtifactBuilder()
    builder.add_batch(_batch())
    artifact = builder.to_artifact(
        manifest=build_manifest(config=config, counters=builder.counters, state=state),
        state=state,
    )
    assert "cumulative_manifest_counters" not in artifact["state"]
    assert "cumulative_manifest_counters" not in artifact["manifest"]
    raw_value = (
        '{"event_timestamp":"2026-01-01 00:00:00.000000",'
        '"user_domain_id":"user-1","click_id":"click-1","event_id":"event-1"}'
    )
    artifact["raw_topics"]["browser_events"][0]["value_json"] = raw_value

    class Publisher:
        def __init__(self):
            self.records = None

        def publish_records(self, topic, records):
            if topic == "browser_events":
                self.records = records
            return len(records), 0

        def flush(self):
            return None

    class CompactWriter:
        def __init__(self):
            self.chunks = []

        def save(self, value):
            self.saved = value

        def save_counter_chunk(self, chunk_sha256, chunk):
            self.chunks.append((chunk_sha256, chunk))

        def flush(self):
            return None

    publisher = Publisher()
    import_startup_history_artifact(
        artifact,
        publisher=publisher,
        state_manager=CompactWriter(),
        manifest_manager=CompactWriter(),
        expected_config=config,
    )

    assert publisher.records[0]["value_json"] == raw_value


def test_artifact_v1_requires_raw_topics_for_import(base_config):
    """Artifact v1.0 без raw value не может обещать byte-for-byte импорт."""
    from clickstream_generator.startup_history_artifact import (
        StartupHistoryArtifactBuilder,
        build_manifest,
        import_startup_history_artifact,
    )

    state = _state()
    config = replace(
        base_config,
        model_t0=state.model_t0,
        model_t_end=state.model_timestamp,
        model_time_speed=1,
        model_timezone="UTC",
        seed=42,
    )
    builder = StartupHistoryArtifactBuilder()
    builder.add_batch(_batch())
    artifact = builder.to_artifact(
        manifest=build_manifest(config=config, counters=builder.counters, state=state),
        state=state,
    )
    del artifact["raw_topics"]

    class Publisher:
        def publish_records(self, topic, records):
            raise AssertionError("publish must not be called")

    with pytest.raises(ValueError, match="raw_topics"):
        import_startup_history_artifact(
            artifact,
            publisher=Publisher(),
            state_manager=None,
            manifest_manager=None,
            expected_config=config,
        )


def test_import_rejects_publisher_without_raw_publish_records(base_config):
    """Импорт не должен пересобирать dict, если publisher не умеет raw records."""
    from clickstream_generator.startup_history_artifact import (
        StartupHistoryArtifactBuilder,
        build_manifest,
        import_startup_history_artifact,
    )

    state = _state()
    config = replace(
        base_config,
        model_t0=state.model_t0,
        model_t_end=state.model_timestamp,
        model_time_speed=1,
        model_timezone="UTC",
        seed=42,
    )
    builder = StartupHistoryArtifactBuilder()
    builder.add_batch(_batch())
    artifact = builder.to_artifact(
        manifest=build_manifest(config=config, counters=builder.counters, state=state),
        state=state,
    )
    artifact["raw_topics"]["browser_events"][0]["value_json"] = (
        '{"event_timestamp":"2026-01-01 00:00:00.000000",'
        '"user_domain_id":"user-1","click_id":"click-1","event_id":"event-1"}'
    )

    class DictPublisher:
        def publish(self, topic, events):
            raise AssertionError("dict publish must not be called")

        def flush(self):
            return None

    with pytest.raises(TypeError, match="publish_records"):
        import_startup_history_artifact(
            artifact,
            publisher=DictPublisher(),
            state_manager=None,
            manifest_manager=None,
            expected_config=config,
        )


def test_import_rejects_dirty_kafka_topics_before_publish(base_config):
    """Повторный импорт не должен дописывать дубли в непустые data-топики."""
    from clickstream_generator.startup_history_artifact import (
        StartupHistoryArtifactBuilder,
        build_manifest,
        import_startup_history_artifact,
    )

    state = _state()
    config = replace(
        base_config,
        model_t0=state.model_t0,
        model_t_end=state.model_timestamp,
        model_time_speed=1,
        model_timezone="UTC",
        seed=42,
    )
    builder = StartupHistoryArtifactBuilder()
    builder.add_batch(_batch())
    artifact = builder.to_artifact(
        manifest=build_manifest(config=config, counters=builder.counters, state=state),
        state=state,
    )

    class Publisher:
        def publish(self, topic, events):
            raise AssertionError("publish must not be called")

    class DirtyTopics:
        def assert_data_topics_empty(self):
            raise RuntimeError("Kafka data topics are not empty")

    with pytest.raises(RuntimeError, match="not empty"):
        import_startup_history_artifact(
            artifact,
            publisher=Publisher(),
            state_manager=None,
            manifest_manager=None,
            expected_config=config,
            topic_inspector=DirtyTopics(),
        )


def test_import_rolls_back_kafka_topics_after_partial_publish(base_config):
    """При частичной публикации импорт откатывает Kafka-топики до повтора."""
    from clickstream_generator.startup_history_artifact import (
        StartupHistoryArtifactBuilder,
        build_manifest,
        import_startup_history_artifact,
    )

    state = _state()
    config = replace(
        base_config,
        model_t0=state.model_t0,
        model_t_end=state.model_timestamp,
        model_time_speed=1,
        model_timezone="UTC",
        seed=42,
    )
    builder = StartupHistoryArtifactBuilder()
    builder.add_batch(_batch())
    artifact = builder.to_artifact(
        manifest=build_manifest(config=config, counters=builder.counters, state=state),
        state=state,
    )

    class PartialPublisher:
        def __init__(self):
            self.calls = []

        def publish_records(self, topic, records):
            self.calls.append(topic)
            if topic == "location_events":
                raise RuntimeError("location down")
            return len(records), 0

    class CompactWriter:
        def save(self, value):
            raise AssertionError("compact topics must not be written")

    class Inspector:
        def __init__(self):
            self.snapshot = None
            self.rollback = None

        def assert_data_topics_empty(self):
            return None

        def snapshot_import_topics(self):
            self.snapshot = {"topics": ["browser_events", "location_events"]}
            return self.snapshot

        def rollback_import_topics(self, snapshot):
            self.rollback = snapshot

    publisher = PartialPublisher()
    inspector = Inspector()

    with pytest.raises(RuntimeError, match="location down"):
        import_startup_history_artifact(
            artifact,
            publisher=publisher,
            state_manager=CompactWriter(),
            manifest_manager=CompactWriter(),
            expected_config=config,
            topic_inspector=inspector,
        )

    assert publisher.calls == ["browser_events", "location_events"]
    assert inspector.rollback == inspector.snapshot


def test_import_replays_events_and_compact_topics(base_config):
    """Импорт пишет события в Kafka и служебные compact-топики, не трогая ClickHouse."""
    from clickstream_generator.startup_history_artifact import (
        ManifestCounters,
        StartupHistoryArtifactBuilder,
        build_manifest,
        cumulative_counter_reference,
        import_startup_history_artifact,
    )

    state = _state()
    config = replace(
        base_config,
        model_t0=state.model_t0,
        model_t_end=state.model_timestamp,
        model_time_speed=1,
        model_timezone="UTC",
        seed=42,
    )
    builder = StartupHistoryArtifactBuilder()
    builder.add_batch(_batch())
    artifact = builder.to_artifact(
        manifest=build_manifest(config=config, counters=builder.counters, state=state),
        state=state,
    )

    class Publisher:
        def __init__(self):
            self.calls = []
            self.flushed = False

        def publish_records(self, topic, records):
            self.calls.append((topic, records))
            return len(records), 0

        def flush(self):
            self.flushed = True

    class CompactWriter:
        def __init__(self):
            self.saved = None
            self.flushed = False
            self.chunks = []

        def save(self, value):
            self.saved = value

        def save_counter_chunk(self, chunk_sha256, chunk):
            self.chunks.append((chunk_sha256, chunk))

        def flush(self):
            self.flushed = True

    publisher = Publisher()
    state_manager = CompactWriter()
    manifest_manager = CompactWriter()

    result = import_startup_history_artifact(
        artifact,
        publisher=publisher,
        state_manager=state_manager,
        manifest_manager=manifest_manager,
        expected_config=config,
    )

    assert [topic for topic, _ in publisher.calls] == [
        "browser_events",
        "location_events",
        "device_events",
        "geo_events",
    ]
    assert publisher.calls[0][1][0]["value_json"].encode("utf-8") == (
        b'{"event_id": "event-1", "click_id": "click-1", '
        b'"user_domain_id": "user-1", '
        b'"event_timestamp": "2026-01-01 00:00:00.000000"}'
    )
    assert result["events"] == 4
    assert state_manager.saved.last_batch_id == state.last_batch_id
    assert state_manager.saved.cumulative_manifest_counters is not None
    assert manifest_manager.saved["state"]["last_batch_id"] == state.last_batch_id
    assert state_manager.saved.cumulative_manifest_counters == (
        cumulative_counter_reference(
            manifest_manager.saved["cumulative_manifest_counters"]
        )
    )
    seeded = ManifestCounters.from_state(
        manifest_manager.saved["cumulative_manifest_counters"],
        known_click_ids={"click-1"},
        known_user_ids={"user-1"},
    )
    assert seeded.click_ids == {"click-1"}
    assert seeded.user_ids == {"user-1"}
    assert len(manifest_manager.chunks) == 1
    assert (
        manifest_manager.saved["cumulative_manifest_counters"]["id_sets"][
            "latest_sha256"
        ]
        == manifest_manager.chunks[0][0]
    )
    assert publisher.flushed
    assert state_manager.flushed
    assert manifest_manager.flushed