add _RequestAdapter as a workaround for the invoke with streamed response method

Author: grdsdev

Sources/Functions/FunctionsClient.swift

@@ -1,7 +1,7 @@
 import ConcurrencyExtras
 import Foundation
-import Helpers
 import HTTPTypes
+import Helpers
 
 #if canImport(FoundationNetworking)
   import FoundationNetworking
@@ -16,6 +16,8 @@ public final class FunctionsClient: Sendable {
     Data, URLResponse
   )
 
+  public typealias _RequestAdapter = @Sendable (URLRequest) async throws -> URLRequest
+
   /// The base URL for the functions.
   let url: URL
 
@@ -29,6 +31,7 @@ public final class FunctionsClient: Sendable {
 
   private let http: any HTTPClientType
   private let mutableState = LockIsolated(MutableState())
+  private let _requestAdapter: _RequestAdapter?
 
   var headers: HTTPFields {
     mutableState.headers
@@ -48,27 +51,32 @@ public final class FunctionsClient: Sendable {
     headers: [String: String] = [:],
     region: String? = nil,
     logger: (any SupabaseLogger)? = nil,
-    fetch: @escaping FetchHandler = { try await URLSession.shared.data(for: $0) }
+    fetch: @escaping FetchHandler = { try await URLSession.shared.data(for: $0) },
+    _requestAdapter: _RequestAdapter? = nil
   ) {
     var interceptors: [any HTTPClientInterceptor] = []
+
     if let logger {
       interceptors.append(LoggerInterceptor(logger: logger))
     }
 
     let http = HTTPClient(fetch: fetch, interceptors: interceptors)
 
-    self.init(url: url, headers: headers, region: region, http: http)
+    self.init(
+      url: url, headers: headers, region: region, http: http, _requestAdapter: _requestAdapter)
   }
 
   init(
     url: URL,
     headers: [String: String],
     region: String?,
-    http: any HTTPClientType
+    http: any HTTPClientType,
+    _requestAdapter: _RequestAdapter?
   ) {
     self.url = url
     self.region = region
     self.http = http
+    self._requestAdapter = _requestAdapter
 
     mutableState.withValue {
       $0.headers = HTTPFields(headers)
@@ -91,9 +99,17 @@ public final class FunctionsClient: Sendable {
     headers: [String: String] = [:],
     region: FunctionRegion? = nil,
     logger: (any SupabaseLogger)? = nil,
-    fetch: @escaping FetchHandler = { try await URLSession.shared.data(for: $0) }
+    fetch: @escaping FetchHandler = { try await URLSession.shared.data(for: $0) },
+    _requestAdapter: _RequestAdapter? = nil
   ) {
-    self.init(url: url, headers: headers, region: region?.rawValue, logger: logger, fetch: fetch)
+    self.init(
+      url: url,
+      headers: headers,
+      region: region?.rawValue,
+      logger: logger,
+      fetch: fetch,
+      _requestAdapter: _requestAdapter
+    )
   }
 
   /// Updates the authorization header.
@@ -164,7 +180,7 @@ public final class FunctionsClient: Sendable {
     let request = buildRequest(functionName: functionName, options: invokeOptions)
     let response = try await http.send(request)
 
-    guard 200 ..< 300 ~= response.statusCode else {
+    guard 200..<300 ~= response.statusCode else {
       throw FunctionsError.httpError(code: response.statusCode, data: response.data)
     }
 
@@ -198,20 +214,37 @@ public final class FunctionsClient: Sendable {
 
     let urlRequest = buildRequest(functionName: functionName, options: invokeOptions).urlRequest
 
-    let task = session.dataTask(with: urlRequest)
-    task.resume()
+    Task {
+      let adaptedRequest: URLRequest
 
-    continuation.onTermination = { _ in
-      task.cancel()
+      if let _requestAdapter {
+        do {
+          adaptedRequest = try await _requestAdapter(urlRequest)
+        } catch {
+          continuation.finish(throwing: error)
+          return
+        }
+      } else {
+        adaptedRequest = urlRequest
+      }
+
+      let task = session.dataTask(with: adaptedRequest)
+      task.resume()
+
+      continuation.onTermination = { _ in
+        task.cancel()
 
-      // Hold a strong reference to delegate until continuation terminates.
-      _ = delegate
+        // Hold a strong reference to delegate until continuation terminates.
+        _ = delegate
+      }
     }
 
     return stream
   }
 
-  private func buildRequest(functionName: String, options: FunctionInvokeOptions) -> Helpers.HTTPRequest {
+  private func buildRequest(functionName: String, options: FunctionInvokeOptions)
+    -> Helpers.HTTPRequest
+  {
     var request = HTTPRequest(
       url: url.appendingPathComponent(functionName),
       method: options.httpMethod ?? .post,
@@ -243,13 +276,16 @@ final class StreamResponseDelegate: NSObject, URLSessionDataDelegate, Sendable {
     continuation.finish(throwing: error)
   }
 
-  func urlSession(_: URLSession, dataTask _: URLSessionDataTask, didReceive response: URLResponse, completionHandler: @escaping (URLSession.ResponseDisposition) -> Void) {
+  func urlSession(
+    _: URLSession, dataTask _: URLSessionDataTask, didReceive response: URLResponse,
+    completionHandler: @escaping (URLSession.ResponseDisposition) -> Void
+  ) {
     guard let httpResponse = response as? HTTPURLResponse else {
       continuation.finish(throwing: URLError(.badServerResponse))
       return
     }
 
-    guard 200 ..< 300 ~= httpResponse.statusCode else {
+    guard 200..<300 ~= httpResponse.statusCode else {
       let error = FunctionsError.httpError(code: httpResponse.statusCode, data: Data())
       continuation.finish(throwing: error)
       return

Sources/Supabase/SupabaseClient.swift

@@ -99,7 +99,8 @@ public final class SupabaseClient: Sendable {
           headers: headers,
           region: options.functions.region,
           logger: options.global.logger,
-          fetch: fetchWithAuth
+          fetch: fetchWithAuth,
+          _requestAdapter: adapt
         )
       }
 
@@ -347,6 +348,7 @@ public final class SupabaseClient: Sendable {
     try await session.upload(for: adapt(request: request), from: data)
   }
 
+  @Sendable
   private func adapt(request: URLRequest) async -> URLRequest {
     let token = try? await _getAccessToken()