"""Trigger auto-creation of every pipeline topic before starting minions.

The Kafka brokers have `KAFKA_AUTO_CREATE_TOPICS_ENABLE=true`, so a producer
send to a non-existent topic creates it. AdminClient.createTopics on this
KRaft cluster is unstable (times out), so we intentionally use produce path.

Run once on a fresh cluster."""
import asyncio, os, sys
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from aiokafka import AIOKafkaProducer  # noqa: E402
from app.config import get_settings    # noqa: E402


TOPICS = [
    "topic_allocate_requested",
    "topic_asset_lookup_done",
    "topic_asset_upserted",
    "topic_asset_docs_uploaded",
    "topic_docs_accepted",
    "topic_allocate_completed",
    "topic_allocate_failed",
    "topic_allocate_blocked",
    "topic_lookup_dlq",
    "topic_upsert_dlq",
    "topic_file_upload_dlq",
    "topic_commit_dlq",
]


async def main() -> None:
    s = get_settings()
    # acks=1 keeps us away from the misconfigured transaction/replica quorum.
    # request_timeout_ms is generous because first send triggers topic creation.
    p = AIOKafkaProducer(
        bootstrap_servers=s.kafka_bootstrap_servers,
        acks=1,
        request_timeout_ms=30000,
        metadata_max_age_ms=1000,
    )
    await p.start()
    try:
        for attr in TOPICS:
            topic = getattr(s, attr, None)
            if not topic:
                print(f"  ! skip {attr}: not in settings")
                continue
            try:
                await asyncio.wait_for(
                    p.send_and_wait(topic, value=b'{"_bootstrap":true}', key=b'bootstrap'),
                    timeout=25,
                )
                print(f"  ✓ {topic}")
            except Exception as e:
                print(f"  ✗ {topic}: {type(e).__name__}: {e}")
    finally:
        await p.stop()

if __name__ == "__main__":
    asyncio.run(main())