128 lines
10 KiB
Swift
128 lines
10 KiB
Swift
import Foundation
|
|
#if canImport(FoundationNetworking)
|
|
import FoundationNetworking
|
|
#endif
|
|
|
|
private final class InboxRedirectPolicy: NSObject, URLSessionTaskDelegate, @unchecked Sendable {
|
|
static let shared=InboxRedirectPolicy()
|
|
func urlSession(_ session: URLSession, task: URLSessionTask, willPerformHTTPRedirection response: HTTPURLResponse, newRequest request: URLRequest, completionHandler: @escaping (URLRequest?) -> Void) {completionHandler(nil)}
|
|
}
|
|
|
|
/// Recipient-only client. Fetch recipient tokens from your own backend; never embed business API keys.
|
|
public final class InboxClient: @unchecked Sendable {
|
|
public let endpoint: URL
|
|
private let token: @Sendable () async throws -> String
|
|
private let session: URLSession
|
|
private let timeout: TimeInterval
|
|
private let lock=NSLock()
|
|
private var closed=false
|
|
private var cancellers=[UUID: @Sendable () -> Void]()
|
|
public init(endpoint: URL = URL(string:"https://notifications.wrnexus.com")!, timeout: TimeInterval = 15, session: URLSession = .shared, token: @escaping @Sendable () async throws -> String) throws {
|
|
guard Self.validEndpoint(endpoint), timeout > 0 else { throw InboxError(code:"invalid-endpoint",message:"Use an HTTPS origin or localhost for development, with a positive timeout.") }
|
|
self.endpoint=endpoint; self.timeout=timeout; self.session=session; self.token=token
|
|
}
|
|
private static func validEndpoint(_ u: URL) -> Bool {
|
|
guard u.host != nil, u.user == nil, u.password == nil, u.query == nil, u.fragment == nil, ["","/"].contains(u.path) else { return false }
|
|
return u.scheme == "https" || (u.scheme == "http" && ["localhost","127.0.0.1","[::1]","::1"].contains(u.host!))
|
|
}
|
|
public func close() { lock.lock(); closed=true; let callbacks=Array(cancellers.values); cancellers.removeAll(); lock.unlock(); for cancel in callbacks {cancel()} }
|
|
private func track(_ id: UUID, _ cancel: @escaping @Sendable () -> Void) {lock.lock(); let stop=closed; if !stop {cancellers[id]=cancel}; lock.unlock(); if stop {cancel()} }
|
|
private func untrack(_ id: UUID) {lock.lock();cancellers.removeValue(forKey:id);lock.unlock()}
|
|
private func isClosed() -> Bool { lock.lock(); defer {lock.unlock()}; return closed }
|
|
private func request(_ path: String, method: String = "GET", body: Data? = nil, stream: Bool = false) async throws -> URLRequest {
|
|
guard !isClosed() else { throw InboxError(code:"closed",message:"The recipient session is closed.") }
|
|
let value=try await token()
|
|
guard !value.isEmpty, !value.contains("\r"), !value.contains("\n") else { throw InboxError(code:"recipient-token-required",message:"Pass a recipient token.") }
|
|
guard !isClosed() else { throw CancellationError() }
|
|
var r=URLRequest(url:URL(string:"/api/v1/inbox/"+path,relativeTo:endpoint)!,timeoutInterval:stream ? 75 : timeout)
|
|
r.httpMethod=method; r.httpBody=body; r.setValue("Bearer "+value,forHTTPHeaderField:"Authorization")
|
|
r.setValue(stream ? "text/event-stream" : "application/json",forHTTPHeaderField:"Accept")
|
|
if body != nil { r.setValue("application/json",forHTTPHeaderField:"Content-Type") }; return r
|
|
}
|
|
static func decoder() -> JSONDecoder { let d=JSONDecoder(); d.keyDecodingStrategy = .convertFromSnakeCase; return d }
|
|
private func check(_ response: URLResponse, data: Data) throws {
|
|
guard let r=response as? HTTPURLResponse else { throw InboxError(code:"network-error",message:"Invalid HTTP response.") }
|
|
guard (200..<300).contains(r.statusCode) else {
|
|
let json=(try? JSONSerialization.jsonObject(with:data)) as? [String:Any]; let error=json?["error"] as? [String:Any]
|
|
throw InboxError(code:error?["code"] as? String ?? "request-failed",status:r.statusCode,message:error?["message"] as? String ?? "Request failed with HTTP \(r.statusCode).")
|
|
}
|
|
}
|
|
private func call<T:Decodable>(_ path: String, method: String = "GET", body: Data? = nil) async throws -> T {
|
|
let r=try await request(path,method:method,body:body)
|
|
let id=UUID(); let task=Task { try await session.data(for:r,delegate:InboxRedirectPolicy.shared) };track(id,{task.cancel()});defer {untrack(id)}
|
|
do {
|
|
let (data,response)=try await withTaskCancellationHandler(operation:{try await task.value},onCancel:{task.cancel()}); try Task.checkCancellation()
|
|
guard !isClosed() else { throw CancellationError() }; try check(response,data:data)
|
|
do { return try Self.decoder().decode(T.self,from:data) } catch {throw InboxError(code:"invalid-response",message:"Invalid inbox response.")}
|
|
} catch let error as URLError {
|
|
if error.code == .cancelled || isClosed() || Task.isCancelled {throw CancellationError()}
|
|
throw InboxError(code:error.code == .timedOut ? "timeout" : "network-error",message:error.code == .timedOut ? "The request timed out." : "Unable to connect to the inbox.")
|
|
}
|
|
}
|
|
private func json(_ value: [String:Any]) throws -> Data { try JSONSerialization.data(withJSONObject:value) }
|
|
public func messages(limit: Int? = nil) async throws -> InboxSnapshot {
|
|
if let limit, !(1...100).contains(limit) { throw InboxError(code:"invalid-limit",message:"Limit must be between 1 and 100.") }
|
|
return try await call("messages"+(limit.map {"?limit=\($0)"} ?? ""))
|
|
}
|
|
public func markRead(_ id: String) async throws -> MessageResult { try await call("messages",method:"POST",body:json(["id":id,"action":"read"])) }
|
|
public func archive(_ id: String) async throws -> MessageResult { try await call("messages",method:"POST",body:json(["id":id,"action":"archive"])) }
|
|
public func preferences() async throws -> InboxPreferences { try await call("preferences") }
|
|
public func savePreferences(_ input: PreferenceInput) async throws { let _:EmptyResult=try await call("preferences",method:"POST",body:JSONEncoder().encode(input)) }
|
|
public func devices() async throws -> InboxDevices { try await call("devices") }
|
|
/// Re-register the same installation ID when FCM/APNs rotates its token.
|
|
public func registerMobilePush(_ input: MobileRegistration) async throws -> DeviceResult {
|
|
var value=try JSONSerialization.jsonObject(with:JSONEncoder().encode(input)) as! [String:Any]; value["action"]="register-mobile"
|
|
return try await call("devices",method:"POST",body:json(value))
|
|
}
|
|
public func unregisterMobilePush(_ id: String) async throws -> DeviceResult { try await call("devices",method:"POST",body:json(["action":"unregister-mobile","deviceId":id])) }
|
|
public func unregisterWebPush(_ id: String) async throws -> DeviceResult { try await call("devices",method:"POST",body:json(["action":"unregister","deviceId":id])) }
|
|
/// Cancelling the consuming task stops streaming. Close on logout and create a new client for a new recipient.
|
|
public func updates() -> AsyncThrowingStream<InboxSnapshot,Error> {
|
|
AsyncThrowingStream { continuation in
|
|
let id=UUID()
|
|
let task=Task {
|
|
defer {self.untrack(id)}
|
|
do {
|
|
while !Task.isCancelled && !self.isClosed() {
|
|
do {
|
|
let r=try await self.request("stream",stream:true)
|
|
#if canImport(Darwin)
|
|
let (bytes,response)=try await self.session.bytes(for:r,delegate:InboxRedirectPolicy.shared); try self.check(response,data:Data())
|
|
var parser=InboxEventParser()
|
|
for try await line in bytes.lines {
|
|
try Task.checkCancellation(); if self.isClosed() { break }
|
|
if let snapshot=try parser.line(line) { continuation.yield(snapshot) }
|
|
}
|
|
#else
|
|
let (data,response)=try await self.session.data(for:r,delegate:InboxRedirectPolicy.shared); try self.check(response,data:data)
|
|
var parser=InboxEventParser()
|
|
for line in String(decoding:data,as:UTF8.self).components(separatedBy:"\n") { if let snapshot=try parser.line(line) { continuation.yield(snapshot) } }
|
|
#endif
|
|
} catch let error as InboxError { throw error }
|
|
catch is CancellationError { throw CancellationError() }
|
|
catch let error as DecodingError { throw error }
|
|
catch { /* Transient connection failures reconnect; writes are never retried. */ }
|
|
try await Task.sleep(nanoseconds:2_000_000_000)
|
|
}
|
|
continuation.finish()
|
|
} catch { if Task.isCancelled || self.isClosed() { continuation.finish() } else { continuation.finish(throwing:error) } }
|
|
}
|
|
self.track(id,{task.cancel()})
|
|
continuation.onTermination={ @Sendable _ in task.cancel() }
|
|
}
|
|
}
|
|
public func logoURL(_ branding: InboxBranding) -> URL? { guard let path=branding.logoUrl, path.range(of:"^/api/v1/inbox/branding-logo/[a-f0-9]{64}$",options:.regularExpression) != nil else {return nil}; return URL(string:path,relativeTo:endpoint)?.absoluteURL }
|
|
public static func safeActionURL(_ text: String?) -> URL? { guard let text, let u=URL(string:text),u.scheme=="https",u.host != nil,u.user==nil,u.password==nil else {return nil}; return u }
|
|
}
|
|
|
|
struct InboxEventParser {
|
|
private var event=""; private var data=[String](); private var size=0
|
|
mutating func line(_ raw: String) throws -> InboxSnapshot? {
|
|
let line=raw.hasSuffix("\r") ? String(raw.dropLast()) : raw
|
|
if line.isEmpty { defer {event="";data=[];size=0}; guard event=="inbox", !data.isEmpty else {return nil}; return try InboxClient.decoder().decode(InboxSnapshot.self,from:Data(data.joined(separator:"\n").utf8)) }
|
|
size += line.utf8.count; guard size < 2_097_152 else { throw InboxError(code:"stream-too-large",message:"Inbox frame exceeded the size limit.") }
|
|
if line.hasPrefix("event:") {event=String(line.dropFirst(6)).trimmingCharacters(in:.whitespaces)}
|
|
if line.hasPrefix("data:") { var value=String(line.dropFirst(5)); if value.hasPrefix(" ") {value.removeFirst()}; data.append(value) }; return nil
|
|
}
|
|
}
|