require 'open3'
require 'timeout'

# Runs solectrus/csv-importer as a detached container against the local
# SOLECTRUS stack. Reads the extracted CSV tree from CsvImportUploader,
# bind-mounts it as /data, and passes through InfluxDB credentials,
# sensor mappings and SENEC_IGNORE from the live `.env` so imported
# points land on the same measurement+field pairs the dashboard reads.
#
# `--rm` is intentionally NOT used: the exit code (and last log lines)
# must remain inspectable across the gap between container exit and the
# next /show poll. `process_completion!` removes the container after
# reading its result.
#
# Lifecycle has four phases the user sees as equal steps:
#   1. preparing  — image pull + `docker run`
#   2. importing  — csv-importer container writes points to InfluxDB
#   3. flushing   — Redis cache flush so the dashboard rebuilds aggregates
#   4. truncating — reset summaries so daily aggregates refill from raw.
#                   When CoveredDates can infer the imported days from
#                   filenames we DELETE only those rows; otherwise we
#                   TRUNCATE the whole table.
#
# Phases 3 and 4 run inside a completion thread (so the /show poll request
# does not block on them) and their results are part of the user-visible
# outcome: the success file is only written if both succeed.
class CsvImportRunner < DetachedRunner # rubocop:disable Metrics/ClassLength
  CONTAINER_NAME = 'helios-csv-import-runner'.freeze
  IMAGE = 'ghcr.io/solectrus/csv-importer:latest'.freeze
  INFLUXDB_SERVICE = 'influxdb'.freeze
  REDIS_SERVICE = 'redis'.freeze
  IMPORT_MOUNT = '/data'.freeze
  I18N_SCOPE = 'csv_imports.runner'.freeze
  UPDATE_PAUSE_REASON = :csv_import

  # Always required for the importer to write into InfluxDB.
  REQUIRED_INFLUX_KEYS = %w[INFLUX_TOKEN_WRITE INFLUX_ORG INFLUX_BUCKET].freeze
  # In collectors_only mode the InfluxDB host is external and must also be
  # set explicitly; otherwise the importer connects to the compose service
  # name and the host is synthesized.
  EXTRA_COLLECTORS_ONLY_INFLUX_KEYS = %w[INFLUX_HOST].freeze

  # Passed through so historical points land on the same measurement+field
  # pairs the dashboard already reads. Names must match what the importer
  # actually reads (csv-importer/app/config.rb#sensors_from_env) — it falls
  # back to the SENEC defaults for anything it doesn't recognize, so a typo
  # here silently imports to the wrong measurement instead of failing.
  SENSOR_MAPPING_KEYS = %w[
    INFLUX_SENSOR_INVERTER_POWER
    INFLUX_SENSOR_HOUSE_POWER
    INFLUX_SENSOR_GRID_IMPORT_POWER
    INFLUX_SENSOR_GRID_EXPORT_POWER
    INFLUX_SENSOR_BATTERY_CHARGING_POWER
    INFLUX_SENSOR_BATTERY_DISCHARGING_POWER
    INFLUX_SENSOR_BATTERY_SOC
  ].freeze

  MUTEX = Mutex.new

  class << self
    delegate :unavailable_reason, :precheck!, to: :new
    delegate :error_message, :success_message, :clear_error!, :clear_success!, to: 'CsvImportRunner::State'

    # True while preparing thread is alive, the container is up, OR the
    # completion thread (flushing + truncating phases) is still running.
    # Keeps the user on the progress page across all phases.
    def in_progress? = preparing? || running? || completing?

    # Request-scoped memo of the same answer, for the read paths that ask
    # more than once per render: /backups reads it for the action gate and
    # again inside BackupRunner.unavailable_reason, and each uncached read
    # forks a `docker inspect`. Validation must keep calling `in_progress?`,
    # which always re-probes — a stale "no import running" there could let a
    # backup launch on top of a live import.
    def in_progress_cached? = Current.instance.fetch(:csv_import_in_progress) { in_progress? }

    # True while the background thread is pulling the image + launching the
    # container. Process-local; killed by a HELIOS restart — the user then
    # simply re-uploads.
    def preparing?
      MUTEX.synchronize { @preparing_thread&.alive? || false }
    end

    # True while the completion thread (post-exit phases 3+4) is still
    # running. Same lifecycle as preparing — survives only within a
    # single HELIOS process.
    def completing?
      MUTEX.synchronize { @completion_thread&.alive? || false }
    end

    # Hands the slow pull + launch work off to a thread so the controller
    # can redirect to the progress page immediately. Mirrors
    # BackupRepository::S3::AsyncWorker (single-flight, Rails-executor
    # wrapped, errors captured into the state file).
    def spawn_preparing_thread!(instance)
      MUTEX.synchronize do
        return if @preparing_thread&.alive?

        @preparing_thread = Thread.new do # rubocop:disable ThreadSafety/NewThread
          Rails.application.executor.wrap { instance.send(:run_preparing!) }
        end
      end
    end

    # Spawns the post-exit completion thread (reads container outcome,
    # runs flushing + truncating phases). Single-flight: a second call
    # while the first thread is alive is a no-op, so two parallel /show
    # polls reaching detect_completion! at the same time don't both run
    # TRUNCATE summaries + Redis FLUSHALL.
    def spawn_completion_thread!(raw)
      MUTEX.synchronize do
        return if @completion_thread&.alive?

        @completion_phase = nil
        @completion_thread = Thread.new do # rubocop:disable ThreadSafety/NewThread
          Rails.application.executor.wrap { new.process_completion!(raw) }
        end
      end
    end

    # Active phase of the completion thread (:flushing or :truncating).
    # Read by `progress` so the UI can highlight the correct step.
    def completion_phase
      MUTEX.synchronize { @completion_phase }
    end

    def completion_phase=(phase)
      MUTEX.synchronize { @completion_phase = phase }
    end

    def log_tail(lines)
      Orchestration::DockerCli.log_tail(CONTAINER_NAME, lines:)
    end

    # Live progress for the UI. Phase follows the active thread:
    # :preparing while the prep thread is alive, :importing while the
    # container is up, then :flushing / :truncating while the completion
    # thread runs phases 3 and 4. During completion the phase is only set
    # by `run_phase!` (success path); on the failure path it stays nil so
    # the UI doesn't pretend flushing/truncating ran — the component falls
    # back to :importing for nil, which matches what just happened.
    def progress
      return { phase: :preparing, done: 0, total: 0 } if preparing?
      return { phase: completion_phase, done: 0, total: 0 } if completing?

      done = LogParser.progress(log_tail(1000))[:done]
      total = Dir.glob(
        "#{CsvImportUploader.extract_directory}/**/*.csv", File::FNM_CASEFOLD
      ).size
      { phase: :importing, done: done, total: total }
    end

    # If the container has exited, hand off to the completion thread.
    # spawn_completion_thread!'s mutex + alive? check is the single-flight
    # gate — two parallel /show polls (or the before_action firing on
    # overlapping requests) cannot both spawn a completion thread.
    def detect_completion!
      raw = Orchestration::DockerCli.inspect_container(CONTAINER_NAME)
      return if raw.nil?
      return if raw.dig('State', 'Running')

      spawn_completion_thread!(raw)
    end
  end

  def start
    validate!
    State.clear_all!
    self.class.spawn_preparing_thread!(self)
  end

  # Cheap pre-flight the controller runs BEFORE accepting an upload so the
  # uploader's destructive cleanup (`rm_rf extract_directory`, which is the
  # bind-mount source for a possibly-running container) cannot fire while
  # an import is in flight or another runner holds the lock. Mirrors
  # `validate!` minus the extracted-CSV check, which can only run after the
  # uploader has finished extracting.
  def precheck!
    check_no_concurrent_runs!

    msg = unavailable_reason
    raise Error, msg if msg
  end

  # Preconditions for the import UI: things the user cannot fix by
  # re-uploading. The CSV-files check stays out — that's a runtime check
  # enforced in `check_ready_to_import!`.
  def unavailable_reason
    return reason(:env_missing) if env.nil?
    return reason(:postgresql_not_running) unless postgresql_running?
    return reason(:influxdb_not_running) unless influxdb_target_running?
    return reason(:influx_target_missing) if influx_credentials_missing?

    nil
  end

  # Body of the preparing thread; surfaces failures into the state file so
  # the UI can show them on the next poll.
  def run_preparing!
    pause_updates!
    pull_image_if_needed!
    run_container!
    self.class.invalidate_in_progress_cache!
  rescue StandardError => e
    logger.warn("preparing failed: #{e.class}: #{e.message}")
    State.write_error!(e.message.presence || "#{e.class}: (no message)")
    CsvImportUploader.cleanup!
  end

  # Body of the completion thread. Reads the just-exited container's
  # state, then — on a clean import — awaits the flushing and truncating
  # phases before writing the final success file so the user-visible
  # outcome reflects whether the dashboard caches were actually reset.
  def process_completion!(raw) # rubocop:disable Metrics/AbcSize
    logger.info('container finished, processing completion')
    state = raw['State'] || {}

    if successful_exit?(state)
      capture_outcome!(raw)
    else
      capture_failure!(raw)
    end
  rescue StandardError => e
    logger.warn("completion failed: #{e.class}: #{e.message}")
    State.write_error!(e.message.presence || "#{e.class}: (no message)")
  ensure
    self.class.completion_phase = nil
    # Each ensure step is wrapped: an exception raised from the ensure block
    # would bypass the `rescue StandardError` above (Ruby semantics), leaving
    # the in_progress cache stale and the UI stuck at "running" forever.
    safely(:remove_container) { remove_container! }
    safely(:cleanup) { CsvImportUploader.cleanup! }
    self.class.invalidate_in_progress_cache!
  end

  private

  def reason(key)
    I18n.t("#{self.class::I18N_SCOPE}.unavailable.#{key}")
  end

  def validate!
    precheck!
    check_ready_to_import!
  end

  # Cross-runner exclusion uses `.in_progress` (struct or nil) rather than
  # `.running?` so it also covers the S3 download window of an external
  # restore (no container yet) and the S3 upload tail of an external
  # backup (container already exited).
  def check_no_concurrent_runs!
    raise Error, error(:already_in_progress) if self.class.in_progress?
    raise Error, error(:exit_pending) if exited_container_pending?
    raise Error, error(:backup_in_progress) if BackupRunner.in_progress
    raise Error, error(:restore_in_progress) if RestoreRunner.in_progress
  end

  def check_ready_to_import!
    raise Error, reason(:no_csv_files) unless extracted_csvs?
  end

  # Defense-in-depth: a stopped container with our reserved name would
  # block `docker run --name`. The controller polls detect_completion!
  # before each form render, so this should rarely fire.
  def exited_container_pending?
    raw = Orchestration::DockerCli.inspect_container(CONTAINER_NAME)
    raw && !raw.dig('State', 'Running')
  end

  def extracted_csvs?
    extracted_csv_paths.any?
  end

  def env
    return @env if defined?(@env)

    @env = Env.load
  end

  def influx_credentials_missing?
    keys = REQUIRED_INFLUX_KEYS
    keys += EXTRA_COLLECTORS_ONLY_INFLUX_KEYS if Configuration.current.collectors_only?
    keys.any? { |key| env[key].blank? }
  end

  # In collectors_only mode InfluxDB lives outside our compose project, so
  # we cannot probe it directly; trust the credentials check and let the
  # importer surface a connection failure if the remote host is down.
  def influxdb_target_running?
    return true if Configuration.current.collectors_only?

    influxdb_container&.running? == true
  end

  # Required for the truncate summaries phase; skipped in collectors_only
  # mode (no local postgres there).
  def postgresql_running?
    return true if Configuration.current.collectors_only?

    Orchestration::Container.find(Orchestration::SummariesReset::SERVICE)&.running? == true
  end

  def influxdb_container
    @influxdb_container ||= Orchestration::Container.find(INFLUXDB_SERVICE)
  end

  def docker_run_command
    [
      'docker', 'run', '-d',
      '--name', CONTAINER_NAME,
      *network_args,
      '--mount', "type=bind,source=#{CsvImportUploader.host_extract_directory},target=#{IMPORT_MOUNT},readonly",
      *env_args,
      IMAGE
    ]
  end

  # Local stack: join the compose-network the InfluxDB container is on so
  # its service name resolves. collectors_only: external `INFLUX_HOST`,
  # the default bridge network is fine.
  def network_args
    return [] if Configuration.current.collectors_only?
    return [] unless influx_network_name

    ['--network', influx_network_name]
  end

  def influx_network_name
    return nil unless influxdb_container

    networks = influxdb_container.json&.dig('NetworkSettings', 'Networks')
    networks&.keys&.first
  end

  def env_args
    importer_env.flat_map { |key, value| ['-e', "#{key}=#{value}"] }
  end

  # ENV the importer respects. INFLUX_HOST is synthesized for the local
  # stack since we attach from outside compose and can't rely on its
  # service discovery.
  def importer_env
    influx_env
      .merge(sensor_mapping_env)
      .merge(senec_ignore_env)
      .compact
  end

  def influx_env
    {
      'INFLUX_ORG' => env['INFLUX_ORG'],
      'INFLUX_BUCKET' => env['INFLUX_BUCKET'],
      'INFLUX_TOKEN_WRITE' => env['INFLUX_TOKEN_WRITE'],
      'INFLUX_HOST' => influx_host,
      'INFLUX_PORT' => env['INFLUX_PORT'].presence || '8086',
      'INFLUX_SCHEMA' => influx_schema,
      'TZ' => env['TZ'].presence || 'Europe/Berlin',
    }
  end

  def influx_schema
    env['INFLUX_SCHEMA'].presence || (Configuration.current.collectors_only? ? 'https' : 'http')
  end

  def senec_ignore_env
    return {} if env['SENEC_IGNORE'].blank?

    { 'SENEC_IGNORE' => env['SENEC_IGNORE'] }
  end

  def influx_host
    if Configuration.current.collectors_only?
      env['INFLUX_HOST']
    else
      influxdb_container&.service_name || INFLUXDB_SERVICE
    end
  end

  def sensor_mapping_env
    SENSOR_MAPPING_KEYS.each_with_object({}) do |key, hash|
      value = env[key]
      hash[key] = value if value.present?
    end
  end

  # A container that exited cleanly: Docker reports it as `exited` (not
  # `created`/`dead`), exit code is 0, and it was not OOM-killed. Checking
  # ExitCode alone would let a `created` container (run -d succeeded, but
  # the OCI runtime never actually started the process) slip through as
  # "success" and trigger the flushing + truncating phases.
  def successful_exit?(state)
    state['Status'] == 'exited' &&
      state['ExitCode'].to_i.zero? &&
      !state['OOMKilled']
  end

  # On a clean container exit: count the imported files from the log; if
  # the count is missing or zero treat it as anomaly (capture_failure!).
  # Otherwise run the remaining phases SYNCHRONOUSLY in the completion
  # thread and only write the success file if BOTH succeed — a partial
  # failure leaves the dashboard cached and is surfaced to the user as
  # a failure rather than a silent warning in the log.
  def capture_outcome!(raw)
    count = LogParser.total_files(container_log_tail(500))
    return capture_failure!(raw) unless count.positive?

    failures = run_remaining_phases!
    if failures.empty?
      State.write_success!(count.to_s)
    else
      State.write_error!([error(:phases_failed_header), *failures].join("\n"))
    end
  end

  def run_remaining_phases!
    failures = []
    failures << error(:redis_flush_failed) unless run_phase!(:flushing) { flush_redis_cache! }
    failures << error(:summaries_reset_failed) unless run_phase!(:truncating) { reset_summaries! }
    failures
  end

  def run_phase!(phase)
    self.class.completion_phase = phase
    yield
  end

  # Dashboard caches series aggregates in Redis; without a flush, imported
  # historical periods stay invisible behind the cached empty results. A
  # missing/stopped redis container counts as success — nothing to flush.
  def flush_redis_cache!
    container = Orchestration::Container.find(REDIS_SERVICE)
    return true unless container&.running?

    Orchestration::RedisCacheFlush.call(container)
  rescue StandardError => e
    logger.warn("redis flush failed: #{e.class}: #{e.message}")
    false
  end

  # Daily summaries are an aggregated cache too — the dashboard would
  # otherwise stick with the pre-import aggregates for the imported range.
  # SOLECTRUS rebuilds the table from raw points on demand, so wiping is
  # the official reset path. We scope the reset to the days CoveredDates
  # extracts from filenames when possible; an unrecognized filename in
  # the upload flips us back to a full TRUNCATE. Skipped in
  # collectors_only mode (no local postgres there).
  def reset_summaries!
    return true if Configuration.current.collectors_only?

    # Defense against re-entry: if a previous process_completion! already
    # ran (extract dir was cleaned up in its ensure) but the container
    # removal failed, detect_completion! would re-process the same
    # container — without this check CoveredDates.scan would return nil
    # (empty dir) and SummariesReset(dates: nil) would TRUNCATE the entire
    # summaries table.
    return true if extracted_csv_paths.empty?

    dates = CoveredDates.scan(CsvImportUploader.extract_directory)
    Orchestration::SummariesReset.call(dates: dates)
  end

  def extracted_csv_paths
    Dir.glob("#{CsvImportUploader.extract_directory}/**/*.csv", File::FNM_CASEFOLD)
  end

  # `State.Error` is populated for OOM/image-pull failures; for the common
  # "importer raised, exited 1" case it is blank, so we fall back to the
  # last log lines so the user sees the actual backtrace, not just "exit 1".
  # The tail must be large enough to capture the exception line at the top
  # of a Ruby backtrace — a bundler/boot failure alone is ~20 `from` frames,
  # so anything smaller leaves the user with only the boilerplate footer.
  def capture_failure!(raw)
    state = raw['State'] || {}
    docker_err = state['Error'].to_s.strip
    tail = container_log_tail(500)
    msg =
      if docker_err.present? && tail.present?
        "#{docker_err}\n\n#{tail}"
      else
        docker_err.presence || tail.presence || error(:nonzero_exit, code: state['ExitCode'])
      end

    State.write_error!(msg)
  end

  def container_log_tail(lines)
    self.class.log_tail(lines).strip
  end

  def safely(label)
    yield
  rescue StandardError => e
    logger.warn("#{label} failed: #{e.class}: #{e.message}")
  end

  # A failed `docker rm` leaves the exited container in place — detect_completion!
  # would then trigger process_completion! again on the next /show poll, which
  # without the extracted_csv_paths guard in reset_summaries! would re-issue
  # destructive cache resets. Log loudly so the precondition stays visible.
  def remove_container!
    output, status = Open3.capture2e('docker', 'rm', '-f', CONTAINER_NAME)
    return if status.success?

    logger.warn(
      "docker rm failed (exit #{status.exitstatus}): #{output.strip}",
    )
  end

  # Override the inherited backup-flavoured mount-source prep: the csv
  # importer needs the runtime directory (for the error file) and an
  # extracted-CSV tree the uploader produces, but no backup destination
  # or influx staging mount.
  def ensure_bind_mount_sources!
    FileUtils.mkdir_p(self.class.runtime_directory)
    FileUtils.mkdir_p(CsvImportUploader.extract_directory)
  end
end