Prd/components/transcription-worker/remote_transcriber.rb

# frozen_string_literal: true

# RemoteTranscriber — drop-in alternative to Transcriber that drives the hosted Whisper Web UI
# (whisper.cajk.org/app) instead of a local WhisperX subprocess. Same public method + result shape,
# so Pipeline can use either. This is what production uses (selected when WHISPER_URL is set).
#
# Job API (the hosted service's FastAPI):
#   POST /api/jobs   (multipart file + model/language/min_speakers/max_speakers) -> 201 {job_id}
#   GET  /api/jobs/{id}                                      -> {status: <stage>, error: ...}
#   GET  /api/jobs/{id}/download                             -> speaker-tagged .txt (when done)
# Terminal stages: done | error | interrupted | cancelled.
#
# The HTTP client is injectable (a fake in tests) so this is verifiable with no network/models/mTLS.
# The real client uses mTLS with the vmin client cert, read from ENV at runtime — never hard-coded.

require "json"

module Helpdesk
  class RemoteTranscriber
    TERMINAL = %w[done error interrupted cancelled].freeze
    MAX_POLL_TRANSIENT = 6 # consecutive non-200 status polls tolerated before giving up (network/5xx/TLS)

    def initialize(base_url: ENV["WHISPER_URL"], language: "cs", client: nil,
                   poll_interval: 3.0, poll_timeout: 3600)
      @base = base_url
      @language = language
      @poll_interval = poll_interval
      @poll_timeout = poll_timeout
      @client = client || MtlsClient.new(base_url)
    end

    # → { state:, retryable:, transcript:, job_id?, reason?, http? }
    #   state ∈ transcribed | no_speech | failed   (parity with Transcriber; remote has no config_error)
    def transcribe(audio, model: "large-v2", min_speakers: 2, max_speakers: 3, timeout: nil)
      code, body = @client.post_multipart("/api/jobs", audio.to_s,
        "model" => model, "language" => @language,
        "min_speakers" => min_speakers.to_s, "max_speakers" => max_speakers.to_s)
      return failed(code, "submit failed", body) unless code == 201
      job_id = (JSON.parse(body)["job_id"] rescue nil)
      return failed(code, "no job_id in submit response", body) unless job_id

      status = poll_to_terminal(job_id, timeout)
      return failed(nil, "poll timeout", nil, job_id) if status == :timeout
      return failed(status[:code], "poll failed", status[:body], job_id) if status.is_a?(Hash)

      case status
      when "done"
        code, text = @client.get("/api/jobs/#{job_id}/download")
        return failed(code, "download failed", text, job_id) unless code == 200
        # The HTTP body arrives tagged ASCII-8BIT, but the transcript is UTF-8 (Czech text + speaker
        # tags). Relabel it so downstream (summariser prompt, wiki RAG, JSON) doesn't raise
        # Encoding::CompatibilityError mixing UTF-8 and ASCII-8BIT. scrub drops any invalid bytes.
        text = text.to_s.dup.force_encoding("UTF-8").scrub
        if text.strip.empty?
          { state: "no_speech", retryable: false, transcript: "", job_id: job_id }
        else
          { state: "transcribed", retryable: false, transcript: text, job_id: job_id }
        end
      else # error | interrupted | cancelled -> transient; safe to re-enqueue
        { state: "failed", retryable: true, transcript: nil, job_id: job_id, reason: "remote #{status}" }
      end
    end

    private

    # Returns the terminal status String, :timeout, or a {code:,body:} on a HARD HTTP error while polling.
    # A single transient blip (5xx, dropped keepalive, read-timeout, TLS hiccup -> code 0) must NOT abort
    # a job the server is still running: that would discard it and force a full re-upload + re-transcribe
    # (an hour of CPU on the no-GPU box). We retry non-200s within the same deadline, with backoff, and
    # only hard-fail after MAX_POLL_TRANSIENT consecutive failures. A definite 404 (job gone) stops now.
    def poll_to_terminal(job_id, timeout)
      deadline = mono + (timeout || @poll_timeout)
      transient = 0
      loop do
        code, body = @client.get("/api/jobs/#{job_id}")
        if code == 200
          transient = 0
          status = (JSON.parse(body)["status"] rescue nil)
          return status if TERMINAL.include?(status)
        elsif code == 404
          return { code: code, body: body }                       # job genuinely gone — don't retry forever
        else
          transient += 1                                          # transient network/5xx/TLS blip — keep polling
          return { code: code, body: body } if transient > MAX_POLL_TRANSIENT
        end
        return :timeout if mono >= deadline
        next unless @poll_interval.positive?
        wait = transient.zero? ? @poll_interval : [@poll_interval * (2**[transient, 4].min), 60.0].min
        sleep [wait, deadline - mono].min
      end
    end

    def failed(code, reason, body = nil, job_id = nil)
      { state: "failed", retryable: true, transcript: nil, reason: reason, http: code, job_id: job_id, body: body }
    end

    def mono = Process.clock_gettime(Process::CLOCK_MONOTONIC)

    # --- real mTLS HTTP client (standard library only; multipart built by hand) ---
    class MtlsClient
      def initialize(base_url,
                     p12_path: ENV["WHISPER_MTLS_P12"], p12_pass: ENV["WHISPER_MTLS_PASS"],
                     ca_file: ENV["WHISPER_CA"])
        require "net/http"
        require "openssl"
        require "uri"
        @base = URI.parse(base_url)
        @p12_path = p12_path
        @p12_pass = p12_pass
        @ca_file = ca_file
      end

      def post_multipart(path, file_path, fields)
        boundary = "----HelpdeskWhisper#{rand(1 << 64).to_s(16)}"
        body = build_multipart(boundary, file_path, fields)
        req = Net::HTTP::Post.new(join(path))
        req["Content-Type"] = "multipart/form-data; boundary=#{boundary}"
        req.body = body
        run(req)
      end

      def get(path)
        run(Net::HTTP::Get.new(join(path)))
      end

      private

      def join(path) = URI.join(@base.to_s.end_with?("/") ? @base.to_s : "#{@base}/", path.sub(%r{\A/}, ""))

      def run(req)
        http = Net::HTTP.new(@base.host, @base.port)
        http.use_ssl = @base.scheme == "https"
        if http.use_ssl?
          http.ca_file = @ca_file if @ca_file
          if @p12_path && File.exist?(@p12_path)
            p12 = OpenSSL::PKCS12.new(File.binread(@p12_path), @p12_pass.to_s)
            http.cert = p12.certificate
            http.key = p12.key
          end
        end
        http.open_timeout = 15
        http.read_timeout = 120
        resp = http.request(req)
        [resp.code.to_i, resp.body]
      rescue => e
        [0, "client error: #{e.class}: #{e.message}"]
      end

      def build_multipart(boundary, file_path, fields)
        parts = +""
        fields.each do |k, v|
          parts << "--#{boundary}\r\nContent-Disposition: form-data; name=\"#{k}\"\r\n\r\n#{v}\r\n"
        end
        # The Whisper Web UI validates the upload by FILE EXTENSION. Backend recordings are stored
        # extensionless as "<uuid>.audio", which the service rejects ("unsupported file type: .audio").
        # Send a content-sniffed, service-accepted extension (the bytes themselves are unchanged, and
        # the server's ffmpeg decodes by content). Survives a recorder-format change (e.g. MP4 → FLAC).
        fname = "recording#{sniff_ext(file_path)}"
        parts << "--#{boundary}\r\nContent-Disposition: form-data; name=\"file\"; filename=\"#{fname}\"\r\n"
        parts << "Content-Type: application/octet-stream\r\n\r\n"
        parts = parts.b
        parts << File.binread(file_path) << "\r\n--#{boundary}--\r\n".b
        parts
      end

      # Map the audio container to an extension the Whisper service accepts, sniffed from magic bytes
      # (the stored recording is extensionless). Defaults to the current recorder format (.m4a/MP4).
      def sniff_ext(path)
        head = File.binread(path, 16).to_s
        return ".m4a"  if head[4, 4] == "ftyp"                                   # ISO-BMFF / MP4 / M4A
        return ".wav"  if head[0, 4] == "RIFF" && head[8, 4] == "WAVE"
        return ".flac" if head[0, 4] == "fLaC"
        return ".ogg"  if head[0, 4] == "OggS"                                   # Ogg (Vorbis/Opus)
        return ".mp3"  if head[0, 3] == "ID3" || (head.bytes[0] == 0xFF && (head.bytes[1].to_i & 0xE0) == 0xE0)
        ".m4a"                                                                   # default: current recorder output
      end
    end
  end
end