//
// OpenRocky — Voice-first AI Agent
// https://github.com/openrocky
//
// Developed by everettjf with the assistance of Claude Code and Codex.
// Date: 2026-03-25
// Copyright (c) 2026 everettjf. All rights reserved.
//

import AVFoundation
import Foundation
@preconcurrency import SwiftOpenAI

@RealtimeActor
final class OpenRockyRealtimeVoiceBridge {
    private var client: (any OpenRockyRealtimeVoiceClient)?
    private var audioController: AudioController?
    private var microphoneTask: Task<Void, Never>?
    private var eventSink: (@Sendable (OpenRockyRealtimeEvent) -> Void)?
    private var activeConfiguration: OpenRockyRealtimeProviderConfiguration?
    private var soulInstructions: String = ""
    private var advancedSettings: OpenRockyRealtimeAdvancedSettings = .default
    /// Captured at startIfNeeded so a reconnect attempt can rebuild the client
    /// with the same surface area (tool whitelist, mic input state, priming).
    private var lastRealtimeTools: [OpenAIRealtimeSessionConfiguration.RealtimeTool] = []
    private var lastVoiceInputEnabled: Bool = true
    private var lastPrimingItems: [OpenRockyVoicePrimingItem] = []

    /// When true, mic audio is not forwarded to the server (echo suppression).
    private var isMicSuspended: Bool = false

    /// Set when the server finishes its response, so we know to resume mic once playback drains.
    private var pendingMicResume: Bool = false

    /// Registration token for AudioSession notifications (route / interruption).
    private var audioSessionObserverToken: UUID?

    /// Wall-clock time when the most recent AVAudioSession interruption began.
    /// Used to distinguish short blips (CarPlay nav prompt, quick Siri query) that
    /// should auto-resume from long ones (phone calls) that should wait for a tap.
    private var interruptionStartedAt: Date?

    /// Wall-clock time when the first audio chunk for the current assistant
    /// response was queued for playback. Used to compute `audio_end_ms` on
    /// barge-in so the server's transcript reflects how much the user actually
    /// heard, not "nothing". Nil between responses.
    private var playbackStartedAt: Date?

    /// In-flight reconnect task. Spawned once on .disconnected; cancelled on
    /// successful reconnect or manual stop. Skips re-entry while running.
    private var reconnectTask: Task<Void, Never>?

    /// Backoff schedule for unexpected drops. Three short tries cover the
    /// common Wi-Fi → cell handoff and brief server hiccups. After that we
    /// give up and ask the user to tap, so we don't burn battery looping.
    private static let reconnectBackoffSeconds: [UInt64] = [1, 3, 9]

    /// Final-transcript turns observed since this session started, captured
    /// from the event stream. Combined with `lastPrimingItems` on reconnect so
    /// the new session sees both the pre-session context AND whatever was said
    /// after — otherwise reconnect "forgets" the live conversation.
    /// Hard-capped so a long session doesn't grow unbounded; the existing
    /// priming budget then re-truncates the combined list before replay.
    private var inSessionTurns: [OpenRockyVoicePrimingItem] = []
    private static let inSessionTurnsHardCap = 100

    /// Cached snapshot of the barge-in preference. Read on every mic buffer
    /// (~50 Hz), so going to UserDefaults each time was a hot-path waste.
    /// Updated via a `UserDefaults.didChangeNotification` async sequence so a
    /// mid-session toggle in Settings still takes effect.
    private var bargeInEnabled: Bool = false
    private var preferencesObserverTask: Task<Void, Never>?
    nonisolated private static let bargeInPreferenceKey = "rocky.pref.voiceInterruptionEnabled"

    /// Maximum interruption duration we'll auto-resume from. Above this we
    /// require an explicit user tap so we don't surprise them with a hot mic
    /// after they've moved on (taken a call, switched apps, etc).
    private static let autoResumeMaxInterruptionDuration: TimeInterval = 10

    func startIfNeeded(
        configuration: OpenRockyRealtimeProviderConfiguration,
        voiceInputEnabled: Bool,
        soulInstructions: String,
        realtimeTools: [OpenAIRealtimeSessionConfiguration.RealtimeTool],
        primingItems: [OpenRockyVoicePrimingItem] = [],
        eventSink: @escaping @Sendable (OpenRockyRealtimeEvent) -> Void
    ) async throws {
        // Wrap the upstream sink to drive reconnect on `.disconnected` and to
        // accumulate `inSessionTurns` from transcript-final events so a reconnect
        // can replay the live conversation, not just the pre-session priming.
        let wrappedSink: @Sendable (OpenRockyRealtimeEvent) -> Void = { [weak self] event in
            switch event {
            case .disconnected(let lastError):
                Task { @RealtimeActor [weak self] in
                    self?.handleUnexpectedDisconnect(lastError: lastError)
                }
            case .userTranscriptFinal(let text):
                Task { @RealtimeActor [weak self] in
                    self?.appendInSessionTurn(role: .user, text: text)
                }
            case .assistantTranscriptFinal(let text):
                Task { @RealtimeActor [weak self] in
                    self?.appendInSessionTurn(role: .assistant, text: text)
                }
            default:
                break
            }
            eventSink(event)
        }

        // Capture pre-update soul/tool signature so the warm-reuse branch can
        // detect what changed and push a `session.update` for it.
        let previousSoul = self.soulInstructions
        let previousToolsSignature = Self.toolsSignature(self.lastRealtimeTools)

        self.eventSink = wrappedSink
        self.soulInstructions = soulInstructions
        self.lastRealtimeTools = realtimeTools
        self.lastVoiceInputEnabled = voiceInputEnabled
        self.lastPrimingItems = primingItems
        self.bargeInEnabled = Self.readBargeInPreference()
        registerPreferencesObserverIfNeeded()
        let normalized = configuration.normalized()
        self.advancedSettings = normalized.effectiveAdvancedSettings

        if activeConfiguration?.identity != normalized.identity || client == nil {
            if client != nil {
                await stop()
                self.eventSink = wrappedSink
                self.lastRealtimeTools = realtimeTools
                self.lastVoiceInputEnabled = voiceInputEnabled
                self.lastPrimingItems = primingItems
            }

            let client = Self.makeClient(
                configuration: normalized,
                soulInstructions: soulInstructions,
                realtimeTools: realtimeTools,
                primingItems: primingItems
            )
            try await client.connect(eventSink: wrappedSink)
            self.client = client
            activeConfiguration = normalized
            // Fresh session — drop any in-session turns from the previous one.
            self.inSessionTurns = []
        } else {
            // Warm reuse — config identity unchanged. But the soul or tool
            // whitelist may still have moved (identity is a narrow hash that
            // misses the system prompt and tool list). Push a `session.update`
            // so the live session reflects the latest values instead of
            // silently running with stale ones.
            let nextToolsSignature = Self.toolsSignature(realtimeTools)
            if previousSoul != soulInstructions || previousToolsSignature != nextToolsSignature {
                rlog.info("Hot-updating realtime session (soul changed=\(previousSoul != soulInstructions) tools changed=\(previousToolsSignature != nextToolsSignature))", category: "Voice")
                try? await client?.updateSession(
                    soulInstructions: soulInstructions,
                    realtimeTools: realtimeTools
                )
            }
        }

        try await configureAudioController(voiceInputEnabled: voiceInputEnabled)
    }

    func stop() async {
        reconnectTask?.cancel()
        reconnectTask = nil
        microphoneTask?.cancel()
        microphoneTask = nil

        audioController?.stop()
        audioController = nil
        await client?.disconnect()
        client = nil
        activeConfiguration = nil
        isMicSuspended = false
        pendingMicResume = false
        playbackStartedAt = nil
        inSessionTurns = []
        if let token = audioSessionObserverToken {
            await MainActor.run { OpenRockyAudioSessionObserver.shared.removeHandler(token) }
            audioSessionObserverToken = nil
        }
        preferencesObserverTask?.cancel()
        preferencesObserverTask = nil
        emit(.microphoneActive(false))
        emit(.status("Voice session stopped."))
    }

    func sendText(_ text: String) async throws {
        guard let client else {
            throw OpenRockyRealtimeVoiceClientError.notConnected
        }
        try await client.sendText(text)
    }

    func sendToolOutput(callID: String, output: String) async throws {
        guard let client else {
            throw OpenRockyRealtimeVoiceClientError.notConnected
        }
        try await client.sendToolOutput(callID: callID, output: output)
    }

    private func configureAudioController(voiceInputEnabled: Bool) async throws {
        let modes: [AudioController.Mode] = voiceInputEnabled ? [.playback, .record] : [.playback]

        if voiceInputEnabled, audioController != nil, microphoneTask != nil {
            emit(.microphoneActive(true))
            return
        }

        audioController?.stop()
        audioController = try await AudioController(modes: modes)
        audioController?.setPlaybackDrainedHandler { [weak self] in
            Task { @RealtimeActor [weak self] in
                guard let self else { return }
                rlog.debug("Bridge: playback drained, pendingMicResume=\(self.pendingMicResume)", category: "Audio")
                guard self.pendingMicResume else { return }
                rlog.debug("Bridge: resuming mic after drain", category: "Audio")
                self.pendingMicResume = false
                self.isMicSuspended = false
                self.playbackStartedAt = nil
            }
        }
        await registerAudioSessionObserverIfNeeded()
        emit(.microphoneActive(voiceInputEnabled))

        if voiceInputEnabled {
            startMicrophoneLoop()
        } else {
            microphoneTask?.cancel()
            microphoneTask = nil
        }
    }

    private func startMicrophoneLoop() {
        guard let client, let audioController else { return }
        microphoneTask?.cancel()
        microphoneTask = Task {
            do {
                let micStream = try audioController.micStream()
                rlog.info("Mic stream started, suspended=\(isMicSuspended)", category: "Audio")
                var micBufferCount = 0
                var skippedCount = 0
                for await buffer in micStream {
                    guard Task.isCancelled == false else { break }
                    micBufferCount += 1
                    // While the assistant is speaking we suspend mic forwarding to prevent the
                    // speaker bleeding back into the model. We still scan for a high-RMS spike
                    // so the user can barge in (gated on the Preferences toggle).
                    if isMicSuspended {
                        skippedCount += 1
                        if bargeInEnabled {
                            let rms = Self.computeRMSFromBuffer(buffer)
                            if rms > advancedSettings.bargeInRMSThreshold {
                                rlog.info("Mic interrupt detected during suspension (rms=\(rms) threshold=\(advancedSettings.bargeInRMSThreshold))", category: "Audio")
                                // Local RMS path: the server hasn't seen audio (mic is suspended),
                                // so it doesn't know to cancel the response. Tell it explicitly.
                                interruptPlayback()
                                cancelInFlightResponse()
                                emit(.inputSpeechStarted)
                            }
                        }
                        if skippedCount % 50 == 1 {
                            rlog.debug("Mic buffer skipped (suspended), skipped=\(skippedCount)", category: "Audio")
                        }
                        continue
                    }
                    guard let base64Audio = AudioUtils.base64EncodeAudioPCMBuffer(from: buffer) else {
                        rlog.warning("Mic buffer encode failed at #\(micBufferCount)", category: "Audio")
                        continue
                    }
                    if micBufferCount % 50 == 1 {
                        rlog.debug("Mic buffer #\(micBufferCount) sending, b64=\(base64Audio.count)", category: "Audio")
                    }
                    try await client.sendAudioChunk(base64Audio: base64Audio)
                }
                rlog.info("Mic stream ended, total buffers=\(micBufferCount)", category: "Audio")
            } catch {
                rlog.error("Mic stream error: \(error.localizedDescription)", category: "Audio")
                emit(.error(error.localizedDescription))
            }
        }
    }

    func handlePlaybackEvent(_ event: OpenRockyRealtimeEvent) {
        guard case .assistantAudioChunk(let audio) = event else { return }
        // First chunk of a response — mark the start so a barge-in can report
        // how many ms the user actually heard.
        if playbackStartedAt == nil {
            playbackStartedAt = Date()
        }
        // Suspend mic while assistant audio is playing to prevent echo feedback.
        if !isMicSuspended {
            rlog.debug("Suspending mic for playback", category: "Audio")
        }
        isMicSuspended = true
        audioController?.playPCM16Audio(base64String: audio)
    }

    /// Flush any buffered audio when the server signals all audio chunks have been sent.
    /// This handles responses shorter than the initial buffer threshold.
    func flushBufferedAudio() {
        audioController?.flushBufferedAudio()
    }

    private func registerAudioSessionObserverIfNeeded() async {
        guard audioSessionObserverToken == nil else { return }
        let token = await MainActor.run { [weak self] () -> UUID in
            OpenRockyAudioSessionObserver.shared.addHandler { event in
                Task { @RealtimeActor [weak self] in
                    self?.handleAudioSessionEvent(event)
                }
            }
        }
        audioSessionObserverToken = token
    }

    private func handleAudioSessionEvent(_ event: OpenRockyAudioSessionObserver.Event) {
        switch event {
        case .routeChanged:
            // Playback buffers made for the previous route can pop / glitch on the new output.
            // Flushing is cheap and keeps the hand-off clean.
            audioController?.flushBufferedAudio()
        case .interruptionBegan:
            // Phone call / Siri / CarPlay nav prompt took over the audio session.
            // Stop local playback, suspend mic, and tell the server to abandon any
            // in-flight response — otherwise on resume we'd play back stale audio
            // generated against context the user never actually heard.
            audioController?.interruptPlayback()
            isMicSuspended = true
            pendingMicResume = false
            interruptionStartedAt = Date()
            cancelInFlightResponse()
            emit(.status("Voice session interrupted."))
        case .interruptionEnded(let shouldResume):
            let elapsed = interruptionStartedAt.map { Date().timeIntervalSince($0) } ?? .infinity
            interruptionStartedAt = nil
            // Short blip + iOS says we may resume → bring the session back so the
            // user can keep talking hands-free (driving with CarPlay nav prompts is
            // the canonical case). Long interruptions still require a tap.
            if shouldResume && elapsed <= Self.autoResumeMaxInterruptionDuration {
                rlog.info("AudioSession auto-resuming after \(String(format: "%.1f", elapsed))s interruption", category: "Audio")
                audioController?.resumeAfterInterruption()
                isMicSuspended = false
                emit(.status("Voice resumed."))
            } else {
                rlog.info("AudioSession interruption ended after \(String(format: "%.1f", elapsed))s shouldResume=\(shouldResume) — waiting for tap", category: "Audio")
                emit(.status("Voice session paused. Tap to resume."))
            }
        }
    }

    /// Local-side interrupt: stop playback and reset mic-suspension state so the
    /// next mic buffer is forwarded again. Does NOT touch the server.
    ///
    /// Called from two places:
    ///   1. Server-VAD path — sessionRuntime forwards `inputSpeechStarted` here
    ///      when the server detects speech. The server is already handling its
    ///      own turn transition (auto-cancelling with `turn_detected`); racing
    ///      it with our own `response.cancel` produces `response_cancel_not_active`
    ///      AND, more importantly, leaves the server's input-audio buffer in a
    ///      state where subsequent `speech_stopped` events stop firing. So we
    ///      stay strictly local on this path.
    ///   2. Audio-session interruption (phone call etc) — pure local stop.
    ///
    /// For the local-RMS barge-in path (mic is suspended, the server can't see
    /// the user), use `cancelInFlightResponse()` in addition to this.
    func interruptPlayback() {
        audioController?.interruptPlayback()
        pendingMicResume = false
        isMicSuspended = false
        playbackStartedAt = nil
    }

    /// Tell the server to stop generating and rewrite the assistant turn. Only
    /// needed when the server hasn't already cancelled itself — i.e., the local
    /// RMS barge-in path where we detected speech that the server can't see.
    /// Internally idempotent: client-side guards make both messages no-ops if
    /// nothing is in flight.
    func cancelInFlightResponse() {
        // Compute how much of the assistant audio the user actually heard, based
        // on wall-clock time since the first chunk played out. Buffering inside
        // AudioController adds a few tens of ms of slack — close enough that the
        // server's transcript stays in sync with what the user remembers.
        let audioEndMs: Int
        if let startedAt = playbackStartedAt {
            audioEndMs = max(0, Int(Date().timeIntervalSince(startedAt) * 1000))
        } else {
            audioEndMs = 0
        }
        playbackStartedAt = nil
        Task { @RealtimeActor in
            try? await client?.cancelResponse()
            try? await client?.truncateAssistantAudio(audioEndMs: audioEndMs)
        }
    }

    /// Called when the assistant finishes its response (transcript final).
    /// Defers mic resume until all queued audio buffers have actually played out.
    func resumeMicAfterPlayback() {
        // Ensure any remaining buffered audio is flushed before checking playback state
        audioController?.flushBufferedAudio()
        if audioController?.isPlaybackActive == true {
            // Audio still playing — defer resume until playback drains
            rlog.debug("resumeMicAfterPlayback: deferring (playback active)", category: "Audio")
            pendingMicResume = true
        } else {
            // No buffered audio left, safe to resume now
            rlog.debug("resumeMicAfterPlayback: resuming now", category: "Audio")
            pendingMicResume = false
            isMicSuspended = false
            playbackStartedAt = nil
        }
    }

    /// Compute RMS energy directly from an AVAudioPCMBuffer without base64 round-trip.
    private static func computeRMSFromBuffer(_ buffer: AVAudioPCMBuffer) -> Double {
        guard buffer.format.channelCount == 1,
              let ptr = buffer.audioBufferList.pointee.mBuffers.mData else { return 0 }
        let byteCount = Int(buffer.audioBufferList.pointee.mBuffers.mDataByteSize)
        let data = Data(bytes: ptr, count: byteCount)
        return AudioUtils.computeRMS(data)
    }

    private func emit(_ event: OpenRockyRealtimeEvent) {
        eventSink?(event)
    }

    nonisolated private static func readBargeInPreference() -> Bool {
        (UserDefaults.standard.object(forKey: bargeInPreferenceKey) as? Bool) ?? false
    }

    private func registerPreferencesObserverIfNeeded() {
        guard preferencesObserverTask == nil else { return }
        preferencesObserverTask = Task { @RealtimeActor [weak self] in
            let stream = NotificationCenter.default.notifications(named: UserDefaults.didChangeNotification)
            for await _ in stream {
                guard let self else { return }
                if Task.isCancelled { return }
                self.bargeInEnabled = Self.readBargeInPreference()
            }
        }
    }

    /// Schedule a reconnect attempt with exponential backoff. Triggered by the
    /// client emitting `.disconnected` (transport drop, not manual stop). At
    /// most one task runs at a time. After the schedule is exhausted we leave
    /// the session stopped so the next user tap rebuilds it from scratch.
    ///
    /// `lastError` is the most recent server-side error attached to the drop;
    /// when it matches an unrecoverable pattern (auth/quota/model-not-found)
    /// we bail out immediately — retrying a 401 three times only produces
    /// three 401s, and meanwhile the user is staring at "reconnecting...".
    private func handleUnexpectedDisconnect(lastError: String?) {
        guard reconnectTask == nil else { return }
        guard let configuration = activeConfiguration else { return }

        // Clear current client/audio so the rebuild path doesn't see a stale
        // session. We keep activeConfiguration so retries know what to dial.
        microphoneTask?.cancel()
        microphoneTask = nil
        audioController?.stop()
        audioController = nil
        client = nil
        isMicSuspended = false
        pendingMicResume = false
        playbackStartedAt = nil

        emit(.microphoneActive(false))

        if let lastError, OpenRockyVoiceErrorTriage.isUnrecoverable(lastError) {
            rlog.warning("Voice session dropped with unrecoverable error — skipping retry: \(lastError)", category: "Voice")
            let hint = OpenRockyVoiceErrorTriage.hint(for: lastError) ?? "Voice session disconnected. Tap to reconnect."
            emit(.status(hint))
            return
        }

        // Combine pre-session priming with anything captured during the live
        // session; truncate back to the priming budget so a long conversation
        // doesn't balloon the replay.
        let combinedPriming = OpenRockyVoicePriming.capped(lastPrimingItems + inSessionTurns)
        let soulInstructions = self.soulInstructions
        let realtimeTools = self.lastRealtimeTools
        let voiceInputEnabled = self.lastVoiceInputEnabled

        emit(.status("Voice session dropped — reconnecting..."))

        reconnectTask = Task { @RealtimeActor [weak self] in
            guard let self else { return }
            for (index, delaySeconds) in Self.reconnectBackoffSeconds.enumerated() {
                do {
                    try await Task.sleep(nanoseconds: delaySeconds * 1_000_000_000)
                } catch {
                    return // cancelled by stop()
                }
                guard !Task.isCancelled else { return }

                rlog.info("Reconnect attempt \(index + 1)/\(Self.reconnectBackoffSeconds.count)", category: "Voice")
                do {
                    let newClient = Self.makeClient(
                        configuration: configuration,
                        soulInstructions: soulInstructions,
                        realtimeTools: realtimeTools,
                        primingItems: combinedPriming
                    )
                    if let sink = self.eventSink {
                        try await newClient.connect(eventSink: sink)
                    }
                    self.client = newClient
                    try await self.configureAudioController(voiceInputEnabled: voiceInputEnabled)
                    // The new session has freshly replayed history — anything
                    // the user said pre-drop is now part of the priming the
                    // server sees, so start the in-session log clean.
                    self.inSessionTurns = []
                    self.emit(.status("Voice reconnected."))
                    self.reconnectTask = nil
                    return
                } catch {
                    rlog.warning("Reconnect attempt \(index + 1) failed: \(error.localizedDescription)", category: "Voice")
                    // Connect-time errors surface as thrown exceptions, not
                    // `.disconnected` events — classify here too so an
                    // unrecoverable failure mid-backoff stops the loop.
                    if OpenRockyVoiceErrorTriage.isUnrecoverable(error.localizedDescription) {
                        let hint = OpenRockyVoiceErrorTriage.hint(for: error.localizedDescription) ?? "Voice session disconnected. Tap to reconnect."
                        self.emit(.status(hint))
                        self.reconnectTask = nil
                        return
                    }
                }
            }
            self.emit(.status("Voice session disconnected. Tap to reconnect."))
            self.reconnectTask = nil
        }
    }

    /// Append a final-transcript turn to the rolling in-session window.
    /// Bounded so we don't blow memory on a marathon conversation; the
    /// priming budget further trims the combined replay on reconnect.
    private func appendInSessionTurn(role: OpenRockyVoicePrimingItem.Role, text: String) {
        let trimmed = text.trimmingCharacters(in: .whitespacesAndNewlines)
        guard !trimmed.isEmpty else { return }
        inSessionTurns.append(OpenRockyVoicePrimingItem(role: role, text: trimmed))
        if inSessionTurns.count > Self.inSessionTurnsHardCap {
            inSessionTurns.removeFirst(inSessionTurns.count - Self.inSessionTurnsHardCap)
        }
    }

    /// Serialize a tool whitelist to a stable signature for change detection.
    /// `RealtimeTool` isn't Equatable upstream, and we only need
    /// "did this list change". JSON of the array does the job; failure falls
    /// back to a count-based signature so an encode error doesn't hide a real
    /// change behind an accidental equality.
    nonisolated private static func toolsSignature(
        _ tools: [OpenAIRealtimeSessionConfiguration.RealtimeTool]
    ) -> String {
        do {
            let data = try JSONEncoder().encode(tools)
            return String(data: data, encoding: .utf8) ?? "count:\(tools.count)"
        } catch {
            return "count:\(tools.count)"
        }
    }

    private static func makeClient(
        configuration: OpenRockyRealtimeProviderConfiguration,
        soulInstructions: String,
        realtimeTools: [OpenAIRealtimeSessionConfiguration.RealtimeTool],
        primingItems: [OpenRockyVoicePrimingItem]
    ) -> any OpenRockyRealtimeVoiceClient {
        let chatConfiguration = OpenRockyProviderConfiguration(
            provider: .openAI,
            modelID: configuration.modelID,
            credential: configuration.credential
        )
        return OpenRockyOpenAIRealtimeVoiceClient(
            configuration: chatConfiguration,
            realtimeConfiguration: configuration,
            soulInstructions: soulInstructions,
            realtimeTools: realtimeTools,
            primingItems: primingItems
        )
    }
}
