Implements signals

kordophone2/XPC/XPCClient.swift

@@ -13,15 +13,47 @@ private let serviceName = "net.buzzert.kordophonecd"
 final class XPCClient
 {
     private let connection: xpc_connection_t
+    private let signalLock = NSLock()
+    private var signalSinks: [UUID: (Signal) -> Void] = [:]
+    private var didSubscribeSignals: Bool = false
 
     init() {
         self.connection = xpc_connection_create_mach_service(serviceName, nil, 0)
 
-        let handler: xpc_handler_t = { _ in }
+        let handler: xpc_handler_t = { [weak self] event in
+            self?.handleIncomingXPCEvent(event)
+        }
+        
         xpc_connection_set_event_handler(connection, handler)
         xpc_connection_resume(connection)
     }
     
+    public func eventStream() -> AsyncStream<Signal> {
+        // Auto-subscribe on first stream creation
+        if !didSubscribeSignals {
+            didSubscribeSignals = true
+            Task {
+                try? await subscribeToSignals()
+            }
+        }
+
+        return AsyncStream { continuation in
+            let id = UUID()
+            signalLock.withLock {
+                signalSinks[id] = { signal in
+                    continuation.yield(signal)
+                }
+            }
+
+            continuation.onTermination = { [weak self] _ in
+                guard let self else { return }
+                _ = self.signalLock.withLock {
+                    self.signalSinks.removeValue(forKey: id)
+                }
+            }
+        }
+    }
+    
     public func getVersion() async throws -> String {
         let req = makeRequest(method: "GetVersion")
         guard let reply = try await sendSync(req), xpc_get_type(reply) == XPC_TYPE_DICTIONARY else { throw Error.typeError }
@@ -46,7 +78,8 @@ final class XPCClient
             if xpc_get_type(element) == XPC_TYPE_DICTIONARY, let conv = Serialized.Conversation(xpc: element) {
                 results.append(conv)
             }
-        return true
+            
+            return true
         }
         return results
     }
@@ -66,6 +99,7 @@ final class XPCClient
             if xpc_get_type(element) == XPC_TYPE_DICTIONARY, let msg = Serialized.Message(xpc: element) {
                 results.append(msg)
             }
+            
             return true
         }
 
@@ -88,10 +122,24 @@ final class XPCClient
         case typeError
         case encodingError
     }
+    
+    enum Signal
+    {
+        case conversationsUpdated
+        case messagesUpdated(conversationId: String)
+        case attachmentDownloaded(attachmentId: String)
+        case attachmentUploaded(uploadGuid: String, attachmentGuid: String)
+        case updateStreamReconnected
+    }
 }
 
 extension XPCClient
 {
+    private func subscribeToSignals() async throws {
+        let req = makeRequest(method: "SubscribeSignals")
+        _ = try await sendSync(req)
+    }
+
     private func makeRequest(method: String, arguments: [String: xpc_object_t]? = nil) -> xpc_object_t {
         let dict = xpc_dictionary_create(nil, nil, 0)
         xpc_dictionary_set_string(dict, "method", method)
@@ -132,6 +180,52 @@ extension XPCClient
             }
         }
     }
+
+    private func handleIncomingXPCEvent(_ event: xpc_object_t) {
+        let type = xpc_get_type(event)
+        switch type {
+            case XPC_TYPE_DICTIONARY:
+                guard let namePtr = xpc_dictionary_get_string(event, "name") else { return }
+                let name = String(cString: namePtr)
+
+                var signal: Signal?
+                if name == "ConversationsUpdated" {
+                    signal = .conversationsUpdated
+                } else if name == "MessagesUpdated" {
+                    if let args = xpc_dictionary_get_value(event, "arguments"), xpc_get_type(args) == XPC_TYPE_DICTIONARY,
+                       let cidPtr = xpc_dictionary_get_string(args, "conversation_id") {
+                        signal = .messagesUpdated(conversationId: String(cString: cidPtr))
+                    }
+                } else if name == "AttachmentDownloadCompleted" {
+                    if let args = xpc_dictionary_get_value(event, "arguments"), xpc_get_type(args) == XPC_TYPE_DICTIONARY,
+                       let aidPtr = xpc_dictionary_get_string(args, "attachment_id") {
+                        signal = .attachmentDownloaded(attachmentId: String(cString: aidPtr))
+                    }
+                } else if name == "AttachmentUploadCompleted" {
+                    if let args = xpc_dictionary_get_value(event, "arguments"), xpc_get_type(args) == XPC_TYPE_DICTIONARY,
+                       let ugPtr = xpc_dictionary_get_string(args, "upload_guid"),
+                       let agPtr = xpc_dictionary_get_string(args, "attachment_guid") {
+                        signal = .attachmentUploaded(uploadGuid: String(cString: ugPtr), attachmentGuid: String(cString: agPtr))
+                    }
+                } else if name == "UpdateStreamReconnected" {
+                    signal = .updateStreamReconnected
+                }
+
+                if let signal {
+                    signalLock.lock()
+                    let sinks = signalSinks.values
+                    signalLock.unlock()
+                    for sink in sinks { sink(signal) }
+                }
+
+            case XPC_TYPE_ERROR:
+                if let errStr = xpc_string_get_string_ptr(event) {
+                    print("XPC event error: \(String(cString: errStr))")
+                }
+            default:
+                break
+        }
+    }
 }