import json,sqlite3,threading
from pathlib import Path
class Outbox:
    def __init__(self,path,limit=1000000):
        Path(path).parent.mkdir(parents=True,exist_ok=True)
        self.db=sqlite3.connect(path,check_same_thread=False)
        self.db.execute('PRAGMA journal_mode=WAL');self.db.execute('PRAGMA synchronous=FULL')
        self.db.execute('CREATE TABLE IF NOT EXISTS outbox (seq INTEGER PRIMARY KEY AUTOINCREMENT, id TEXT UNIQUE, body TEXT NOT NULL)')
        self.db.commit();self.lock=threading.Lock();self.limit=limit
    def add(self,event):
        with self.lock:
            if self.db.execute('SELECT COUNT(*) FROM outbox').fetchone()[0]>=self.limit:
                raise RuntimeError('Outbox full: collector stopped to avoid silent data loss')
            with self.db:self.db.execute('INSERT OR IGNORE INTO outbox(id,body) VALUES (?,?)',(event['id'],json.dumps(event,allow_nan=False)))
    def batch(self,limit=200):
        with self.lock:return [(r[0],json.loads(r[1])) for r in self.db.execute('SELECT seq,body FROM outbox ORDER BY seq LIMIT ?',(limit,))]
    def ack(self,ids):
        with self.lock,self.db:self.db.executemany('DELETE FROM outbox WHERE seq=?',[(i,) for i in ids])
    def count(self):
        with self.lock:return self.db.execute('SELECT COUNT(*) FROM outbox').fetchone()[0]
    def close(self): self.db.close()