Logging and Data Files

self.data_dir: str = os.environ.get("WEBHOOK_SERVER_DATA_DIR", "/home/podman/data")
self.config_path: str = os.path.join(self.data_dir, "config.yaml")

volumes:
  - "./webhook_server_data_dir:/home/podman/data:Z" # Should include config.yaml and webhook-server.private-key.pem
  # Mount temporary directories to prevent boot ID mismatch issues
  - "/tmp/podman-storage-${USER:-1000}:/tmp/storage-run-1000"
environment:
  - PUID=1000
  - PGID=1000
  - TZ=Asia/Jerusalem
  - MAX_WORKERS=50 # Defaults to 10 if not set
  - WEBHOOK_SERVER_IP_BIND=0.0.0.0 # IP to listen
  - WEBHOOK_SERVER_PORT=5000 # Port to listen
  - WEBHOOK_SECRET=<secret> # If set verify hook is a valid hook from Github
  - VERIFY_GITHUB_IPS=1 # Verify hook request is from GitHub IPs
  - VERIFY_CLOUDFLARE_IPS=1 # Verify hook request is from Cloudflare IPs
  - ENABLE_LOG_SERVER=true # Enable log viewer endpoints (default: false)
  - ENABLE_MCP_SERVER=false # Enable MCP server for AI agent integration (default: false)

def get_repository_github_app_api(config_: Config, repository_name: str) -> Github | None:
    LOGGER.debug("Getting repositories GitHub app API")

    with open(os.path.join(config_.data_dir, "webhook-server.private-key.pem")) as fd:
        private_key = fd.read()

    github_app_id: int = config_.root_data["github-app-id"]

log-level: INFO # Set global log level, change take effect immediately without server restart
log-file: webhook-server.log # Set global log file, change take effect immediately without server restart
mcp-log-file: mcp_server.log # Set global MCP log file, change take effect immediately without server restart
logs-server-log-file: logs_server.log # Set global Logs Server log file, change take effect immediately without server restart
mask-sensitive-data: true # Mask sensitive data in logs (default: true). Set to false for debugging (NOT recommended in production)

repositories:
  my-repository:
    name: my-org/my-repository
    log-level: DEBUG # Override global log-level for repository
    log-file: my-repository.log # Override global log-file for repository
    mask-sensitive-data: false # Override global setting - disable masking for debugging this specific repo (NOT recommended in production)

def get_log_file_path(config: Config, log_file_name: str | None) -> str | None:
    """
    Resolve the full path for a log file using the configuration data directory.

    Args:
        config: Config object containing data_dir
        log_file_name: Name of the log file (e.g., "server.log")

    Returns:
        Full path to the log file, or None if log_file_name is None
    """
    if log_file_name and not log_file_name.startswith("/"):
        log_file_path = os.path.join(config.data_dir, "logs")

        if not os.path.isdir(log_file_path):
            os.makedirs(log_file_path, exist_ok=True)

        return os.path.join(log_file_path, log_file_name)

    return log_file_name

# Patch simple_logger to use SafeRotatingFileHandler to prevent crashes
# when backup log files are missing during rollover
simple_logger.logger.RotatingFileHandler = SafeRotatingFileHandler

logger = get_logger(
    name=logger_cache_key,
    filename=log_file_path_resolved,
    level=log_level,
    file_max_bytes=1024 * 1024 * 10,
    mask_sensitive=mask_sensitive,
    mask_sensitive_patterns=mask_sensitive_patterns,
    console=True,  # Enable console output for docker logs with FORCE_COLOR support
)

if self.backupCount > 0:
    # Remove backup files that exceed backupCount, handle missing files
    for i in range(self.backupCount - 1, 0, -1):
        sfn = self.rotation_filename(f"{self.baseFilename}.{i}")
        dfn = self.rotation_filename(f"{self.baseFilename}.{i + 1}")
        if os.path.exists(sfn):
            try:
                if os.path.exists(dfn):
                    os.remove(dfn)
                os.rename(sfn, dfn)
            except FileNotFoundError:
                # File was deleted between exists check and operation - ignore
                pass
            except OSError:
                # Broad suppression intentional: logging must never crash.
                # See module docstring for full rationale.
                pass

    dfn = self.rotation_filename(f"{self.baseFilename}.1")
    try:
        if os.path.exists(dfn):
            os.remove(dfn)
    except FileNotFoundError:
        # File was deleted between exists check and remove - ignore
        pass
    except OSError:
        # Broad suppression intentional: logging must never crash.
        # See module docstring for full rationale.
        pass

    try:
        self.rotate(self.baseFilename, dfn)
    except FileNotFoundError:
        # Base file was deleted - just create a new one
        pass
    except OSError:
        # Broad suppression intentional: logging must never crash.
        # See module docstring for full rationale.
        pass

if not self.delay:
    try:
        self.stream = self._open()
    except OSError:
        # Cannot open new log file - leave stream as None.
        # FileHandler.emit() will attempt to open on next log entry.
        pass

def _get_log_file_path(self, date: datetime | None = None) -> Path:
    """Get log file path for the specified date.

    Args:
        date: Date for the log file (defaults to current UTC date)

    Returns:
        Path to the log file (e.g., {log_dir}/webhooks_2026-01-05.json)
    """
    if date is None:
        date = datetime.now(UTC)
    date_str = date.strftime("%Y-%m-%d")
    return self.log_dir / f"webhooks_{date_str}.json"

def write_log(self, context: WebhookContext) -> None:
    """Write webhook context as JSONL entry to date-based log file.

    Writes a compact JSON entry (single line, no indentation) containing complete webhook execution context.
    Each entry is terminated by a newline character.
    Uses atomic write pattern (temp file + rename) with file locking for safety.

    Args:
        context: WebhookContext to serialize and write

    Note:
        Uses context.completed_at as source of truth, falls back to datetime.now(UTC)
    """
    # Prefer context.completed_at as source of truth, fall back to current time
    completed_at = context.completed_at if context.completed_at else datetime.now(UTC)

    # Get context dict and update timing locally (without mutating context)
    context_dict = context.to_dict()
    context_dict["type"] = "webhook_summary"
    if "timing" in context_dict:
        context_dict["timing"]["completed_at"] = completed_at.isoformat()
        if context.started_at:
            duration_ms = int((completed_at - context.started_at).total_seconds() * 1000)
            context_dict["timing"]["duration_ms"] = duration_ms

    # Get log file path
    log_file = self._get_log_file_path(completed_at)

    # Serialize context to JSON (compact JSONL format - single line, no indentation)
    log_entry = json.dumps(context_dict, ensure_ascii=False)

message = record.getMessage()
message = _ANSI_ESCAPE_RE.sub("", message)

exc_text: str | None = None
if record.exc_info and record.exc_info[0] is not None:
    exc_text = "".join(traceback.format_exception(*record.exc_info))

entry: dict[str, object] = {
    "type": "log_entry",
    "timestamp": datetime.fromtimestamp(record.created, tz=UTC).isoformat(),
    "level": record.levelname,
    "logger_name": record.name,
    "message": message,
}

if exc_text:
    entry["exc_info"] = exc_text

# Enrich with webhook context when available
ctx = get_context()
if ctx is not None:
    entry["hook_id"] = ctx.hook_id
    entry["event_type"] = ctx.event_type
    entry["repository"] = ctx.repository
    entry["pr_number"] = ctx.pr_number
    entry["api_user"] = ctx.api_user

return {
    "hook_id": self.hook_id,
    "level": self._derive_level(),
    "status": self._derive_status(),
    "event_type": self.event_type,
    "action": self.action,
    "sender": self.sender,
    "repository": self.repository,
    "repository_full_name": self.repository_full_name,
    "pr": {
        "number": self.pr_number,
        "title": self.pr_title,
        "author": self.pr_author,
    }
    if self.pr_number
    else None,
    "api_user": self.api_user,
    "timing": {
        "started_at": self.started_at.isoformat(),
        "completed_at": (self.completed_at.isoformat() if self.completed_at else None),
        "duration_ms": int((self.completed_at - self.started_at).total_seconds() * 1000)
        if self.completed_at
        else None,
    },
    "workflow_steps": self.workflow_steps,
    "token_spend": self.token_spend,
    "initial_rate_limit": self.initial_rate_limit,
    "final_rate_limit": self.final_rate_limit,
    "success": self.success,
    "error": self.error,
    "summary": self._build_summary(),
}

mask_sensitive_patterns: list[str] = [
    # Passwords and secrets
    "container_repository_password",
    "password",
    "secret",
    # Tokens and API keys
    "token",
    "apikey",
    "api_key",
    "github_token",
    "GITHUB_TOKEN",
    "pypi",
    # Authentication credentials
    "username",
    "login",
    "-u",
    "-p",
    "--username",
    "--password",
    "--creds",
    # Private keys and sensitive IDs
    "private_key",
    "private-key",
    "webhook_secret",
    "webhook-secret",
    "github-app-id",
    # Slack webhooks (contain sensitive URLs)
    "slack-webhook-url",
    "slack_webhook_url",
    "webhook-url",
    "webhook_url",
]

# Attach JsonLogHandler for writing log records to the webhook JSONL file.
# Only attach when:
# - A log file path is configured (skip console-only loggers)
# - The logger is for the main webhook log (log_file_name not explicitly set)
#   Infrastructure loggers (mcp_server.log, logs_server.log) must NOT write
#   to webhooks_*.json because their entries lack webhook context (hook_id,
#   event_type, etc.) and pollute the webhook log with noise entries.
# - Only once per logger instance to avoid duplicate handlers.
# Uses _config.data_dir/logs (same directory as StructuredLogWriter) instead
# of deriving from the text log file path, which may differ for absolute paths.
if log_file_path_resolved and not log_file_name:
    log_dir = os.path.join(_config.data_dir, "logs")
    with _JSON_HANDLER_LOCK:
        if not any(isinstance(h, JsonLogHandler) and h.log_dir == Path(log_dir) for h in logger.handlers):
            logger.addHandler(
                JsonLogHandler(
                    log_dir=log_dir,
                    level=getattr(logging, log_level.upper(), logging.DEBUG),
                )
            )

if _log_viewer_controller_singleton is None:
    # Use global LOGGER for config operations
    config = Config(logger=LOGGER)
    logs_server_log_file = config.get_value("logs-server-log-file", return_on_none="logs_server.log")

    # Create dedicated logger for log viewer
    log_viewer_logger = get_logger_with_params(log_file_name=logs_server_log_file)
    _log_viewer_controller_singleton = LogViewerController(logger=log_viewer_logger)

# Configure MCP logging separation
if MCP_SERVER_ENABLED:
    mcp_log_file = root_config.get("mcp-log-file", "mcp_server.log")

    # Use get_logger_with_params to reuse existing logging configuration logic
    # (rotation, sensitive data masking, formatting)
    # This returns a logger configured for the specific file
    mcp_file_logger = get_logger_with_params(log_file_name=mcp_log_file)

    # Add the configured handler to the actual MCP logger and stop propagation
    # This ensures MCP logs go ONLY to mcp_server.log and not webhook_server.log
    if mcp_file_logger.handlers:
        for handler in mcp_file_logger.handlers:
            mcp_logger.addHandler(handler)

        mcp_logger.propagate = False

# Find all log files including rotated ones and JSON files
log_files: list[Path] = []
log_files.extend(log_dir.glob("*.log"))
log_files.extend(log_dir.glob("*.log.*"))
log_files.extend(log_dir.glob("webhooks_*.json"))

# Sort log files to prioritize JSON webhook files first (primary data source),
# then other files by modification time (newest first)
# This ensures webhook data is displayed before internal log files
def sort_key(f: Path) -> tuple[int, float]:
    is_json_webhook = f.suffix == ".json" and f.name.startswith("webhooks_")
    # JSON webhook files: (0, -mtime) - highest priority, newest first
    # Other files: (1, -mtime) - lower priority, newest first
    return (0 if is_json_webhook else 1, -f.stat().st_mtime)

...
async with aiofiles.open(log_file, encoding="utf-8") as f:
    # Use appropriate parser based on file type
    if log_file.suffix == ".json":
        # JSONL files: one compact JSON object per line
        # Process both "log_entry" and "webhook_summary" entries
        # Skip infrastructure logger entries that lack webhook context
        async for line in f:
            entry = self.log_parser.parse_json_log_entry(line)
            if entry and not LogViewerController._is_infrastructure_noise(entry):
                buffer.append(entry)
    else:
        # Text log files: parse line by line
        # Skip infrastructure logger entries that lack webhook context
        async for line in f:
            entry = self.log_parser.parse_log_entry(line)
            if entry and not LogViewerController._is_infrastructure_noise(entry):
                buffer.append(entry)

@staticmethod
def _is_infrastructure_noise(entry: LogEntry) -> bool:
    """Check if a log entry is infrastructure noise that should be excluded.

    Infrastructure loggers (MCP server, log viewer) produce high-frequency
    entries without webhook context. These are filtered out to prevent them
    from drowning actual webhook processing entries in unfiltered queries.

    Only excludes entries that have NO webhook context (hook_id is None),
    preserving any infrastructure log that happens to correlate with a webhook.

    Args:
        entry: LogEntry to check

    Returns:
        True if the entry is infrastructure noise and should be excluded

    """
    return entry.logger_name in LogViewerController._INFRASTRUCTURE_LOGGERS and entry.hook_id is None