"""Sequential safe-offset consumer with local SQLite inbox and confirmed DLQ."""
import argparse
import base64
import json
from pathlib import Path
import sqlite3
import time


def commit_confirmed(consumer, *, message=None, offsets=None):
    """Raise unless the synchronous response confirms every requested offset."""
    if (message is None) == (offsets is None):
        raise ValueError("Specify exactly one of message and offsets")
    if message is not None:
        expected = [(message.topic(), message.partition(), message.offset() + 1)]
        results = consumer.commit(message=message, asynchronous=False)
    else:
        expected = [(item.topic, item.partition, item.offset) for item in offsets]
        if not expected:
            raise ValueError("Specify at least one offset")
        results = consumer.commit(offsets=offsets, asynchronous=False)

    if not results:
        raise RuntimeError("Kafka offset commit returned no confirmation")
    failures = [
        f"{item.topic}[{item.partition}]: {item.error}"
        for item in results if item.error is not None
    ]
    if failures:
        raise RuntimeError("Kafka offset commit failed: " + "; ".join(failures))
    actual = [(item.topic, item.partition, item.offset) for item in results]
    if sorted(actual) != sorted(expected):
        raise RuntimeError("Kafka offset commit response does not match the requested offsets")

def publish_quarantine(producer,message,reason,topic):
    payload={"source_topic":message.topic(),"partition":message.partition(),
             "offset":message.offset(),"error":type(reason).__name__,
             "raw_base64":base64.b64encode(message.value() or b"").decode()}
    errors=[]; acknowledged=[]
    def done(error,record):
        if error:errors.append(str(error))
        else:acknowledged.append(True)
    key=f"{message.topic()}:{message.partition()}:{message.offset()}"
    producer.produce(topic,key=key,value=json.dumps(payload),on_delivery=done)
    pending=producer.flush(30)
    if pending or errors or not acknowledged:
        raise RuntimeError("DLQ outcome not confirmed; source offset stays uncommitted")

def main():
    ap=argparse.ArgumentParser()
    ap.add_argument("--brokers",default="localhost:19092")
    ap.add_argument("--topic",default="handbook.events")
    ap.add_argument("--group",default="handbook-projection-v1")
    ap.add_argument("--database",type=Path,default=Path(__file__).parent/"output/projection.sqlite")
    ap.add_argument("--max-records",type=int,default=5)
    ap.add_argument("--dlq",help="Enable owned quarantine; duplicates by source coordinate possible")
    ap.add_argument("--crash-after-effect",action="store_true")
    a=ap.parse_args()
    from confluent_kafka import Consumer, Producer, KafkaError
    from jsonschema import ValidationError
    from projection import apply_event, open_projection

    a.database.parent.mkdir(parents=True,exist_ok=True)
    con=open_projection(a.database)
    consumer=Consumer({"bootstrap.servers":a.brokers,"group.id":a.group,
       "group.protocol":"classic","enable.auto.commit":False,
       "enable.auto.offset.store":False,"auto.offset.reset":"earliest",
       "isolation.level":"read_committed","max.poll.interval.ms":120000})
    dlq=Producer({"bootstrap.servers":a.brokers,"enable.idempotence":True,
                  "acks":"all","delivery.timeout.ms":30000}) if a.dlq else None
    consumer.subscribe([a.topic])
    count=0; idle_deadline=time.monotonic()+60
    try:
        while count<a.max_records:
            message=consumer.poll(1)
            if message is None:
                if time.monotonic()>idle_deadline:raise TimeoutError("No required records; inspect group/topic")
                continue
            if message.error():
                if message.error().code()==KafkaError._PARTITION_EOF:continue
                raise RuntimeError(str(message.error()))
            idle_deadline=time.monotonic()+60
            try:
                event=json.loads(message.value())
                for attempt in range(3):
                    try:
                        apply_event(con,event)
                        break
                    except sqlite3.OperationalError:
                        if attempt==2:raise
                        time.sleep(0.2*(2**attempt))
                if a.crash_after_effect:
                    raise RuntimeError("Injected stop after durable effect before offset commit")
            except (ValueError,UnicodeDecodeError,ValidationError) as error:
                if dlq is None:raise
                publish_quarantine(dlq,message,error,a.dlq)
            # One record at a time; never commit beyond an unresolved earlier
            # record in this partition. Rebalance commit rejection is fatal here:
            # already committed DB effect remains safe on replay.
            commit_confirmed(consumer, message=message)
            count+=1
        print(json.dumps({"records_disposed":count,"totals":con.execute(
            "SELECT * FROM totals ORDER BY tenant_id,currency").fetchall()}))
    finally:
        consumer.close()
        con.close()

if __name__=="__main__":main()