#!/usr/bin/env python3
"""Recreate Test2 Integration from canonical environment and Runtime authority."""
from __future__ import annotations

import argparse
import base64
import json
import os
from pathlib import Path
import re
import ssl
import subprocess
import time
import urllib.error
import urllib.parse
import urllib.request

from prepare_test2_l0_state import load_object, render_env, write_private
from common import (
    activate_nginx_conf,
    load_env_file,
    provision_hubspot_private_app_named_credential_token,
    provision_hubspot_webhook_secret_values,
)
from hubspot_credential_admission import (
    enforce_hubspot_credential_admission,
    expected_nonprod_hubspot_identity,
    hubspot_account_identity,
    nonprod_runtime_environment,
)
from test2_standalone import ENV_ROOT, INTEGRATION_CONTAINER, docker, require_standalone

IMAGE_RE = r'ghcr\.io/pingcap-inc/nexuscrm-integration-service@sha256:[0-9a-f]{64}'
OWNER = 'io.nexuscrm.test2.component'
TEST2_NGINX_CONF = Path('/etc/nginx/conf.d/00-nexuscrm-test.conf')
CONNECTOR_SETTINGS = (
    'FLOW_DEADLINE_MS', 'INBOUND_MAX_PROCESSING_ROWS', 'INBOUND_QUEUE_PROFILE',
    'INBOUND_STAGE_TIMING', 'INBOUND_WORKERS_PER_PACK', 'RECEIPT_POLL_INTERVAL_MS',
    'RECEIPT_ROUTER_BATCH_SIZE', 'RECEIPT_ROUTER_PREPARE_CONCURRENCY',
    'RECEIPT_ROUTER_PREPARE_TIMEOUT_MS', 'WEBHOOK_ACK_DROP_ENDPOINTS',
    'WEBHOOK_ACK_DROP_TIMEOUT_MS', 'WEBHOOK_RETRY_AFTER_SECONDS',
    'WEBHOOK_VERIFICATION_LIMIT', 'WEBHOOK_VERIFICATION_WAIT_LIMIT',
    'WEBHOOK_VERIFICATION_WAIT_MS',
)


def inspect(name: str) -> dict | None:
    ids = docker('ps', '-aq', '--filter', f'name=^/{name}$').split()
    return json.loads(docker('inspect', ids[0]))[0] if ids else None


def require_image(image: str) -> None:
    if not re.fullmatch(IMAGE_RE, image):
        raise RuntimeError('deploy requires an immutable published Integration image')


def require_owner(item: dict) -> None:
    labels = item['Config'].get('Labels') or {}
    if not (labels.get(OWNER) == 'integration-service'
            or labels.get('io.nexuscrm.serviceification-d3') == 'test2'):
        raise RuntimeError('not an owned Test2 Integration container')


def stop_container(item: dict | None) -> None:
    if item is None:
        return
    require_owner(item)
    docker('stop', '--time', '30', item['Id'])
    state = json.loads(docker('inspect', item['Id']))[0]['State']
    if any(state.get(key) for key in ('Running', 'Restarting', 'Paused')):
        raise RuntimeError('Integration did not stop')


def required(values: dict, key: str) -> str:
    value = values.get(key)
    if not isinstance(value, str) or not value or value.strip() != value or any(c in value for c in '\n\r\0'):
        raise RuntimeError(f'missing or invalid Integration input: {key}')
    return value


def _nginx_server_block_bounds(content: str, server_name: str) -> tuple[int, int]:
    matches = []
    for match in re.finditer(r'(?m)^server\s*\{\s*$', content):
        depth = 0
        for index in range(match.start(), len(content)):
            if content[index] == '{':
                depth += 1
            elif content[index] == '}':
                depth -= 1
                if depth == 0:
                    end = index + 1
                    block = content[match.start():end]
                    names = re.findall(r'(?m)^\s*server_name\s+([^;]+);\s*$', block)
                    if any(server_name in value.split() for value in names):
                        matches.append((match.start(), end))
                    break
        else:
            raise RuntimeError('Test2 nginx contains an unterminated server block')
    if len(matches) != 1:
        raise RuntimeError(
            f'Test2 nginx must contain exactly one server block for {server_name}')
    return matches[0]


def render_public_integration_route(
    content: str, server_name: str, backend_port: str, integration_port: str = '3403',
) -> str:
    """Point the Test2 webhook route at its standalone Integration owner."""
    start, end = _nginx_server_block_bounds(content, server_name)
    block = content[start:end]
    route = re.compile(
        r'(location /webhooks/ \{\s*proxy_pass http://127\.0\.0\.1:)([0-9]+)(;)')
    matches = list(route.finditer(block))
    if len(matches) != 1 or matches[0].group(2) not in {backend_port, integration_port}:
        raise RuntimeError('Test2 nginx webhook route has unexpected authority')
    rendered = route.sub(r'\g<1>' + integration_port + r'\g<3>', block, count=1)
    return content[:start] + rendered + content[end:]


def preflight_public_integration_route(
    root: Path, nginx_path: Path = TEST2_NGINX_CONF,
) -> tuple[Path, str, str, str]:
    envv = load_env_file(root / 'env/runtime.env')
    public_origin = required(envv, 'PUBLIC_BASE_URL').rstrip('/')
    server_name = urllib.parse.urlsplit(public_origin).hostname or ''
    backend_port = required(envv, 'BACKEND_PORT')
    if not server_name or nginx_path.is_symlink() or not nginx_path.is_file():
        raise RuntimeError('Test2 public nginx authority is unavailable')
    # Validate the route authority before disrupting the running process. The
    # activation phase reads it again under the workflow's shared Test2 lock so
    # it cannot publish a stale snapshot of unrelated ingress changes.
    render_public_integration_route(nginx_path.read_text(), server_name, backend_port)
    return nginx_path, server_name, backend_port, public_origin


def activate_public_integration_route(
    prepared: tuple[Path, str, str, str], *, activate=activate_nginx_conf,
    probe=None,
) -> None:
    path, server_name, backend_port, public_origin = prepared
    rendered = render_public_integration_route(
        path.read_text(), server_name, backend_port)
    # Always reload the prepared configuration. A previous attempt may have
    # written the file and then failed before nginx accepted it; comparing only
    # on-disk bytes would make every retry skip the actual activation.
    activate(path, rendered, healthcheck_url=public_origin + '/health')
    (probe or probe_public_integration_route)(public_origin)


def probe_public_integration_route(public_origin: str) -> None:
    request = urllib.request.Request(
        public_origin + '/webhooks/hubspot-crm-connector/webhook', data=b'{}', method='POST')
    try:
        urllib.request.urlopen(request, timeout=10)
    except urllib.error.HTTPError as exc:
        if exc.code == 401:
            return
        raise RuntimeError(
            f'Test2 public webhook route returned unexpected status {exc.code}') from exc
    raise RuntimeError('Test2 public webhook route accepted an unsigned request')


def reconcile_hubspot_credential(
    root: Path,
    access_token: str,
    client_secret: str,
    *,
    identify=hubspot_account_identity,
    provision=provision_hubspot_private_app_named_credential_token,
    provision_webhooks=provision_hubspot_webhook_secret_values,
    admit=enforce_hubspot_credential_admission,
) -> None:
    """Verify provider identity, then install and admit all runtime credentials."""
    token = required({'TEST2_HUBSPOT_ACCESS_TOKEN': access_token}, 'TEST2_HUBSPOT_ACCESS_TOKEN')
    secret = required({'TEST2_HUBSPOT_CLIENT_SECRET': client_secret}, 'TEST2_HUBSPOT_CLIENT_SECRET')
    if len(secret) < 16:
        raise RuntimeError('TEST2_HUBSPOT_CLIENT_SECRET must contain at least 16 characters')
    envv = load_env_file(root / 'env/runtime.env')
    expected = expected_nonprod_hubspot_identity(nonprod_runtime_environment(envv))
    actual = identify(token)
    if actual != expected:
        raise RuntimeError(
            'TEST2_HUBSPOT_ACCESS_TOKEN identity does not match the committed Test2 policy')
    envv['EXP_HUBSPOT_ACCESS_TOKEN'] = token
    provision(envv)
    provision_webhooks(envv, secret)
    admit(envv)


def configuration(root: Path, address: str, runtime_target: str) -> tuple[bytes, bytes, Path]:
    """Read only authorities; never infer configuration from the old container."""
    source = load_env_file(root / 'env/runtime.env')
    public_origin = required(source, 'PUBLIC_BASE_URL').rstrip('/')
    try:
        parsed = urllib.parse.urlsplit(public_origin)
        valid_origin = (parsed.scheme == 'https' and bool(parsed.hostname)
                        and parsed.username is None and parsed.port != 0
                        and not parsed.netloc.endswith(':')
                        and not any(char.isspace() for char in public_origin)
                        and public_origin == f'https://{parsed.netloc}')
    except ValueError:
        valid_origin = False
    if not valid_origin:
        raise RuntimeError('PUBLIC_BASE_URL must be an HTTPS origin with a host and no userinfo, path, query or fragment')
    l0 = root / '.runtime/l0-standalone'
    serving = load_object(l0 / 'config/serving.json', 'Runtime serving config')
    identities = load_object(l0 / 'secrets/workload-identities.json', 'Runtime workload identities')
    if (serving.get('schema') != 'nexuscrm.runtime-api-serving/v4'
            or 'integration-service' not in serving.get('authorized_workload_ids', [])
            or identities.get('schema') != 'nexuscrm.runtime-workload-identities/v1'):
        raise RuntimeError('Runtime does not authorize Integration')
    tokens = [required(row, 'token') for row in identities.get('identities', [])
              if row.get('workloadId') == 'integration-service']
    if len(tokens) != 1:
        raise RuntimeError('Runtime must own one Integration workload token')
    ca = l0 / 'secrets/tls/server.crt'
    ssl.create_default_context(cafile=str(ca))
    values = {key: required(source, key) for key in (
        'TIDB_HOST', 'TIDB_PORT', 'TIDB_USER', 'TIDB_PASSWORD', 'TIDB_DATABASE',
        'NEXUSCRM_NC_MASTER_KEY', 'NEXUSCRM_CONNECTOR_VERIFIED_INGRESS_RECEIPTS',
    )}
    if len(base64.b64decode(values['NEXUSCRM_NC_MASTER_KEY'], validate=True)) != 32:
        raise RuntimeError('Integration requires an AES-256 named-credential key')
    if values['NEXUSCRM_CONNECTOR_VERIFIED_INGRESS_RECEIPTS'] != 'true':
        raise RuntimeError('Integration requires verified ingress receipts')
    # Never silently drop a provider configuration needing a declared secret mount.
    if any(source.get(key) for key in source if key.startswith('NEXUSCRM_INTEGRATION_CONNECTED_MAIL_')):
        raise RuntimeError('connected-mail configuration needs an explicit Test2 deployment contract')
    for key in ('TIDB_TLS_DISABLE', *('NEXUSCRM_CONNECTOR_' + suffix for suffix in CONNECTOR_SETTINGS)):
        if source.get(key):
            values[key] = required(source, key)
    values.update({
        'NEXUSCRM_DB_MAX_OPEN_CONNS': '8', 'NEXUSCRM_DB_MAX_IDLE_CONNS': '4',
        'WEBHOOK_BASE_URL': public_origin,
        'NEXUSCRM_INTEGRATION_HTTP_ADDR': address,
        'NEXUSCRM_INTEGRATION_ORG_ID': required(serving, 'org_id'),
        'NEXUSCRM_INTEGRATION_APP_INSTALLATION_ID': required(serving, 'app_installation_id'),
        'NEXUSCRM_INTEGRATION_AUTHORIZED_WORKLOAD_ID': 'integration-service',
        'NEXUSCRM_INTEGRATION_WORKLOAD_TOKEN_FILE': '/run/secrets/workload.token',
        'NEXUSCRM_INTEGRATION_RUNTIME_TARGET': runtime_target,
        'NEXUSCRM_INTEGRATION_RUNTIME_CA_FILE': '/run/trust/runtime-ca.crt',
        'NEXUSCRM_INTEGRATION_RUNTIME_SERVER_NAME': required(
            {'server_name': (l0 / 'config/runtime-server-name').read_text().strip()}, 'server_name'),
        'NEXUSCRM_INTEGRATION_RUNTIME_ATTEMPT_TIMEOUT': '4s',
        'NEXUSCRM_INTEGRATION_RUNTIME_STARTUP_TIMEOUT': '60s',
        'NEXUSCRM_INTEGRATION_SHUTDOWN_TIMEOUT': '20s',
        'NEXUSCRM_INTEGRATION_READ_HEADER_TIMEOUT': '10s',
        'NEXUSCRM_INTEGRATION_REQUEST_TIMEOUT': '30s',
        'NEXUSCRM_INTEGRATION_IDLE_TIMEOUT': '60s',
    })
    return render_env(values), (tokens[0] + '\n').encode(), ca


def preflight_container(image: str, root: Path, *, name: str = INTEGRATION_CONTAINER,
                        address: str = '127.0.0.1:3403',
                        runtime_target: str = '127.0.0.1:3312') -> tuple:
    """Resolve every local deployment input without changing files or processes."""
    selected = json.loads(docker('image', 'inspect', image))[0]
    config = selected['Config']
    if config.get('Entrypoint') != ['/usr/local/bin/integration-service'] or config.get('Cmd'):
        raise RuntimeError('unsupported Integration process command')
    revision = (config.get('Labels') or {}).get('org.opencontainers.image.revision', '')
    if not re.fullmatch('[0-9a-f]{40}', revision):
        raise RuntimeError('Integration image has no exact source identity')
    env, token, ca = configuration(root, address, runtime_target)
    state = root / '.runtime/integration'
    if state.is_symlink() or (state.exists() and not state.is_dir()):
        raise RuntimeError('invalid Integration configuration directory')
    for path in (state / 'integration.env', state / 'integration.token'):
        if path.is_symlink() or (path.exists() and not path.is_file()):
            raise RuntimeError('invalid Integration configuration output path')
    current = inspect(name)
    if current:
        require_owner(current)
        if any(m['Destination'] not in ('/run/secrets/workload.token', '/run/trust/runtime-ca.crt')
               for m in current['Mounts']):
            raise RuntimeError('existing Integration has undeclared mounts')
    return selected, revision, env, token, ca, state, current


def deploy_container(image: str, root: Path, *, name: str = INTEGRATION_CONTAINER,
                     address: str = '127.0.0.1:3403', runtime_target: str = '127.0.0.1:3312',
                     network: tuple[str, ...] = ('--network', 'host'),
                     readiness_url: str = 'http://127.0.0.1:3403/readyz', timeout: float = 120,
                     prepared: tuple | None = None) -> str:
    """Shared stop/recreate/readiness runner for production and local proof."""
    if prepared is None:
        prepared = preflight_container(
            image, root, name=name, address=address, runtime_target=runtime_target)
    selected, revision, env, token, ca, state, current = prepared
    stop_container(current)
    # Write after stopping, but before removal. A failed write leaves the old
    # container stopped; a later retry re-derives these files from authority.
    write_private(state / 'integration.env', env, mode=0o600)
    write_private(state / 'integration.token', token)
    if current:
        docker('rm', current['Id'])
    cid = docker('create', '--platform', 'linux/amd64', '--name', name,
                 *network, '--restart', 'unless-stopped', '--user', f'{os.getuid()}:{os.getgid()}',
                 '--label', OWNER + '=integration-service',
                 '--env-file', str(state / 'integration.env'),
                 '--mount', f'type=bind,src={state}/integration.token,dst=/run/secrets/workload.token,readonly',
                 '--mount', f'type=bind,src={ca},dst=/run/trust/runtime-ca.crt,readonly', image)
    try:
        docker('start', cid)
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            actual = json.loads(docker('inspect', cid))[0]
            if actual['State']['Running'] and actual['Image'] == selected['Id']:
                try:
                    with urllib.request.urlopen(readiness_url, timeout=3) as response:
                        if response.status == 200:
                            print(f'Integration ready: {image} (source {revision})')
                            return cid
                except OSError:
                    pass
            time.sleep(min(1, max(0, deadline - time.monotonic())))
        raise RuntimeError('Integration did not become ready; inspect the stopped process')
    except BaseException:
        stop_container(inspect(name))
        raise


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('operation', choices=('stop', 'deploy'))
    parser.add_argument('--image', default='')
    args = parser.parse_args()
    if args.operation == 'deploy':
        require_image(args.image)
    elif args.image:
        parser.error('image selection is only allowed for deploy')
    require_standalone(ENV_ROOT)
    current = inspect(INTEGRATION_CONTAINER)
    if current:
        require_owner(current)
        require_image(current['Config']['Image'])
    if args.operation == 'stop':
        stop_container(current)
    else:
        prepared = preflight_container(args.image, ENV_ROOT)
        public_route = preflight_public_integration_route(ENV_ROOT)
        reconcile_hubspot_credential(
            ENV_ROOT,
            os.environ.get('TEST2_HUBSPOT_ACCESS_TOKEN', ''),
            os.environ.get('TEST2_HUBSPOT_CLIENT_SECRET', ''),
        )
        deploy_container(args.image, ENV_ROOT, prepared=prepared)
        activate_public_integration_route(public_route)


if __name__ == '__main__':
    try:
        main()
    except subprocess.CalledProcessError as exc:
        raise SystemExit(f'Docker operation failed (exit {exc.returncode}); inspect Integration privately') from None