"""Leased local integration outbox. Owner composition supplies bounded transports.

Unknown adapters leave rows pending; hub resolutions remain pending while the
write-back flag is off. Dispatch failure never rolls back a human disposition.
Audit/notification receivers must deduplicate using the stable delivery id.
"""
import json
import time
import uuid

from .hub_writeback import Answer, enabled, write_back
from .audit import OutboxWorker


class IntakeDeliveryWorker:
    def __init__(self, store, *, audit=None, notify=None, hub_transport=None,
                 hub_base_url='', env=None, clock=time.time):
        self.store, self.audit, self.notify = store, audit, notify
        self.hub_transport, self.hub_base_url = hub_transport, hub_base_url
        self.env, self.clock = env, clock

    def run_once(self, limit=10):
        if type(limit) is not int or not 1 <= limit <= 100:
            raise ValueError('delivery limit must be 1..100')
        delivered = 0
        if self.audit:
            self.store.stage_intake_audit()
            OutboxWorker(self.store, self.audit).run_once(limit=limit)
            delivered += self.store.settle_intake_audit()
        kinds = []
        if self.notify:
            kinds.append('notification')
        if self.hub_transport and enabled(self.env):
            kinds.append('hub-resolution')
        if not kinds:
            return delivered
        for _ in range(limit):
            now, lease = self.clock(), str(uuid.uuid4())
            with self.store._transaction() as db:
                placeholders = ','.join('?' for kind in kinds)
                row = db.execute(f"SELECT * FROM intake_deliveries WHERE kind IN ({placeholders}) "
                    "AND state IN ('pending','in_flight') AND next_attempt<=? "
                    "AND (lease_until IS NULL OR lease_until<=?) ORDER BY rowid LIMIT 1", (*kinds, now, now)).fetchone()
                if row is None:
                    break
                db.execute("UPDATE intake_deliveries SET state='in_flight',lease=?,lease_until=?,attempts=attempts+1 WHERE id=?",
                           (lease, now + 30, row['id']))
            state = 'pending'
            try:
                payload = json.loads(row['payload'])
                if row['kind'] == 'notification':
                    # notify(id, metadata) returns only after a deduplicated receiver receipt.
                    if self.notify(row['id'], payload) is True:
                        state = 'delivered'
                else:
                    result = write_back(Answer(**payload['answer']), payload['serves'],
                                        base_url=self.hub_base_url, transport=self.hub_transport, env=self.env)
                    if result.status in ('resolved', 'already_resolved'):
                        state = 'delivered'
                    elif result.status == 'conflict':
                        state = 'blocked'
            except Exception:
                pass  # Never persist upstream bodies, tokens, or exception text.
            with self.store._transaction() as db:
                db.execute("UPDATE intake_deliveries SET state=?,lease=NULL,lease_until=NULL,next_attempt=? "
                           "WHERE id=? AND lease=?", (state, now + min(3600, 2 ** min(row['attempts'] + 1, 11)), row['id'], lease))
            delivered += state == 'delivered'
        return delivered