Function bodies 405 total

        def self.buildkit_available?
          system("docker", "buildx", "version", out: File::NULL, err: File::NULL) == true
        end

        def self.aws_credentials_valid?
          Aws::STS::Client.new.get_caller_identity
          true
        rescue Aws::STS::Errors::ServiceError
          false
        end

        def self.git_clean?
          `git status --porcelain`.strip.empty?
        end

        def self.warn_running_executions(sfn_client, state_machine_arn)
          running = sfn_client.list_executions(state_machine_arn: state_machine_arn, status_filter: "RUNNING")
          if running.executions.any?
            puts "WARNING: #{running.executions.size} execution(s) currently running."
            puts "Running executions will continue with previous job definitions."
          end

  class CLI < Thor
    module Deploy
      def self.call(pipeline_name:, stage:, dry_run: false)
        turbofans_root = "turbofans"
        pipeline_file = File.join(turbofans_root, "pipelines", "#{pipeline_name}.rb")
        schemas_dir = File.join(turbofans_root, "schemas")

        load_result = Turbofan::Deploy::PipelineLoader.load(pipeline_file, turbofans_root: turbofans_root)
        pipeline_class = load_result.pipeline
        steps = load_result.steps
        step_dirs = load_result.step_dirs

        # Load stage config
        config_path = File.join(turbofans_root, "config", "#{stage}.yml")
        config = File.exist?(config_path) ? YAML.safe_load_file(config_path, symbolize_names: true) : {}

        # Pre-flight checks
        raise "BuildKit not available. Install Docker 23.0+ or enable BuildKit." unless Preflight.buildkit_available?
        raise "AWS credentials invalid. Run 'aws sts get-caller-identity' to debug." unless Preflight.aws_credentials_valid?
        if 

      def self.call(pipeline_name:, stage:, dry_run: false)
        turbofans_root = "turbofans"
        pipeline_file = File.join(turbofans_root, "pipelines", "#{pipeline_name}.rb")
        schemas_dir = File.join(turbofans_root, "schemas")

        load_result = Turbofan::Deploy::PipelineLoader.load(pipeline_file, turbofans_root: turbofans_root)
        pipeline_class = load_result.pipeline
        steps = load_result.steps
        step_dirs = load_result.step_dirs

        # Load stage config
        config_path = File.join(turbofans_root, "config", "#{stage}.yml")
        config = File.exist?(config_path) ? YAML.safe_load_file(config_path, symbolize_names: true) : {}

        # Pre-flight checks
        raise "BuildKit not available. Install Docker 23.0+ or enable BuildKit." unless Preflight.buildkit_available?
        raise "AWS credentials invalid. Run 'aws sts get-caller-identity' to debug." unless Preflight.aws_credentials_valid?
        if CLI::PROTECTED_STAGES.include?(stage)

      def self.build_and_push_all(step_dirs:, schemas_dir:, stack_name:, registry:, ecr_client:, image_tags:)
        configs = step_dirs.map do |step_name, step_dir|
          {
            step_dir: step_dir,
            schemas_dir: schemas_dir,
            ecr_client: ecr_client,
            repository_name: "#{stack_name}-ecr-#{step_name}",
            repository_uri: "#{registry}/#{stack_name}-ecr-#{step_name}",
            tag: image_tags[step_name]
          }
        end

  class CLI < Thor
    module Destroy
      def self.call(pipeline_name:, stage:, force: false, cf_client: Aws::CloudFormation::Client.new)
        dash_name = pipeline_name.tr("_", "-")
        stack_name = "turbofan-#{dash_name}-#{stage}"

        if CLI::PROTECTED_STAGES.include?(stage) && !force
          unless Turbofan::CLI::Prompt.confirm_destructive(
            "WARNING: '#{stage}' is a protected stage.\nStack '#{stack_name}' will be permanently deleted.",
            expected_input: stack_name
          )
            raise Thor::Error, "Use --force to destroy protected stacks in non-interactive mode." unless Turbofan::CLI::Prompt.tty?
            return
          end
        end

        resources = cf_client.describe_stack_resources(stack_name: stack_name).stack_resources
        $stdout.puts "Resources in #{stack_name}:"
        resources.each do |r|
          $stdout.puts "  #{r.resource_type}  #{r.logical_resource_id}  #{r.physical_resource_id}"
        end

      def self.call(pipeline_name:, stage:, force: false, cf_client: Aws::CloudFormation::Client.new)
        dash_name = pipeline_name.tr("_", "-")
        stack_name = "turbofan-#{dash_name}-#{stage}"

        if CLI::PROTECTED_STAGES.include?(stage) && !force
          unless Turbofan::CLI::Prompt.confirm_destructive(
            "WARNING: '#{stage}' is a protected stage.\nStack '#{stack_name}' will be permanently deleted.",
            expected_input: stack_name
          )
            raise Thor::Error, "Use --force to destroy protected stacks in non-interactive mode." unless Turbofan::CLI::Prompt.tty?
            return
          end

  class CLI < Thor
    module History
      def self.call(pipeline_name:, stage:, limit: 20)
        cf = Aws::CloudFormation::Client.new
        sfn = Aws::States::Client.new
        stack_name = Turbofan::Naming.stack_name(pipeline_name, stage)

        sm_arn = Turbofan::Deploy::StackManager.stack_output(cf, stack_name, "StateMachineArn")

        response = sfn.list_executions(
          state_machine_arn: sm_arn,
          max_results: limit
        )

        executions = response.executions

        if executions.empty?
          puts "No executions found for #{pipeline_name} (#{stage})"
          return
        end

        executions.each do |exec|
          duration = if exec.stop_date && exec.start_date
            elapsed = exec.stop_date - exec.start_date
            format_duration(elapsed)
          else
            "running"
          end

          puts "#{exec.name}  #{exec.status}  #{exec.start_date}  #{duration}"
        end

      def self.call(pipeline_name:, stage:, limit: 20)
        cf = Aws::CloudFormation::Client.new
        sfn = Aws::States::Client.new
        stack_name = Turbofan::Naming.stack_name(pipeline_name, stage)

        sm_arn = Turbofan::Deploy::StackManager.stack_output(cf, stack_name, "StateMachineArn")

        response = sfn.list_executions(
          state_machine_arn: sm_arn,
          max_results: limit
        )

        executions = response.executions

        if executions.empty?
          puts "No executions found for #{pipeline_name} (#{stage})"
          return
        end

      def self.format_duration(seconds)
        if seconds < 60
          "#{seconds.round(1)}s"
        elsif seconds < 3600
          "#{(seconds / 60).round(1)}m"
        else
          "#{(seconds / 3600).round(1)}h"
        end

  class CLI < Thor
    module Logs
      def self.call(pipeline_name:, stage:, step:, execution: nil, item: nil, query: nil, logs_client: Aws::CloudWatchLogs::Client.new)
        dash_name = pipeline_name.tr("_", "-")

        log_group = if step.nil? || step.empty?
          # When no step is specified, query a general log group for the pipeline
          "turbofan-#{dash_name}-#{stage}-logs"
        else
          "turbofan-#{dash_name}-#{stage}-logs-#{step}"
        end

        insights = Turbofan::Observability::InsightsQuery.new(log_group: log_group)
        insights = insights.execution(execution) if execution
        insights = insights.item(item) if item
        insights = insights.expression(query) if query

        query_string = insights.build

        now = Time.now
        response = logs_client.start_query(
          log_group_name: log_group,
          start_time: (now - 86_400).to_i,
          end_time: now.to_i,
          query_string: query_string
        )

        

      def self.call(pipeline_name:, stage:, step:, execution: nil, item: nil, query: nil, logs_client: Aws::CloudWatchLogs::Client.new)
        dash_name = pipeline_name.tr("_", "-")

        log_group = if step.nil? || step.empty?
          # When no step is specified, query a general log group for the pipeline
          "turbofan-#{dash_name}-#{stage}-logs"
        else
          "turbofan-#{dash_name}-#{stage}-logs-#{step}"
        end

      def self.poll_results(logs_client, query_id)
        deadline = Time.now + POLL_TIMEOUT
        loop do
          response = logs_client.get_query_results(query_id: query_id)
          case response.status
          when "Complete" then return response.results
          when "Failed", "Cancelled", "Timeout"
            raise "CloudWatch Insights query #{response.status}"
          end

      def self.format_results(results)
        results.each do |row|
          fields = row.each_with_object({}) { |f, h| h[f.field] = f.value }
          timestamp = fields["@timestamp"] || ""
          message = fields["@message"] || ""
          puts "#{timestamp}  #{message}"
        end

  class CLI < Thor
    module New
      def self.call(name)
        Dir.chdir(Turbofan::CLI.project_root) do
          class_name = name.split("_").map(&:capitalize).join

          pipelines_dir = File.join("turbofans", "pipelines")
          schemas_dir = File.join("turbofans", "schemas")
          config_dir = File.join("turbofans", "config")

          FileUtils.mkdir_p(pipelines_dir)
          FileUtils.mkdir_p(schemas_dir)
          FileUtils.mkdir_p(config_dir)

          write_pipeline_rb(pipelines_dir, name, class_name)
          write_config(config_dir, "production")
          write_config(config_dir, "staging")
        end
      end

      def self.write_pipeline_rb(dir, name, class_name)
        File.write(File.join(dir, "#{name}.rb"), <<~RUBY)
          class #{class_name}
            include Turbofan::Pipeline

            pipeline_name "#{name}"

            pipeline do |input|
              # Add steps with: turbofan step new STEP_NAME
            end
          end

      def self.call(name)
        Dir.chdir(Turbofan::CLI.project_root) do
          class_name = name.split("_").map(&:capitalize).join

          pipelines_dir = File.join("turbofans", "pipelines")
          schemas_dir = File.join("turbofans", "schemas")
          config_dir = File.join("turbofans", "config")

          FileUtils.mkdir_p(pipelines_dir)
          FileUtils.mkdir_p(schemas_dir)
          FileUtils.mkdir_p(config_dir)

          write_pipeline_rb(pipelines_dir, name, class_name)
          write_config(config_dir, "production")
          write_config(config_dir, "staging")
        end

      def self.write_pipeline_rb(dir, name, class_name)
        File.write(File.join(dir, "#{name}.rb"), <<~RUBY)
          class #{class_name}
            include Turbofan::Pipeline

            pipeline_name "#{name}"

            pipeline do |input|
              # Add steps with: turbofan step new STEP_NAME
            end

      def self.write_config(config_dir, stage)
        path = File.join(config_dir, "#{stage}.yml")
        return if File.exist?(path)

        File.write(path, <<~YAML)
          subnets: []
          security_groups: []
        YAML
      end

      def self.write_step(step_dir, class_name, duckdb:, step_name: nil, compute_environment: :compute, cpu: 1, extensions: [])
        step_name ||= File.basename(step_dir)
        write_worker(step_dir, class_name, duckdb: duckdb, step_name: step_name, compute_environment: compute_environment, cpu: cpu)
        write_gemfile(step_dir, duckdb: duckdb)
        write_dockerfile(step_dir, duckdb: duckdb, extensions: extensions)
        write_entrypoint(step_dir, class_name)
      end

      def self.write_worker(step_dir, class_name, duckdb:, step_name:, compute_environment: :compute, cpu: 1)
        File.write(File.join(step_dir, "worker.rb"), <<~RUBY)
          class #{class_name}
            include Turbofan::Step

            compute_environment :#{compute_environment}
            cpu #{cpu}
            ram 2048
            input_schema "#{step_name}_input.json"
            output_schema "#{step_name}_output.json"

            def call(inputs, context)
              # TODO: implement
            end
          end

            def call(inputs, context)
              # TODO: implement
            end

      def self.write_gemfile(step_dir, duckdb:)
        lines = ['source "https://rubygems.org"', "", 'gem "turbofan"']
        lines << 'gem "duckdb"' if duckdb
        File.write(File.join(step_dir, "Gemfile"), lines.join("\n") + "\n")
      end

      def self.write_dockerfile(step_dir, duckdb: false, extensions: [])
        all_extensions = if duckdb
          ([:postgres_scanner] + extensions.map(&:to_sym)).uniq
        else
          []
        end

      def self.write_entrypoint(step_dir, class_name)
        File.write(File.join(step_dir, "entrypoint.rb"), <<~RUBY)
          require "turbofan/runtime/wrapper"
          require_relative "worker"

          Turbofan::Runtime::Wrapper.run(#{class_name})
        RUBY
      end

      def self.write_schema(schemas_dir, step_name)
        File.write(File.join(schemas_dir, "#{step_name}_input.json"), '{"type": "object"}' + "\n")
        File.write(File.join(schemas_dir, "#{step_name}_output.json"), '{"type": "object"}' + "\n")
      end

        def input = @input || $stdin
        def output = @output || $stdout
        def tty? = input.respond_to?(:tty?) && input.tty?

        def ask(question, default: nil)
          return default unless tty?
          output.print default ? "#{question} [#{default}]: " : "#{question}: "
          answer = input.gets&.chomp
          (answer.nil? || answer.empty?) ? default : answer
        end

        def yes?(question, default: true)
          return default unless tty?
          output.print "#{question} #{default ? "[Y/n]" : "[y/N]"} "
          answer = input.gets&.chomp&.downcase
          return default if answer.nil? || answer.empty?
          %w[y yes].include?(answer)
        end

        def select(question, choices, default: nil)
          default ||= choices.first
          return default unless tty?
          output.puts question
          choices.each_with_index { |c, i| output.puts "  #{i + 1}) #{c}" }
          output.print "Choice [#{choices.index(default).to_i +

        def output = @output || $stdout
        def tty? = input.respond_to?(:tty?) && input.tty?

        def ask(question, default: nil)
          return default unless tty?
          output.print default ? "#{question} [#{default}]: " : "#{question}: "
          answer = input.gets&.chomp
          (answer.nil? || answer.empty?) ? default : answer
        end

        def yes?(question, default: true)
          return default unless tty?
          output.print "#{question} #{default ? "[Y/n]" : "[y/N]"} "
          answer = input.gets&.chomp&.downcase
          return default if answer.nil? || answer.empty?
          %w[y yes].include?(answer)
        end

        def select(question, choices, default: nil)
          default ||= choices.first
          return default unless tty?
          output.puts question
          choices.each_with_index { |c, i| output.puts "  #{i + 1}) #{c}" }
          output.print "Choice [#{choices.index(default).to_i + 1}]: "
          answer = input.gets

        def tty? = input.respond_to?(:tty?) && input.tty?

        def ask(question, default: nil)
          return default unless tty?
          output.print default ? "#{question} [#{default}]: " : "#{question}: "
          answer = input.gets&.chomp
          (answer.nil? || answer.empty?) ? default : answer
        end

        def yes?(question, default: true)
          return default unless tty?
          output.print "#{question} #{default ? "[Y/n]" : "[y/N]"} "
          answer = input.gets&.chomp&.downcase
          return default if answer.nil? || answer.empty?
          %w[y yes].include?(answer)
        end

        def select(question, choices, default: nil)
          default ||= choices.first
          return default unless tty?
          output.puts question
          choices.each_with_index { |c, i| output.puts "  #{i + 1}) #{c}" }
          output.print "Choice [#{choices.index(default).to_i + 1}]: "
          answer = input.gets&.chomp
          return default if answ

        def ask(question, default: nil)
          return default unless tty?
          output.print default ? "#{question} [#{default}]: " : "#{question}: "
          answer = input.gets&.chomp
          (answer.nil? || answer.empty?) ? default : answer
        end

        def yes?(question, default: true)
          return default unless tty?
          output.print "#{question} #{default ? "[Y/n]" : "[y/N]"} "
          answer = input.gets&.chomp&.downcase
          return default if answer.nil? || answer.empty?
          %w[y yes].include?(answer)
        end

        def select(question, choices, default: nil)
          default ||= choices.first
          return default unless tty?
          output.puts question
          choices.each_with_index { |c, i| output.puts "  #{i + 1}) #{c}" }
          output.print "Choice [#{choices.index(default).to_i + 1}]: "
          answer = input.gets&.chomp
          return default if answer.nil? || answer.empty?
          idx = answer.to_i - 1
          (idx >= 0 && idx < choices.size) ? choices[idx] : default
        end

        def confirm_destructive(message, expected_input:)
          return false unless tty?
          output.puts message
          output.print "Type '#{expected_input}' to confirm: "
          input.gets&.chomp == expected_input
        end

        def reset!
          @input = nil
          @output = nil
        end

  class CLI < Thor
    PROTECTED_STAGES = %w[production staging].freeze

    desc "new NAME", "Create a new Turbofan pipeline"
    def new(name)
      Turbofan::CLI::New.call(name)
    end

    desc "check PIPELINE STAGE", "Run validation checks on a pipeline"
    def check(pipeline_name, stage)
      Turbofan::CLI::Check.call(
        pipeline_name: pipeline_name,
        stage: stage
      )
    end

    desc "destroy PIPELINE STAGE", "Destroy a deployed pipeline stack"
    option :force, type: :boolean, default: false
    def destroy(pipeline_name, stage)
      Turbofan::CLI::Destroy.call(
        pipeline_name: pipeline_name,
        stage: stage,
        force: options[:force]
      )
    end

    desc "logs PIPELINE STAGE", "Query CloudWatch logs for a pipeline"
    option :step, type: :string
    option :execution, type: :string
    option :item, type: :string
    option :query, type: :string
    def logs(pipeline_name, stage)
      Turbofan::CLI::Logs.call(
        pipeline_nam

    def new(name)
      Turbofan::CLI::New.call(name)
    end

    def check(pipeline_name, stage)
      Turbofan::CLI::Check.call(
        pipeline_name: pipeline_name,
        stage: stage
      )
    end

    def destroy(pipeline_name, stage)
      Turbofan::CLI::Destroy.call(
        pipeline_name: pipeline_name,
        stage: stage,
        force: options[:force]
      )
    end

    def logs(pipeline_name, stage)
      Turbofan::CLI::Logs.call(
        pipeline_name: pipeline_name,
        stage: stage,
        step: options[:step],
        execution: options[:execution],
        item: options[:item],
        query: options[:query]
      )
    end

    def deploy(pipeline_name, stage)
      Turbofan::CLI::Deploy.call(
        pipeline_name: pipeline_name,
        stage: stage,
        dry_run: options[:dry_run]
      )
    end

    def rollback(pipeline_name, stage)
      Turbofan::CLI::Rollback.call(
        pipeline_name: pipeline_name,
        stage: stage
      )
    end

    def start(pipeline_name, stage)
      Turbofan::CLI::Run.call(
        pipeline_name: pipeline_name,
        stage: stage,
        input: options[:input],
        input_file: options[:input_file],
        dry_run: options[:dry_run]
      )
    end

    def history(pipeline_name, stage)
      Turbofan::CLI::History.call(
        pipeline_name: pipeline_name,
        stage: stage,
        limit: options[:limit]
      )
    end

    def cost(pipeline_name, stage)
      Turbofan::CLI::Cost.call(
        pipeline_name: pipeline_name,
        stage: stage
      )
    end

    def status(pipeline_name, stage)
      Turbofan::CLI::Status.call(
        pipeline_name: pipeline_name,
        stage: stage,
        watch: options[:watch]
      )
    end

      def new(name)
        Turbofan::CLI::Ce.new_ce(name)
      end

      def deploy(stage)
        Turbofan::CLI::Ce.deploy(stage: stage)
      end

      def list
        Turbofan::CLI::Ce.list
      end

      def deploy(stage)
        Turbofan::CLI::Resources.deploy(stage: stage)
      end