copilotkit__copilotkit/showcase/bin/railway

#!/usr/bin/env ruby
# frozen_string_literal: true

# bin/railway — single-file Ruby tooling for showcase Railway operations.
#
# Subcommands:
#   snapshot           Capture current service config to a YAML snapshot.
#   restore            Restore a snapshot to an environment (force-redeploy).
#   rollback           Roll a single service back to its previous deploy.
#   rollback-commit    Roll all services back to digests captured at a git SHA.
#   promote            Promote staging snapshot to production with prechecks.
#   pin                Pin a service to a specific image digest.
#   env-diff           Diff two environments and exit non-zero on drift.
#   resolve-digest     Resolve an image reference to its content digest via GHCR.
#   lint-prod          Verify production is fully digest-pinned (CI gate).
#
# Stdlib only — no Bundler, no Gemfile. Ruby 3.x.
#
# Auth: RAILWAY_TOKEN env var, or ~/.railway/config.json (token field).
# Never invokes `railway login` / `railway logout` / `op` / any external CLI.
#
# Production protection: --yes + typed env confirmation (unless --non-interactive).
#
# Exit codes:
#   0  clean / success
#   1  drift / findings
#   2  error (network, auth, schema, etc.)

require "json"
require "net/http"
require "optparse"
require "set"
require "uri"
require "yaml"
require "io/console"
require "fileutils"
require "time"

module Railway
    VERSION = "0.1.0"

    # ── Constants ──────────────────────────────────────────────────────────────

    WORKSPACE              = "CopilotKit"
    PROJECT_ID             = "6f8c6bff-a80d-4f8f-b78d-50b32bcf4479"
    PRODUCTION_ENV_ID      = "b14919f4-6417-429f-848d-c6ae2201e04f"
    STAGING_ENV_ID         = "8edfef02-ea09-4a20-8689-261f21cc2849"
    GHCR_ORG               = "copilotkit"

    GRAPHQL_ENDPOINT       = "https://backboard.railway.app/graphql/v2"

    ENV_IDS = {
        "production" => PRODUCTION_ENV_ID,
        "prod"       => PRODUCTION_ENV_ID,
        "staging"    => STAGING_ENV_ID,
        "stage"      => STAGING_ENV_ID,
    }.freeze

    # Env vars required to be present (parity-checked) before any promote.
    CRITICAL_ENV_KEYS = %w[
        RAILWAY_TOKEN
        GHCR_TOKEN
        SHARED_SECRET
        OPS_TRIGGER_TOKEN
        POCKETBASE_SUPERUSER_EMAIL
        POCKETBASE_SUPERUSER_PASSWORD
        GITHUB_APP_PRIVATE_KEY
        OPENAI_API_KEY
        ANTHROPIC_API_KEY
        GOOGLE_API_KEY
    ].freeze

    # EXPECTED_DOMAINS is derived from showcase/scripts/railway-envs.generated.json
    # (canonical source: showcase/scripts/railway-envs.ts). The CI guard
    # `npx tsx showcase/scripts/emit-railway-envs-json.ts --check` ensures the
    # JSON artifact stays in sync with the TS SSOT on every PR. Only public
    # (non-`*.up.railway.app`) hosts are included, preserving the original set.
    # Shared SSOT load — read once at class-load, derive both EXPECTED_DOMAINS
    # and STAGING_SERVICES from the same generated.json so they cannot drift.
    SSOT_DATA = begin
        ssot_json = File.expand_path("../scripts/railway-envs.generated.json", __dir__)
        unless File.exist?(ssot_json)
            raise "railway-envs.generated.json not found at #{ssot_json}. " \
                  "Re-run: npx tsx showcase/scripts/emit-railway-envs-json.ts"
        end
        JSON.parse(File.read(ssot_json)).freeze
    end

    EXPECTED_DOMAINS = begin
        by_env = { PRODUCTION_ENV_ID => [], STAGING_ENV_ID => [] }
        SSOT_DATA.fetch("services").each do |svc|
            domains = svc.fetch("domains")
            prod = domains.fetch("prod")
            staging = domains.fetch("staging")
            by_env[PRODUCTION_ENV_ID] << prod unless prod.end_with?(".up.railway.app")
            by_env[STAGING_ENV_ID] << staging unless staging.end_with?(".up.railway.app")
        end
        by_env.transform_values { |v| v.sort.uniq.freeze }.freeze
    end

    # Canonical staging service names — used to validate the optional
    # positional `bin/railway promote <service>` argument. Sourced from the
    # same SSOT as EXPECTED_DOMAINS so they cannot drift.
    STAGING_SERVICES = SSOT_DATA.fetch("services").map { |s| s.fetch("name") }.sort.freeze

    # Heuristic markers for env-scoped URLs (ignored in env-diff and promote).
    ENV_SCOPED_URL_MARKERS = [
        ".staging.copilotkit.ai",
        ".copilotkit.ai",
        "staging-",
        "prod-",
    ].freeze

    # ── Token / Auth ───────────────────────────────────────────────────────────

    module Auth
        module_function

        def token
            t = ENV["RAILWAY_TOKEN"]
            return t.strip if t && !t.strip.empty?

            cfg = File.expand_path("~/.railway/config.json")
            if File.exist?(cfg)
                begin
                    data = JSON.parse(File.read(cfg))
                    # Railway CLI stores the bearer in `user.accessToken` (43+ chars).
                    # `user.token` is a short legacy CLI session token that does
                    # NOT authenticate to the public GraphQL API. Prefer accessToken.
                    candidate =
                        data.dig("user", "accessToken") ||
                        data["accessToken"] ||
                        data.dig("user", "token") ||
                        data["token"] ||
                        data.dig("projects", PROJECT_ID, "token")
                    return candidate.strip if candidate.is_a?(String) && !candidate.strip.empty?
                rescue JSON::ParserError
                    # fall through
                end
            end

            nil
        end

        def require_token!
            t = token
            if t.nil? || t.empty?
                Railway.die!("RAILWAY_TOKEN not set and ~/.railway/config.json not usable. " \
                    "Export RAILWAY_TOKEN before running.")
            end
            t
        end

        # GHCR bearer. Distinct from the Railway token (Auth.token).
        # Resolution order:
        #   1. GHCR_TOKEN — explicit PAT (local dev or CI override).
        #   2. GITHUB_TOKEN — GitHub Actions automatic token (needs packages:read).
        # Returns nil if no token is available; caller MUST refuse rather than
        # silently fall through to anonymous (which works for public images but
        # not for digest-existence verification against private repos).
        def ghcr_token
            t = ENV["GHCR_TOKEN"]
            return t.strip if t && !t.strip.empty?
            t = ENV["GITHUB_TOKEN"]
            return t.strip if t && !t.strip.empty?
            nil
        end
    end

    # ── GraphQL Client ─────────────────────────────────────────────────────────

    class GraphQL
        class Error < StandardError; end

        def initialize(token: nil, endpoint: GRAPHQL_ENDPOINT, http: nil)
            @token = token || Auth.require_token!
            @endpoint = endpoint
            @http = http # optional injection for tests
        end

        # Execute a query. Returns parsed data hash; raises on errors.
        def query(query, variables = {})
            body = { query: query, variables: variables }.to_json

            if @http
                resp = @http.call(endpoint: @endpoint, token: @token, body: body)
            else
                uri = URI(@endpoint)
                req = Net::HTTP::Post.new(uri)
                req["Content-Type"] = "application/json"
                req["Authorization"] = "Bearer #{@token}"
                req.body = body

                resp = Net::HTTP.start(uri.host, uri.port, use_ssl: true,
                                        open_timeout: 10, read_timeout: 30) do |h|
                    h.request(req)
                end
            end

            status = resp.respond_to?(:code) ? resp.code.to_i : resp[:status].to_i
            body_str = resp.respond_to?(:body) ? resp.body : resp[:body]

            raise Error, "HTTP #{status}: #{body_str}" if status >= 400

            parsed = JSON.parse(body_str)
            if parsed["errors"] && !parsed["errors"].empty?
                msgs = parsed["errors"].map { |e| e["message"] }.join("; ")
                raise Error, "GraphQL: #{msgs}"
            end
            parsed["data"]
        end
    end

    # ── GHCR Client ────────────────────────────────────────────────────────────
    #
    # We need to resolve a tag (e.g. `ghcr.io/copilotkit/showcase-shell:latest`)
    # to its content-addressable digest (`sha256:...`). GHCR exposes the
    # OCI Distribution Spec at https://ghcr.io/v2/<org>/<image>/manifests/<tag>.
    # Requires a bearer token; for public images, a token issued via the
    # /token endpoint works.
    class GHCR
        class Error < StandardError; end

        ACCEPT_MANIFEST = [
            "application/vnd.oci.image.index.v1+json",
            "application/vnd.oci.image.manifest.v1+json",
            "application/vnd.docker.distribution.manifest.list.v2+json",
            "application/vnd.docker.distribution.manifest.v2+json",
        ].join(", ").freeze

        def initialize(token: Railway::Auth.ghcr_token, http: nil)
            @token = token
            @http = http
        end

        # Resolve "ghcr.io/copilotkit/showcase-shell:latest" to "sha256:<...>".
        # Returns nil if the tag does not exist.
        def resolve_digest(image_ref)
            parts = parse_image_ref(image_ref)
            return parts[:digest] if parts[:digest] # already pinned

            org   = parts[:org]
            name  = parts[:name]
            tag   = parts[:tag] || "latest"

            bearer = bearer_for(org, name)
            url = "https://ghcr.io/v2/#{org}/#{name}/manifests/#{tag}"

            if @http
                resp = @http.call(method: :head, url: url, headers: headers(bearer))
            else
                resp = http_head(url, headers: headers(bearer))
            end

            status = resp[:status]
            return nil if status == 404
            raise Error, "GHCR manifest HEAD #{status} for #{image_ref}" if status >= 400

            digest = resp[:headers]["docker-content-digest"] ||
                resp[:headers]["Docker-Content-Digest"]
            raise Error, "GHCR did not return Docker-Content-Digest for #{image_ref}" if digest.nil?

            digest
        end

        # Verify a digest-pinned image exists in GHCR. Returns:
        #   :exists       — HEAD returned 200.
        #   :missing      — HEAD returned 404 (digest GC'd or never built).
        #   :auth_failed  — HEAD returned 401/403 (token missing/insufficient).
        # Raises GHCR::Error on 5xx or transport failure.
        #
        # The caller MUST pass an image ref already pinned to @sha256:<digest>;
        # this is the "P1 precondition" check from the showcase deploy spec, not
        # a tag-resolution. We do not chase a tag here — we verify the exact
        # bytes about to be pinned to prod.
        def manifest_exists(image_ref)
            parts = parse_image_ref(image_ref)
            digest = parts[:digest]
            raise ArgumentError, "manifest_exists requires a digest-pinned ref, got #{image_ref}" if digest.nil?

            org  = parts[:org]
            name = parts[:name]

            bearer = bearer_for(org, name)
            url = "https://ghcr.io/v2/#{org}/#{name}/manifests/#{digest}"

            resp =
                if @http
                    @http.call(method: :head, url: url, headers: headers(bearer))
                else
                    http_head(url, headers: headers(bearer))
                end

            status = resp[:status]
            case status
            when 200       then :exists
            when 404       then :missing
            when 401, 403  then :auth_failed
            else
                raise Error, "GHCR manifest HEAD #{status} for #{image_ref}"
            end
        end

        # Parse image ref into { registry, org, name, tag, digest }.
        def parse_image_ref(ref)
            r = ref.to_s.strip
            registry = nil
            if r.start_with?("ghcr.io/")
                registry = "ghcr.io"
                r = r.sub(/^ghcr\.io\//, "")
            end

            digest = nil
            if r.include?("@")
                r, digest = r.split("@", 2)
            end

            tag = nil
            if r.include?(":")
                r, tag = r.rsplit_colon
            end

            org, name = r.split("/", 2)
            { registry: registry, org: org, name: name, tag: tag, digest: digest }
        end

        private

        def headers(bearer)
            h = { "Accept" => ACCEPT_MANIFEST }
            h["Authorization"] = "Bearer #{bearer}" if bearer
            h
        end

        def bearer_for(org, name)
            return @token if @token && !@token.empty?
            # Public images: anonymous token from /token endpoint.
            url = "https://ghcr.io/token?service=ghcr.io&scope=repository:#{org}/#{name}:pull"
            resp = http_get(url, headers: {})
            return nil if resp[:status] >= 400
            JSON.parse(resp[:body])["token"]
        rescue StandardError
            nil
        end

        def http_get(url, headers: {})
            uri = URI(url)
            req = Net::HTTP::Get.new(uri)
            headers.each { |k, v| req[k] = v }
            resp = Net::HTTP.start(uri.host, uri.port, use_ssl: true,
                                    open_timeout: 10, read_timeout: 30) do |h|
                h.request(req)
            end
            { status: resp.code.to_i, headers: resp.to_hash.transform_values(&:first), body: resp.body }
        end

        def http_head(url, headers: {})
            uri = URI(url)
            req = Net::HTTP::Head.new(uri)
            headers.each { |k, v| req[k] = v }
            resp = Net::HTTP.start(uri.host, uri.port, use_ssl: true,
                                    open_timeout: 10, read_timeout: 30) do |h|
                h.request(req)
            end
            hdrs = {}
            resp.each_header { |k, v| hdrs[k] = v; hdrs[k.downcase] = v }
            { status: resp.code.to_i, headers: hdrs, body: resp.body }
        end
    end

    # ── Helpers ────────────────────────────────────────────────────────────────

    module_function

    def die!(msg, code: 2)
        warn "railway: #{msg}"
        exit code
    end

    def env_id_for(name)
        ENV_IDS[name.to_s.downcase] || die!("Unknown env: #{name.inspect}. Use staging or production.")
    end

    def env_label(env_id)
        case env_id
        when PRODUCTION_ENV_ID then "production"
        when STAGING_ENV_ID    then "staging"
        else env_id
        end
    end

    def production?(env_id)
        env_id == PRODUCTION_ENV_ID
    end

    # Prompt for typed confirmation. Returns true if confirmed.
    def confirm_destructive!(env_label:, action:, non_interactive: false, yes: false)
        return true unless production?(env_id_for(env_label))

        unless yes
            die!("Refusing #{action} on production without --yes.", code: 2)
        end

        if non_interactive
            warn "[non-interactive] proceeding with #{action} on production (--yes given)."
            return true
        end

        $stderr.print "Type 'production' to confirm #{action}: "
        line = $stdin.gets&.strip
        unless line == "production"
            die!("Confirmation phrase mismatch. Aborting.", code: 2)
        end
        true
    end

    # Find service entry in a snapshot by name.
    def find_service(snapshot, name)
        (snapshot["services"] || []).find { |s| s["name"] == name }
    end

    # Strip env-scoped url-ish values when comparing two envs.
    def env_scoped?(value)
        return false unless value.is_a?(String)
        ENV_SCOPED_URL_MARKERS.any? { |m| value.include?(m) }
    end

    # ── Snapshot Schema ────────────────────────────────────────────────────────
    #
    # snapshot:
    #   version: 1
    #   captured_at: <ISO8601>
    #   project_id: <uuid>
    #   environment:
    #     id: <uuid>
    #     name: production|staging
    #   services:
    #     - name: showcase-shell
    #       service_id: <uuid>
    #       image: ghcr.io/copilotkit/showcase-shell@sha256:...
    #       image_tag: ghcr.io/copilotkit/showcase-shell:latest
    #       digest: sha256:...
    #       start_command: <string|nil>
    #       auto_updates_disabled: <bool>
    #       latest_deployment_id: <uuid|nil>
    #       env_keys: [KEY1, KEY2, ...]   # keys only, never values
    #       custom_domains: [...]
    #
    # We capture KEYS only for env vars (never values) so snapshots are safe
    # to commit/share.
    class SnapshotIO
        SCHEMA_VERSION = 2

        # v1: initial fields (name, service_id, image, image_tag, digest,
        #     start_command, auto_updates_disabled, latest_deployment_id,
        #     env_keys, custom_domains).
        # v2: adds healthcheck_path, region, replicas, restart_policy
        #     (for the P6 parity matrix in promote). We accept v1 reads
        #     for backwards-compat with historical committed snapshots —
        #     promote always uses live snapshots, but rollback-commit
        #     replays snapshots from arbitrary SHAs.
        SUPPORTED_VERSIONS = [1, SCHEMA_VERSION].freeze

        def self.write(path, snapshot)
            FileUtils.mkdir_p(File.dirname(path)) unless path == "-"
            yaml = YAML.dump(snapshot)
            if path == "-"
                $stdout.write(yaml)
            else
                File.write(path, yaml)
            end
        end

        def self.read(path)
            raw = path == "-" ? $stdin.read : File.read(path)
            data = YAML.safe_load(raw, permitted_classes: [Time, Symbol], aliases: false)
            unless data.is_a?(Hash) && SUPPORTED_VERSIONS.include?(data["version"])
                Railway.die!("Snapshot schema mismatch (expected version in #{SUPPORTED_VERSIONS.inspect}).")
            end
            data
        end
    end

    # ── Service Inventory ──────────────────────────────────────────────────────
    #
    # GraphQL fragments used by snapshot/env-diff/promote/lint-prod.
    #
    # Railway's public schema notes (verified via introspection 2026-05):
    #   * Project has NO `domains` field. Custom domains are reached either via
    #     top-level `domains(projectId, environmentId, serviceId)` returning
    #     `AllDomains { customDomains, serviceDomains }`, or via
    #     `serviceInstance.domains` (same shape).
    #   * Service has NO `serviceInstances` field. To get an instance's
    #     image/startCommand/etc. for a given env, use
    #     `serviceInstance(serviceId, environmentId)` directly.
    #   * `serviceInstanceDeployV2` takes (serviceId, environmentId, commitSha)
    #     ONLY — no `image` arg. To pin a service to a specific image, use
    #     `serviceInstanceUpdate(serviceId, environmentId, input: { source: { image } })`
    #     followed by `serviceInstanceRedeploy`.
    #   * `Environment.variables` returns an `EnvironmentVariablesConnection`
    #     whose edges include `node.serviceId` — we filter by service id to get
    #     per-service env-key sets.
    #
    # The query below enumerates all services in the project and, for each one,
    # the per-environment serviceInstance and that env's variables (keys only).
    # We use GraphQL field aliases to fetch all per-service data in a single
    # round-trip rather than N+1.

    SERVICES_LIST_QUERY = <<~GQL
        query ProjectServices($projectId: String!) {
            project(id: $projectId) {
                id
                name
                services {
                    edges {
                        node { id name }
                    }
                }
            }
        }
    GQL

    SERVICE_INSTANCE_QUERY = <<~GQL
        query ServiceInstance($serviceId: String!, $envId: String!) {
            serviceInstance(serviceId: $serviceId, environmentId: $envId) {
                id
                serviceId
                environmentId
                startCommand
                healthcheckPath
                region
                numReplicas
                restartPolicyType
                source { image repo }
                latestDeployment { id status }
                domains {
                    customDomains { id domain }
                    serviceDomains { id domain }
                }
            }
        }
    GQL

    ENVIRONMENT_VARIABLES_QUERY = <<~GQL
        query EnvVariables($envId: String!) {
            environment(id: $envId) {
                id
                name
                variables(first: 1000) {
                    edges {
                        node { name serviceId isSealed }
                    }
                }
            }
        }
    GQL

    # ── Subcommands ────────────────────────────────────────────────────────────

    class BaseCommand
        attr_reader :argv, :options

        def initialize(argv)
            @argv = argv.dup
            @options = default_options
        end

        def default_options
            {
                env: nil,
                yes: false,
                non_interactive: false,
                dry_run: false,
                output: nil,
            }
        end

        def self.call(argv)
            new(argv).run
        end

        # Each subcommand must implement run and parser.
        def run
            raise NotImplementedError
        end

        def parser
            raise NotImplementedError
        end

        # Returns a Railway::GraphQL client (mockable via @gql=).
        def gql
            @gql ||= GraphQL.new
        end

        # Returns a Railway::GHCR client.
        def ghcr
            @ghcr ||= GHCR.new
        end
    end

    # snapshot — capture an environment's state into a YAML file.
    class SnapshotCommand < BaseCommand
        def parser
            OptionParser.new do |o|
                o.banner = <<~BANNER
                    Usage: bin/railway snapshot --env <staging|production> [--output FILE] [--dry-run]

                      Capture current image digests, start commands, env-var KEYS,
                      and custom domains for every service in the given env.

                      Exits 0 on success. Exits 2 on error.
                BANNER
                o.on("--env ENV", "Environment (staging|production)") { |v| options[:env] = v }
                o.on("--output FILE", "Write to FILE (default stdout)") { |v| options[:output] = v }
                o.on("--dry-run", "Capture but do not write to disk") { options[:dry_run] = true }
                o.on("-h", "--help", "Show this help") { puts o; exit 0 }
            end
        end

        def run
            parser.parse!(argv)
            Railway.die!("--env is required") unless options[:env]
            env_id = Railway.env_id_for(options[:env])

            snap = build_snapshot(env_id)

            if options[:dry_run] && options[:output].nil?
                # always dump to stdout for dry-run with no output
                puts YAML.dump(snap)
                return 0
            end

            path = options[:output] ||
                "showcase/.railway-snapshots/#{Time.now.utc.strftime('%Y%m%dT%H%M%SZ')}-#{options[:env]}.yaml"

            SnapshotIO.write(path, snap)
            warn "wrote snapshot: #{path}" unless path == "-"
            0
        end

        def build_snapshot(env_id)
            # 1. List all services in the project.
            list = gql.query(SERVICES_LIST_QUERY, projectId: PROJECT_ID)
            service_nodes = (list.dig("project", "services", "edges") || []).map { |e| e["node"] }

            # 2. Fetch the env's variables once, group keys by serviceId.
            env_data = gql.query(ENVIRONMENT_VARIABLES_QUERY, envId: env_id)
            keys_by_service = Hash.new { |h, k| h[k] = [] }
            (env_data.dig("environment", "variables", "edges") || []).each do |edge|
                n = edge["node"]
                next unless n && n["serviceId"]
                keys_by_service[n["serviceId"]] << n["name"]
            end

            # 3. For each service, fetch its serviceInstance for this env.
            #    Some services may not exist in this env (returns nil); skip them.
            services = []
            service_nodes.each do |node|
                instance_data = gql.query(SERVICE_INSTANCE_QUERY,
                    serviceId: node["id"], envId: env_id)
                inst = instance_data["serviceInstance"]
                next if inst.nil?

                image_ref = inst.dig("source", "image")
                digest = nil
                tag_ref = image_ref
                if image_ref&.include?("@sha256:")
                    tag_ref, digest = image_ref.split("@", 2)
                end

                custom_domains = (inst.dig("domains", "customDomains") || [])
                    .map { |d| d["domain"] }.compact.sort

                services << {
                    "name"                  => node["name"],
                    "service_id"            => node["id"],
                    "image"                 => image_ref,
                    "image_tag"             => tag_ref,
                    "digest"                => digest,
                    "start_command"         => inst["startCommand"],
                    "healthcheck_path"      => inst["healthcheckPath"],
                    "region"                => inst["region"],
                    "replicas"              => inst["numReplicas"],
                    "restart_policy"        => inst["restartPolicyType"],
                    "auto_updates_disabled" => nil,
                    "latest_deployment_id"  => inst.dig("latestDeployment", "id"),
                    "env_keys"              => (keys_by_service[node["id"]] || []).sort.uniq,
                    "custom_domains"        => custom_domains,
                }
            end

            {
                "version"     => SnapshotIO::SCHEMA_VERSION,
                "captured_at" => Time.now.utc.iso8601,
                "project_id"  => PROJECT_ID,
                "environment" => { "id" => env_id, "name" => Railway.env_label(env_id) },
                "services"    => services.sort_by { |s| s["name"].to_s },
            }
        end
    end

    # restore — given a snapshot YAML, force-redeploy each service to its
    # captured digest via serviceInstanceUpdate(source.image) + redeploy.
    #
    # Railway's `serviceInstanceDeployV2` mutation does NOT accept an `image`
    # arg — its signature is (serviceId, environmentId, commitSha). To pin a
    # service to a specific image digest we must:
    #   1. update the service instance's source.image to the desired ref
    #   2. trigger a redeploy
    class RestoreCommand < BaseCommand
        UPDATE_IMAGE_MUTATION = <<~GQL
            mutation UpdateImage($serviceId: String!, $envId: String!, $image: String!) {
                serviceInstanceUpdate(
                    serviceId: $serviceId
                    environmentId: $envId
                    input: { source: { image: $image } }
                )
            }
        GQL

        REDEPLOY_MUTATION = <<~GQL
            mutation Redeploy($serviceId: String!, $envId: String!) {
                serviceInstanceRedeploy(
                    serviceId: $serviceId
                    environmentId: $envId
                )
            }
        GQL

        # Helper used by RestoreCommand, PinCommand, PromoteCommand: pin then redeploy.
        def self.pin_and_redeploy(gql, service_id:, env_id:, image:)
            gql.query(UPDATE_IMAGE_MUTATION,
                serviceId: service_id, envId: env_id, image: image)
            gql.query(REDEPLOY_MUTATION,
                serviceId: service_id, envId: env_id)
        end

        def parser
            OptionParser.new do |o|
                o.banner = <<~BANNER
                    Usage: bin/railway restore --env ENV --snapshot FILE [--yes] [--non-interactive] [--dry-run] [--service NAME]

                      Restore an environment to the digests captured in a snapshot.
                      Production requires --yes + typed 'production' confirmation
                      (or --non-interactive to skip the prompt).

                      Exits 0 on success. Exits 2 on auth/network/refused errors.
                BANNER
                o.on("--env ENV") { |v| options[:env] = v }
                o.on("--snapshot FILE") { |v| options[:snapshot] = v }
                o.on("--service NAME", "Restrict to a single service") { |v| options[:service] = v }
                o.on("--yes") { options[:yes] = true }
                o.on("--non-interactive") { options[:non_interactive] = true }
                o.on("--dry-run") { options[:dry_run] = true }
                o.on("-h", "--help") { puts o; exit 0 }
            end
        end

        def run
            parser.parse!(argv)
            Railway.die!("--env required") unless options[:env]
            Railway.die!("--snapshot required") unless options[:snapshot]

            env_id = Railway.env_id_for(options[:env])
            snap = SnapshotIO.read(options[:snapshot])

            Railway.confirm_destructive!(
                env_label: options[:env],
                action: "restore",
                non_interactive: options[:non_interactive],
                yes: options[:yes],
            )

            services = snap["services"] || []
            services = services.select { |s| s["name"] == options[:service] } if options[:service]
            Railway.die!("No matching services in snapshot.") if services.empty?

            services.each do |svc|
                image = svc["image"] || svc["image_tag"]
                Railway.die!("Service #{svc['name']} has no image in snapshot.") if image.nil?

                if options[:dry_run]
                    puts "[dry-run] would redeploy #{svc['name']} -> #{image}"
                    next
                end

                RestoreCommand.pin_and_redeploy(gql,
                    service_id: svc["service_id"], env_id: env_id, image: image)
                puts "redeployed #{svc['name']} -> #{image}"
            end
            0
        end
    end

    # rollback — roll a single service back one deploy (its previous deploy).
    class RollbackCommand < BaseCommand
        DEPLOYMENTS_QUERY = <<~GQL
            query Deployments($serviceId: String!, $envId: String!) {
                deployments(
                    first: 10
                    input: { serviceId: $serviceId, environmentId: $envId }
                ) { edges { node { id status meta createdAt } } }
            }
        GQL

        # deploymentRollback returns a scalar Boolean — no selection set.
        ROLLBACK_MUTATION = <<~GQL
            mutation Rollback($id: String!) { deploymentRollback(id: $id) }
        GQL

        def parser
            OptionParser.new do |o|
                o.banner = <<~BANNER
                    Usage: bin/railway rollback --env ENV --service NAME [--to DEPLOYMENT_ID] [--yes]

                      Rolls a single service back to its previous successful
                      deploy (or to a specific deployment id with --to).
                      Production requires --yes + typed confirmation.
                BANNER
                o.on("--env ENV") { |v| options[:env] = v }
                o.on("--service NAME") { |v| options[:service] = v }
                o.on("--to ID", "Specific deployment id to roll to") { |v| options[:to] = v }
                o.on("--yes") { options[:yes] = true }
                o.on("--non-interactive") { options[:non_interactive] = true }
                o.on("--dry-run") { options[:dry_run] = true }
                o.on("-h", "--help") { puts o; exit 0 }
            end
        end

        def run
            parser.parse!(argv)
            Railway.die!("--env required") unless options[:env]
            Railway.die!("--service required") unless options[:service]

            env_id = Railway.env_id_for(options[:env])
            Railway.confirm_destructive!(
                env_label: options[:env],
                action: "rollback",
                non_interactive: options[:non_interactive],
                yes: options[:yes],
            )

            service_id = resolve_service_id(env_id, options[:service])

            target_id = options[:to] || find_previous_deployment(service_id, env_id)
            Railway.die!("No previous deployment found for #{options[:service]}.") unless target_id

            if options[:dry_run]
                puts "[dry-run] would rollback #{options[:service]} -> deployment #{target_id}"
                return 0
            end

            gql.query(ROLLBACK_MUTATION, id: target_id)
            puts "rolled back #{options[:service]} -> #{target_id}"
            0
        end

        def resolve_service_id(_env_id, name)
            data = gql.query(SERVICES_LIST_QUERY, projectId: PROJECT_ID)
            (data.dig("project", "services", "edges") || []).each do |e|
                node = e["node"]
                return node["id"] if node["name"] == name
            end
            Railway.die!("Service #{name.inspect} not found.")
        end

        def find_previous_deployment(service_id, env_id)
            data = gql.query(DEPLOYMENTS_QUERY, serviceId: service_id, envId: env_id)
            deployments = (data.dig("deployments", "edges") || []).map { |e| e["node"] }
            # Pick the second SUCCESS in reverse-chronological order.
            successes = deployments.select { |d| d["status"] == "SUCCESS" }
            return nil if successes.size < 2
            successes[1]["id"]
        end
    end

    # rollback-commit — roll all services back to the digests captured at a
    # given git SHA's snapshot file.
    class RollbackCommitCommand < BaseCommand
        def parser
            OptionParser.new do |o|
                o.banner = <<~BANNER
                    Usage: bin/railway rollback-commit --env ENV --sha SHA [--yes]

                      Looks up the snapshot file checked into git at SHA, then
                      redeploys every service to the digests recorded there.
                      Effectively a "restore to point-in-time" using committed
                      snapshots.
                BANNER
                o.on("--env ENV") { |v| options[:env] = v }
                o.on("--sha SHA") { |v| options[:sha] = v }
                o.on("--yes") { options[:yes] = true }
                o.on("--non-interactive") { options[:non_interactive] = true }
                o.on("--dry-run") { options[:dry_run] = true }
                o.on("-h", "--help") { puts o; exit 0 }
            end
        end

        def run
            parser.parse!(argv)
            Railway.die!("--env required") unless options[:env]
            Railway.die!("--sha required") unless options[:sha]

            # Locate the most recent snapshot file for env at SHA via `git show`.
            env = options[:env]
            sha = options[:sha]

            # We look in showcase/.railway-snapshots/*-<env>.yaml at SHA, take last.
            list_cmd = %(git ls-tree -r --name-only #{sha} -- 'showcase/.railway-snapshots/*-#{env}.yaml')
            entries = `#{list_cmd}`.lines.map(&:strip).reject(&:empty?).sort
            Railway.die!("No snapshot for env=#{env} at #{sha}.") if entries.empty?
            path = entries.last

            yaml = `git show #{sha}:#{path}`
            Railway.die!("git show failed for #{sha}:#{path}") if yaml.nil? || yaml.empty?

            snap = YAML.safe_load(yaml, permitted_classes: [Time, Symbol], aliases: false)

            # Hand off to RestoreCommand's logic by writing to a tmpfile.
            tmp = "/tmp/railway-rollback-commit-#{Process.pid}.yaml"
            File.write(tmp, YAML.dump(snap))
            restore_argv = ["--env", env, "--snapshot", tmp]
            restore_argv << "--yes" if options[:yes]
            restore_argv << "--non-interactive" if options[:non_interactive]
            restore_argv << "--dry-run" if options[:dry_run]
            RestoreCommand.new(restore_argv).run
        ensure
            FileUtils.rm_f(tmp) if defined?(tmp) && tmp
        end
    end

    # promote — copy staging digests into production with prechecks.
    class PromoteCommand < BaseCommand
        class MutationError < StandardError; end

        SERVICE_INSTANCE_RECHECK_QUERY = <<~GQL
            query ServiceInstanceRecheck($serviceId: String!, $envId: String!) {
                serviceInstance(serviceId: $serviceId, environmentId: $envId) {
                    id
                    source { image }
                    updatedAt
                }
            }
        GQL

        RETRY_COUNT     = 3
        RETRY_DELAY_SEC = 10

        # Pin + verify: confirms the boolean mutation result AND re-queries
        # serviceInstance to confirm BOTH source.image advanced to the new
        # digest AND updatedAt strictly advanced past the pre-mutation value.
        # 3 retries 10s apart absorb Railway's eventual consistency.
        # `sleeper:` is injected for tests.
        def self.pin_and_verify(gql, service_id:, env_id:, image:, sleeper: ->(n) { sleep(n) })
            # Upfront guard: this method's contract is to pin a DIGEST-form
            # image. Pre-fix, a stray tag ref would fall through to retries
            # (since expected_digest stayed nil and image_ok was always false)
            # and ultimately surface as a misleading "did not observe image
            # advance" error after 30s of futile waits.
            unless image.include?("@sha256:")
                raise ArgumentError,
                    "pin_and_verify requires an @sha256:-pinned image, got #{image.inspect}"
            end

            # Capture pre-update serviceInstance.updatedAt so we can gate on a
            # strict advance after the mutation (spec §7.2 P5). Image-equality
            # alone is insufficient — a no-op re-pin to the current value would
            # otherwise appear green. nil pre_ts means "no prior instance" and
            # is treated as a permanent advance below.
            pre_data = gql.query(SERVICE_INSTANCE_RECHECK_QUERY,
                serviceId: service_id, envId: env_id)
            pre_inst = pre_data && pre_data["serviceInstance"]
            pre_update_ts = pre_inst && pre_inst["updatedAt"]

            updated = gql.query(RestoreCommand::UPDATE_IMAGE_MUTATION,
                serviceId: service_id, envId: env_id, image: image)
            unless updated && updated["serviceInstanceUpdate"] == true
                raise MutationError,
                    "P5: serviceInstanceUpdate returned #{updated.inspect} (expected true) for #{service_id} -> #{image}"
            end

            # Symmetric assertion: serviceInstanceUpdate has already advanced
            # source.image+updatedAt, so a failed redeploy could otherwise
            # sneak through verification. Require truthy redeploy result.
            redeployed = gql.query(RestoreCommand::REDEPLOY_MUTATION,
                serviceId: service_id, envId: env_id)
            unless redeployed && redeployed["serviceInstanceRedeploy"]
                raise MutationError,
                    "P5: serviceInstanceRedeploy returned #{redeployed.inspect} (expected truthy) for #{service_id} -> #{image}"
            end

            expected_digest = image.include?("@") ? image.split("@", 2).last : nil
            last_seen_image = nil
            last_seen_ts    = nil

            RETRY_COUNT.times do |i|
                data = gql.query(SERVICE_INSTANCE_RECHECK_QUERY,
                    serviceId: service_id, envId: env_id)
                inst = data && data["serviceInstance"]
                actual_image  = inst && inst.dig("source", "image")
                actual_ts     = inst && inst["updatedAt"]
                actual_digest = actual_image && actual_image.include?("@") ? actual_image.split("@", 2).last : nil
                last_seen_image = actual_image
                last_seen_ts    = actual_ts

                image_ok = expected_digest && actual_digest == expected_digest
                # Non-vacuous timestamp gate: a non-nil observed updatedAt is
                # ALWAYS required. Pre-fix this collapsed to (pre_ts.nil? ||
                # ...), which made the gate vacuous for new prod services and
                # fell back to digest-equality alone — the exact weakness this
                # gate was added to prevent.
                ts_ok    = actual_ts && (pre_update_ts.nil? || actual_ts > pre_update_ts)
                return inst if image_ok && ts_ok

                sleeper.call(RETRY_DELAY_SEC) if i < RETRY_COUNT - 1
            end

            raise MutationError,
                "P5: re-query did not observe image advance to #{image} AND updatedAt > #{pre_update_ts.inspect} " \
                "after #{RETRY_COUNT} retries (last seen image=#{last_seen_image.inspect}, " \
                "updatedAt=#{last_seen_ts.inspect}). Refusing to declare promote successful."
        end

        def default_options
            super.merge(
                confirm_divergence:    false,
                require_staging_green: true,   # default-on per spec §7.2 P3
                service:               nil,    # optional positional: restrict to ONE service
                digest:                nil,    # override resolved prod-image ref (single-service only)
            )
        end

        def parser
            OptionParser.new do |o|
                o.banner = <<~BANNER
                    Usage: bin/railway promote [SERVICE] [--digest REF] [--yes] [--non-interactive] [--dry-run]

                      Promote staging snapshot to production.

                      SERVICE (optional positional): restrict the promote to a single
                        staging service by name. Without it, the entire staging fleet
                        is promoted (interactive operator use).
                      --digest REF: pin SERVICE's prod image to REF instead of the
                        digest resolved from staging's :latest tag. Only valid with a
                        positional SERVICE; errors otherwise.

                      MOVES: image digests (resolved to @sha256: at promote time);
                             autoUpdate=disabled flag.
                      VERIFY-REFUSE: service-set parity, critical env-key parity,
                             startCommand parity, PB superuser auth,
                             PB collection parity, cross-env URL leak scan.
                      WARN: missing/extra custom domains, sealed-var heuristics.
                      IGNORE: env-scoped URLs, volumes.

                      Exit 0 on clean promotion, 1 on refuse/findings, 2 on error.
                BANNER
                o.on("--confirm-divergence", "Proceed past WARN findings (region/replicas/etc.)") { options[:confirm_divergence] = true }
                o.on("--require-staging-green", "Require live staging probe green at promote time (default)") { options[:require_staging_green] = true }
                o.on("--no-require-staging-green", "Skip live staging probe (NOT recommended)") { options[:require_staging_green] = false }
                o.on("--digest REF", "Pin SERVICE's prod image to REF (requires positional SERVICE).") { |v| options[:digest] = v }
                o.on("--yes") { options[:yes] = true }
                o.on("--non-interactive") { options[:non_interactive] = true }
                o.on("--dry-run") { options[:dry_run] = true }
                o.on("-h", "--help") { puts o; exit 0 }
            end
        end

        def run
            parser.parse!(argv)

            # Optional positional service name; remaining argv after parse! is
            # whatever the OptionParser left behind. The workflow shape is
            # `promote <svc> [--digest REF]` — one positional only.
            if argv.length > 1
                Railway.die!("promote: too many positional args #{argv.inspect}; expected at most one service name.")
            end
            options[:service] = argv.first if argv.first && !argv.first.empty?

            # --digest is meaningful only with a positional service. Without
            # one it would otherwise silently promote the whole fleet pinned
            # to one (likely wrong) digest — fail fast instead.
            if options[:digest] && options[:service].nil?
                Railway.die!("promote: --digest requires a positional service argument " \
                             "(e.g. `bin/railway promote <service> --digest <ref>`). " \
                             "Refusing to promote the fleet with a single digest.")
            end

            # Validate the positional service against the SSOT
            # (railway-envs.generated.json). Unknown names must hard-fail with
            # the valid set listed so the operator can self-correct; falling
            # through would silently fleet-promote (the original bug).
            if options[:service] && !STAGING_SERVICES.include?(options[:service])
                Railway.die!("promote: unknown service #{options[:service].inspect}. " \
                             "Valid staging services: #{STAGING_SERVICES.join(', ')}.")
            end

            capture_snapshots

            # When restricted to a single service, narrow PER-SERVICE views
            # of both snapshots BEFORE preflight so per-service checks
            # (P1..P3/P6, critical env keys, execute_promotion) operate
            # only on the targeted service. FLEET-SCOPED invariants
            # (check_expected_prod_domains, check_service_set_parity) MUST
            # continue to evaluate against the FULL un-narrowed snapshots
            # — they describe properties of the fleet itself, not the
            # promote target. Co-narrowing them produces two failure
            # modes that the per-service workflow can't tolerate:
            #
            #   (1) check_expected_prod_domains diffs the fleet-wide
            #       EXPECTED_DOMAINS against the union of custom_domains
            #       in `prod["services"]`. Narrowed prod carries at most
            #       one service's domains → the other ~4 fleet hosts look
            #       "missing" → spurious WARN → run_with_preflight_only
            #       refuses (workflow doesn't pass --confirm-divergence).
            #   (2) check_service_set_parity is a fleet-shape invariant
            #       (no env has services the other lacks); evaluating it
            #       on the narrowed pair makes it tautological when both
            #       contain the same single name, and only accidentally
            #       fires in the target-absent-from-prod case.
            #
            # We therefore retain references to the full snapshots and
            # only narrow when the per-service branch is taken. The
            # fleet-scoped checks always read @full_*_snapshot; everything
            # else reads @staging_snapshot/@prod_snapshot (narrowed when
            # a positional service is given, full otherwise).
            @full_staging_snapshot = @staging_snapshot
            @full_prod_snapshot    = @prod_snapshot
            if options[:service]
                narrow_snapshots_to_single_service!(options[:service])
            end

            run_with_preflight_only
        end

        # Narrow @staging_snapshot and @prod_snapshot to only the named
        # service. Staging presence is mandatory (validated against SSOT
        # already); prod absence is tolerated here so the fleet-scoped
        # service-set-parity REFUSE (now reading @full_prod_snapshot) can
        # surface as the user-facing error rather than a silent rc=0
        # 'success' (find_service returns nil, so execute_promotion would
        # otherwise skip the absent service with no mutation).
        def narrow_snapshots_to_single_service!(name)
            staging_match = (@staging_snapshot["services"] || []).select { |s| s["name"] == name }
            if staging_match.empty?
                Railway.die!("promote: service #{name.inspect} not present in staging snapshot " \
                             "(SSOT lists it but Railway env returned no instance).")
            end
            @staging_snapshot = @staging_snapshot.merge("services" => staging_match)
            prod_match = (@prod_snapshot["services"] || []).select { |s| s["name"] == name }
            @prod_snapshot = @prod_snapshot.merge("services" => prod_match)
        end

        # Test seam — capture_snapshots may be skipped by injecting
        # @staging_snapshot / @prod_snapshot directly.
        def capture_snapshots
            @staging_snapshot ||= SnapshotCommand.new(["--env", "staging", "--dry-run"]).build_snapshot(STAGING_ENV_ID)
            @prod_snapshot    ||= SnapshotCommand.new(["--env", "production", "--dry-run"]).build_snapshot(PRODUCTION_ENV_ID)
        end

        # All preconditions, then (if clean) the actual promote.
        def run_with_preflight_only
            findings = []

            findings.concat(check_p1_ghcr_digests(@staging_snapshot))
            findings.concat(check_p2_staging_deployments(@staging_snapshot))
            findings.concat(check_p3_staging_live_green(@staging_snapshot))
            findings.concat(check_p6_parity(@staging_snapshot, @prod_snapshot))
            # FLEET-SCOPED invariants — must see the FULL fleet so they
            # produce the same verdict regardless of whether this run is
            # full-fleet or restricted to a single service. See `run` for
            # the @full_*_snapshot rationale. When `run` was bypassed
            # (test seam — preflight invoked directly without capture),
            # fall back to the current snapshot views so legacy tests
            # that only set @staging_snapshot/@prod_snapshot continue to
            # work (full-fleet semantics = full == narrowed).
            full_staging = @full_staging_snapshot || @staging_snapshot
            full_prod    = @full_prod_snapshot    || @prod_snapshot
            findings.concat(check_service_set_parity(full_staging, full_prod))
            findings.concat(check_critical_env_key_parity(@staging_snapshot, @prod_snapshot))
            findings.concat(check_expected_prod_domains(full_prod))

            # Mandatory parity-NOTE that env var VALUES are not compared.
            puts "NOTE: env var VALUES are not compared between staging and prod " \
                 "(intentional — staging and prod hold different secrets/URLs). " \
                 "Only the set of keys is compared."

            refuses = findings.select { |f| f.start_with?("REFUSE") }
            warns   = findings.select { |f| f.start_with?("WARN") }

            unless refuses.empty?
                refuses.each { |f| puts f }
                puts "Promote refused due to #{refuses.size} REFUSE finding(s)."
                return 1
            end

            unless warns.empty?
                warns.each { |f| puts f }
                unless options[:confirm_divergence]
                    puts "Promote refused: #{warns.size} WARN finding(s). " \
                         "Re-run with --confirm-divergence after inspecting."
                    return 1
                end
                puts "[--confirm-divergence set] proceeding past #{warns.size} WARN finding(s)."
            end

            Railway.confirm_destructive!(
                env_label: "production",
                action: "promote",
                non_interactive: options[:non_interactive],
                yes: options[:yes],
            )

            execute_promotion(@staging_snapshot, @prod_snapshot)
        end

        # Resolve a staging service's image to the DIGEST-form ref that should
        # be pinned to prod. The showcase deploy model is STAGING = mutable
        # `:latest` tag, PROD = immutable `@sha256:<digest>`. Pinning prod to
        # a tag would defeat the entire pipeline invariant — and verify-railway-
        # image-refs PROD_SHAPE would later REFUSE.
        #
        # Returns the canonical `<ghcr.io/org/name>@sha256:<digest>` ref.
        # Returns nil if the staging tag cannot be resolved (e.g. GHCR 404 /
        # auth failed); the caller MUST refuse rather than pin a tag.
        def resolved_prod_image(svc)
            # --digest override: when the operator supplied an explicit ref AND
            # we are restricted to a single service, use that ref verbatim. The
            # `run` entry-point already enforces that --digest requires a
            # positional service, and snapshots are narrowed to that service
            # before preflight, so `svc["name"] == options[:service]` here.
            if options[:digest] && options[:service] && svc["name"] == options[:service]
                return options[:digest]
            end

            image = svc["image"] || svc["image_tag"]
            return nil if image.nil? || image.empty?
            return image if image.include?("@sha256:")  # already pinned

            digest = ghcr.resolve_digest(image)
            return nil if digest.nil?

            # Strip the tag (`:latest`) and append the digest. parse_image_ref
            # returns the structured parts; rebuild the canonical pinned ref.
            parts = ghcr.parse_image_ref(image)
            registry = parts[:registry] || "ghcr.io"
            "#{registry}/#{parts[:org]}/#{parts[:name]}@#{digest}"
        end

        # P1 — every staging digest about to be promoted must exist in GHCR.
        # Verifies the DIGEST-form ref that will actually be pinned to prod
        # (resolves tag refs first; refuses if the tag can't be resolved).
        #
        # Side-effect: populates @promote_refs (service_name => digest-pinned
        # ref) so execute_promotion pins the EXACT ref P1 verified — staging
        # `:latest` is mutable, so a second resolve_digest could return a
        # different digest (TOCTOU) and prod would be pinned to a digest P1
        # never verified.
        def check_p1_ghcr_digests(staging)
            findings = []
            # RESET (not memoize): a reused PromoteCommand instance running a
            # second preflight against a different snapshot must not carry
            # stale A-era refs into a B-era promote.
            @promote_refs = {}
            (staging["services"] || []).each do |svc|
                image = svc["image"] || svc["image_tag"]
                if image.nil? || image.empty?
                    # Make this loud — pre-fix this was a silent `next`, and
                    # execute_promotion later emitted a misleading "internal
                    # error" for the same service.
                    findings << "REFUSE: P1 (#{svc['name']}): no image recorded in staging snapshot; " \
                                "cannot promote."
                    next
                end

                # Per-service rescue: an error on one service must not
                # discard findings already accumulated for earlier services.
                # Broadened to StandardError — a non-GHCR error (e.g. an
                # ArgumentError from parse_image_ref, or a network error
                # wrapped as something else) would otherwise bypass the
                # per-service rescue and crash the whole loop.
                begin
                    resolved = resolved_prod_image(svc)
                    if resolved.nil?
                        findings << "REFUSE: P1 (#{svc['name']}): cannot resolve #{image} to a GHCR digest; " \
                                    "refusing to pin prod to a mutable tag."
                        next
                    end

                    case ghcr.manifest_exists(resolved)
                    when :exists
                        # Record the verified ref for use by execute_promotion
                        # — guarantees pin parity with what P1 verified.
                        @promote_refs[svc["name"]] = resolved
                    when :missing
                        findings << "REFUSE: P1 (#{svc['name']}): #{resolved} not found in GHCR " \
                                    "(digest may have been garbage-collected; staging build may not have pushed)."
                    when :auth_failed
                        findings << "REFUSE: P1 (#{svc['name']}): GHCR auth failed for #{resolved}. " \
                                    "Set GHCR_TOKEN (local: 'gh auth token') or GITHUB_TOKEN (CI: workflow token with packages:read)."
                    end
                rescue StandardError => e
                    findings << "REFUSE: P1 (#{svc['name']}): unexpected #{e.class}: #{e.message}"
                end
            end
            findings
        end

        # P2 — for each staging service we are promoting, the most recent
        # staging deployment must be SUCCESS AND its image digest must match
        # the digest we are about to promote (otherwise a newer build is
        # in flight; refuse to avoid the race).
        def check_p2_staging_deployments(staging)
            findings = []
            (staging["services"] || []).each do |svc|
                next if svc["service_id"].nil? || svc["image"].nil?
                deployments = fetch_latest_staging_deployments(svc["service_id"])
                latest = deployments.first
                if latest.nil?
                    findings << "REFUSE: P2 (#{svc['name']}): no staging deployments found."
                    next
                end
                status = latest["status"]
                if status != "SUCCESS"
                    findings << "REFUSE: P2 (#{svc['name']}): latest staging deployment status is #{status}, not SUCCESS."
                    next
                end
                # Railway's Deployment.meta is a JSON scalar that may deserialize
                # as a String (not a Hash). Parse it first; fall back to the
                # WARN branch only if it is still non-Hash after the parse.
                meta = latest["meta"]
                meta = (JSON.parse(meta) rescue meta) if meta.is_a?(String)
                if meta.is_a?(Hash)
                    deployed_image = meta["image"].to_s
                    deployed_digest = deployed_image.include?("@") ? deployed_image.split("@", 2).last : nil
                    # Compare against the digest P1 resolved+verified (recorded
                    # in @promote_refs), NOT svc["digest"] — staging images are
                    # tag-form, so svc["digest"] is nil and the race-check would
                    # otherwise be dead code. P2 runs after P1 in the preflight,
                    # so the entry must exist; if it doesn't, P1 has already
                    # REFUSEd this service and we skip the race comparison.
                    promote_ref = (@promote_refs || {})[svc["name"]]
                    promote_digest = promote_ref && promote_ref.include?("@sha256:") ? promote_ref.split("@", 2).last : nil
                    if promote_digest && deployed_digest && deployed_digest != promote_digest
                        findings << "REFUSE: P2 (#{svc['name']}): in-flight race — latest staging deployment is " \
                                    "#{deployed_digest} but P1 resolved #{promote_digest}. Re-snapshot and retry."
                    end
                else
                    findings << "WARN: P2 (#{svc['name']}): in-flight race check skipped — " \
                                "deployment meta is #{meta.class}, expected Hash."
                end
            end
            findings
        end

        def fetch_latest_staging_deployments(service_id)
            data = gql.query(RollbackCommand::DEPLOYMENTS_QUERY,
                serviceId: service_id, envId: STAGING_ENV_ID)
            nodes = (data.dig("deployments", "edges") || []).map { |e| e["node"] }
            # Railway's deployments query has no explicit order; sort client-side
            # by createdAt descending so `.first` is genuinely the newest. nil
            # createdAt sorts last (treated as oldest).
            nodes.sort_by { |n| n["createdAt"].to_s }.reverse
        end

        # P3 — re-probe staging live at promote time. Authoritative; CI history
        # is not, because showcase_deploy.yml uses cancel-in-progress.
        # Default-on; can be disabled with --no-require-staging-green.
        def check_p3_staging_live_green(staging)
            unless options[:require_staging_green]
                puts "P3 SKIPPED (--no-require-staging-green set; staging is NOT being live-re-probed)."
                return []
            end
            services = (staging["services"] || []).map { |s| s["name"] }.compact.uniq
            return [] if services.empty?
            result = run_staging_probe(services: services)
            return [] if result[:ok]
            ["REFUSE: P3: staging is not green for #{services.join(', ')}: #{result[:summary]}"]
        end

        # Shell out to Workstream A's parameterized probe entrypoint.
        # Contract: exit 0 = green, non-zero = red; stdout is the human summary.
        #
        # Explicitly forward RAILWAY_TOKEN (the probe enumerates Railway state
        # over GraphQL) and GHCR_TOKEN / GITHUB_TOKEN (the probe may need to
        # cross-check GHCR for the digest under test). All three are inherited
        # from the parent process env when set; the explicit hash below makes
        # the dependency visible and survives any future child-env sanitizing.
        def run_staging_probe(services:)
            probe_bin = File.expand_path("../scripts/verify-deploy.ts", __dir__)
            unless File.exist?(probe_bin)
                return { ok: false, summary: "verify-deploy.ts not found at #{probe_bin} — Workstream A dependency missing." }
            end
            services_arg = services.join(",")
            child_env = {
                "RAILWAY_TOKEN" => ENV["RAILWAY_TOKEN"],
                "GHCR_TOKEN"    => ENV["GHCR_TOKEN"],
                "GITHUB_TOKEN"  => ENV["GITHUB_TOKEN"],
                "PATH"          => ENV["PATH"],
                "HOME"          => ENV["HOME"],
            }.compact
            # Use `tsx` (workspace dev dependency) for stdlib-free TS execution.
            # IO.popen preserves a clean child env (Kernel#`` would inherit the
            # parent shell verbatim — fine, but explicit is better for audit).
            # Wrap the launch in a rescue: a missing `npx` (Errno::ENOENT) or
            # any other spawn-time failure should produce a clean REFUSE
            # rather than a raw stack trace bubbling up out of P3.
            begin
                output = IO.popen(child_env, ["npx", "--yes", "tsx", probe_bin,
                    "--env", "staging",
                    "--services", services_arg,
                    err: [:child, :out]]) { |io| io.read }
            rescue Errno::ENOENT, StandardError => e
                return { ok: false, summary: "staging probe failed to launch: #{e.class}: #{e.message}" }
            end
            ok = $?.exitstatus == 0
            { ok: ok, summary: output.lines.last(10).join.strip }
        end

        # P6 — parity matrix.
        #   REFUSE: startCommand, healthcheckPath, image shape.
        #   WARN:   region, replicas, restartPolicy, env-var KEY set.
        #   IGNORE: env-var VALUES (printed as NOTE every run; see
        #           run_with_preflight_only).
        def check_p6_parity(staging, prod)
            findings = []
            (staging["services"] || []).each do |svc|
                pmatch = Railway.find_service(prod, svc["name"])
                next unless pmatch  # service-set parity catches this separately.
                name = svc["name"]

                # REFUSE — startCommand
                if svc["start_command"] != pmatch["start_command"]
                    findings << "REFUSE: P6 (#{name}): startCommand divergence " \
                                "(staging=#{svc['start_command'].inspect} prod=#{pmatch['start_command'].inspect})"
                end

                # REFUSE — healthcheckPath
                if svc["healthcheck_path"] != pmatch["healthcheck_path"]
                    findings << "REFUSE: P6 (#{name}): healthcheckPath divergence " \
                                "(staging=#{svc['healthcheck_path'].inspect} prod=#{pmatch['healthcheck_path'].inspect})"
                end

                # REFUSE — image shape (staging=:tag mutable, prod=@sha256 pinned).
                staging_shape = image_shape(svc["image"])
                prod_shape    = image_shape(pmatch["image"])
                expected_staging_shape = :tag
                expected_prod_shape    = :digest
                if staging_shape != expected_staging_shape || prod_shape != expected_prod_shape
                    findings << "REFUSE: P6 (#{name}): image shape wrong " \
                                "(staging=#{staging_shape} expected=#{expected_staging_shape}; " \
                                "prod=#{prod_shape} expected=#{expected_prod_shape})"
                end

                # WARN — region
                if svc["region"] != pmatch["region"]
                    findings << "WARN: P6 (#{name}): region divergence " \
                                "(staging=#{svc['region'].inspect} prod=#{pmatch['region'].inspect})"
                end

                # WARN — replicas
                if svc["replicas"] != pmatch["replicas"]
                    findings << "WARN: P6 (#{name}): replicas divergence " \
                                "(staging=#{svc['replicas']} prod=#{pmatch['replicas']})"
                end

                # WARN — restartPolicy
                if svc["restart_policy"] != pmatch["restart_policy"]
                    findings << "WARN: P6 (#{name}): restartPolicy divergence " \
                                "(staging=#{svc['restart_policy'].inspect} prod=#{pmatch['restart_policy'].inspect})"
                end

                # WARN — env var KEY set
                staging_keys = (svc["env_keys"] || []).sort
                prod_keys    = (pmatch["env_keys"] || []).sort
                if staging_keys != prod_keys
                    only_staging = staging_keys - prod_keys
                    only_prod    = prod_keys - staging_keys
                    findings << "WARN: P6 (#{name}): env key set divergence " \
                                "(only-in-staging=#{only_staging.inspect} only-in-prod=#{only_prod.inspect})"
                end
            end
            findings
        end

        # Classify an image ref. :tag (mutable), :digest (immutable @sha256:),
        # :missing, or :other.
        def image_shape(ref)
            return :missing if ref.nil? || ref.empty?
            return :digest  if ref.include?("@sha256:")
            return :tag     if ref.include?(":") && ref.split(":", 2).last !~ /\A\s*\z/
            :other
        end

        def check_service_set_parity(staging, prod)
            findings = []
            s_names = (staging["services"] || []).map { |s| s["name"] }.sort
            p_names = (prod["services"] || []).map { |s| s["name"] }.sort
            findings << "REFUSE: services in staging not in prod: #{(s_names - p_names).join(', ')}" unless (s_names - p_names).empty?
            findings << "REFUSE: services in prod not in staging: #{(p_names - s_names).join(', ')}" unless (p_names - s_names).empty?
            findings
        end

        def check_critical_env_key_parity(staging, prod)
            findings = []
            (staging["services"] || []).each do |svc|
                pmatch = Railway.find_service(prod, svc["name"])
                next unless pmatch
                missing = (CRITICAL_ENV_KEYS & (svc["env_keys"] || [])) - (pmatch["env_keys"] || [])
                findings << "REFUSE: #{svc['name']}: critical env keys missing in prod: #{missing.join(', ')}" unless missing.empty?
            end
            findings
        end

        def check_expected_prod_domains(prod)
            expected = EXPECTED_DOMAINS[PRODUCTION_ENV_ID] || []
            actual = (prod["services"] || []).flat_map { |s| s["custom_domains"] || [] }.uniq.sort
            missing = expected - actual
            return [] if missing.empty?
            ["WARN: production missing expected custom domains: #{missing.join(', ')}"]
        end

        def execute_promotion(staging, prod)
            # Hard guard: @promote_refs is populated by check_p1_ghcr_digests.
            # If it is nil, preflight did not run — refusing to silently pin
            # nothing (which a memoize default `||= {}` would have done).
            raise "internal error: execute_promotion invoked without preflight (@promote_refs nil)" if @promote_refs.nil?

            # Pre-validation pass: every prod-matched staging service MUST
            # have a digest-shaped @promote_refs entry BEFORE we pin anything.
            # Pre-fix, a missing entry was detected lazily mid-loop, which
            # could leave production partially-promoted (some services pinned
            # to new digest, others still on prior digest). Fail fast here.
            missing = []
            (staging["services"] || []).each do |svc|
                next unless Railway.find_service(prod, svc["name"])
                ref = @promote_refs[svc["name"]]
                if ref.nil? || !ref.include?("@sha256:")
                    missing << "#{svc['name']}=#{ref.inspect}"
                end
            end
            unless missing.empty?
                warn "REFUSE: promote: P1-verified digest ref missing or non-digest for " \
                     "#{missing.join(', ')} (internal error — preflight did not capture these services). " \
                     "Refusing to pin (no mutations issued)."
                return 1
            end

            already_pinned = []
            (staging["services"] || []).each do |svc|
                pmatch = Railway.find_service(prod, svc["name"])
                next unless pmatch
                # Use the EXACT ref P1 resolved+verified. Re-resolving here
                # would risk TOCTOU on staging's mutable `:latest`.
                image = @promote_refs[svc["name"]]
                if options[:dry_run]
                    puts "[dry-run] promote #{svc['name']} -> #{image}"
                    next
                end
                begin
                    PromoteCommand.pin_and_verify(gql,
                        service_id: pmatch["service_id"],
                        env_id: PRODUCTION_ENV_ID,
                        image: image)
                    puts "promoted #{svc['name']} -> #{image}"
                    already_pinned << svc["name"]
                # Broadened rescue: a transient GraphQL or network error
                # mid-loop must NOT crash the script (which would lose the
                # PARTIAL-PROMOTION report — its entire reason for existing).
                rescue PromoteCommand::MutationError, Railway::GraphQL::Error, StandardError => e
                    warn "PARTIAL PROMOTION: already pinned #{already_pinned.inspect}; " \
                         "FAILED on #{svc['name']}: #{e.class}: #{e.message}. " \
                         "Production is in a mixed state — note that serviceInstanceUpdate " \
                         "runs BEFORE serviceInstanceRedeploy, so #{svc['name']}'s prod " \
                         "source.image may already be partially advanced on Railway's side. " \
                         "Run 'bin/railway rollback-commit' against the prior snapshot to revert, " \
                         "or re-run promote to retry."
                    return 1
                end
            end
            0
        end
    end

    # pin — pin a service to a specific image digest.
    class PinCommand < BaseCommand
        def parser
            OptionParser.new do |o|
                o.banner = <<~BANNER
                    Usage: bin/railway pin --env ENV --service NAME --image REF [--yes] [--dry-run]

                      Pin a service to a specific image (tag or @sha256:... digest).
                      If a tag is given, resolves it via GHCR first.
                BANNER
                o.on("--env ENV") { |v| options[:env] = v }
                o.on("--service NAME") { |v| options[:service] = v }
                o.on("--image REF") { |v| options[:image] = v }
                o.on("--yes") { options[:yes] = true }
                o.on("--non-interactive") { options[:non_interactive] = true }
                o.on("--dry-run") { options[:dry_run] = true }
                o.on("-h", "--help") { puts o; exit 0 }
            end
        end

        def run
            parser.parse!(argv)
            %i[env service image].each do |k|
                Railway.die!("--#{k} required") unless options[k]
            end

            env_id = Railway.env_id_for(options[:env])
            image = options[:image]

            unless image.include?("@sha256:")
                digest = ghcr.resolve_digest(image)
                Railway.die!("Could not resolve digest for #{image}.") unless digest
                base = image.split(":", 2).first
                image = "#{base}@#{digest}"
            end

            Railway.confirm_destructive!(
                env_label: options[:env],
                action: "pin",
                non_interactive: options[:non_interactive],
                yes: options[:yes],
            )

            service_id = RollbackCommand.new([]).resolve_service_id(env_id, options[:service])

            if options[:dry_run]
                puts "[dry-run] would pin #{options[:service]} -> #{image}"
                return 0
            end

            RestoreCommand.pin_and_redeploy(gql,
                service_id: service_id, env_id: env_id, image: image)
            puts "pinned #{options[:service]} -> #{image}"
            0
        end
    end

    # env-diff — diff two environments and exit 1 on drift.
    class EnvDiffCommand < BaseCommand
        def parser
            OptionParser.new do |o|
                o.banner = <<~BANNER
                    Usage: bin/railway env-diff ENV_A ENV_B [--ignore-env-scoped]

                      Compare image digests, startCommand, env-var key sets, and
                      custom domains between two envs. Exits 0 if equal (modulo
                      ignored markers), 1 if drift, 2 on error.
                BANNER
                o.on("--ignore-env-scoped") { options[:ignore_env_scoped] = true }
                o.on("-h", "--help") { puts o; exit 0 }
            end
        end

        def run
            parser.parse!(argv)
            Railway.die!("two env args required") if argv.length < 2

            a, b = argv[0], argv[1]
            id_a = Railway.env_id_for(a)
            id_b = Railway.env_id_for(b)

            snap_a = SnapshotCommand.new(["--env", a, "--dry-run"]).build_snapshot(id_a)
            snap_b = SnapshotCommand.new(["--env", b, "--dry-run"]).build_snapshot(id_b)

            drift = []
            names = ((snap_a["services"] + snap_b["services"]).map { |s| s["name"] }).uniq.sort
            names.each do |name|
                sa = Railway.find_service(snap_a, name)
                sb = Railway.find_service(snap_b, name)
                if sa.nil?
                    drift << "service #{name}: missing in #{a}"
                    next
                end
                if sb.nil?
                    drift << "service #{name}: missing in #{b}"
                    next
                end

                if sa["digest"] != sb["digest"]
                    drift << "service #{name}: digest #{sa['digest']} != #{sb['digest']}"
                end
                if sa["start_command"] != sb["start_command"]
                    drift << "service #{name}: startCommand differs"
                end
                missing_in_b = sa["env_keys"] - sb["env_keys"]
                missing_in_a = sb["env_keys"] - sa["env_keys"]
                drift << "service #{name}: env keys missing in #{b}: #{missing_in_b.join(', ')}" unless missing_in_b.empty?
                drift << "service #{name}: env keys missing in #{a}: #{missing_in_a.join(', ')}" unless missing_in_a.empty?
            end

            drift.each { |line| puts line }
            puts drift.empty? ? "OK: #{a} and #{b} agree." : "DRIFT: #{drift.size} finding(s)."
            drift.empty? ? 0 : 1
        end
    end

    # resolve-digest — resolve an image reference to its digest via GHCR.
    class ResolveDigestCommand < BaseCommand
        def parser
            OptionParser.new do |o|
                o.banner = <<~BANNER
                    Usage: bin/railway resolve-digest IMAGE_REF

                      Resolve a tag like 'ghcr.io/copilotkit/showcase-shell:latest' to its
                      Docker-Content-Digest. Prints sha256:... on stdout. Exits 2 if absent.
                BANNER
                o.on("-h", "--help") { puts o; exit 0 }
            end
        end

        def run
            parser.parse!(argv)
            Railway.die!("image ref required") if argv.empty?

            digest = ghcr.resolve_digest(argv[0])
            Railway.die!("Could not resolve digest for #{argv[0]}.") unless digest
            puts digest
            0
        end
    end

    # lint-prod — verify every prod service is pinned to a digest.
    class LintProdCommand < BaseCommand
        def initialize(argv)
            super
            @exit_zero = false
            @format = "text"
        end

        def parser
            OptionParser.new do |o|
                o.banner = <<~BANNER
                    Usage: bin/railway lint-prod [--exit-zero] [--format text|json]

                      Fails (exit 1) if any production service is NOT pinned to
                      an immutable image digest (must be ghcr.io/...@sha256:...).
                      Intended as a CI gate on every PR touching showcase/.

                      --exit-zero      Always exit 0 even on findings (advisory mode).
                                       Findings still print to stdout.
                      --format FORMAT  Output format: 'text' (default) or 'json'.
                                       JSON shape:
                                         {services:[{name,source,status}],
                                          findings:N, timestamp:"ISO8601"}
                                       'source' is the raw Source.image / image-ref
                                       string. 'status' is 'pinned' or 'mutable-tag'.
                BANNER
                o.on("--exit-zero", "Advisory: exit 0 even when findings exist") { @exit_zero = true }
                o.on("--format FORMAT", %w[text json], "Output format (text|json)") { |v| @format = v }
                o.on("-h", "--help") { puts o; exit 0 }
            end
        end

        def run
            parser.parse!(argv)

            prod = SnapshotCommand.new(["--env", "production", "--dry-run"])
                .build_snapshot(PRODUCTION_ENV_ID)

            services = (prod["services"] || []).map do |svc|
                image = svc["image"].to_s
                status =
                    if image.empty?
                        "mutable-tag"
                    elsif !image.include?("@sha256:")
                        "mutable-tag"
                    else
                        "pinned"
                    end
                { "name" => svc["name"], "source" => image, "status" => status }
            end
            findings = services.reject { |s| s["status"] == "pinned" }

            if @format == "json"
                payload = {
                    "services" => services,
                    "findings" => findings.size,
                    "timestamp" => Time.now.utc.iso8601,
                }
                puts JSON.generate(payload)
                return 0 if findings.empty?
                return 0 if @exit_zero
                return 1
            end

            if findings.empty?
                puts "OK: all production services digest-pinned."
                return 0
            end

            findings.each do |f|
                src = f["source"]
                if src.empty?
                    puts "#{f['name']}: no image set"
                else
                    puts "#{f['name']}: not digest-pinned (image=#{src})"
                end
            end
            puts "DRIFT: #{findings.size} production service(s) not digest-pinned."
            if @exit_zero
                puts "(advisory mode: --exit-zero set; exiting 0)"
                return 0
            end
            1
        end
    end

    # ── Dispatcher ─────────────────────────────────────────────────────────────

    SUBCOMMANDS = {
        "snapshot"         => SnapshotCommand,
        "restore"          => RestoreCommand,
        "rollback"         => RollbackCommand,
        "rollback-commit"  => RollbackCommitCommand,
        "promote"          => PromoteCommand,
        "pin"              => PinCommand,
        "env-diff"         => EnvDiffCommand,
        "resolve-digest"   => ResolveDigestCommand,
        "lint-prod"        => LintProdCommand,
    }.freeze

    def self.usage
        <<~USAGE
            bin/railway — showcase Railway operations (Ruby, stdlib-only)

            Subcommands:
              snapshot          Capture an env's services + config into a YAML snapshot.
              restore           Restore an env to a snapshot (force-redeploy).
              rollback          Roll a single service back one deploy.
              rollback-commit   Restore an env to the snapshot committed at a given SHA.
              promote           Promote staging digests to production with prechecks.
              pin               Pin a service to a specific image digest.
              env-diff          Diff two envs; exit 1 if drift.
              resolve-digest    Resolve an image tag to its GHCR digest.
              lint-prod         CI gate: fail if any prod service is not digest-pinned.

            Run any subcommand with --help for full flag list.

            Auth: RAILWAY_TOKEN env var (or ~/.railway/config.json).
            Exit codes: 0 clean, 1 drift/findings, 2 error.
        USAGE
    end

    def self.run(argv)
        if argv.empty? || %w[-h --help help].include?(argv.first)
            puts usage
            return 0
        end

        if argv.first == "--version"
            puts "railway #{VERSION}"
            return 0
        end

        cmd = argv.shift
        klass = SUBCOMMANDS[cmd]
        if klass.nil?
            warn "Unknown subcommand: #{cmd}"
            warn usage
            return 2
        end

        klass.call(argv)
    rescue GraphQL::Error => e
        warn "graphql error: #{e.message}"
        2
    rescue GHCR::Error => e
        warn "ghcr error: #{e.message}"
        2
    rescue StandardError => e
        warn "error: #{e.class}: #{e.message}"
        warn e.backtrace.first(5).join("\n") if ENV["RAILWAY_DEBUG"]
        2
    end
end

# String#rsplit_colon — split on last ':' (so tags with port-style refs are handled).
class String
    def rsplit_colon
        idx = rindex(":")
        return [self, nil] unless idx
        [self[0...idx], self[(idx + 1)..]]
    end
end

# Only run if invoked as a script (not when required by tests).
if $PROGRAM_NAME == __FILE__
    exit Railway.run(ARGV)
end