Nexus/server/application/services/sync_service.py

"""Nexus — Sync Service (Batch push engine with rsync + Agent coordination)
Application layer — orchestrates Server Repository + asyncssh pool + Redis progress + WebSocket broadcast.

W3: Migrated from paramiko pool to asyncssh pool (consistent with sync_engine_v2.py).
"""

import asyncio
import json
import logging
import shlex
from datetime import datetime, timezone
from typing import Optional, List, Dict

from server.domain.models import Server, SyncLog, AuditLog
from server.domain.repositories import (
    ServerRepository, SyncLogRepository, AuditLogRepository,
    PushRetryJobRepository,
)
from server.infrastructure.ssh.asyncssh_pool import exec_ssh_command
from server.infrastructure.redis.client import get_redis

logger = logging.getLogger("nexus.sync_service")

# Batch push configuration
DEFAULT_BATCH_SIZE = 50       # Servers per batch
DEFAULT_CONCURRENCY = 10      # Max concurrent SSH per batch
DEFAULT_BATCH_INTERVAL = 30   # Seconds between batches
DEFAULT_TIMEOUT = 300         # 5 minutes per server


class SyncService:
    """Business logic for batch file push with rsync and progress tracking"""

    def __init__(
        self,
        server_repo: ServerRepository,
        sync_log_repo: SyncLogRepository,
        audit_repo: AuditLogRepository,
        retry_repo: PushRetryJobRepository,
    ):
        self.server_repo = server_repo
        self.sync_log_repo = sync_log_repo
        self.audit_repo = audit_repo
        self.retry_repo = retry_repo

    async def batch_push(
        self,
        server_ids: List[int],
        source_path: str,
        sync_mode: str = "incremental",
        operator: str = "admin",
        batch_size: int = DEFAULT_BATCH_SIZE,
        concurrency: int = DEFAULT_CONCURRENCY,
        batch_interval: int = DEFAULT_BATCH_INTERVAL,
    ) -> Dict[int, SyncLog]:
        """Push files to multiple servers in batches with concurrency control.

        Flow:
        1. Divide server_ids into batches (50-100 per batch)
        2. Each batch: up to 10 concurrent rsync connections
        3. Batch interval: 30 seconds between batches
        4. Failed servers: add to retry queue (max 100 retries)
        5. Progress tracked in Redis for WebSocket broadcast
        """
        results: Dict[int, SyncLog] = {}
        servers = []

        # Resolve all servers
        for sid in server_ids:
            server = await self.server_repo.get_by_id(sid)
            if server:
                servers.append(server)
            else:
                logger.warning(f"Server ID {sid} not found, skipping")

        # Divide into batches
        batches = [servers[i:i + batch_size] for i in range(0, len(servers), batch_size)]
        total = len(servers)
        completed = 0

        redis = get_redis()

        # Initialize progress in Redis (TTL 1 hour — auto-cleanup after push completes)
        progress_key = f"sync:progress:{operator}:{int(asyncio.get_event_loop().time())}"
        progress_data = json.dumps({
            "total": total,
            "completed": 0,
            "failed": 0,
            "current_batch": 0,
            "total_batches": len(batches),
            "status": "running",
        })
        await redis.set(progress_key, progress_data, ex=3600)

        for batch_idx, batch in enumerate(batches):
            logger.info(f"Batch {batch_idx + 1}/{len(batches)}: pushing to {len(batch)} servers")

            # Update progress
            await redis.set(progress_key, json.dumps({
                    "total": total,
                    "completed": completed,
                    "failed": sum(1 for r in results.values() if r.status == "failed"),
                    "current_batch": batch_idx + 1,
                    "total_batches": len(batches),
                    "status": "running",
                }), ex=3600)

            # Execute batch with concurrency control
            semaphore = asyncio.Semaphore(concurrency)
            batch_tasks = []

            async def _push_to_server(server: Server):
                async with semaphore:
                    log = await self._push_single(server, source_path, sync_mode, operator)
                    results[server.id] = log
                    if log.status == "success":
                        completed += 1
                    elif log.status == "failed":
                        # Add to retry queue
                        await self._add_retry(server, source_path, operator)

            for server in batch:
                batch_tasks.append(_push_to_server(server))

            await asyncio.gather(*batch_tasks)

            # Batch interval (pause between batches)
            if batch_idx < len(batches) - 1 and batch_interval > 0:
                logger.info(f"Batch {batch_idx + 1} complete, waiting {batch_interval}s before next batch")
                await asyncio.sleep(batch_interval)

        # Final progress update (shorter TTL — data is complete)
        await redis.set(progress_key, json.dumps({
            "total": total,
            "completed": completed,
            "failed": sum(1 for r in results.values() if r.status == "failed"),
            "current_batch": len(batches),
            "total_batches": len(batches),
            "status": "completed",
        }), ex=600)

        await self._audit(
            "batch_push", "sync", 0,
            f"Pushed {source_path} to {total} servers: {completed} success, {total - completed} failed",
        )

        return results

    async def _push_single(
        self, server: Server, source_path: str, sync_mode: str, operator: str,
    ) -> SyncLog:
        """Push files to a single server via rsync-over-SSH (asyncssh pool)

        W3: Uses asyncssh pool instead of paramiko. rsync command built with shlex.quote().
        """
        target_path = server.target_path or "/tmp/nexus-sync"

        # Build rsync flags based on sync_mode
        rsync_flags = self._rsync_flags(sync_mode)

        # Create sync log
        log = SyncLog(
            server_id=server.id,
            source_path=source_path,
            target_path=target_path,
            trigger_type="batch",
            operator=operator,
            status="running",
            sync_mode=sync_mode,
        )
        log = await self.sync_log_repo.create(log)

        start_time = datetime.now(timezone.utc)

        try:
            # Build rsync command with safe quoting (W3: shlex.quote prevents injection)
            rsync_cmd = (
                f"rsync {rsync_flags} "
                f"-e 'ssh -p {server.port} -o StrictHostKeyChecking=no' "
                f"{shlex.quote(source_path)}/ "
                f"{shlex.quote(server.username or 'root')}@{shlex.quote(server.domain)}:{shlex.quote(target_path)}/"
            )

            # Execute rsync via asyncssh pool
            result = await exec_ssh_command(server, rsync_cmd, timeout=DEFAULT_TIMEOUT)

            duration = int((datetime.now(timezone.utc) - start_time).total_seconds())

            # Parse rsync output for file statistics
            stdout = result.get("stdout", "")
            files_transferred = self._parse_rsync_transferred(stdout)

            if result.get("exit_code", -1) == 0:
                log = await self.sync_log_repo.update_status(
                    log.id, "success",
                    files_transferred=files_transferred,
                    duration_seconds=duration,
                    diff_summary=stdout[:2000],
                    finished_at=datetime.now(timezone.utc),
                )
            else:
                log = await self.sync_log_repo.update_status(
                    log.id, "failed",
                    error_message=result.get("stderr", "Unknown error")[:2000],
                    duration_seconds=duration,
                    finished_at=datetime.now(timezone.utc),
                )

        except Exception as e:
            logger.error(f"Push to server {server.id} ({server.name}) failed: {e}")
            log = await self.sync_log_repo.update_status(
                log.id, "failed",
                error_message=str(e)[:2000],
                finished_at=datetime.now(timezone.utc),
            )

        return log

    async def get_sync_logs(self, server_id: int, limit: int = 50) -> List[SyncLog]:
        return await self.sync_log_repo.get_by_server_id(server_id, limit)

    async def _add_retry(self, server: Server, source_path: str, operator: str):
        """Add failed push to retry queue"""
        if not self.retry_repo:
            return
        from server.domain.models import PushRetryJob
        from datetime import timedelta

        job = PushRetryJob(
            server_id=server.id,
            server_name=server.name,
            operator=operator,
            source_path=source_path,
            target_path=server.target_path or "/tmp/nexus-sync",
            status="pending",
            next_retry_at=datetime.now(timezone.utc) + timedelta(minutes=5),
        )
        await self.retry_repo.create(job)

    # ── Private helpers ──

    def _rsync_flags(self, sync_mode: str) -> str:
        """Build rsync flags based on sync mode"""
        base = "-avz"
        mode_flags = {
            "incremental": base,                          # Default: only sync modified files
            "full": f"{base} --delete",                    # Sync + delete extraneous files
            "overwrite": f"{base} --inplace",              # Direct overwrite without temp files
            "checksum": f"{base} --checksum",              # Checksum-based comparison
        }
        return mode_flags.get(sync_mode, base)

    def _parse_rsync_transferred(self, stdout: str) -> int:
        """Parse rsync output to count transferred files"""
        for line in stdout.split("\n"):
            if "Number of regular files transferred:" in line:
                try:
                    return int(line.split(":")[1].strip())
                except (ValueError, IndexError):
                    pass
        return 0

    async def _audit(self, action: str, target_type: str, target_id: int, detail: str):
        if not self.audit_repo:
            return
        log = AuditLog(action=action, target_type=target_type, target_id=target_id, detail=detail)
        await self.audit_repo.create(log)