"""Harmless queue probe. Requires an external TLS Redis broker in real use."""

from datetime import datetime, timezone
import json
import os
import re
import ssl
import time
from urllib.parse import parse_qs, urlsplit

from celery import Celery


QUEUE = 'daemon-probe'


def checked_broker_url(value):
    if not value:
        raise ValueError('CELERY_BROKER_URL is required')
    try:
        parsed = urlsplit(value)
        valid = parsed.scheme == 'rediss' and bool(parsed.hostname) and not parsed.fragment
        # A URL query can override connection options; do not accept weaker TLS.
        cert_mode = parse_qs(parsed.query, keep_blank_values=True).get('ssl_cert_reqs')
        if not valid or (cert_mode is not None and cert_mode != ['required']):
            raise ValueError
        parsed.port  # Validate the optional numeric port without displaying the URL.
    except ValueError:
        raise ValueError('Use a valid rediss URL with certificate verification') from None
    return value


app = Celery('daemon_probe', broker=checked_broker_url(os.environ.get('CELERY_BROKER_URL')))
app.conf.update(
    broker_use_ssl={'ssl_cert_reqs': ssl.CERT_REQUIRED},
    broker_connection_retry_on_startup=True,
    broker_transport_options={'visibility_timeout': 120},
    result_backend_transport_options={'visibility_timeout': 120},
    visibility_timeout=120,
    task_default_queue=QUEUE,
    task_serializer='json',
    accept_content=['json'],
    task_ignore_result=True,
    task_acks_late=True,
    task_reject_on_worker_lost=True,
    worker_prefetch_multiplier=1,
)


def validate_probe(job_id, seconds):
    if not isinstance(job_id, str) or re.fullmatch(r'[0-9a-f]{32}', job_id) is None:
        raise ValueError('job_id must be a lowercase UUID hex value')
    if type(seconds) is not int or not 0 <= seconds <= 90:
        raise ValueError('seconds must be an integer from 0 to 90')


@app.task(bind=True, name='daemon_probe.check')
def probe(self, job_id, seconds=2):
    validate_probe(job_id, seconds)
    delivery = self.request.delivery_info or {}
    record = {
        'task_id': self.request.id,
        'job_id': job_id,
        'worker': self.request.hostname,
        'redelivered': delivery.get('redelivered'),
        'retries': self.request.retries,
    }
    for stage in ('started', 'completed'):
        if stage == 'completed':
            time.sleep(seconds)
        print(json.dumps({**record, 'event': stage,
                          'at': datetime.now(timezone.utc).isoformat()}), flush=True)
    # No email, database writes, or result backend. Logs are the remote evidence.
    return {'job_id': job_id, 'completed': True}