airbyte_ops_mcp.cli.registry

   1# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
   2"""CLI commands for connector registry operations.
   3
   4This module provides CLI wrappers for registry operations. The core logic
   5lives in the `airbyte_ops_mcp.registry` capability module.
   6
   7Command groups:
   8    airbyte-ops registry connector list - List all connectors in registry
   9    airbyte-ops registry connector-version list - List versions for a connector
  10    airbyte-ops registry connector-version next - Compute next version tag (prerelease/RC)
  11    airbyte-ops registry connector-version yank - Mark a connector version as yanked
  12    airbyte-ops registry connector-version unyank - Remove yank marker from a connector version
  13    airbyte-ops registry connector-version metadata get - Read connector metadata from GCS
  14    airbyte-ops registry connector-version artifacts generate - Generate version artifacts locally via docker
  15    airbyte-ops registry connector-version artifacts publish - Publish version artifacts to GCS
  16    airbyte-ops registry store mirror --local|--gcs-bucket|--s3-bucket - Mirror entire registry for testing
  17    airbyte-ops registry store compile --store coral:dev|coral:prod - Compile registry indexes and sync latest/ dirs
  18    airbyte-ops registry marketing-stubs sync --store coral:prod - Sync connector_stubs.json to GCS
  19    airbyte-ops registry marketing-stubs check --store coral:prod - Compare local file with GCS
  20
  21## CLI reference
  22
  23The commands below are regenerated by `poe docs-generate` via cyclopts's
  24programmatic docs API; see `docs/generate_cli.py`.
  25
  26.. include:: ../../../docs/generated/cli/registry.md
  27   :start-line: 2
  28"""
  29
  30from __future__ import annotations
  31
  32# Hide Python-level members from the pdoc page for this module; the rendered
  33# docs for this CLI group come entirely from the grafted `.. include::` in
  34# the module docstring above.
  35__all__: list[str] = []
  36
  37import sys
  38from pathlib import Path
  39from typing import Annotated, Any, Literal
  40
  41import yaml
  42from cyclopts import Parameter
  43from fastmcp_extensions.cli import (
  44    exit_with_error,
  45    print_error,
  46    print_json,
  47    print_success,
  48)
  49from rich.console import Console
  50
  51from airbyte_ops_mcp.cli._base import App, app
  52from airbyte_ops_mcp.github_api import (
  53    get_file_contents_at_ref,
  54    resolve_default_github_token,
  55)
  56from airbyte_ops_mcp.mcp.connector_versions import (
  57    compute_prerelease_docker_image_tag,
  58)
  59from airbyte_ops_mcp.registry import (
  60    resolve_registry_store,
  61)
  62from airbyte_ops_mcp.registry._constants import PROD_METADATA_SERVICE_BUCKET_NAME
  63from airbyte_ops_mcp.registry._enums import (
  64    ConnectorLanguage,
  65    ConnectorType,
  66    SupportLevel,
  67)
  68from airbyte_ops_mcp.registry.compare import compare_stores
  69from airbyte_ops_mcp.registry.connector_stubs import (
  70    CONNECTOR_STUBS_FILE,
  71)
  72from airbyte_ops_mcp.registry.generate import generate_version_artifacts
  73from airbyte_ops_mcp.registry.operations import _read_cloud_registry_index
  74from airbyte_ops_mcp.registry.progressive_rollout_status import (
  75    get_connector_rollout_status,
  76)
  77from airbyte_ops_mcp.registry.registry_store_base import (
  78    Registry,
  79    get_registry,
  80)
  81
  82error_console = Console(stderr=True)
  83
  84# Create the registry sub-app
  85registry_app = App(
  86    name="registry", help="Connector registry operations (GCS metadata service)."
  87)
  88app.command(registry_app)
  89
  90# Create the connector sub-app under registry
  91connector_app = App(name="connector", help="Connector listing operations.")
  92registry_app.command(connector_app)
  93
  94# Create the connector-version sub-app under registry
  95connector_version_app = App(
  96    name="connector-version",
  97    help="Connector version operations (list, yank, unyank, artifacts).",
  98)
  99registry_app.command(connector_version_app)
 100
 101# Create the metadata sub-app under connector-version
 102metadata_app = App(
 103    name="metadata",
 104    help="Connector metadata inspection.",
 105)
 106connector_version_app.command(metadata_app)
 107
 108# Create the artifacts sub-app under connector-version
 109artifacts_app = App(
 110    name="artifacts",
 111    help="Version artifact generation and publishing.",
 112)
 113connector_version_app.command(artifacts_app)
 114
 115# Create the store sub-app under registry (whole-registry operations)
 116store_app = App(
 117    name="store",
 118    help="Whole-registry store operations (mirror, compile).",
 119)
 120registry_app.command(store_app)
 121
 122# Create the progressive-rollout sub-app under registry
 123progressive_rollout_app = App(
 124    name="progressive-rollout",
 125    help="Progressive rollout lifecycle operations.",
 126)
 127registry_app.command(progressive_rollout_app)
 128
 129# Create the marketing-stubs sub-app under registry (for whole-file GCS operations)
 130marketing_stubs_app = App(
 131    name="marketing-stubs",
 132    help="Marketing connector stubs GCS operations (whole-file sync).",
 133)
 134registry_app.command(marketing_stubs_app)
 135
 136
 137AIRBYTE_REPO_OWNER = "airbytehq"
 138AIRBYTE_ENTERPRISE_REPO_NAME = "airbyte-enterprise"
 139AIRBYTE_REPO_NAME = "airbyte"
 140CONNECTOR_PATH_PREFIX = "airbyte-integrations/connectors"
 141
 142
 143def _resolve_store(store: str) -> Registry:
 144    """Resolve `--store` to a `Registry` instance.
 145
 146    Wraps `resolve_registry_store` and `get_registry`, converting
 147    `ValueError` into a user-friendly CLI error via `exit_with_error`.
 148    """
 149    try:
 150        resolved = resolve_registry_store(store=store)
 151    except ValueError as e:
 152        exit_with_error(str(e))
 153        raise  # unreachable; satisfies type checker
 154    return get_registry(resolved)
 155
 156
 157def _get_connector_version_from_github(
 158    connector_name: str,
 159    ref: str,
 160    token: str | None = None,
 161) -> str | None:
 162    """Fetch connector version from metadata.yaml via GitHub API.
 163
 164    Args:
 165        connector_name: Connector name (e.g., "source-github")
 166        ref: Git ref (commit SHA, branch name, or tag)
 167        token: GitHub API token (optional for public repos)
 168
 169    Returns:
 170        Version string from metadata.yaml, or None if not found.
 171    """
 172    path = f"{CONNECTOR_PATH_PREFIX}/{connector_name}/metadata.yaml"
 173    contents = get_file_contents_at_ref(
 174        owner=AIRBYTE_REPO_OWNER,
 175        repo=AIRBYTE_REPO_NAME,
 176        path=path,
 177        ref=ref,
 178        token=token,
 179    )
 180    if contents is None:
 181        return None
 182
 183    metadata = yaml.safe_load(contents)
 184    return metadata.get("data", {}).get("dockerImageTag")
 185
 186
 187@connector_version_app.command(name="next")
 188def compute_next_version(
 189    name: Annotated[
 190        str,
 191        Parameter(help="Connector name (e.g., 'source-github')."),
 192    ],
 193    sha: Annotated[
 194        str,
 195        Parameter(help="Git commit SHA (full or at least 7 characters)."),
 196    ],
 197    base_version: Annotated[
 198        str | None,
 199        Parameter(
 200            help="Base version override. If not provided, fetched from metadata.yaml at the given SHA."
 201        ),
 202    ] = None,
 203) -> None:
 204    """Compute the next version tag for a connector.
 205
 206    Outputs the version tag to stdout for easy capture in shell scripts.
 207    This is the single source of truth for pre-release version format.
 208
 209    The command fetches the connector's metadata.yaml from GitHub at the given SHA
 210    to determine the base version. It also compares against the master branch and
 211    prints a warning to stderr if no version bump is detected.
 212
 213    If --base-version is provided, it is used directly instead of fetching from GitHub.
 214
 215    Examples:
 216        airbyte-ops registry connector-version next --name source-github --sha abcdef1234567
 217        # Output: 1.2.3-preview.abcdef1
 218
 219        airbyte-ops registry connector-version next --name source-github --sha abcdef1234567 --base-version 1.2.3
 220        # Output: 1.2.3-preview.abcdef1 (uses provided version, skips GitHub API)
 221    """
 222    # Try to get a GitHub token (optional, but helps avoid rate limiting)
 223    # Token resolution may fail if no token is configured, which is fine for public repos
 224    token: str | None = None
 225    token = resolve_default_github_token(allow_none=True)
 226
 227    # Determine base version
 228    version: str
 229    if base_version:
 230        version = base_version
 231    else:
 232        # Fetch version from metadata.yaml at the given SHA
 233        fetched_version = _get_connector_version_from_github(name, sha, token)
 234        if fetched_version is None:
 235            print(
 236                f"Error: Could not fetch metadata.yaml for {name} at ref {sha}",
 237                file=sys.stderr,
 238            )
 239            sys.exit(1)
 240        version = fetched_version
 241
 242    # Compare with master branch version and warn if no bump detected
 243    master_version = _get_connector_version_from_github(name, "master", token)
 244    if master_version and master_version == version:
 245        print(
 246            f"Warning: No version bump detected for {name}. "
 247            f"Version {version} matches master branch.",
 248            file=sys.stderr,
 249        )
 250
 251    # Compute and output the prerelease tag
 252    tag = compute_prerelease_docker_image_tag(version, sha)
 253    print(tag)
 254
 255
 256@progressive_rollout_app.command(name="list")
 257def progressive_rollout_list() -> None:
 258    """List all connectors with active release candidates in the compiled registry."""
 259    try:
 260        entries = _read_cloud_registry_index(
 261            bucket_name=PROD_METADATA_SERVICE_BUCKET_NAME,
 262        )
 263    except FileNotFoundError as e:
 264        exit_with_error(str(e))
 265
 266    rc_entries: list[dict[str, Any]] = []
 267    for entry in entries:
 268        releases = entry.get("releases", {})
 269        candidates = releases.get("releaseCandidates")
 270        if not candidates:
 271            continue
 272        docker_repo = entry.get("dockerRepository", "")
 273        connector_name = (
 274            docker_repo.split("/", 1)[1] if "/" in docker_repo else docker_repo
 275        )
 276        rc_entries.append(
 277            {
 278                "connector": connector_name,
 279                "rc_versions": list(candidates.keys()),
 280            }
 281        )
 282
 283    rc_entries.sort(key=lambda x: x["connector"])
 284    print_json(rc_entries)
 285
 286
 287@progressive_rollout_app.command(name="status")
 288def progressive_rollout_status(
 289    name: Annotated[
 290        str,
 291        Parameter(help="Connector technical name (e.g., source-github)."),
 292    ],
 293    *,
 294    repo_path: Annotated[
 295        Path,
 296        Parameter(help="Path to the Airbyte monorepo. Defaults to current directory."),
 297    ] = Path.cwd(),
 298    active_only: Annotated[
 299        bool,
 300        Parameter(
 301            help="Only return active non-terminal rollouts.",
 302            negative="--with-terminal",
 303        ),
 304    ] = True,
 305    limit: Annotated[
 306        int,
 307        Parameter(help="Maximum number of rollout records to return."),
 308    ] = 100,
 309) -> None:
 310    """Get progressive rollout status for a connector."""
 311    if not repo_path.exists():
 312        exit_with_error(f"Repository path not found: {repo_path}")
 313    try:
 314        result = get_connector_rollout_status(
 315            repo_path=repo_path,
 316            connector_name=name,
 317            active_only=active_only,
 318            limit=limit,
 319        )
 320    except (FileNotFoundError, ValueError) as e:
 321        exit_with_error(str(e))
 322
 323    print_json(result.model_dump())
 324
 325
 326@progressive_rollout_app.command(name="finalize-marker")
 327def progressive_rollout_finalize_marker(
 328    name: Annotated[
 329        str,
 330        Parameter(help="Connector technical name (e.g., source-github)."),
 331    ],
 332    store: Annotated[
 333        str,
 334        Parameter(help="Store target (e.g. 'coral:dev', 'coral:prod')."),
 335    ],
 336    outcome: Annotated[
 337        Literal["promoted", "aborted"],
 338        Parameter(help="Marker outcome used in the dated audit filename."),
 339    ],
 340    *,
 341    version: Annotated[
 342        str | None,
 343        Parameter(
 344            help="Version to finalize. If omitted, exactly one active marker must exist."
 345        ),
 346    ] = None,
 347    dry_run: Annotated[
 348        bool,
 349        Parameter(help="Show what would be done without making changes."),
 350    ] = False,
 351) -> None:
 352    """Rename an active `progressive-rollout.yml` marker to an audit marker."""
 353    registry = _resolve_store(store)
 354    result = registry.finalize_progressive_rollout_marker(
 355        connector_name=name,
 356        outcome=outcome,
 357        version=version,
 358        dry_run=dry_run,
 359    )
 360    print_json(result.to_dict())
 361    if result.success:
 362        print_success(result.message)
 363    else:
 364        exit_with_error(result.message, code=1)
 365
 366
 367# =============================================================================
 368# REGISTRY I/O - READ COMMANDS
 369# =============================================================================
 370
 371
 372@metadata_app.command(name="get")
 373def get_connector_version_metadata_cmd(
 374    name: Annotated[
 375        str,
 376        Parameter(
 377            help="Connector name (e.g., 'source-faker', 'destination-postgres')."
 378        ),
 379    ],
 380    store: Annotated[
 381        str,
 382        Parameter(
 383            help="Store target (e.g. 'coral:dev', 'coral:prod').",
 384        ),
 385    ],
 386    *,
 387    version: Annotated[
 388        str,
 389        Parameter(help="Version to read (e.g., 'latest', '1.2.3')."),
 390    ] = "latest",
 391    format: Annotated[
 392        Literal["json", "raw"],
 393        Parameter(help="Output format: 'json' for JSON, 'raw' for YAML."),
 394    ] = "json",
 395) -> None:
 396    """Read a connector version's metadata from the registry.
 397
 398    Returns the full metadata.yaml content for a connector at the specified version.
 399
 400    Requires GCS_CREDENTIALS environment variable to be set.
 401
 402    Examples:
 403        airbyte-ops registry connector-version metadata get --name source-faker --store coral:dev
 404        airbyte-ops registry connector-version metadata get --name source-faker --store coral:dev --version 6.2.38
 405        airbyte-ops registry connector-version metadata get --name source-faker --store coral:prod
 406    """
 407    registry = _resolve_store(store)
 408
 409    try:
 410        metadata = registry.get_connector_metadata(
 411            connector_name=name,
 412            version=version,
 413        )
 414    except FileNotFoundError as e:
 415        exit_with_error(str(e), code=1)
 416    except Exception as e:
 417        exit_with_error(f"Error reading metadata: {e}", code=1)
 418
 419    if format == "json":
 420        print_json(metadata)
 421    else:
 422        print(yaml.dump(metadata, default_flow_style=False))
 423
 424
 425@connector_app.command(name="list")
 426def list_connectors_cmd(
 427    store: Annotated[
 428        str,
 429        Parameter(
 430            help="Store target (e.g. 'coral:dev', 'coral:prod').",
 431        ),
 432    ],
 433    *,
 434    certified_only: Annotated[
 435        bool,
 436        Parameter(help="Include only certified connectors."),
 437    ] = False,
 438    support_level: Annotated[
 439        SupportLevel | None,
 440        Parameter(
 441            help=(
 442                "Exact support level to match "
 443                "(e.g. `certified`, `community`, `archived`)."
 444            )
 445        ),
 446    ] = None,
 447    min_support_level: Annotated[
 448        SupportLevel | None,
 449        Parameter(
 450            help=(
 451                "Minimum support level (inclusive). "
 452                "Levels from lowest to highest: `archived`, `community`, `certified`."
 453            )
 454        ),
 455    ] = None,
 456    connector_type: Annotated[
 457        ConnectorType | None,
 458        Parameter(help="Filter by connector type: `source` or `destination`."),
 459    ] = None,
 460    language: Annotated[
 461        ConnectorLanguage | None,
 462        Parameter(
 463            help=(
 464                "Filter by implementation language "
 465                "(e.g. `python`, `java`, `manifest-only`)."
 466            )
 467        ),
 468    ] = None,
 469    format: Annotated[
 470        Literal["json", "text"],
 471        Parameter(
 472            help="Output format: 'json' for JSON array, 'text' for newline-separated."
 473        ),
 474    ] = "json",
 475) -> None:
 476    """List connectors in the registry.
 477
 478    When filters are applied, reads the compiled `cloud_registry.json` index
 479    for fast lookups. Without filters, falls back to scanning individual
 480    metadata blobs (captures all connectors including OSS-only).
 481
 482    Requires GCS_CREDENTIALS environment variable to be set.
 483    """
 484    registry = _resolve_store(store)
 485
 486    # `--certified-only` is sugar for `--support-level certified`.
 487    effective_support_level = support_level
 488    if certified_only:
 489        if support_level and support_level != SupportLevel.CERTIFIED:
 490            exit_with_error(
 491                "`--certified-only` conflicts with `--support-level "
 492                f"{support_level}`. Use one or the other.",
 493                code=1,
 494            )
 495        effective_support_level = SupportLevel.CERTIFIED
 496
 497    try:
 498        connectors = registry.list_connectors(
 499            support_level=effective_support_level,
 500            min_support_level=min_support_level,
 501            connector_type=connector_type,
 502            language=language,
 503        )
 504    except Exception as e:
 505        exit_with_error(f"Error listing connectors: {e}", code=1)
 506
 507    if format == "json":
 508        print_json({"connectors": connectors, "count": len(connectors)})
 509    else:
 510        for connector in connectors:
 511            print(connector)
 512
 513
 514@connector_version_app.command(name="list")
 515def list_connector_versions_cmd(
 516    name: Annotated[
 517        str,
 518        Parameter(help="Connector name (e.g., 'source-faker')."),
 519    ],
 520    store: Annotated[
 521        str,
 522        Parameter(
 523            help="Store target (e.g. 'coral:dev', 'coral:prod').",
 524        ),
 525    ],
 526    *,
 527    format: Annotated[
 528        Literal["json", "text"],
 529        Parameter(
 530            help="Output format: 'json' for JSON array, 'text' for newline-separated."
 531        ),
 532    ] = "json",
 533) -> None:
 534    """List all versions of a connector in the registry.
 535
 536    Scans the registry bucket to find all versions of a specific connector.
 537
 538    Requires GCS_CREDENTIALS environment variable to be set.
 539    """
 540    registry = _resolve_store(store)
 541
 542    try:
 543        versions = registry.list_connector_versions(
 544            connector_name=name,
 545        )
 546    except Exception as e:
 547        exit_with_error(f"Error listing versions: {e}", code=1)
 548
 549    if format == "json":
 550        print_json({"connector": name, "versions": versions, "count": len(versions)})
 551    else:
 552        for v in versions:
 553            print(v)
 554
 555
 556@marketing_stubs_app.command(name="check")
 557def marketing_stubs_check(
 558    store: Annotated[
 559        str,
 560        Parameter(
 561            help="Store target (e.g. 'coral:dev', 'coral:prod').",
 562        ),
 563    ],
 564    *,
 565    repo_root: Annotated[
 566        Path,
 567        Parameter(
 568            help="Path to the airbyte-enterprise repository root. Defaults to current directory."
 569        ),
 570    ] = Path.cwd(),
 571) -> None:
 572    """Compare local connector_stubs.json with the version in GCS.
 573
 574    This command reads the entire local connector_stubs.json file and compares it
 575    with the version currently published in GCS.
 576
 577    Exit codes:
 578        0: Local file matches GCS (check passed)
 579        1: Differences found (check failed)
 580
 581    Output:
 582        STDOUT: JSON representation of the comparison result
 583        STDERR: Informational messages and comparison details
 584
 585    Example:
 586        airbyte-ops registry marketing-stubs check --store coral:prod --repo-root /path/to/airbyte-enterprise
 587        airbyte-ops registry marketing-stubs check --store coral:dev
 588    """
 589    registry = _resolve_store(store)
 590
 591    try:
 592        result = registry.marketing_stubs_check(repo_root=repo_root)
 593    except FileNotFoundError as e:
 594        exit_with_error(str(e))
 595    except ValueError as e:
 596        exit_with_error(str(e))
 597
 598    error_console.print(
 599        f"Comparing local {CONNECTOR_STUBS_FILE} with {result.get('bucket', '')}/{result.get('path', '')}"
 600    )
 601
 602    differences = result.get("differences", [])
 603    if differences:
 604        error_console.print(
 605            f"[yellow]Warning:[/yellow] {len(differences)} difference(s) found:"
 606        )
 607        for diff in differences:
 608            error_console.print(f"  {diff['id']}: {diff['status']}")
 609        print_json(result)
 610        sys.exit(1)
 611
 612    error_console.print(
 613        f"[green]Local file is in sync with GCS ({result.get('local_count', 0)} stubs)[/green]"
 614    )
 615    print_json(result)
 616
 617
 618@marketing_stubs_app.command(name="sync")
 619def marketing_stubs_sync(
 620    store: Annotated[
 621        str,
 622        Parameter(
 623            help="Store target (e.g. 'coral:dev', 'coral:prod').",
 624        ),
 625    ],
 626    *,
 627    repo_root: Annotated[
 628        Path,
 629        Parameter(
 630            help="Path to the airbyte-enterprise repository root. Defaults to current directory."
 631        ),
 632    ] = Path.cwd(),
 633    dry_run: Annotated[
 634        bool,
 635        Parameter(help="Show what would be uploaded without making changes."),
 636    ] = False,
 637) -> None:
 638    """Sync local connector_stubs.json to GCS.
 639
 640    This command uploads the entire local connector_stubs.json file to GCS,
 641    replacing the existing file. Use this after merging changes to master
 642    in the airbyte-enterprise repository.
 643
 644    Exit codes:
 645        0: Sync successful (or dry-run completed)
 646        1: Error (file not found, validation failed, etc.)
 647
 648    Output:
 649        STDOUT: JSON representation of the sync result
 650        STDERR: Informational messages and status updates
 651
 652    Example:
 653        airbyte-ops registry marketing-stubs sync --store coral:prod --repo-root /path/to/airbyte-enterprise
 654        airbyte-ops registry marketing-stubs sync --store coral:dev
 655        airbyte-ops registry marketing-stubs sync --store coral:dev --dry-run
 656    """
 657    registry = _resolve_store(store)
 658
 659    try:
 660        result = registry.marketing_stubs_sync(
 661            repo_root=repo_root,
 662            dry_run=dry_run,
 663        )
 664    except FileNotFoundError as e:
 665        exit_with_error(str(e))
 666    except ValueError as e:
 667        exit_with_error(str(e))
 668
 669    bucket_name = result.get("bucket", "")
 670    path = result.get("path", "")
 671    stub_count = result.get("stub_count", 0)
 672
 673    if dry_run:
 674        error_console.print(
 675            f"[DRY RUN] Would upload {stub_count} stubs to {bucket_name}/{path}"
 676        )
 677    else:
 678        error_console.print(
 679            f"[green]Synced {stub_count} stubs to {bucket_name}/{path}[/green]"
 680        )
 681    print_json(result)
 682
 683
 684# =============================================================================
 685# REGISTRY REBUILD COMMANDS
 686# =============================================================================
 687
 688
 689@store_app.command(name="mirror")
 690def mirror_cmd(
 691    local: Annotated[
 692        bool,
 693        Parameter(
 694            help="Write output to a local directory. Mutually exclusive with --gcs-bucket and --s3-bucket.",
 695            negative="",
 696        ),
 697    ] = False,
 698    gcs_bucket: Annotated[
 699        str | None,
 700        Parameter(
 701            help="Write output to a GCS bucket. Must not be the prod bucket. "
 702            "Mutually exclusive with --local and --s3-bucket.",
 703        ),
 704    ] = None,
 705    s3_bucket: Annotated[
 706        str | None,
 707        Parameter(
 708            help="Write output to an S3 bucket. "
 709            "Mutually exclusive with --local and --gcs-bucket.",
 710        ),
 711    ] = None,
 712    output_path_root: Annotated[
 713        str | None,
 714        Parameter(
 715            help="Root path/prefix for the output. For --local, this is a directory path "
 716            "(defaults to a new temp dir if omitted). For --gcs-bucket/--s3-bucket, "
 717            "this prefix is prepended to all blob paths.",
 718        ),
 719    ] = None,
 720    dry_run: Annotated[
 721        bool,
 722        Parameter(help="Show what would be rebuilt without writing any files."),
 723    ] = False,
 724    source_store: Annotated[
 725        str,
 726        Parameter(
 727            help="Source store to read from (e.g. 'coral:prod', 'coral:dev').",
 728        ),
 729    ] = "coral:prod",
 730    connector_name: Annotated[
 731        tuple[str, ...] | None,
 732        Parameter(
 733            help="Only rebuild these connectors (by name). "
 734            "Can be specified multiple times, e.g. "
 735            "--connector-name source-faker --connector-name destination-bigquery.",
 736        ),
 737    ] = None,
 738) -> None:
 739    """Create a mirror of the connector registry from the source store.
 740
 741    Reads all connector metadata from the source store and copies it
 742    to the specified output target. Supports local filesystem, GCS, and S3
 743    as output targets via fsspec.
 744
 745    Output targets are mutually exclusive: specify exactly one of
 746    --local, --gcs-bucket, or --s3-bucket.
 747
 748    The production bucket is categorically disallowed as an output target.
 749
 750    To clean up legacy artifacts (e.g. disabled strict-encrypt connectors)
 751    after mirroring, run compile with `--with-legacy-migration v1`::
 752
 753        airbyte-ops registry store compile --store coral:dev/my-prefix \\
 754            --with-legacy-migration v1
 755
 756    Examples:
 757        airbyte-ops registry store mirror --local
 758        airbyte-ops registry store mirror --local --source-store coral:dev
 759        airbyte-ops registry store mirror --gcs-bucket dev-airbyte-cloud-connector-metadata-service-2 \\
 760            --output-path-root test-run-123
 761        airbyte-ops registry store mirror --s3-bucket my-test-bucket --dry-run
 762    """
 763    # Validate mutually exclusive output targets
 764    targets = [local, gcs_bucket is not None, s3_bucket is not None]
 765    if sum(targets) != 1:
 766        exit_with_error(
 767            "Specify exactly one output target: --local, --gcs-bucket, or --s3-bucket."
 768        )
 769
 770    if local:
 771        output_mode = "local"
 772    elif gcs_bucket is not None:
 773        output_mode = "gcs"
 774    else:
 775        output_mode = "s3"
 776
 777    registry = _resolve_store(source_store)
 778
 779    result = registry.mirror(
 780        output_mode=output_mode,
 781        output_path_root=output_path_root,
 782        gcs_bucket=gcs_bucket,
 783        s3_bucket=s3_bucket,
 784        dry_run=dry_run,
 785        connector_name=list(connector_name) if connector_name else None,
 786    )
 787
 788    print_json(
 789        {
 790            "status": result.status,
 791            "source_bucket": result.source_bucket,
 792            "output_mode": result.output_mode,
 793            "output_root": result.output_root,
 794            "connectors_processed": result.connectors_processed,
 795            "blobs_copied": result.blobs_copied,
 796            "blobs_skipped": result.blobs_skipped,
 797            "error_count": len(result.errors),
 798            "dry_run": result.dry_run,
 799        }
 800    )
 801
 802    if result.errors:
 803        for err in result.errors[:10]:
 804            print_error(err)
 805        if len(result.errors) > 10:
 806            print_error(f"... and {len(result.errors) - 10} more errors")
 807
 808    if result.status == "success":
 809        print_success(result.summary())
 810    elif result.status == "dry-run":
 811        print_success(f"[DRY RUN] {result.summary()}")
 812    else:
 813        error_console.print(f"[yellow]{result.summary()}[/yellow]")
 814
 815
 816# =============================================================================
 817# VERSION YANK COMMANDS
 818# =============================================================================
 819
 820
 821@connector_version_app.command(name="yank")
 822def yank_cmd(
 823    name: Annotated[
 824        str,
 825        Parameter(help="Connector name (e.g., 'source-faker')."),
 826    ],
 827    version: Annotated[
 828        str,
 829        Parameter(help="Version to yank (e.g., '1.2.3')."),
 830    ],
 831    store: Annotated[
 832        str,
 833        Parameter(
 834            help="Store target (e.g. 'coral:dev', 'coral:prod').",
 835        ),
 836    ],
 837    *,
 838    reason: Annotated[
 839        str,
 840        Parameter(help="Reason for yanking this version."),
 841    ] = "",
 842    approval_url: Annotated[
 843        str,
 844        Parameter(help="Approval evidence URL to record in the yank marker."),
 845    ] = "",
 846    dry_run: Annotated[
 847        bool,
 848        Parameter(help="Show what would be done without making changes."),
 849    ] = False,
 850) -> None:
 851    """Mark a connector version as yanked.
 852
 853    Writes a version-yank.yml marker file to the version's directory in GCS.
 854    Yanked versions are excluded when determining the latest version of a
 855    connector.
 856
 857    Requires GCS_CREDENTIALS environment variable to be set.
 858
 859    Examples:
 860        airbyte-ops registry connector-version yank --name source-faker --version 1.2.3 --store coral:dev
 861        airbyte-ops registry connector-version yank --name source-faker --version 1.2.3 --store coral:dev --reason "Critical bug"
 862        airbyte-ops registry connector-version yank --name source-faker --version 1.2.3 --store coral:prod
 863    """
 864    registry = _resolve_store(store)
 865
 866    result = registry.yank(
 867        connector_name=name,
 868        version=version,
 869        reason=reason,
 870        approval_url=approval_url,
 871        dry_run=dry_run,
 872    )
 873
 874    print_json(result.to_dict())
 875
 876    if result.success:
 877        print_success(result.message)
 878        if not dry_run:
 879            error_console.print(
 880                "\n[yellow]Note:[/yellow] The registry indexes are now stale. "
 881                "To update them, run:\n\n"
 882                f"    airbyte-ops registry store compile --store {store}\n\n"
 883                "Or wait for the next scheduled compile operation."
 884            )
 885    else:
 886        exit_with_error(result.message, code=1)
 887
 888
 889@connector_version_app.command(name="unyank")
 890def unyank_cmd(
 891    name: Annotated[
 892        str,
 893        Parameter(help="Connector name (e.g., 'source-faker')."),
 894    ],
 895    version: Annotated[
 896        str,
 897        Parameter(help="Version to unyank (e.g., '1.2.3')."),
 898    ],
 899    store: Annotated[
 900        str,
 901        Parameter(
 902            help="Store target (e.g. 'coral:dev', 'coral:prod').",
 903        ),
 904    ],
 905    *,
 906    dry_run: Annotated[
 907        bool,
 908        Parameter(help="Show what would be done without making changes."),
 909    ] = False,
 910) -> None:
 911    """Rename the active yank marker to a dated audit marker.
 912
 913    Moves `version-yank.yml` to `version-unyanked-yyyymmdd.yml`, making the
 914    version eligible again when determining the latest version.
 915
 916    Requires GCS_CREDENTIALS environment variable to be set.
 917
 918    Examples:
 919        airbyte-ops registry connector-version unyank --name source-faker --version 1.2.3 --store coral:dev
 920        airbyte-ops registry connector-version unyank --name source-faker --version 1.2.3 --store coral:prod
 921    """
 922    registry = _resolve_store(store)
 923
 924    result = registry.unyank(
 925        connector_name=name,
 926        version=version,
 927        dry_run=dry_run,
 928    )
 929
 930    print_json(result.to_dict())
 931
 932    if result.success:
 933        print_success(result.message)
 934        if not dry_run:
 935            error_console.print(
 936                "\n[yellow]Note:[/yellow] The registry indexes are now stale. "
 937                "To update them, run:\n\n"
 938                f"    airbyte-ops registry store compile --store {store}\n\n"
 939                "Or wait for the next scheduled compile operation."
 940            )
 941    else:
 942        exit_with_error(result.message, code=1)
 943
 944
 945@connector_version_app.command(name="list-yanked")
 946def list_yanked_cmd(
 947    store: Annotated[
 948        str,
 949        Parameter(
 950            help="Store target (e.g. 'coral:dev', 'coral:prod').",
 951        ),
 952    ],
 953    *,
 954    format: Annotated[
 955        Literal["json", "text"],
 956        Parameter(
 957            help="Output format: 'json' for JSON array, 'text' for newline-separated."
 958        ),
 959    ] = "json",
 960) -> None:
 961    """List all yanked connector versions in the registry.
 962
 963    Returns one entry per active `version-yank.yml` marker, sorted by connector
 964    name then version. Historical `version-unyanked-*.yml` audit markers are
 965    ignored.
 966
 967    Requires GCS_CREDENTIALS environment variable to be set.
 968
 969    Examples:
 970        airbyte-ops registry connector-version list-yanked --store coral:prod
 971        airbyte-ops registry connector-version list-yanked --store coral:prod --format text
 972    """
 973    registry = _resolve_store(store)
 974    yanked = registry.list_yanked_versions()
 975
 976    if format == "json":
 977        print_json(
 978            {
 979                "yanked_versions": [item.to_dict() for item in yanked],
 980                "count": len(yanked),
 981            }
 982        )
 983    else:
 984        for item in yanked:
 985            print(f"{item.connector_name}\t{item.version}")
 986
 987
 988@connector_version_app.command(name="yank-status")
 989def yank_status_cmd(
 990    name: Annotated[
 991        str,
 992        Parameter(help="Connector name (e.g., 'source-faker')."),
 993    ],
 994    version: Annotated[
 995        str,
 996        Parameter(help="Version to inspect (e.g., '1.2.3')."),
 997    ],
 998    store: Annotated[
 999        str,
1000        Parameter(
1001            help="Store target (e.g. 'coral:dev', 'coral:prod').",
1002        ),
1003    ],
1004) -> None:
1005    """Show the active yank marker for a single connector version.
1006
1007    Prints the marker's parsed fields (`yanked_at`, `reason`, `approval_url`)
1008    when the version currently has an active `version-yank.yml` marker, or
1009    reports that the version is not yanked. Historical
1010    `version-unyanked-*.yml` audit markers are ignored.
1011
1012    Requires GCS_CREDENTIALS environment variable to be set.
1013
1014    Examples:
1015        airbyte-ops registry connector-version yank-status --name source-faker --version 1.2.3 --store coral:prod
1016    """
1017    registry = _resolve_store(store)
1018    marker = registry.get_yank_marker(connector_name=name, version=version)
1019
1020    if marker is None:
1021        print_json(
1022            {
1023                "connector_name": name,
1024                "version": version,
1025                "yanked": False,
1026            }
1027        )
1028        print_success(f"{name} {version} is not yanked.")
1029        return
1030
1031    print_json({"yanked": True, **marker.to_dict()})
1032    print_success(f"{name} {version} is yanked.")
1033
1034
1035# =============================================================================
1036# ARTIFACT GENERATION COMMANDS
1037# =============================================================================
1038
1039
1040@artifacts_app.command(name="generate")
1041def generate_version_artifacts_cmd(
1042    metadata_file: Annotated[
1043        Path,
1044        Parameter(help="Path to the connector's metadata.yaml file."),
1045    ],
1046    docker_image: Annotated[
1047        str,
1048        Parameter(
1049            help="Docker image to run spec against (e.g., 'airbyte/source-faker:6.2.38')."
1050        ),
1051    ],
1052    output_dir: Annotated[
1053        Path | None,
1054        Parameter(
1055            help="Directory to write artifacts to. If not specified, a temp directory is created."
1056        ),
1057    ] = None,
1058    repo_root: Annotated[
1059        Path | None,
1060        Parameter(
1061            help=(
1062                "Root of the Airbyte repo checkout (for resolving doc.md). "
1063                "If not specified, inferred by walking up from metadata-file."
1064            ),
1065        ),
1066    ] = None,
1067    dry_run: Annotated[
1068        bool,
1069        Parameter(
1070            help="Show what would be generated without running docker or writing files."
1071        ),
1072    ] = False,
1073    with_validate: Annotated[
1074        bool,
1075        Parameter(
1076            help=(
1077                "Run metadata validators after generation (default: enabled). "
1078                "Use --no-validate to skip."
1079            ),
1080            negative="--no-validate",
1081        ),
1082    ] = True,
1083    with_sbom: Annotated[
1084        bool,
1085        Parameter(
1086            help=(
1087                "Generate spdx.json (SBOM) for connectors "
1088                "(default: enabled). Use --no-sbom to skip."
1089            ),
1090            negative="--no-sbom",
1091        ),
1092    ] = True,
1093    with_dependency_dump: Annotated[
1094        bool,
1095        Parameter(
1096            help=(
1097                "Generate dependencies.json for Python connectors "
1098                "(default: enabled). Use --no-dependency-dump to skip."
1099            ),
1100            negative="--no-dependency-dump",
1101        ),
1102    ] = True,
1103) -> None:
1104    """Generate version artifacts for a connector locally.
1105
1106    Runs the connector's docker image with DEPLOYMENT_MODE=cloud and
1107    DEPLOYMENT_MODE=oss to obtain both spec variants, then generates
1108    the registry entries (cloud.json, oss.json) by applying
1109    registryOverrides from the metadata.
1110
1111    The generated metadata.yaml is enriched with git commit info, SBOM URL,
1112    and (when applicable) components SHA before writing.  Validation is run
1113    after generation by default; pass `--no-validate` to skip.
1114
1115    This is a local-only operation -- no files are uploaded to GCS.
1116    Use `artifacts publish` to upload generated artifacts to GCS.
1117
1118    Examples:
1119        airbyte-ops registry connector-version artifacts generate \\
1120            --metadata-file path/to/metadata.yaml \\
1121            --docker-image airbyte/source-faker:6.2.38
1122
1123        airbyte-ops registry connector-version artifacts generate \\
1124            --metadata-file path/to/metadata.yaml \\
1125            --docker-image airbyte/source-faker:6.2.38 \\
1126            --output-dir ./artifacts --with-validate
1127    """
1128    result = generate_version_artifacts(
1129        metadata_file=metadata_file,
1130        docker_image=docker_image,
1131        output_dir=output_dir,
1132        repo_root=repo_root,
1133        dry_run=dry_run,
1134        with_validate=with_validate,
1135        with_dependency_dump=with_dependency_dump,
1136        with_sbom=with_sbom,
1137    )
1138
1139    print_json(result.to_dict())
1140
1141    if result.success:
1142        print_success(
1143            f"Generated {len(result.artifacts_written)} artifacts to {result.output_dir}"
1144        )
1145    else:
1146        all_errors = result.errors + result.validation_errors
1147        exit_with_error(
1148            f"Generation completed with {len(all_errors)} error(s): "
1149            + "; ".join(all_errors),
1150            code=1,
1151        )
1152
1153
1154@artifacts_app.command(name="publish")
1155def publish_version_artifacts_cmd(
1156    name: Annotated[
1157        str,
1158        Parameter(help="Connector name (e.g., 'source-faker')."),
1159    ],
1160    version: Annotated[
1161        str,
1162        Parameter(help="Version to publish artifacts for (e.g., '1.2.3')."),
1163    ],
1164    artifacts_dir: Annotated[
1165        Path,
1166        Parameter(
1167            help="Directory containing generated artifacts to publish (from 'artifacts generate')."
1168        ),
1169    ],
1170    store: Annotated[
1171        str,
1172        Parameter(
1173            help="Store target (e.g. 'coral:dev', 'coral:prod', 'coral:dev/prefix').",
1174        ),
1175    ],
1176    *,
1177    dry_run: Annotated[
1178        bool,
1179        Parameter(help="Show what would be published without writing to GCS."),
1180    ] = False,
1181    with_validate: Annotated[
1182        bool,
1183        Parameter(
1184            help=(
1185                "Validate metadata before uploading (default: enabled). "
1186                "Use --no-validate to skip."
1187            ),
1188            negative="--no-validate",
1189        ),
1190    ] = True,
1191) -> None:
1192    """Publish version artifacts to GCS using fsspec rsync.
1193
1194    Uploads locally generated artifacts (from `artifacts generate`) to the
1195    versioned path in GCS.  By default, metadata is validated before upload;
1196    pass `--no-validate` to skip.
1197
1198    Uses `--store` to select the destination store and environment:
1199
1200    * `coral:dev`              → coral dev bucket at root
1201    * `coral:prod`             → coral prod bucket at root
1202    * `coral:dev/aj-test100`   → coral dev bucket under `aj-test100/` prefix
1203
1204    Requires GCS_CREDENTIALS environment variable to be set.
1205
1206    Examples:
1207        airbyte-ops registry connector-version artifacts publish \\
1208            --name source-faker --version 6.2.38 \\
1209            --artifacts-dir ./artifacts --store coral:dev --with-validate
1210
1211        airbyte-ops registry connector-version artifacts publish \\
1212            --name source-faker --version 6.2.38 \\
1213            --artifacts-dir ./artifacts --store coral:prod
1214
1215        airbyte-ops registry connector-version artifacts publish \\
1216            --name source-faker --version 6.2.38 \\
1217            --artifacts-dir ./artifacts --store coral:dev/aj-test100
1218    """
1219    registry = _resolve_store(store)
1220
1221    result = registry.publish_version_artifacts(
1222        connector_name=name,
1223        version=version,
1224        artifacts_dir=artifacts_dir,
1225        dry_run=dry_run,
1226        with_validate=with_validate,
1227    )
1228
1229    print_json(
1230        {
1231            "status": result.status,
1232            "connector_name": result.connector_name,
1233            "version": result.version,
1234            "target": result.target,
1235            "gcs_destination": result.gcs_destination,
1236            "files_uploaded": result.files_uploaded,
1237            "errors": result.errors,
1238            "validation_errors": result.validation_errors,
1239            "dry_run": result.dry_run,
1240        }
1241    )
1242
1243    if result.success:
1244        print_success(
1245            f"Published {len(result.files_uploaded)} artifacts for "
1246            f"{result.connector_name}@{result.version} → {result.gcs_destination}"
1247        )
1248    else:
1249        all_errors = result.errors + result.validation_errors
1250        exit_with_error(
1251            f"Publish completed with {len(all_errors)} error(s): "
1252            + "; ".join(all_errors),
1253            code=1,
1254        )
1255
1256
1257# =============================================================================
1258# COMPILE COMMAND
1259# =============================================================================
1260
1261
1262@store_app.command(name="compile")
1263def compile_cmd(
1264    store: Annotated[
1265        str,
1266        Parameter(
1267            help="Store target (e.g. 'coral:dev', 'coral:prod', 'coral:dev/prefix').",
1268        ),
1269    ],
1270    *,
1271    connector_name: Annotated[
1272        tuple[str, ...] | None,
1273        Parameter(
1274            help="Only compile these connectors (can be repeated).",
1275        ),
1276    ] = None,
1277    dry_run: Annotated[
1278        bool,
1279        Parameter(help="Show what would be done without writing."),
1280    ] = False,
1281    with_secrets_mask: Annotated[
1282        bool,
1283        Parameter(
1284            help=(
1285                "Also regenerate specs_secrets_mask.yaml by scanning all "
1286                "connector specs for airbyte_secret properties."
1287            ),
1288        ),
1289    ] = False,
1290    with_legacy_migration: Annotated[
1291        str | None,
1292        Parameter(
1293            help=(
1294                "Run a one-time legacy migration step during compile. "
1295                "Currently supported: 'v1' — delete cloud.json / oss.json "
1296                "files for connectors whose registryOverrides.cloud.enabled "
1297                "or registryOverrides.oss.enabled is false. This cleans up "
1298                "artifacts produced by the legacy pipeline that did not "
1299                "respect the enabled flag."
1300            ),
1301        ),
1302    ] = None,
1303    with_metrics: Annotated[
1304        bool,
1305        Parameter(
1306            help=(
1307                "Inject latest connector quality metrics from the analytics "
1308                "JSONL export into generated.metrics."
1309            ),
1310            negative="--no-metrics",
1311        ),
1312    ] = True,
1313    force: Annotated[
1314        bool,
1315        Parameter(
1316            help=(
1317                "Force resync of latest/ directories even if version markers are current. "
1318                "Useful when metadata changes without a version bump."
1319            ),
1320        ),
1321    ] = False,
1322) -> None:
1323    """Compile the registry: sync latest/ dirs, write global and per-connector indexes.
1324
1325    Scans all version directories in the target store, determines the latest GA
1326    semver per connector (excluding yanked and pre-release versions), ensures
1327    each `latest/` directory matches the computed latest, and writes:
1328
1329    * `registries/v0/cloud_registry.json` -- global cloud registry index
1330    * `registries/v0/oss_registry.json`   -- global OSS registry index
1331    * `metadata/airbyte/<connector>/versions.json` -- per-connector version index
1332
1333    With `--with-secrets-mask`, also regenerates:
1334
1335    * `registries/v0/specs_secrets_mask.yaml` -- properties marked as secrets
1336
1337    With `--with-legacy-migration=v1`, deletes `cloud.json` / `oss.json`
1338    files for connectors whose `registryOverrides.cloud.enabled` or
1339    `registryOverrides.oss.enabled` is `false`.
1340
1341    By default, injects connector quality metrics from the latest analytics
1342    JSONL export into `generated.metrics`. Use `--no-metrics` for offline
1343    scenarios.
1344
1345    With `--force`, resyncs all latest/ directories even if the version marker
1346    matches the computed latest version.
1347
1348    Uses efficient glob patterns for scanning (no file downloads during discovery).
1349
1350    Requires GCS_CREDENTIALS environment variable to be set.
1351
1352    Examples:
1353        airbyte-ops registry store compile --store coral:dev --dry-run
1354
1355        airbyte-ops registry store compile --store coral:dev/aj-test100 \\
1356            --connector-name source-faker --connector-name destination-bigquery
1357
1358        airbyte-ops registry store compile --store coral:prod --with-secrets-mask
1359
1360        airbyte-ops registry store compile --store coral:dev \\
1361            --with-legacy-migration v1
1362    """
1363    registry = _resolve_store(store)
1364
1365    result = registry.compile(
1366        connector_name=list(connector_name) if connector_name else None,
1367        dry_run=dry_run,
1368        with_secrets_mask=with_secrets_mask,
1369        with_legacy_migration=with_legacy_migration,
1370        with_metrics=with_metrics,
1371        force=force,
1372    )
1373
1374    print_json(
1375        {
1376            "status": result.status,
1377            "target": result.target,
1378            "connectors_scanned": result.connectors_scanned,
1379            "versions_found": result.versions_found,
1380            "yanked_versions": result.yanked_versions,
1381            "latest_updated": result.latest_updated,
1382            "latest_already_current": result.latest_already_current,
1383            "cloud_registry_entries": result.cloud_registry_entries,
1384            "oss_registry_entries": result.oss_registry_entries,
1385            "composite_registry_entries": result.composite_registry_entries,
1386            "metrics_connector_count": result.metrics_connector_count,
1387            "metrics_registry_entries": result.metrics_registry_entries,
1388            "metrics_source": result.metrics_source,
1389            "metrics_error": result.metrics_error,
1390            "version_indexes_written": result.version_indexes_written,
1391            "specs_secrets_mask_properties": result.specs_secrets_mask_properties,
1392            "errors": result.errors,
1393            "dry_run": result.dry_run,
1394        }
1395    )
1396
1397    if result.status == "success" or result.status == "dry-run":
1398        print_success(result.summary())
1399    else:
1400        exit_with_error(
1401            f"Compile completed with {len(result.errors)} error(s): "
1402            + "; ".join(result.errors),
1403            code=1,
1404        )
1405
1406
1407# =============================================================================
1408# DELETE-DEV-LATEST COMMAND
1409# =============================================================================
1410
1411
1412@store_app.command(name="delete-dev-latest")
1413def delete_dev_latest_cmd(
1414    store: Annotated[
1415        str,
1416        Parameter(
1417            help="Store target (must begin with 'coral:dev').",
1418        ),
1419    ],
1420    *,
1421    connector_name: Annotated[
1422        tuple[str, ...] | None,
1423        Parameter(
1424            help="Only delete latest/ for these connectors (can be repeated).",
1425        ),
1426    ] = None,
1427    dry_run: Annotated[
1428        bool,
1429        Parameter(help="Show what would be done without deleting."),
1430    ] = False,
1431) -> None:
1432    """Delete all latest/ directories from a dev registry store.
1433
1434    Discovers every connector that has a `latest/` directory and
1435    deletes each one in parallel using a thread pool.
1436
1437    This is useful before a full re-compile to prove that latest/
1438    directories can be correctly regenerated from versioned data.
1439
1440    Only dev stores are allowed (store must begin with 'coral:dev').
1441
1442    Requires GCS_CREDENTIALS environment variable to be set.
1443
1444    Examples:
1445        airbyte-ops registry store delete-dev-latest --store coral:dev --dry-run
1446
1447        airbyte-ops registry store delete-dev-latest --store coral:dev/aj-test100
1448
1449        airbyte-ops registry store delete-dev-latest --store coral:dev \\
1450            --connector-name source-faker --connector-name destination-bigquery
1451    """
1452    if not store.startswith("coral:dev"):
1453        exit_with_error(
1454            "delete-dev-latest only supports dev stores "
1455            f"(store must begin with 'coral:dev', got '{store}').",
1456            code=1,
1457        )
1458
1459    registry = _resolve_store(store)
1460
1461    result = registry.delete_dev_latest(
1462        connector_name=list(connector_name) if connector_name else None,
1463        dry_run=dry_run,
1464    )
1465
1466    print_json(
1467        {
1468            "status": result.status,
1469            "target": result.target,
1470            "connectors_found": result.connectors_found,
1471            "latest_dirs_deleted": result.latest_dirs_deleted,
1472            "errors": result.errors,
1473            "dry_run": result.dry_run,
1474        }
1475    )
1476
1477    if result.status in ("success", "dry-run"):
1478        print_success(result.summary())
1479    else:
1480        exit_with_error(
1481            f"Delete completed with {len(result.errors)} error(s): "
1482            + "; ".join(result.errors[:5]),
1483            code=1,
1484        )
1485
1486
1487# =============================================================================
1488# STORE COMPARE COMMAND
1489# =============================================================================
1490
1491
1492@store_app.command(name="compare")
1493def compare_cmd(
1494    store: Annotated[
1495        str,
1496        Parameter(
1497            help="Store target being evaluated (e.g. 'coral:dev/20260306-mirror-compile').",
1498        ),
1499    ],
1500    reference_store: Annotated[
1501        str,
1502        Parameter(
1503            help="Known-good reference store to compare against.",
1504        ),
1505    ],
1506    *,
1507    connector_name: Annotated[
1508        tuple[str, ...] | None,
1509        Parameter(
1510            help="Only compare these connectors (can be repeated).",
1511        ),
1512    ] = None,
1513    with_artifacts: Annotated[
1514        bool,
1515        Parameter(
1516            help="Compare per-connector artifact files "
1517            "(metadata.yaml, cloud.json, oss.json, spec.json).",
1518            negative="--no-artifacts",
1519        ),
1520    ] = True,
1521    with_indexes: Annotated[
1522        bool,
1523        Parameter(
1524            help="Compare global registry index files "
1525            "(cloud_registry.json, oss_registry.json).",
1526            negative="--no-indexes",
1527        ),
1528    ] = True,
1529) -> None:
1530    """Compare a store against a reference store and report differences.
1531
1532    Evaluates the `--store` target against `--reference-store` and reports
1533    per-connector artifact diffs and global index diffs.
1534
1535    Requires GCS_CREDENTIALS environment variable to be set.
1536
1537    Examples:
1538        airbyte-ops registry store compare --store coral:dev/20260306-mirror \\
1539            --reference-store coral:prod
1540
1541        airbyte-ops registry store compare --store coral:dev/my-test \\
1542            --connector-name source-faker --no-indexes
1543
1544        airbyte-ops registry store compare --store coral:dev/my-test \\
1545            --no-artifacts
1546    """
1547    store_target = _resolve_store(store)
1548    ref_target = _resolve_store(reference_store)
1549
1550    store_prefix = f"{store_target.prefix}/" if store_target.prefix else ""
1551    ref_prefix = f"{ref_target.prefix}/" if ref_target.prefix else ""
1552
1553    result = compare_stores(
1554        store_bucket=store_target.bucket_name,
1555        store_prefix=store_prefix,
1556        reference_bucket=ref_target.bucket_name,
1557        reference_prefix=ref_prefix,
1558        connector_name=list(connector_name) if connector_name else None,
1559        with_artifacts=with_artifacts,
1560        with_indexes=with_indexes,
1561    )
1562
1563    print_json(result.to_dict())
1564
1565    if result.status == "match":
1566        print_success(result.summary())
1567    elif result.status == "differences-found":
1568        error_console.print(f"[yellow]{result.summary()}[/yellow]")
1569
1570        # Print a concise per-connector diff summary
1571        for diff in result.connector_diffs:
1572            if diff.status in ("only_in_store", "only_in_reference"):
1573                error_console.print(f"  {diff.connector}: {diff.status}")
1574            else:
1575                for ad in diff.artifact_diffs:
1576                    error_console.print(
1577                        f"  {diff.connector}/{ad.file}: {ad.status}"
1578                        + (f" ({ad.details})" if ad.details else "")
1579                    )
1580
1581        for idx_diff in result.index_diffs:
1582            if idx_diff.status != "match":
1583                error_console.print(
1584                    f"  [index] {idx_diff.file}: {idx_diff.status}"
1585                    + (
1586                        f" (store={idx_diff.entry_count_store},"
1587                        f" ref={idx_diff.entry_count_reference})"
1588                        if idx_diff.entry_count_store or idx_diff.entry_count_reference
1589                        else ""
1590                    )
1591                )
1592
1593        sys.exit(1)
1594    else:
1595        exit_with_error(
1596            f"Compare completed with {len(result.errors)} error(s): "
1597            + "; ".join(result.errors[:5]),
1598            code=1,
1599        )