inbound-library/inbound_library/notify.py

"""Matching, idempotent notices and their three channels.

A notice is an immutable record {repo, delivery, item, source}. Its content is
deterministic, so matching the same delivery twice yields the same notice id.
Hub delivery state is kept separately, so a hub outage leaves notices pending.
"""
import json
import urllib.error
import urllib.request
from datetime import date, datetime, timedelta, timezone
from pathlib import Path

from . import registry as reg
from . import store

HUB_STATE = Path('data') / 'state' / 'hub-sent.json'


def match(root, registry=None):
    """Create notices for every routed delivery and matching subscriber."""
    registry = registry or reg.load(root)
    existing = {(n['repo'], n['delivery']) for n in store.records(root, 'notices')}
    created = []
    directory = store.data_dir(root) / 'deliveries'
    for path in sorted(directory.glob('*.json')) if directory.is_dir() else []:
        delivery = store.get_record(root, 'deliveries', path.stem)
        if delivery['status'] != 'routed' or delivery.get('channel') == 'attachment':
            continue
        for repo in reg.subscribers_for(registry, delivery['source']):
            if (repo, path.stem) in existing:
                continue
            created.append(store.put_record(root, 'notices', {
                'schema': store.SCHEMA, 'repo': repo, 'delivery': path.stem,
                'item': delivery['item'], 'source': delivery['source'],
                'received_at': delivery.get('received_at')}))
    return created


def notices_for(root, repo, topic=None, since=None, registry=None):
    registry = registry or reg.load(root)
    found = []
    for notice in store.records(root, 'notices'):
        if notice['repo'] != repo:
            continue
        source = registry['sources'].get(notice['source'], {})
        if topic and topic not in source.get('topics', []):
            continue
        if since and (notice.get('received_at') or '') < since:
            continue
        derived = store.get_derived(root, notice['item']) or {}
        found.append({**notice, 'topics': source.get('topics', []),
                      'subject': derived.get('subject', '')})
    return sorted(found, key=lambda n: n.get('received_at') or '')


def _notice_ids(root):
    directory = store.data_dir(root) / 'notices'
    return {p.stem: store.get_record(root, 'notices', p.stem)
            for p in sorted(directory.glob('*.json'))} if directory.is_dir() else {}


def pending_hub(root, registry=None):
    registry = registry or reg.load(root)
    path = Path(root) / HUB_STATE
    sent = set(json.loads(path.read_text())) if path.exists() else set()
    pending = {}
    for ident, notice in _notice_ids(root).items():
        subscription = registry['subscriptions'].get(notice['repo'], {})
        if 'hub' in subscription.get('channels', []) and ident not in sent:
            pending.setdefault(notice['repo'], []).append(ident)
    return pending, sent


def post_message(api_base, payload, timeout=15):
    request = urllib.request.Request(
        api_base.rstrip('/') + '/messages/', data=json.dumps(payload).encode(),
        headers={'Content-Type': 'application/json'}, method='POST')
    with urllib.request.urlopen(request, timeout=timeout) as response:
        if response.status >= 300:
            raise OSError(f'hub returned {response.status}')


def deliver_hub(root, api_base, from_agent='inbound-library', registry=None, post=post_message):
    """Send one message per subscriber with pending notices. Failures stay pending."""
    registry = registry or reg.load(root)
    pending, sent = pending_hub(root, registry)
    notices = _notice_ids(root)
    report = {}
    for repo, idents in sorted(pending.items()):
        subscription = registry['subscriptions'][repo]
        lines = []
        for ident in idents:
            notice = notices[ident]
            subject = (store.get_derived(root, notice['item']) or {}).get('subject', '')
            lines.append(f"- {notice['source']}: {subject or '(no subject)'}  item {notice['item'][:12]}")
        cli = f'cd {Path(root).resolve()} && python3 -m inbound_library'
        body = (f'{len(idents)} new item(s) for {repo}:\n' + '\n'.join(lines) +
                f'\n\nList: {cli} new --repo {repo}\n'
                f'Read: {cli} show <item>   Original bytes: {cli} get <item>')
        try:
            post(api_base, {'from_agent': from_agent, 'to_agent': subscription['agent'],
                            'subject': f'inbound-library: {len(idents)} new item(s)',
                            'body': body})
        except (OSError, urllib.error.URLError, ValueError) as exc:
            report[repo] = {'ok': False, 'pending': len(idents), 'error': str(exc)}
            continue
        sent.update(idents)
        store.atomic(Path(root) / HUB_STATE, json.dumps(sorted(sent), indent=2).encode() + b'\n')
        report[repo] = {'ok': True, 'sent': len(idents)}
    return report


def period_of(moment, cadence):
    day = moment.date() if isinstance(moment, datetime) else moment
    if cadence == 'weekly':
        year, week, _ = day.isocalendar()
        return f'{year}-W{week:02}'
    return day.isoformat()


def write_digests(root, now=None, registry=None):
    """Write the current-period digest per digest subscriber. Digests are derived."""
    registry = registry or reg.load(root)
    now = now or datetime.now(timezone.utc)
    written = []
    for repo, subscription in registry['subscriptions'].items():
        if 'digest' not in subscription.get('channels', []):
            continue
        cadence = subscription['cadence']
        period = period_of(now, cadence)
        start = now.date() - timedelta(days=now.weekday()) if cadence == 'weekly' else now.date()
        notices = notices_for(root, repo, since=start.isoformat(), registry=registry)
        lines = [f'# inbound-library digest — {repo} — {period}', '',
                 'Unreviewed material routed by topic. Review belongs to the consumer.', '']
        if not notices:
            lines.append('No new items in this period.')
        for notice in notices:
            lines.append(f"- **{notice['subject'] or '(no subject)'}** — {notice['source']} "
                         f"({', '.join(notice['topics'])}), received {notice.get('received_at')}, "
                         f"item `{notice['item']}`")
        path = Path(root) / 'data' / 'digests' / repo / cadence / f'{period}.md'
        store.atomic(path, ('\n'.join(lines) + '\n').encode())
        written.append(str(path.relative_to(root)))
    return written


SILENCE = {'daily': timedelta(hours=36), 'weekly': timedelta(days=9)}


def health(root, now=None, registry=None):
    registry = registry or reg.load(root)
    now = now or datetime.now(timezone.utc)
    deliveries = store.records(root, 'deliveries')
    last_seen, counts = {}, {}
    routed_items = {d['item'] for d in deliveries if d['status'] == 'routed'}
    quarantine = {d['item'] for d in deliveries
                  if d['status'] == 'quarantined' and d['item'] not in routed_items}
    for delivery in deliveries:
        if delivery['status'] != 'routed' or delivery.get('channel') == 'attachment':
            continue
        counts[delivery['source']] = counts.get(delivery['source'], 0) + 1
        stamp = delivery.get('received_at') or ''
        last_seen[delivery['source']] = max(last_seen.get(delivery['source'], ''), stamp)
    silent = []
    for ident, source in registry['sources'].items():
        window = SILENCE.get(source.get('cadence'))
        if source.get('channel') != 'email' or window is None:
            continue
        seen = last_seen.get(ident)
        if not seen or datetime.fromisoformat(seen) < now - window:
            silent.append({'source': ident, 'last_seen': seen})
    cursor_path = Path(root) / 'data' / 'state' / 'imap-cursor.json'
    cursor = json.loads(cursor_path.read_text()) if cursor_path.exists() else {}
    pending, _ = pending_hub(root, registry)
    return {'last_scan': cursor.get('scanned_at'), 'items_per_source': counts,
            'silent_sources': silent, 'quarantined_items': len(quarantine),
            'pending_hub_notices': sum(len(v) for v in pending.values()),
            'integrity_problems': store.verify(root)}