# consumer/run.py
# Streaming consumer that enriches orders with user_country and writes
# to the warehouse. Auto-commit was disabled weeks ago (see git blame on
# this file) but offsets still don't survive a rebalance.

from consumer.client import KafkaConsumer
from consumer.warehouse import write_enriched
from consumer.enrich import enrich

BATCH_SIZE = 500


def run(consumer: KafkaConsumer, max_messages: int) -> int:
    processed = 0
    batch = []
    for msg in consumer:
        batch.append(msg)
        if len(batch) >= BATCH_SIZE:
            enriched = [enrich(m) for m in batch]
            write_enriched(enriched)
            consumer.commit()
            processed += len(batch)
            batch = []
        if processed >= max_messages:
            break
    if batch:
        enriched = [enrich(m) for m in batch]
        write_enriched(enriched)
        consumer.commit()
        processed += len(batch)
    return processed