"""Idempotent daily/weekly source capture and digest jobs, with durable run health."""
from datetime import datetime, timedelta
import fcntl
import json
from pathlib import Path
from zoneinfo import ZoneInfo

from .ingest import ingest
from .store import atomic, put


def run_due(root, config, now=None, runner=ingest):
    from .__main__ import draft
    root = Path(root)
    zone = ZoneInfo(config['timezone'])
    now = now or datetime.now(zone)
    if now.tzinfo is None:
        raise ValueError('scheduler time requires a timezone')
    local = now.astimezone(zone)
    state_dir = root / '.runtime'
    state_dir.mkdir(parents=True, exist_ok=True)
    with (state_dir / 'scheduler.lock').open('a') as lock:
        try:
            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError:
            return {'ok': True, 'status': 'already_running', 'jobs': []}
        state_path = state_dir / 'scheduler.json'
        state = json.loads(state_path.read_text()) if state_path.exists() else {}
        results = []
        for job in config['jobs']:
            hour, minute = (int(x) for x in job['at'].split(':'))
            if not (0 <= hour < 24 and 0 <= minute < 60):
                raise ValueError('invalid schedule time')
            due = local.replace(hour=hour, minute=minute, second=0, microsecond=0)
            if job['cadence'] == 'daily':
                if due > local:
                    due -= timedelta(days=1)
                period = due.date().isoformat()
            elif job['cadence'] == 'weekly':
                due -= timedelta(days=(due.weekday()-job['weekday']) % 7)
                if due > local:
                    due -= timedelta(days=7)
                year, week, _ = due.isocalendar()
                period = f'{year}-W{week:02}'
            else:
                raise ValueError('unsupported cadence')
            key = job['id'] + ':' + period
            if state.get(key, {}).get('ok'):
                continue
            try:
                registry_path = (root / job['sources']).resolve()
                if not registry_path.is_relative_to(root.resolve()):
                    raise ValueError('source registry escapes repository')
                registry = json.loads(registry_path.read_text())
                result = runner(root, registry)
                if not result['ok']:
                    raise ValueError('one or more sources failed; see ingestion-runs')
                # Capture-only changes remain unreviewed. Do not fabricate factual events.
                try:
                    output = draft(root, period)
                    digest_status = 'created'
                except FileExistsError:
                    output = root / 'digests' / job['cadence'] / f'{period}.md'
                    digest_status = 'existing_preserved'
                entry = {'ok': True, 'finished_at': local.isoformat(), 'period': period,
                         'digest': str(output.relative_to(root)), 'digest_status': digest_status}
            except (ValueError, OSError, KeyError) as exc:
                entry = {'ok': False, 'finished_at': local.isoformat(), 'period': period,
                         'error': str(exc)}
            state[key] = entry
            put(root, 'scheduler-runs', {'job_id': job['id'], **entry})
            atomic(state_path, json.dumps(state, indent=2).encode() + b'\n')
            results.append({'job_id': job['id'], **entry})
        return {'ok': all(r['ok'] for r in results), 'status': 'checked', 'jobs': results}