28167379888

Committed 25 Jun 2026 11:38AM UTC coverage: 80.743%. First build

Build # 28167379888

Build Type

Pull #987

github

Committed by

web-flow

Commit Message

Merge 693c0012d into c08ceb649

Pull Request Pull Request #987: feat(storage)!: storage v3

Coverage Stats

1784 of 2046 new or added lines in 14 files covered. (87.19%)

8264 of 10235 relevant lines covered (80.74%)

34.21 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

78.03

/Sources/Storage/MultipartUploadEngine.swift

//
//  MultipartUploadEngine.swift
//  Storage
//
//  Created by Guilherme Souza on 04/05/26.
//

import ConcurrencyExtras
import Foundation
import Helpers

#if canImport(FoundationNetworking)
  import FoundationNetworking
#endif

private struct MultipartServerResponse: Decodable {
  let Key: String
  let Id: UUID
}

actor MultipartUploadEngine {
  enum State {
    case idle
    case uploading
    case completed(FileUploadResponse)
    case failed(StorageError)
    case cancelled

    var isTerminal: Bool {
      switch self {
      case .completed, .failed, .cancelled: return true
      default: return false
      }
    }
  }

  private let bucketId: String
  private let path: String
  private let source: UploadSource
  private let options: FileOptions
  private let httpMethod: HTTPMethod
  private let client: StorageClient
  private let eventsContinuation: AsyncStream<TransferEvent<FileUploadResponse>>.Continuation
  private let resultContinuation: AsyncStream<Result<FileUploadResponse, any Error>>.Continuation

  private var state: State = .idle
  private var currentUploadTask: Task<Void, Never>?

  init(
    bucketId: String,
    path: String,
    source: UploadSource,
    options: FileOptions,
    httpMethod: HTTPMethod = .post,
    client: StorageClient,
    eventsContinuation: AsyncStream<TransferEvent<FileUploadResponse>>.Continuation,
    resultContinuation: AsyncStream<Result<FileUploadResponse, any Error>>.Continuation
  ) {
    self.bucketId = bucketId
    self.path = path
    self.source = source
    self.options = options
    self.httpMethod = httpMethod
    self.client = client
    self.eventsContinuation = eventsContinuation
    self.resultContinuation = resultContinuation
  }

  func start() {
    guard case .idle = state else { return }
    state = .uploading
    currentUploadTask = Task { await run() }
  }

  // Multipart uploads do not support pause/resume — the single-shot request cannot be
  // interrupted and resumed mid-flight. These are intentional no-ops; callers that need
  // pause/resume should use the TUS upload path instead.
  func pause() {}
  func resume() {}

  func cancel() {
    guard !state.isTerminal else { return }
    currentUploadTask?.cancel()
    state = .cancelled
    let error = StorageError.cancelled
    eventsContinuation.yield(.failed(error))
    eventsContinuation.finish()
    resultContinuation.yield(.failure(error))
    resultContinuation.finish()
  }

  // MARK: - Private

  private func run() async {
    do {
      try Task.checkCancellation()
      let response = try await performUpload()
      finish(with: .success(response))
    } catch {
      handleError(error)
    }
  }

  private func performUpload() async throws -> FileUploadResponse {
    #if DEBUG
      let builder = MultipartBuilder(
        boundary: testingBoundary.value ?? "----sb-\(UUID().uuidString)"
      )
    #else
      let builder = MultipartBuilder()
    #endif

    let multipart = source.append(to: builder, withPath: path, options: options)

    var headers: [String: String] = [:]
    headers["Content-Type"] = multipart.contentType
    if options.upsert {
      headers["x-upsert"] = "true"
    }

    var url = client.url.appendingPathComponent("object").appendingPathComponent(bucketId)
    for component in path.split(separator: "/") {
      url = url.appendingPathComponent(String(component))
    }

    let request = try await client.http.createRequest(
      httpMethod,
      url: url,
      headers: client.mergedHeaders(headers)
    )

    do {
      let (data, urlResponse) = try await uploadWithProgress(request: request, multipart: multipart)
      let httpResponse = try client.http.validateResponse(urlResponse, data: data)
      client.logResponse(httpResponse, data: data)
      let serverResponse = try client.decoder.decode(MultipartServerResponse.self, from: data)
      return FileUploadResponse(id: serverResponse.Id, path: path, fullPath: serverResponse.Key)
    } catch {
      client.logFailure(error)
      throw client.translateStorageError(error)
    }
  }

  private func uploadWithProgress(
    request: URLRequest,
    multipart: MultipartBuilder
  ) async throws -> (Data, URLResponse) {
    let progressContinuation = eventsContinuation

    let progressDelegate = UploadProgressDelegate { sent, total in
      progressContinuation.yield(
        .progress(TransferProgress(bytesTransferred: sent, totalBytes: total))
      )
    }

    if source.usesTempFileUpload {
      let tempFile = try multipart.buildToTempFile()
      defer { try? FileManager.default.removeItem(at: tempFile) }
      #if canImport(Darwin)
        return try await client.http.session.upload(
          for: request, fromFile: tempFile, delegate: progressDelegate)
      #else
        let result = try await client.http.session.upload(for: request, fromFile: tempFile)
        let totalBytes = (try? source.totalBytes()) ?? 0
        progressContinuation.yield(
          .progress(TransferProgress(bytesTransferred: totalBytes, totalBytes: totalBytes))
        )
        return result
      #endif
    } else {
      let body = try multipart.buildInMemory()
      #if canImport(Darwin)
        return try await client.http.session.upload(
          for: request, from: body, delegate: progressDelegate)
      #else
        let result = try await client.http.session.upload(for: request, from: body)
        let totalBytes = Int64(body.count)
        progressContinuation.yield(
          .progress(TransferProgress(bytesTransferred: totalBytes, totalBytes: totalBytes))
        )
        return result
      #endif
    }
  }

  private func handleError(_ error: any Error) {
    let isCancellation =
      error is CancellationError || (error as? URLError)?.code == .cancelled
    if isCancellation {
      switch state {
      case .uploading:
        cancel()
      default:
        return
      }
    } else {
      finish(with: .failure(StorageError.from(error)))
    }
  }

  private func finish(with result: Result<FileUploadResponse, any Error>) {
    switch result {
    case .success(let response):
      state = .completed(response)
      eventsContinuation.yield(.completed(response))
    case .failure(let error):
      let storageError =
        error as? StorageError ?? StorageError.networkError(underlying: error)
      state = .failed(storageError)
      eventsContinuation.yield(.failed(storageError))
    }
    eventsContinuation.finish()
    resultContinuation.yield(result.mapError { $0 })
    resultContinuation.finish()
  }
}

// MARK: - Factory

extension MultipartUploadEngine {
  static func makeTask(
    bucketId: String,
    path: String,
    source: UploadSource,
    options: FileOptions,
    httpMethod: HTTPMethod = .post,
    client: StorageClient
  ) -> StorageUploadTask {
    let (eventStream, eventsContinuation) =
      AsyncStream<TransferEvent<FileUploadResponse>>.makeStream()
    let (resultStream, resultContinuation) =
      AsyncStream<Result<FileUploadResponse, any Error>>.makeStream(
        bufferingPolicy: .bufferingNewest(1))

    let engine = MultipartUploadEngine(
      bucketId: bucketId,
      path: path,
      source: source,
      options: options,
      httpMethod: httpMethod,
      client: client,
      eventsContinuation: eventsContinuation,
      resultContinuation: resultContinuation
    )

    eventsContinuation.onTermination = { reason in
      guard case .cancelled = reason else { return }
      Task { await engine.cancel() }
    }

    let resultTask = Task<FileUploadResponse, any Error> {
      for await r in resultStream { return try r.get() }
      throw StorageError.cancelled
    }

    let task = StorageUploadTask(
      events: eventStream,
      resultTask: resultTask,
      pause: { await engine.pause() },
      resume: { await engine.resume() },
      cancel: { await engine.cancel() }
    )

    Task { await engine.start() }

    return task
  }
}

// MARK: - Progress delegate

private final class UploadProgressDelegate: NSObject, URLSessionTaskDelegate, Sendable {
  let handler: @Sendable (Int64, Int64) -> Void

  init(handler: @Sendable @escaping (Int64, Int64) -> Void) {
    self.handler = handler
  }

  func urlSession(
    _ session: URLSession,
    task: URLSessionTask,
    didSendBodyData bytesSent: Int64,
    totalBytesSent: Int64,
    totalBytesExpectedToSend: Int64
  ) {
    handler(totalBytesSent, totalBytesExpectedToSend)
  }
}

1	//
2	// MultipartUploadEngine.swift
3	// Storage
4	//
5	// Created by Guilherme Souza on 04/05/26.
6	//
7
8	import ConcurrencyExtras
9	import Foundation
10	import Helpers
11
12	#if canImport(FoundationNetworking)
13	import FoundationNetworking
14	#endif
15
16	private struct MultipartServerResponse: Decodable {
17	let Key: String
18	let Id: UUID
19	}
20
21	actor MultipartUploadEngine {
22	enum State {
23	case idle
24	case uploading
25	case completed(FileUploadResponse)
26	case failed(StorageError)
27	case cancelled
28
29	var isTerminal: Bool {	6✔
30	switch self {	6✔
31	case .completed, .failed, .cancelled: return true	6✔
32	default: return false	6✔
33	}	6✔
34	}	6✔
35	}
36
37	private let bucketId: String
38	private let path: String
39	private let source: UploadSource
40	private let options: FileOptions
41	private let httpMethod: HTTPMethod
42	private let client: StorageClient
43	private let eventsContinuation: AsyncStream<TransferEvent<FileUploadResponse>>.Continuation
44	private let resultContinuation: AsyncStream<Result<FileUploadResponse, any Error>>.Continuation
45
46	private var state: State = .idle	12✔
47	private var currentUploadTask: Task<Void, Never>?
48
49	init(
50	bucketId: String,
51	path: String,
52	source: UploadSource,
53	options: FileOptions,
54	httpMethod: HTTPMethod = .post,
55	client: StorageClient,
56	eventsContinuation: AsyncStream<TransferEvent<FileUploadResponse>>.Continuation,
57	resultContinuation: AsyncStream<Result<FileUploadResponse, any Error>>.Continuation
58	) {	12✔
59	self.bucketId = bucketId	12✔
60	self.path = path	12✔
61	self.source = source	12✔
62	self.options = options	12✔
63	self.httpMethod = httpMethod	12✔
64	self.client = client	12✔
65	self.eventsContinuation = eventsContinuation	12✔
66	self.resultContinuation = resultContinuation	12✔
67	}	12✔
68
69	func start() {	12✔
70	guard case .idle = state else { return }	12✔
71	state = .uploading	11✔
72	currentUploadTask = Task { await run() }	11✔
73	}	11✔
74
75	// Multipart uploads do not support pause/resume — the single-shot request cannot be
76	// interrupted and resumed mid-flight. These are intentional no-ops; callers that need
77	// pause/resume should use the TUS upload path instead.
NEW 78	func pause() {}	×
NEW 79	func resume() {}	×
80
81	func cancel() {	1✔
82	guard !state.isTerminal else { return }	1✔
83	currentUploadTask?.cancel()	1✔
84	state = .cancelled	1✔
85	let error = StorageError.cancelled	1✔
86	eventsContinuation.yield(.failed(error))	1✔
87	eventsContinuation.finish()	1✔
88	resultContinuation.yield(.failure(error))	1✔
89	resultContinuation.finish()	1✔
90	}	1✔
91
92	// MARK: - Private
93
94	private func run() async {	11✔
95	do {	11✔
96	try Task.checkCancellation()	11✔
97	let response = try await performUpload()	11✔
98	finish(with: .success(response))	11✔
99	} catch {	11✔
NEW 100	handleError(error)	×
101	}	11✔
102	}	11✔
103
104	private func performUpload() async throws -> FileUploadResponse {	11✔
105	#if DEBUG
106	let builder = MultipartBuilder(	11✔
107	boundary: testingBoundary.value ?? "----sb-\(UUID().uuidString)"	11✔
108	)	11✔
109	#else
110	let builder = MultipartBuilder()
111	#endif
112		11✔
113	let multipart = source.append(to: builder, withPath: path, options: options)	11✔
114		11✔
115	var headers: [String: String] = [:]	11✔
116	headers["Content-Type"] = multipart.contentType	11✔
117	if options.upsert {	11✔
118	headers["x-upsert"] = "true"	1✔
119	}	1✔
120		11✔
121	var url = client.url.appendingPathComponent("object").appendingPathComponent(bucketId)	11✔
122	for component in path.split(separator: "/") {	11✔
123	url = url.appendingPathComponent(String(component))	11✔
124	}	11✔
125		11✔
126	let request = try await client.http.createRequest(	11✔
127	httpMethod,	11✔
128	url: url,	11✔
129	headers: client.mergedHeaders(headers)	11✔
130	)	11✔
131		11✔
132	do {	11✔
133	let (data, urlResponse) = try await uploadWithProgress(request: request, multipart: multipart)	11✔
134	let httpResponse = try client.http.validateResponse(urlResponse, data: data)	11✔
135	client.logResponse(httpResponse, data: data)	11✔
136	let serverResponse = try client.decoder.decode(MultipartServerResponse.self, from: data)	11✔
137	return FileUploadResponse(id: serverResponse.Id, path: path, fullPath: serverResponse.Key)	11✔
138	} catch {	11✔
NEW 139	client.logFailure(error)	×
NEW 140	throw client.translateStorageError(error)	×
NEW 141	}	×
142	}	11✔
143
144	private func uploadWithProgress(
145	request: URLRequest,
146	multipart: MultipartBuilder
147	) async throws -> (Data, URLResponse) {	11✔
148	let progressContinuation = eventsContinuation	11✔
149		11✔
150	let progressDelegate = UploadProgressDelegate { sent, total in	11✔
NEW 151	progressContinuation.yield(	×
NEW 152	.progress(TransferProgress(bytesTransferred: sent, totalBytes: total))	×
NEW 153	)	×
NEW 154	}	×
155		11✔
156	if source.usesTempFileUpload {	11✔
NEW 157	let tempFile = try multipart.buildToTempFile()	×
NEW 158	defer { try? FileManager.default.removeItem(at: tempFile) }	×
159	#if canImport(Darwin)
NEW 160	return try await client.http.session.upload(	×
NEW 161	for: request, fromFile: tempFile, delegate: progressDelegate)	×
162	#else
163	let result = try await client.http.session.upload(for: request, fromFile: tempFile)
164	let totalBytes = (try? source.totalBytes()) ?? 0
165	progressContinuation.yield(
166	.progress(TransferProgress(bytesTransferred: totalBytes, totalBytes: totalBytes))
167	)
168	return result
169	#endif
170	} else {	11✔
171	let body = try multipart.buildInMemory()	11✔
172	#if canImport(Darwin)
173	return try await client.http.session.upload(	11✔
174	for: request, from: body, delegate: progressDelegate)	11✔
175	#else
176	let result = try await client.http.session.upload(for: request, from: body)
177	let totalBytes = Int64(body.count)
178	progressContinuation.yield(
179	.progress(TransferProgress(bytesTransferred: totalBytes, totalBytes: totalBytes))
180	)
181	return result
182	#endif
183	}	11✔
184	}	11✔
185
NEW 186	private func handleError(_ error: any Error) {	×
NEW 187	let isCancellation =	×
NEW 188	error is CancellationError \|\| (error as? URLError)?.code == .cancelled	×
NEW 189	if isCancellation {	×
NEW 190	switch state {	×
NEW 191	case .uploading:	×
NEW 192	cancel()	×
NEW 193	default:	×
NEW 194	return	×
NEW 195	}	×
NEW 196	} else {	×
NEW 197	finish(with: .failure(StorageError.from(error)))	×
NEW 198	}	×
NEW 199	}	×
200
201	private func finish(with result: Result<FileUploadResponse, any Error>) {	11✔
202	switch result {	11✔
203	case .success(let response):	11✔
204	state = .completed(response)	11✔
205	eventsContinuation.yield(.completed(response))	11✔
206	case .failure(let error):	11✔
NEW 207	let storageError =	×
NEW 208	error as? StorageError ?? StorageError.networkError(underlying: error)	×
NEW 209	state = .failed(storageError)	×
NEW 210	eventsContinuation.yield(.failed(storageError))	×
211	}	11✔
212	eventsContinuation.finish()	11✔
213	resultContinuation.yield(result.mapError { $0 })	11✔
214	resultContinuation.finish()	11✔
215	}	11✔
216	}
217
218	// MARK: - Factory
219
220	extension MultipartUploadEngine {
221	static func makeTask(
222	bucketId: String,
223	path: String,
224	source: UploadSource,
225	options: FileOptions,
226	httpMethod: HTTPMethod = .post,
227	client: StorageClient
228	) -> StorageUploadTask {	12✔
229	let (eventStream, eventsContinuation) =	12✔
230	AsyncStream<TransferEvent<FileUploadResponse>>.makeStream()	12✔
231	let (resultStream, resultContinuation) =	12✔
232	AsyncStream<Result<FileUploadResponse, any Error>>.makeStream(	12✔
233	bufferingPolicy: .bufferingNewest(1))	12✔
234		12✔
235	let engine = MultipartUploadEngine(	12✔
236	bucketId: bucketId,	12✔
237	path: path,	12✔
238	source: source,	12✔
239	options: options,	12✔
240	httpMethod: httpMethod,	12✔
241	client: client,	12✔
242	eventsContinuation: eventsContinuation,	12✔
243	resultContinuation: resultContinuation	12✔
244	)	12✔
245		12✔
246	eventsContinuation.onTermination = { reason in	12✔
247	guard case .cancelled = reason else { return }	12✔
NEW 248	Task { await engine.cancel() }	×
NEW 249	}	×
250		12✔
251	let resultTask = Task<FileUploadResponse, any Error> {	12✔
252	for await r in resultStream { return try r.get() }	12✔
NEW 253	throw StorageError.cancelled	×
254	}	12✔
255		12✔
256	let task = StorageUploadTask(	12✔
257	events: eventStream,	12✔
258	resultTask: resultTask,	12✔
259	pause: { await engine.pause() },	12✔
260	resume: { await engine.resume() },	12✔
261	cancel: { await engine.cancel() }	12✔
262	)	12✔
263		12✔
264	Task { await engine.start() }	12✔
265		12✔
266	return task	12✔
267	}	12✔
268	}
269
270	// MARK: - Progress delegate
271
272	private final class UploadProgressDelegate: NSObject, URLSessionTaskDelegate, Sendable {
273	let handler: @Sendable (Int64, Int64) -> Void
274
275	init(handler: @Sendable @escaping (Int64, Int64) -> Void) {	11✔
276	self.handler = handler	11✔
277	}	11✔
278
279	func urlSession(
280	_ session: URLSession,
281	task: URLSessionTask,
282	didSendBodyData bytesSent: Int64,
283	totalBytesSent: Int64,
284	totalBytesExpectedToSend: Int64
NEW 285	) {	×
NEW 286	handler(totalBytesSent, totalBytesExpectedToSend)	×
NEW 287	}	×
288	}

supabase / supabase-swift / 28167379888

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous