Integrate live phone and watch remote prototype with guarded motion control

2026-09-18 19:11:52 +04:00
parent 3e787543c0
commit 8131f4f873
54 changed files with 4355 additions and 329 deletions
@@ -0,0 +1,387 @@
+import Foundation
+import RemoteCore
+import Testing
+@testable import TVTransport
+
+private let baseDate = Date(timeIntervalSince1970: 10_000)
+
+private func envelope(
+    sessionID: UUID,
+    televisionID: UUID,
+    sequence: UInt64,
+    command: RemoteCommand,
+    id: UUID = UUID(),
+    issuedAt: Date = baseDate,
+    expiresAt: Date = baseDate.addingTimeInterval(0.8)
+) -> CommandEnvelope {
+    CommandEnvelope(
+        sessionID: sessionID,
+        televisionID: televisionID,
+        sequence: sequence,
+        issuedAt: issuedAt,
+        expiresAt: expiresAt,
+        command: command,
+        id: id
+    )
+}
+
+@Test func explicitPairThenTokenReconnectAuthorizesBeforeReady() async throws {
+    let tv = TelevisionTarget(id: UUID(), host: "tv.synthetic.test")
+    let pairSocket = FakeSocket(events: [Data(#"{"event":"ms.channel.connect","data":{"token":"redacted-token"}}"#.utf8)])
+    let reconnectSocket = FakeSocket(events: [Data(#"{"event":"ms.channel.connect","data":{}}"#.utf8)])
+    let factory = FakeSocketFactory(sockets: [pairSocket, reconnectSocket])
+    let store = FakeAuthorizationStore()
+    let transport = makeTransport(factory: factory, store: store)
+
+    let firstSession = UUID()
+    #expect(await transport.pair(target: tv, sessionID: firstSession) == .ready(televisionID: tv.id))
+    #expect(await store.savedToken(for: tv.id) == "redacted-token")
+    let firstEndpoint = try #require(await factory.endpoint(at: 0))
+    #expect(URLComponents(url: firstEndpoint, resolvingAgainstBaseURL: false)?.queryItems?.contains(where: { $0.name == "token" }) == false)
+
+    await transport.disconnect()
+    let secondSession = UUID()
+    #expect(await transport.connect(target: tv, sessionID: secondSession) == .ready(televisionID: tv.id))
+    let secondEndpoint = try #require(await factory.endpoint(at: 1))
+    #expect(URLComponents(url: secondEndpoint, resolvingAgainstBaseURL: false)?.queryItems?.first(where: { $0.name == "token" })?.value == "redacted-token")
+}
+
+@Test func pairingDenialTimeoutAndMissingInitialTokenFailClosed() async {
+    let tv = TelevisionTarget(id: UUID(), host: "tv.synthetic.test")
+    let deniedSocket = FakeSocket(events: [Data(#"{"event":"ms.channel.unauthorized"}"#.utf8)])
+    let missingTokenSocket = FakeSocket(events: [Data(#"{"event":"ms.channel.connect","data":{}}"#.utf8)])
+    let timeoutSocket = FakeSocket(events: [])
+    let factory = FakeSocketFactory(sockets: [deniedSocket, missingTokenSocket, timeoutSocket])
+    let store = FakeAuthorizationStore()
+    let transport = makeTransport(factory: factory, store: store, pairingTimeout: 0.01)
+
+    #expect(await transport.pair(target: tv, sessionID: UUID()) == .denied(televisionID: tv.id))
+    #expect(await transport.pair(target: tv, sessionID: UUID()) == .unauthorized(televisionID: tv.id))
+    #expect(await transport.pair(target: tv, sessionID: UUID()) == .timedOut(televisionID: tv.id))
+}
+
+@Test func sendsInOrderConsumesAdmissionAndExpiresQueuedWork() async throws {
+    let tv = TelevisionTarget(id: UUID(), host: "tv.synthetic.test")
+    let session = UUID()
+    let clock = TestClock(wall: baseDate, monotonic: 50)
+    let socket = FakeSocket(
+        events: [Data(#"{"event":"ms.channel.connect","data":{"token":"redacted-token"}}"#.utf8)],
+        blockedSendCount: 1
+    )
+    let factory = FakeSocketFactory(sockets: [socket])
+    let transport = makeTransport(factory: factory, store: FakeAuthorizationStore(), clock: clock.value)
+    #expect(await transport.pair(target: tv, sessionID: session) == .ready(televisionID: tv.id))
+
+    let first = envelope(sessionID: session, televisionID: tv.id, sequence: 1, command: .right)
+    let second = envelope(sessionID: session, televisionID: tv.id, sequence: 2, command: .left)
+    let firstTask = Task { await transport.send(first) }
+    await socket.waitForSendCount(1)
+    let secondTask = Task { await transport.send(second) }
+    try? await Task.sleep(for: .milliseconds(10))
+    clock.advanceMonotonic(seconds: 1)
+    await socket.releaseBlockedSend()
+
+    #expect(await firstTask.value.status == .sentToTransport)
+    #expect(await secondTask.value.status == .expired)
+    #expect(await socket.sentKeyNames() == ["KEY_RIGHT"])
+    #expect(await transport.send(first).status == .duplicate)
+}
+
+@Test func failedWriteIsConsumedAndNeverRetried() async {
+    let tv = TelevisionTarget(id: UUID(), host: "tv.synthetic.test")
+    let session = UUID()
+    let socket = FakeSocket(
+        events: [Data(#"{"event":"ms.channel.connect","data":{"token":"redacted-token"}}"#.utf8)],
+        failingSendCount: 1
+    )
+    let transport = makeTransport(
+        factory: FakeSocketFactory(sockets: [socket]),
+        store: FakeAuthorizationStore()
+    )
+    #expect(await transport.pair(target: tv, sessionID: session) == .ready(televisionID: tv.id))
+    let command = envelope(sessionID: session, televisionID: tv.id, sequence: 1, command: .up)
+    #expect(await transport.send(command).status == .unavailable)
+    #expect(await transport.send(command).status == .unavailable)
+    #expect(await socket.sentKeyNames() == ["KEY_UP"])
+}
+
+@Test func targetChangeCancelsInflightAndQueuedOldGenerationCommands() async {
+    let firstTV = TelevisionTarget(id: UUID(), host: "first.synthetic.test")
+    let secondTV = TelevisionTarget(id: UUID(), host: "second.synthetic.test")
+    let firstSession = UUID()
+    let firstSocket = FakeSocket(
+        events: [Data(#"{"event":"ms.channel.connect","data":{"token":"redacted-one"}}"#.utf8)],
+        blockedSendCount: 1
+    )
+    let secondSocket = FakeSocket(events: [Data(#"{"event":"ms.channel.connect","data":{"token":"redacted-two"}}"#.utf8)])
+    let factory = FakeSocketFactory(sockets: [firstSocket, secondSocket])
+    let transport = makeTransport(factory: factory, store: FakeAuthorizationStore())
+    #expect(await transport.pair(target: firstTV, sessionID: firstSession) == .ready(televisionID: firstTV.id))
+
+    let first = envelope(sessionID: firstSession, televisionID: firstTV.id, sequence: 1, command: .right)
+    let second = envelope(sessionID: firstSession, televisionID: firstTV.id, sequence: 2, command: .left)
+    let firstTask = Task { await transport.send(first) }
+    await firstSocket.waitForSendCount(1)
+    let secondTask = Task { await transport.send(second) }
+    await Task.yield()
+
+    #expect(await transport.pair(target: secondTV, sessionID: UUID()) == .ready(televisionID: secondTV.id))
+    #expect(await firstTask.value.status == .unavailable)
+    #expect(await secondTask.value.status == .unavailable)
+    #expect(await firstSocket.sentKeyNames() == ["KEY_RIGHT"])
+}
+
+@Test func reconnectRequiresTokenAndRevocationDeletesIt() async {
+    let tv = TelevisionTarget(id: UUID(), host: "tv.synthetic.test")
+    let store = FakeAuthorizationStore()
+    let transport = makeTransport(factory: FakeSocketFactory(sockets: []), store: store)
+    #expect(await transport.connect(target: tv, sessionID: UUID()) == .unauthorized(televisionID: tv.id))
+
+    await store.setAuthorization(.init(token: "redacted-token"), for: tv.id)
+    await transport.revokeAuthorization()
+    #expect(await store.savedToken(for: tv.id) == nil)
+}
+
+@Test func forgetDeletesAuthorizationAndSelectedTarget() async {
+    let tv = TelevisionTarget(id: UUID(), host: "tv.synthetic.test")
+    let store = FakeAuthorizationStore()
+    await store.setAuthorization(.init(token: "redacted-token"), for: tv.id)
+    let socket = FakeSocket(events: [Data(#"{"event":"ms.channel.connect","data":{}}"#.utf8)])
+    let transport = makeTransport(factory: FakeSocketFactory(sockets: [socket]), store: store)
+    #expect(await transport.connect(target: tv, sessionID: UUID()) == .ready(televisionID: tv.id))
+    await transport.forgetTelevision()
+    #expect(await transport.connectionState() == .disconnected)
+    #expect(await store.savedToken(for: tv.id) == nil)
+}
+
+private func makeTransport(
+    factory: FakeSocketFactory,
+    store: FakeAuthorizationStore,
+    clock: TVTransportClock = TestClock(wall: baseDate, monotonic: 50).value,
+    pairingTimeout: TimeInterval = 1
+) -> TVConnectionActor {
+    TVConnectionActor(
+        socketFactory: factory,
+        authorizationStore: store,
+        clock: clock,
+        configuration: .init(pairingTimeout: pairingTimeout)
+    )
+}
+
+private final class TestClock: @unchecked Sendable {
+    private let lock = NSLock()
+    private var wall: Date
+    private var monotonic: TimeInterval
+
+    init(wall: Date, monotonic: TimeInterval) {
+        self.wall = wall
+        self.monotonic = monotonic
+    }
+
+    var value: TVTransportClock {
+        TVTransportClock(
+            wallNow: { [self] in lock.withLock { wall } },
+            monotonicNow: { [self] in lock.withLock { monotonic } }
+        )
+    }
+
+    func advance(seconds: TimeInterval) {
+        lock.withLock {
+            wall = wall.addingTimeInterval(seconds)
+            monotonic += seconds
+        }
+    }
+
+    func advanceMonotonic(seconds: TimeInterval) {
+        lock.withLock { monotonic += seconds }
+    }
+}
+
+private actor FakeAuthorizationStore: TVAuthorizationStore {
+    private var values: [UUID: StoredTVAuthorization] = [:]
+
+    func authorization(for televisionID: UUID) async throws -> StoredTVAuthorization {
+        values[televisionID] ?? StoredTVAuthorization(token: nil)
+    }
+
+    func saveToken(_ token: String, for televisionID: UUID) async throws {
+        let trust = values[televisionID]?.trust ?? .systemDefault
+        values[televisionID] = StoredTVAuthorization(token: token, trust: trust)
+    }
+
+    func deleteToken(for televisionID: UUID) async throws {
+        let trust = values[televisionID]?.trust ?? .systemDefault
+        values[televisionID] = StoredTVAuthorization(token: nil, trust: trust)
+    }
+
+    func deleteTrust(for televisionID: UUID) async throws {
+        let token = values[televisionID]?.token
+        values[televisionID] = StoredTVAuthorization(token: token, trust: .systemDefault)
+    }
+
+    func savedToken(for televisionID: UUID) -> String? { values[televisionID]?.token }
+
+    func setAuthorization(_ authorization: StoredTVAuthorization, for televisionID: UUID) {
+        values[televisionID] = authorization
+    }
+}
+
+private actor FakeSocketFactory: TVWebSocketFactory {
+    private var sockets: [FakeSocket]
+    private var endpoints: [URL] = []
+
+    init(sockets: [FakeSocket]) { self.sockets = sockets }
+
+    func open(endpoint: URL, trust: TVTrustPolicy, maximumMessageBytes: Int) async throws -> any TVWebSocket {
+        endpoints.append(endpoint)
+        guard !sockets.isEmpty else { throw TVSocketError.unavailable }
+        return sockets.removeFirst()
+    }
+
+    func endpoint(at index: Int) -> URL? {
+        endpoints.indices.contains(index) ? endpoints[index] : nil
+    }
+}
+
+private actor FakeSocket: TVWebSocket {
+    private var events: [Data]
+    private var receiveContinuation: CheckedContinuation<Data, Error>?
+    private var sendContinuation: CheckedContinuation<Void, Error>?
+    private var sent: [Data] = []
+    private var blockedSendCount: Int
+    private var failingSendCount: Int
+    private var closed = false
+
+    init(events: [Data], blockedSendCount: Int = 0, failingSendCount: Int = 0) {
+        self.events = events
+        self.blockedSendCount = blockedSendCount
+        self.failingSendCount = failingSendCount
+    }
+
+    func receive() async throws -> Data {
+        if closed { throw TVSocketError.closed }
+        if !events.isEmpty { return events.removeFirst() }
+        return try await withCheckedThrowingContinuation { receiveContinuation = $0 }
+    }
+
+    func send(_ data: Data) async throws {
+        if closed { throw TVSocketError.closed }
+        sent.append(data)
+        if failingSendCount > 0 {
+            failingSendCount -= 1
+            throw TVSocketError.unavailable
+        }
+        if blockedSendCount > 0 {
+            blockedSendCount -= 1
+            try await withCheckedThrowingContinuation { sendContinuation = $0 }
+        }
+    }
+
+    func close() async {
+        closed = true
+        receiveContinuation?.resume(throwing: TVSocketError.closed)
+        receiveContinuation = nil
+        sendContinuation?.resume(throwing: TVSocketError.closed)
+        sendContinuation = nil
+    }
+
+    func deliver(_ data: Data) {
+        if let pending = receiveContinuation {
+            receiveContinuation = nil
+            pending.resume(returning: data)
+        } else { events.append(data) }
+    }
+
+    func releaseBlockedSend() {
+        sendContinuation?.resume()
+        sendContinuation = nil
+    }
+
+    func waitForSendCount(_ count: Int) async {
+        while sent.count < count { await Task.yield() }
+    }
+
+    func sentKeyNames() -> [String] {
+        sent.compactMap { data in
+            guard let object = try? JSONSerialization.jsonObject(with: data) as? [String: Any],
+                  let parameters = object["params"] as? [String: String] else { return nil }
+            return parameters["DataOfCmd"]
+        }
+    }
+}
+
+@Test func watchNegotiationKeepsIdleSocketAndPhoneCommandsDoNotStealGrant() async throws {
+    let tv = TelevisionTarget(id: UUID(), host: "tv.synthetic.test")
+    let socket = FakeSocket(events: [Data(#"{"event":"ms.channel.connect","data":{"token":"redacted-token"}}"#.utf8)])
+    let transport = makeTransport(factory: FakeSocketFactory(sockets: [socket]), store: FakeAuthorizationStore())
+    _ = await transport.pair(target: tv, sessionID: UUID())
+    let peer = WatchPeerGeneration()
+    await transport.watchBridgeDidActivate(peer: peer)
+    let grant = try await transport.beginWatchSession(.init(), peer: peer).get()
+    #expect(await transport.sendPhoneCommand(.left).status == .sentToTransport)
+    let command = envelope(sessionID: grant.sessionID, televisionID: tv.id, sequence: 1, command: .right)
+    #expect(await transport.submitWatchCommand(command, peer: peer).status == .sentToTransport)
+    #expect(await transport.submitWatchCommand(command, peer: peer).status == .duplicate)
+    #expect(await socket.sentKeyNames() == ["KEY_LEFT", "KEY_RIGHT"])
+    await transport.disconnect()
+}
+
+@Test func stalePeerAndStaleEndCannotInvalidateCurrentGrant() async throws {
+    let tv = TelevisionTarget(id: UUID(), host: "tv.synthetic.test")
+    let socket = FakeSocket(events: [Data(#"{"event":"ms.channel.connect","data":{"token":"redacted-token"}}"#.utf8)])
+    let transport = makeTransport(factory: FakeSocketFactory(sockets: [socket]), store: FakeAuthorizationStore())
+    _ = await transport.pair(target: tv, sessionID: UUID())
+    let oldPeer = WatchPeerGeneration(), peer = WatchPeerGeneration()
+    await transport.watchBridgeDidActivate(peer: oldPeer)
+    let old = try await transport.beginWatchSession(.init(), peer: oldPeer).get()
+    await transport.watchBridgeDidActivate(peer: peer)
+    let current = try await transport.beginWatchSession(.init(), peer: peer).get()
+    await transport.watchBridgeDidDeactivate(peer: oldPeer)
+    #expect(await transport.endWatchSession(.init(sessionID: old.sessionID), peer: peer) == .failure(.init(code: .sessionInvalidated)))
+    #expect(await transport.submitWatchCommand(envelope(sessionID: old.sessionID, televisionID: tv.id, sequence: 1, command: .left), peer: oldPeer).status == .rejected)
+    #expect(await transport.submitWatchCommand(envelope(sessionID: current.sessionID, televisionID: tv.id, sequence: 1, command: .right), peer: peer).status == .sentToTransport)
+    await transport.disconnect()
+}
+
+@Test func rotatingGrantDuringWriteClosesSocketBeforeNewGrant() async throws {
+    let tv = TelevisionTarget(id: UUID(), host: "tv.synthetic.test")
+    let socket = FakeSocket(events: [Data(#"{"event":"ms.channel.connect","data":{"token":"redacted-token"}}"#.utf8)], blockedSendCount: 1)
+    let transport = makeTransport(factory: FakeSocketFactory(sockets: [socket]), store: FakeAuthorizationStore())
+    _ = await transport.pair(target: tv, sessionID: UUID())
+    let peer = WatchPeerGeneration()
+    await transport.watchBridgeDidActivate(peer: peer)
+    let grant = try await transport.beginWatchSession(.init(), peer: peer).get()
+    let command = envelope(sessionID: grant.sessionID, televisionID: tv.id, sequence: 1, command: .right)
+    let pending = Task { await transport.submitWatchCommand(command, peer: peer) }
+    await socket.waitForSendCount(1)
+    #expect(await transport.beginWatchSession(.init(), peer: peer) == .failure(.init(code: .unavailable)))
+    #expect(await pending.value.status == .unavailable)
+    #expect(await transport.connectionState() == .disconnected)
+}
+
+@Test func readyConnectionConsumesRevocationAndStopsCommands() async throws {
+    let tv = TelevisionTarget(id: UUID(), host: "tv.synthetic.test")
+    let socket = FakeSocket(events: [Data(#"{"event":"ms.channel.connect","data":{"token":"redacted-token"}}"#.utf8)])
+    let store = FakeAuthorizationStore()
+    let transport = makeTransport(factory: FakeSocketFactory(sockets: [socket]), store: store)
+    _ = await transport.pair(target: tv, sessionID: UUID())
+    await socket.deliver(Data(#"{"event":"ms.channel.unauthorized"}"#.utf8))
+    for _ in 0..<1000 {
+        if await transport.connectionState() == .unauthorized(televisionID: tv.id) { break }
+        await Task.yield()
+    }
+    #expect(await transport.connectionState() == .unauthorized(televisionID: tv.id))
+    #expect(await transport.sendPhoneCommand(.left).status == .unavailable)
+}
+
+@Test func sendDeadlineClosesNonCompletingSocket() async {
+    let tv = TelevisionTarget(id: UUID(), host: "tv.synthetic.test")
+    let socket = FakeSocket(events: [Data(#"{"event":"ms.channel.connect","data":{"token":"redacted-token"}}"#.utf8)], blockedSendCount: 1)
+    let transport = makeTransport(factory: FakeSocketFactory(sockets: [socket]), store: FakeAuthorizationStore())
+    let session = UUID()
+    _ = await transport.pair(target: tv, sessionID: session)
+    let command = envelope(sessionID: session, televisionID: tv.id, sequence: 1, command: .up, expiresAt: baseDate.addingTimeInterval(0.02))
+    #expect(await transport.send(command).status == .unavailable)
+    #expect(await transport.connectionState() == .unavailable(televisionID: tv.id, failure: .socket))
+    #expect(await socket.sentKeyNames() == ["KEY_UP"])
+}