Musubi/Packages/MusubiKit/Sources/Intercom/IntercomServer.swift

import CoreVideo
import Darwin
import Foundation
import Network
import Observation
import OSLog

/// Intercom stage 1 (M4): phones on the clinic interface open a page, hear talkback and see PGM.
/// Web server and media port bind only to the chosen interface.
@MainActor @Observable
public final class IntercomServer {
    public struct Participant: Identifiable, Equatable, Sendable {
        public let id: UUID
        public let name: String
        public let role: String
        public let connected: Bool
    }

    public static let httpPort: UInt16 = 8080
    public static let mediaPort: UInt16 = 50_000

    public private(set) var participants: [Participant] = []
    /// Address phones open (also encoded in the QR code).
    public private(set) var url: URL?
    public private(set) var error: String?
    public private(set) var isRunning = false

    /// Thread-safe media input (PGM frames, talkback audio).
    public nonisolated let input: IntercomInput

    private var engine: IntercomEngine?
    private var http: HTTPServer?
    private var retryTask: Task<Void, Never>?
    private let log = Logger(subsystem: "studio.zielke.musubi", category: "intercom")

    public init() {
        input = IntercomInput()
    }

    public func start(interfaceName: String) async {
        stop()
        guard let interface = await Self.interface(named: interfaceName), let address = Self.ipv4Address(of: interfaceName) else {
            error = "Klinik-Schnittstelle \(interfaceName) hat keine Adresse. Neuer Versuch in 5 s."
            log.error("Intercom: interface \(interfaceName, privacy: .public) unavailable")
            scheduleRetry(interfaceName)
            return
        }
        do {
            let engine = IntercomEngine(certificate: try DTLSCertificate())
            engine.onSessionsChanged = { [weak self, input] infos in
                input.hasListeners = infos.contains { $0.connected }
                Task { @MainActor in
                    self?.participants = infos.map { Participant(id: $0.id, name: $0.name, role: $0.role, connected: $0.connected) }
                }
            }
            engine.onKeyFrameNeeded = { [input] in input.requestKeyFrame() }
            try await engine.start(interface: interface, port: Self.mediaPort)

            let mediaPort = Self.mediaPort
            let http = HTTPServer { [weak engine] request in
                guard let engine else { return .text(503, "Musubi startet neu.") }
                return Self.route(request, engine: engine, address: address, mediaPort: mediaPort)
            }
            try await http.start(interface: interface, port: Self.httpPort)

            input.attach(engine)
            self.engine = engine
            self.http = http
            url = URL(string: "http://\(address):\(Self.httpPort)/")
            isRunning = true
            error = nil
            log.info("Intercom on \(interfaceName, privacy: .public) at \(address, privacy: .public)")
        } catch {
            stop()
            self.error = "Intercom lässt sich auf \(interfaceName) nicht starten. Neuer Versuch in 5 s."
            log.error("Intercom start failed: \(error.localizedDescription, privacy: .public)")
            scheduleRetry(interfaceName)
        }
    }

    public func stop() {
        retryTask?.cancel()
        retryTask = nil
        input.attach(nil)
        http?.stop()
        http = nil
        engine?.stop()
        engine = nil
        participants = []
        url = nil
        isRunning = false
    }

    private func scheduleRetry(_ interfaceName: String) {
        retryTask = Task { [weak self] in
            try? await Task.sleep(for: .seconds(5))
            guard !Task.isCancelled else { return }
            await self?.start(interfaceName: interfaceName)
        }
    }

    // MARK: - Signalling

    private struct OfferRequest: Decodable { let name: String; let role: String; let sdp: String }
    private struct OfferResponse: Encodable { let id: UUID; let sdp: String }
    private struct LeaveRequest: Decodable { let id: UUID }

    nonisolated private static func route(_ request: HTTPRequest, engine: IntercomEngine, address: String, mediaPort: UInt16) -> HTTPResponse {
        switch (request.method, request.path) {
        case ("GET", "/"), ("GET", "/index.html"):
            return HTTPResponse(status: 200, contentType: "text/html; charset=utf-8", body: Array(PhonePage.html.utf8))
        case ("POST", "/api/offer"):
            guard let offer = try? JSONDecoder().decode(OfferRequest.self, from: Data(request.body)) else {
                return .text(400, "Angebot unlesbar.")
            }
            let name = String(offer.name.trimmingCharacters(in: .whitespacesAndNewlines).prefix(40))
            let role = String(offer.role.trimmingCharacters(in: .whitespacesAndNewlines).prefix(40))
            guard !name.isEmpty, !role.isEmpty else { return .text(400, "Name und Rolle fehlen.") }
            do {
                guard let (id, sdp) = try engine.createSession(offer: offer.sdp, name: name, role: role,
                                                               address: address, port: mediaPort) else {
                    return .text(400, "Dein Browser kann weder Opus noch H.264 empfangen.")
                }
                return .json(OfferResponse(id: id, sdp: sdp))
            } catch {
                return .text(400, "Angebot unvollständig.")
            }
        case ("POST", "/api/leave"):
            if let leave = try? JSONDecoder().decode(LeaveRequest.self, from: Data(request.body)) { engine.removeSession(leave.id) }
            return HTTPResponse(status: 204, contentType: "text/plain", body: [])
        case (_, "/"), (_, "/api/offer"), (_, "/api/leave"):
            return .text(405, "Methode nicht erlaubt.")
        default:
            return .text(404, "Nicht gefunden.")
        }
    }

    // MARK: - Interfaces

    private static func interface(named name: String) async -> NWInterface? {
        await withCheckedContinuation { continuation in
            let monitor = NWPathMonitor()
            let once = OnceFlag()
            monitor.pathUpdateHandler = { path in
                guard once.claim() else { return }
                monitor.cancel()
                continuation.resume(returning: path.availableInterfaces.first { $0.name == name })
            }
            monitor.start(queue: .global(qos: .utility))
        }
    }

    /// First IPv4 address of a BSD interface (e.g. "en0").
    nonisolated static func ipv4Address(of interfaceName: String) -> String? {
        var list: UnsafeMutablePointer<ifaddrs>?
        guard getifaddrs(&list) == 0, let first = list else { return nil }
        defer { freeifaddrs(list) }
        for entry in sequence(first: first, next: { $0.pointee.ifa_next }) {
            let ifa = entry.pointee
            guard String(cString: ifa.ifa_name) == interfaceName, let addr = ifa.ifa_addr,
                  addr.pointee.sa_family == UInt8(AF_INET) else { continue }
            var host = [CChar](repeating: 0, count: Int(NI_MAXHOST))
            guard getnameinfo(addr, socklen_t(addr.pointee.sa_len), &host, socklen_t(host.count), nil, 0, NI_NUMERICHOST) == 0 else { continue }
            return String(decoding: host.prefix { $0 != 0 }.map { UInt8(bitPattern: $0) }, as: UTF8.self)
        }
        return nil
    }
}

/// Media input for the intercom; safe to call from any thread.
public final class IntercomInput: @unchecked Sendable {
    private let lock = NSLock()
    private var engine: IntercomEngine?
    private var listeners = false
    private var talkback = TalkbackEncoder()
    private let preview = PreviewEncoder()
    private var resamplePosition: Double = 0
    private var resampleLast: Float = 0
    private var scratch: [Float] = []

    init() {
        preview.onAccessUnit = { [weak self] nalUnits, timestamp in
            self?.currentEngine?.sendVideo(nalUnits: nalUnits, timestamp: timestamp)
        }
    }

    var hasListeners: Bool {
        get { lock.withLock { listeners } }
        set { lock.withLock { listeners = newValue } }
    }

    private var currentEngine: IntercomEngine? { lock.withLock { engine } }

    func attach(_ engine: IntercomEngine?) {
        lock.withLock {
            self.engine = engine
            listeners = false
        }
    }

    func requestKeyFrame() { preview.requestKeyFrame() }

    /// One PGM frame (any size, BGRA or UYVY). Dropped while no phone is connected.
    public func pushVideo(_ pixelBuffer: CVPixelBuffer) {
        guard hasListeners else { return }
        preview.encode(pixelBuffer)
    }

    /// Talkback audio, mono, any sample rate. Pass silence while the talk key is up so the stream keeps its timing.
    /// Call from one thread only (not the audio render thread).
    public func pushTalkback(_ samples: UnsafeBufferPointer<Float>, sampleRate: Double) {
        guard let engine = currentEngine, hasListeners, let talkback else { return }
        let packets: [([UInt8], UInt32)]
        if sampleRate == TalkbackEncoder.sampleRate {
            packets = talkback.append(samples)
        } else {
            resample(samples, from: sampleRate)
            packets = scratch.withUnsafeBufferPointer { talkback.append($0) }
        }
        for (packet, timestamp) in packets { engine.sendAudio(opus: packet, timestamp: timestamp) }
    }

    /// Linear interpolation to 48 kHz; good enough for speech.
    private func resample(_ input: UnsafeBufferPointer<Float>, from rate: Double) {
        scratch.removeAll(keepingCapacity: true)
        let step = rate / TalkbackEncoder.sampleRate
        var position = resamplePosition
        while position < Double(input.count) {
            let index = Int(position)
            let fraction = Float(position - Double(index))
            let previous = index == 0 ? resampleLast : input[index - 1]
            scratch.append(previous + (input[index] - previous) * fraction)
            position += step
        }
        resamplePosition = position - Double(input.count)
        resampleLast = input.last ?? resampleLast
    }
}