inbound-library/inbound_library/registry.py

"""File-backed registries: topics, sources and subscriptions.

Consumers change these with a commit. `validate` is the gate that check runs.
"""
import json
from pathlib import Path

CHANNELS = {'hub', 'pull', 'digest'}
CADENCES = {'daily', 'weekly', 'irregular'}
VISIBILITY = {'shared', 'restricted'}


def _load(path):
    return json.loads(Path(path).read_text())


def load(root):
    base = Path(root) / 'registry'
    topics_file = base / 'topics.json'
    topics = _load(topics_file) if topics_file.exists() else {'version': 0, 'topics': {}}
    sources = {}
    for path in sorted((base / 'sources').glob('*.json')):
        source = _load(path)
        sources[source.get('id', path.stem)] = {**source, '_file': path.name}
    subscriptions = {}
    for path in sorted((base / 'subscriptions').glob('*.json')):
        subscription = _load(path)
        subscriptions[subscription.get('repo', path.stem)] = {**subscription, '_file': path.name}
    return {'topics': topics, 'sources': sources, 'subscriptions': subscriptions}


def validate(registry):
    problems = []
    known = set(registry['topics'].get('topics', {}))
    rules = {}
    for ident, source in registry['sources'].items():
        where = f"sources/{source['_file']}"
        if source['_file'] != f'{ident}.json':
            problems.append(f'{where}: file name must be <id>.json')
        for field in ('id', 'owner_repo', 'channel', 'topics', 'cadence', 'visibility'):
            if field not in source:
                problems.append(f'{where}: missing {field}')
        if source.get('cadence') not in CADENCES:
            problems.append(f'{where}: cadence must be one of {sorted(CADENCES)}')
        if source.get('visibility') not in VISIBILITY:
            problems.append(f'{where}: visibility must be shared or restricted')
        for topic in source.get('topics', []):
            if topic not in known:
                problems.append(f'{where}: unknown topic {topic}')
        if source.get('channel') == 'email':
            match = source.get('match', {})
            if not match.get('from'):
                problems.append(f'{where}: email source needs match.from')
            key = (match.get('from', '').lower(), match.get('subject', ''), match.get('list_id', ''))
            if key in rules:
                problems.append(f'{where}: duplicate match rule with {rules[key]}')
            rules[key] = ident
        elif source.get('channel') == 'url' and not source.get('url'):
            problems.append(f'{where}: url source needs url')
    for repo, subscription in registry['subscriptions'].items():
        where = f"subscriptions/{subscription['_file']}"
        if subscription['_file'] != f'{repo}.json':
            problems.append(f'{where}: file name must be <repo>.json')
        if not subscription.get('agent'):
            problems.append(f'{where}: missing agent')
        channels = set(subscription.get('channels', []))
        if not channels or not channels <= CHANNELS:
            problems.append(f'{where}: channels must be a non-empty subset of {sorted(CHANNELS)}')
        if 'digest' in channels and subscription.get('cadence') not in ('daily', 'weekly'):
            problems.append(f'{where}: digest channel needs cadence daily or weekly')
        for topic in subscription.get('topics', []):
            if topic not in known:
                problems.append(f'{where}: unknown topic {topic}')
        for ident in subscription.get('sources', []):
            source = registry['sources'].get(ident)
            if source is None:
                problems.append(f'{where}: unknown source {ident}')
            elif source.get('visibility') == 'restricted' and source.get('owner_repo') != repo:
                problems.append(f'{where}: restricted source {ident} belongs to {source["owner_repo"]}')
    return problems


def source_for_mail(registry, sender, subject='', list_id=''):
    """Return the email source whose match rule fits, or None (quarantine)."""
    sender = (sender or '').lower()
    for ident, source in registry['sources'].items():
        if source.get('channel') != 'email':
            continue
        match = source.get('match', {})
        if match.get('from', '').lower() != sender:
            continue
        if match.get('subject') and match['subject'].lower() not in (subject or '').lower():
            continue
        if match.get('list_id') and match['list_id'] != list_id:
            continue
        return ident
    return None


def visible_sources(registry, repo):
    """Sources a repo may see: shared ones, plus restricted ones it owns."""
    return {ident: source for ident, source in registry['sources'].items()
            if source.get('visibility') == 'shared' or source.get('owner_repo') == repo}


def subscribers_for(registry, source_id):
    """Repos whose subscription matches a source by id or by topic."""
    source = registry['sources'].get(source_id)
    if source is None:
        return []
    matched = []
    for repo, subscription in registry['subscriptions'].items():
        if source_id not in visible_sources(registry, repo):
            continue
        if source_id in subscription.get('sources', []) or \
                set(source.get('topics', [])) & set(subscription.get('topics', [])):
            matched.append(repo)
    return matched