25547203894

Committed 08 May 2026 09:07AM UTC coverage: 80.974% (+0.004%) from 80.97%

Build # 25547203894

Build Type

push

github

Committed by

grdsdev

Commit Message

feat(storage): add integration tests, update examples, and CI configuration

- Add new integration tests in Tests/IntegrationTests/Storage/:
  StorageClientIntegrationTests, StorageFileIntegrationTests,
  StorageTransferIntegrationTests (TUS and download scenarios)
- Update example app to demonstrate pause/resume/cancel with the new API
- Update CI workflow configuration
- Update Auth and docs

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

Coverage Stats

7831 of 9671 relevant lines covered (80.97%)

29.15 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
//

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

supabase / supabase-swift / 25547203894

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