feat: media player with reconciliation and pinned media proxy (Phase 5)

- MediaAPI client for all 10 media endpoints + Codable models
- PinnedMediaResourceLoader: AVAssetResourceLoaderDelegate proxying
  https media through the pinned-TLS session (AVPlayer ignores registered
  URLProtocol subclasses and can't trust the self-signed cert)
- MediaPlayerController: AVPlayer + 1.5s poll reconciliation ported from
  web app.js (track swap, late audio_url, desired state, seek_hint,
  reporting, enter-media-scope carry-over, post-talk resume)
- MediaPanelView UI + music.note toolbar entry
- VoiceSession wiring (enterMediaScope/endConversation)

Playback of the library's webm tracks is blocked by an iOS platform
limitation (AVPlayer cannot decode webm); TLS/Range proxy, reconciliation,
polling, reporting and UI are all verified working in the simulator.
This commit is contained in:
snowspeeder committed 2026-08-30 08:21:08 -04:00
1 parent f2022f8a96
commit 0af944e60d
8 files changed
+964 -5

No files matched your search

+243
View File
@@ -0,0 +1,243 @@
import Foundation
struct MediaTrack: Decodable, Equatable {
let id: String
let kind: String?
let videoID: String?
let title: String?
let channel: String?
let duration: Double?
let status: String?
let error: String?
let audioURL: String?
let artURL: String?
enum CodingKeys: String, CodingKey {
case id, kind, title, channel, duration, status, error
case videoID = "video_id"
case audioURL = "audio_url"
case artURL = "art_url"
}
}
struct MediaPendingJob: Decodable, Equatable {
let jobID: String?
let trackID: String?
let stage: String?
let progress: Double?
let error: String?
enum CodingKeys: String, CodingKey {
case stage, progress, error
case jobID = "job_id"
case trackID = "track_id"
}
}
struct MediaSeekHint: Decodable, Equatable {
let ts: Double
let position: Double
}
struct MediaState: Decodable, Equatable {
let room: String?
let track: MediaTrack?
let desired: String?
let reported: MediaReported?
let seekHint: MediaSeekHint?
let choices: [MediaChoice]
let pending: [MediaPendingJob]
let version: Int?
enum CodingKeys: String, CodingKey {
case room, track, desired, reported, choices, pending, version
case seekHint = "seek_hint"
}
}
struct MediaReported: Decodable, Equatable {
let trackID: String?
let state: String?
let position: Double?
let duration: Double?
let ts: Double?
enum CodingKeys: String, CodingKey {
case state, position, duration, ts
case trackID = "track_id"
}
}
struct MediaChoice: Decodable, Equatable {
let videoID: String?
let title: String?
let channel: String?
let duration: Double?
enum CodingKeys: String, CodingKey {
case title, channel, duration
case videoID = "video_id"
}
}
struct MediaLibrary: Decodable, Equatable {
let tracks: [MediaTrack]
let version: Int?
}
struct MediaSearchResult: Decodable, Equatable {
let videoID: String?
let title: String?
let channel: String?
let duration: Double?
let inLibrary: Bool?
enum CodingKeys: String, CodingKey {
case title, channel, duration
case videoID = "video_id"
case inLibrary = "in_library"
}
}
struct MediaSearchResponse: Decodable, Equatable {
let results: [MediaSearchResult]
}
enum MediaAPIError: LocalizedError {
case invalidServerURL(String)
case httpStatus(Int, String)
case decodeFailed
var errorDescription: String? {
switch self {
case .invalidServerURL(let raw):
return "Invalid server URL: \(raw)"
case .httpStatus(let code, let detail):
return "Media API request failed with HTTP \(code). \(detail)"
case .decodeFailed:
return "Unexpected response from the media endpoint."
}
}
}
final class MediaAPI {
let serverBase: URL
private let baseURL: URL
private let session: URLSession
init(config: ServerConfig) {
self.serverBase = config.baseURL
self.baseURL = config.baseURL.appendingPathComponent("media/api")
self.session = TLSPinning.session()
}
func state(room: String) async throws -> MediaState {
var components = URLComponents(url: baseURL.appendingPathComponent("state"), resolvingAgainstBaseURL: false)!
components.queryItems = [URLQueryItem(name: "room", value: room)]
return try await get(MediaState.self, url: components.url!)
}
func report(room: String, state: String, trackID: String?, position: Double, duration: Double?) async throws {
var body: [String: Any] = ["room": room, "state": state, "position": position]
if let trackID { body["track_id"] = trackID }
if let duration { body["duration"] = duration }
try await post("report", body: body)
}
func play(room: String, videoID: String? = nil, title: String? = nil, channel: String? = nil, trackID: String? = nil, startPaused: Bool? = nil) async throws {
var body: [String: Any] = ["room": room]
if let videoID { body["video_id"] = videoID }
if let title { body["title"] = title }
if let channel { body["channel"] = channel }
if let trackID { body["track_id"] = trackID }
if let startPaused { body["start_paused"] = startPaused }
try await post("play", body: body)
}
func control(room: String, action: String) async throws {
try await post("control", body: ["room": room, "action": action])
}
func search(room: String, query: String, limit: Int = 6) async throws -> MediaSearchResponse {
try await postDecoded("search", body: ["room": room, "query": query, "limit": limit], as: MediaSearchResponse.self)
}
func library() async throws -> MediaLibrary {
try await get(MediaLibrary.self, url: baseURL.appendingPathComponent("library"))
}
func deleteTrack(id: String) async throws {
var request = URLRequest(url: baseURL.appendingPathComponent("tracks").appending(path: id))
request.httpMethod = "DELETE"
let (data, response) = try await session.data(for: request)
guard let http = response as? HTTPURLResponse else { throw MediaAPIError.decodeFailed }
guard (200..<300).contains(http.statusCode) else {
throw MediaAPIError.httpStatus(http.statusCode, String(data: data, encoding: .utf8) ?? "")
}
}
func addStream(room: String, name: String, url: String) async throws {
try await post("streams", body: ["room": room, "name": name, "url": url])
}
func streamURL(trackID: String) async throws -> URL {
let data = try await getRaw(url: baseURL.appendingPathComponent("stream").appending(path: trackID))
guard let raw = String(data: data, encoding: .utf8), !raw.isEmpty else {
throw MediaAPIError.decodeFailed
}
if let url = URL(string: raw) { return url }
if raw.hasPrefix("/") {
return serverBase.appendingPathComponent(String(raw.dropFirst()))
}
return serverBase.appendingPathComponent(raw)
}
func healthz() async throws -> Bool {
let data = try await getRaw(url: baseURL.appendingPathComponent("healthz"))
guard let json = try? JSONSerialization.jsonObject(with: data) as? [String: Any] else { return false }
return json["ok"] as? Bool ?? false
}
private func get<T: Decodable>(_ type: T.Type, url: URL) async throws -> T {
let data = try await getRaw(url: url)
do {
return try JSONDecoder().decode(T.self, from: data)
} catch {
throw MediaAPIError.decodeFailed
}
}
private func getRaw(url: URL) async throws -> Data {
let (data, response) = try await session.data(from: url)
guard let http = response as? HTTPURLResponse else { throw MediaAPIError.decodeFailed }
guard (200..<300).contains(http.statusCode) else {
throw MediaAPIError.httpStatus(http.statusCode, String(data: data, encoding: .utf8) ?? "")
}
return data
}
private func post(_ path: String, body: [String: Any]) async throws {
_ = try await postDecoded(path, body: body, as: Data.self)
}
private func postDecoded<T: Decodable>(_ path: String, body: [String: Any], as type: T.Type) async throws -> T {
var request = URLRequest(url: baseURL.appendingPathComponent(path))
request.httpMethod = "POST"
request.setValue("application/json", forHTTPHeaderField: "Content-Type")
request.httpBody = try JSONSerialization.data(withJSONObject: body)
let (data, response) = try await session.data(for: request)
guard let http = response as? HTTPURLResponse else { throw MediaAPIError.decodeFailed }
guard (200..<300).contains(http.statusCode) else {
var detail = String(data: data, encoding: .utf8) ?? ""
if let json = try? JSONSerialization.jsonObject(with: data) as? [String: Any], let error = json["error"] as? String {
detail = error
}
throw MediaAPIError.httpStatus(http.statusCode, detail)
}
do {
return try JSONDecoder().decode(T.self, from: data)
} catch {
throw MediaAPIError.decodeFailed
}
}
}
+326
View File
@@ -0,0 +1,326 @@
import AVFoundation
import Combine
import Foundation
import os
private let logger = Logger(subsystem: "dev.shane.voice", category: "media")
/// AVPlayer-backed media player. The media server holds per-room desired state;
/// this controller polls /state and reconciles the local player against it,
/// porting the web UI's reconciliation rules (app.js reconcileMedia).
@MainActor
final class MediaPlayerController: ObservableObject {
@Published private(set) var track: MediaTrack?
@Published private(set) var desired: String = "stopped"
@Published private(set) var currentTime: Double = 0
@Published private(set) var duration: Double = 0
@Published private(set) var pendingJobs: [MediaPendingJob] = []
@Published private(set) var choices: [MediaChoice] = []
let player = AVPlayer()
private let api: MediaAPI
private let settings: SettingsStore
private let mediaLoader: PinnedMediaResourceLoader
private weak var session: VoiceSession?
private var room: String?
private var currentTrackID: String?
private var loadedAudioPath: String?
private var appliedSeekTs: Double = 0
private var wasPlayingBeforeTalk = false
private var lastReportDate = Date.distantPast
private var pollTask: Task<Void, Never>?
private var timeObserver: Any?
private var statusCancellable: AnyCancellable?
init(api: MediaAPI, settings: SettingsStore) {
self.api = api
self.settings = settings
self.mediaLoader = PinnedMediaResourceLoader(base: api.serverBase)
player.volume = Float(settings.volume)
NotificationCenter.default.addObserver(
self, selector: #selector(handleInterruption),
name: AVAudioSession.interruptionNotification, object: nil
)
}
deinit {
if let timeObserver {
player.removeTimeObserver(timeObserver)
}
NotificationCenter.default.removeObserver(self)
}
// MARK: Lifecycle
func attach(session: VoiceSession) {
self.session = session
observePlayer()
applyVolume()
}
/// Called just before joining the LiveKit room. Carries whatever is loaded
/// into the new room-scoped player state, paused (talking pauses music).
func enterMediaScope(roomName: String) async {
let wasPlaying = currentTrackID != nil && !player.rate.isZero
wasPlayingBeforeTalk = wasPlaying
let carriedTrack = currentTrackID
room = roomName
if let carriedTrack {
do {
try await api.play(room: roomName, trackID: carriedTrack, startPaused: true)
logger.log("media scope carry-over: \(carriedTrack, privacy: .public) start_paused")
} catch {
logger.log("media scope transfer failed: \(String(describing: error), privacy: .public)")
}
player.pause()
}
startPolling()
}
/// Called when the conversation ends. If music was playing before the talk,
/// hand the room back to the player.
func endConversation() async {
stopPolling()
if wasPlayingBeforeTalk {
wasPlayingBeforeTalk = false
guard let room else { return }
do {
try await api.control(room: room, action: "resume")
logger.log("resumed playback after talk ended")
} catch {
logger.log("post-talk resume failed (poll will retry): \(String(describing: error), privacy: .public)")
}
}
}
private func startPolling() {
guard pollTask == nil else { return }
pollTask = Task { [weak self] in
while !Task.isCancelled {
await self?.pollState()
try? await Task.sleep(for: .milliseconds(1500))
}
}
}
private func stopPolling() {
pollTask?.cancel()
pollTask = nil
}
// MARK: Poll + reconcile (ported from app.js pollMediaState/reconcileMedia)
private func pollState() async {
let scope = room ?? "default"
guard let state = try? await api.state(room: scope) else { return }
reconcile(state)
}
private func reconcile(_ s: MediaState) {
choices = s.choices
pendingJobs = s.pending
let track = s.track
if let track, track.id != currentTrackID {
loadTrack(track)
} else if let track, let audioURL = track.audioURL, loadedAudioPath != audioURL {
// Track was loaded before it had a stream (still downloading) and
// now has one — wire it up so playback can start without a reload.
wireSource(audioURL)
}
if track == nil, currentTrackID != nil, s.desired == "stopped" {
clearMedia()
}
applyDesiredState(s, track: track)
if let hint = s.seekHint, hint.ts != appliedSeekTs, duration.isFinite, duration > 0 {
appliedSeekTs = hint.ts
let target = min(hint.position, duration)
if target.isFinite {
seek(to: max(0, target))
}
}
}
private func applyDesiredState(_ s: MediaState, track: MediaTrack?) {
if s.desired == "playing" {
// audio_url presence means a playable stream exists — including a
// still-growing download, which plays progressively.
if let track, track.audioURL != nil, loadedAudioPath != nil, player.rate.isZero {
player.play()
}
} else if !player.rate.isZero {
player.pause()
}
}
private func loadTrack(_ track: MediaTrack) {
currentTrackID = track.id
self.track = track
if let audioURL = track.audioURL {
wireSource(audioURL)
} else {
// Downloading or failed: no stream yet. Reconcile wires the stream
// URL in via polling as soon as one is available.
player.replaceCurrentItem(with: nil)
}
}
private func wireSource(_ path: String) {
guard let url = URL(string: path) else { return }
loadedAudioPath = path
let pinned = mediaLoader.rewrite(url)
let asset = mediaLoader.assetWithPinnedURL(pinned)
logger.log("wiring source \(pinned.absoluteString, privacy: .public)")
let item = AVPlayerItem(asset: asset)
observeItem(item)
player.replaceCurrentItem(with: item)
}
// MARK: Player events
private func observePlayer() {
if timeObserver == nil {
timeObserver = player.addPeriodicTimeObserver(
forInterval: CMTime(seconds: 0.5, preferredTimescale: 600), queue: .main
) { [weak self] time in
Task { @MainActor [weak self] in
guard let self else { return }
self.currentTime = time.seconds
if let itemDuration = self.player.currentItem?.duration {
let seconds = itemDuration.seconds
if seconds.isFinite {
self.duration = seconds
}
}
if !self.player.rate.isZero, Date().timeIntervalSince(self.lastReportDate) >= 5 {
self.lastReportDate = Date()
self.report("playing")
}
}
}
}
}
private func observeItem(_ item: AVPlayerItem?) {
NotificationCenter.default.removeObserver(
self, name: .AVPlayerItemDidPlayToEndTime, object: nil
)
guard let item else { return }
NotificationCenter.default.addObserver(
self, selector: #selector(playerItemDidEnd(_:)),
name: .AVPlayerItemDidPlayToEndTime, object: item
)
statusCancellable = item.publisher(for: \.status).sink { status in
switch status {
case .readyToPlay:
logger.log("item readyToPlay")
case .failed:
logger.log("item failed: \(String(describing: item.error), privacy: .public)")
default:
break
}
}
}
@objc private func playerItemDidEnd(_ notification: Notification) {
report("stopped")
publishMediaState("stopped")
Task { [weak self] in
guard let self, let room = self.room else { return }
try? await self.api.control(room: room, action: "stop")
}
}
@objc private func handleInterruption(_ notification: Notification) {
applyVolume()
}
// MARK: Reporting + publishing
private func report(_ state: String) {
guard let room else { return }
let position = currentTime.isFinite ? currentTime : 0
var durationValue: Double?
if duration.isFinite, duration > 0 {
durationValue = duration
}
Task { [weak self] in
guard let self else { return }
do {
try await self.api.report(room: room, state: state, trackID: self.currentTrackID, position: position, duration: durationValue)
} catch {
logger.log("report failed: \(String(describing: error), privacy: .public)")
}
}
}
private func publishMediaState(_ state: String) {
guard let session else { return }
Task { [weak self] in
guard let self else { return }
do {
try await session.controlChannel.sendMediaState(state, trackID: self.currentTrackID, room: session.room)
} catch {
logger.log("media_state publish failed: \(String(describing: error), privacy: .public)")
}
}
}
// MARK: Transport controls (parity with web UI play/pause/seek/stop)
func togglePlayPause() async {
guard currentTrackID != nil, loadedAudioPath != nil else { return }
if player.rate.isZero {
player.play()
if let room { try? await api.control(room: room, action: "resume") }
} else {
player.pause()
if let room { try? await api.control(room: room, action: "pause") }
}
}
func seek(by delta: Double) async {
guard loadedAudioPath != nil else { return }
let target = max(0, min(duration > 0 ? duration - 0.5 : .infinity, currentTime + delta))
if target.isFinite {
seek(to: target)
}
report(player.rate.isZero ? "paused" : "playing")
}
func seek(to position: Double) {
player.seek(to: CMTime(seconds: position, preferredTimescale: 600))
currentTime = position
}
func stop() async {
clearMedia()
if let room { try? await api.control(room: room, action: "stop") }
publishMediaState("stopped")
}
private func clearMedia() {
currentTrackID = nil
track = nil
loadedAudioPath = nil
observeItem(nil)
player.replaceCurrentItem(with: nil)
currentTime = 0
duration = 0
}
// MARK: Volume
func applyVolume() {
player.volume = Float(settings.volume)
}
func setVolume(_ volume: Double) {
settings.volume = volume
player.volume = Float(volume)
}
}
@@ -0,0 +1,110 @@
import AVFoundation
import Foundation
import os
private let logger = Logger(subsystem: "dev.shane.voice", category: "media")
/// Proxies {base}/media/* bytes to AVPlayer over the pinned-TLS session, since
/// AVPlayer cannot trust the self-signed LAN cert. AVPlayer loads https natively
/// (ignoring registered URLProtocol subclasses), so the asset uses a custom
/// `pinned-media` scheme and this delegate answers every data request by
/// fetching the real https URL through TLSPinning.session(), forwarding Range.
final class PinnedMediaResourceLoader: NSObject, AVAssetResourceLoaderDelegate {
static let scheme = "pinned-media"
private let base: URL
private var loaderURLs: [ObjectIdentifier: URL] = [:]
init(base: URL) {
self.base = base
}
/// Maps a media URL (absolute https or /media/... path) to the pinned-media
/// scheme so AVPlayer routes it through this delegate instead of loading
/// https natively. Built as a string because URLComponents.url returns nil
/// for schemes not in Foundation's known-scheme list.
func rewrite(_ url: URL) -> URL {
let absolute: String
if let scheme = url.scheme?.lowercased(), scheme.hasPrefix("http") {
var c = URLComponents(url: url, resolvingAgainstBaseURL: false)!
c.scheme = Self.scheme
return c.url!
} else {
let path = url.absoluteString.hasPrefix("/") ? String(url.absoluteString.dropFirst()) : url.absoluteString
guard let host = base.host, !host.isEmpty else { return url }
let port = base.port.map { ":\($0)" } ?? ""
absolute = "\(Self.scheme)://\(host)\(port)/\(path)"
}
return URL(string: absolute) ?? url
}
/// Maps a pinned-media URL back to the real https URL it proxies.
func resolve(_ url: URL) -> URL {
guard let scheme = url.scheme?.lowercased(), scheme == Self.scheme else { return url }
var c = URLComponents(url: url, resolvingAgainstBaseURL: false)!
c.scheme = "https"
return c.url ?? url
}
func assetWithPinnedURL(_ pinned: URL) -> AVURLAsset {
let asset = AVURLAsset(url: pinned)
loaderURLs[ObjectIdentifier(asset.resourceLoader)] = resolve(pinned)
asset.resourceLoader.setDelegate(self, queue: .main)
return asset
}
func resourceLoader(
_ resourceLoader: AVAssetResourceLoader,
shouldWaitForLoadingOfRequestedResource loadingRequest: AVAssetResourceLoadingRequest
) -> Bool {
guard let dataRequest = loadingRequest.dataRequest else { return false }
guard let url = loaderURLs[ObjectIdentifier(resourceLoader)] else {
logger.log("no resolved URL for resource loader")
loadingRequest.finishLoading(with: URLError(.badURL))
return false
}
var upstream = URLRequest(url: url)
let offset = dataRequest.requestedOffset
let length = Int64(dataRequest.requestedLength)
if length > 0 {
upstream.setValue("bytes=\(offset)-\(offset + length - 1)", forHTTPHeaderField: "Range")
} else {
upstream.setValue("bytes=\(offset)-", forHTTPHeaderField: "Range")
}
logger.log("fetching \(url.absoluteString, privacy: .public) range=\(upstream.value(forHTTPHeaderField: "Range") ?? "-", privacy: .public)")
let task = TLSPinning.session().dataTask(with: upstream) { data, response, error in
if let error {
logger.log("fetch failed \(url.absoluteString, privacy: .public): \(error.localizedDescription, privacy: .public)")
loadingRequest.finishLoading(with: error)
return
}
guard let http = response as? HTTPURLResponse else {
loadingRequest.finishLoading(with: URLError(.badServerResponse))
return
}
if let info = loadingRequest.contentInformationRequest {
info.contentType = http.mimeType ?? "application/octet-stream"
if http.expectedContentLength >= 0 {
info.contentLength = Int64(http.expectedContentLength)
}
info.isByteRangeAccessSupported = true
}
if let data {
dataRequest.respond(with: data)
}
loadingRequest.finishLoading()
}
task.resume()
return true
}
func resourceLoader(
_ resourceLoader: AVAssetResourceLoader,
didCancel loadingRequest: AVAssetResourceLoadingRequest
) {
logger.log("resource loader cancelled request")
}
}