25488992304

Committed 07 May 2026 09:58AM UTC coverage: 79.987%. First build

Build # 25488992304

Build Type

Pull #991

github

Committed by

grdsdev

Commit Message

style: apply swift-format to changed files

Pull Request Pull Request #991: feat(storage): add StorageTransferTask, MultipartUploadEngine, and upload API

Coverage Stats

307 of 422 new or added lines in 6 files covered. (72.75%)

7302 of 9129 relevant lines covered (79.99%)

29.62 hits per line

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

77.78

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

supabase / supabase-swift / 25488992304

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