25488993349

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

Build # 25488993349

Build Type

Pull #991

github

Committed by

web-flow

Commit Message

Merge 0eefab23c into 6fa2a16c0

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%)

7282 of 9129 relevant lines covered (79.77%)

29.57 hits per line

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

87.34

/Sources/Storage/StorageTransferTask.swift

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

import Foundation

/// A handle to an in-flight upload or download.
///
/// ## Usage patterns
///
/// **Fire and forget**
/// ```swift
/// storage.from("avatars").upload("user.jpg", data: imageData)
/// ```
///
/// **Await the result**
/// ```swift
/// let response = try await storage.from("avatars")
///   .upload("user.jpg", data: imageData)
///   .value
/// ```
///
/// **Observe progress**
/// ```swift
/// let task = storage.from("avatars").upload("user.jpg", data: imageData)
/// for await event in task.events {
///   switch event {
///   case .progress(let p):
///     updateProgressBar(p.fractionCompleted)
///   case .completed(let response):
///     print("Uploaded to \(response.fullPath)")
///   case .failed(let error):
///     print("Upload failed: \(error.message)")
///   }
/// }
/// ```
///
/// **Pause, resume, and cancel (TUS uploads only)**
/// ```swift
/// // TUS (resumable) upload — use method: .resumable to force TUS
/// let upload = storage.from("videos").upload("clip.mp4", fileURL: fileURL, method: .resumable)
/// await upload.pause()   // suspend mid-upload
/// await upload.resume()  // continue from where it left off
/// await upload.cancel()  // abort entirely
///
/// // Download
/// let download = storage.from("videos").download(path: "clip.mp4")
/// await download.pause()   // suspend; captures resume data if server supports range requests
/// await download.resume()  // continue from last byte, or restart if no resume data
/// await download.cancel()  // abort entirely
/// ```
///
/// Both ``events`` and ``value`` are independent: consuming one does not affect the other.
public final class StorageTransferTask<Success: Sendable>: Sendable {

  /// A stream of transfer lifecycle events.
  ///
  /// Emits ``TransferEvent/progress(_:)`` events while the transfer is active, then terminates
  /// with either ``TransferEvent/completed(_:)`` or ``TransferEvent/failed(_:)``.
  public let events: AsyncStream<TransferEvent<Success>>

  private let _resultTask: Task<Success, any Error>
  private let _pause: @Sendable () async -> Void
  private let _resume: @Sendable () async -> Void
  private let _cancel: @Sendable () async -> Void

  init(
    events: AsyncStream<TransferEvent<Success>>,
    resultTask: Task<Success, any Error>,
    pause: @Sendable @escaping () async -> Void,
    resume: @Sendable @escaping () async -> Void,
    cancel: @Sendable @escaping () async -> Void
  ) {
    self.events = events
    self._resultTask = resultTask
    self._pause = pause
    self._resume = resume
    self._cancel = cancel
  }

  /// The transfer outcome as a `Result`. Never throws — inspect `.success` / `.failure` directly.
  public var result: Result<Success, any Error> {
    get async { await _resultTask.result }
  }

  /// Awaits the success value. Throws `StorageError` on failure or cancellation.
  public var value: Success {
    get async throws { try await result.get() }
  }

  /// Suspends the transfer.
  ///
  /// Only supported for TUS (resumable) uploads. For multipart uploads this is a no-op —
  /// use ``cancel()`` and re-upload from scratch if you need to stop a multipart transfer.
  /// For TUS uploads the current in-flight chunk is drained before the task suspends.
  public func pause() async { await _pause() }

  /// Resumes a previously paused transfer.
  ///
  /// Only supported for TUS (resumable) uploads. For multipart uploads this is a no-op.
  /// For TUS uploads the server is HEAD-queried to re-sync the byte offset before uploading resumes.
  public func resume() async { await _resume() }

  /// Cancels the transfer immediately.
  ///
  /// Cancelling a transfer that has already completed or failed is a no-op.
  /// After cancellation, ``value`` throws a ``StorageError`` with code ``StorageErrorCode/cancelled``,
  /// and the ``events`` stream ends with a ``TransferEvent/failed(_:)`` event.
  public func cancel() async {
    // Order is load-bearing: _cancel() must run first.
    // It sets the engine's state to .cancelled and finishes the continuations before
    // Swift's structured cancellation propagates. If _resultTask.cancel() fired first,
    // the resulting CancellationError would reach handleRunError while the engine is
    // still in .uploading/.creating, causing it to call cancel() a second time and
    // race with this explicit cancellation path.
    await _cancel()
    _resultTask.cancel()
  }
}

extension StorageTransferTask {
  /// Returns a new task that applies `transform` to the success value.
  /// Progress events pass through unchanged. Pause/resume/cancel delegate to `self`.
  func mapResult<NewSuccess: Sendable>(
    _ transform: @Sendable @escaping (Success) throws -> NewSuccess
  ) -> StorageTransferTask<NewSuccess> {
    let (newStream, newContinuation) = AsyncStream<TransferEvent<NewSuccess>>.makeStream()
    let (resultStream, resultContinuation) = AsyncStream<Result<NewSuccess, any Error>>.makeStream(
      bufferingPolicy: .bufferingNewest(1))

    let bridgeTask = Task {
      for await event in self.events {
        switch event {
        case .progress(let p):
          newContinuation.yield(.progress(p))
        case .completed(let value):
          do {
            let mapped = try transform(value)
            newContinuation.yield(.completed(mapped))
            newContinuation.finish()
            resultContinuation.yield(.success(mapped))
            resultContinuation.finish()
          } catch {
            let storageError = StorageError.fileSystemError(underlying: error)
            newContinuation.yield(.failed(storageError))
            newContinuation.finish()
            resultContinuation.yield(.failure(storageError))
            resultContinuation.finish()
          }
        case .failed(let error):
          newContinuation.yield(.failed(error))
          newContinuation.finish()
          resultContinuation.yield(.failure(error))
          resultContinuation.finish()
        }
      }
      newContinuation.finish()
    }

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

    return StorageTransferTask<NewSuccess>(
      events: newStream,
      resultTask: newResultTask,
      pause: self._pause,
      resume: self._resume,
      cancel: {
        await self._cancel()
        bridgeTask.cancel()
        // Ensure resultContinuation is finished so newResultTask exits via the stream
        // (yielding StorageError.cancelled) rather than via Task cancellation
        // (which would throw CancellationError), keeping the error type deterministic.
        resultContinuation.finish()
      }
    )
  }
}

/// An event emitted during a transfer.
public enum TransferEvent<Success: Sendable>: Sendable {
  /// Periodic progress update during an active transfer.
  case progress(TransferProgress)
  /// The transfer finished successfully. Terminal — the stream ends after this event.
  case completed(Success)
  /// The transfer failed or was cancelled. Terminal — the stream ends after this event.
  case failed(StorageError)
}

/// Byte-level progress for a transfer.
public struct TransferProgress: Sendable {
  /// Number of bytes sent or received so far.
  public let bytesTransferred: Int64

  /// Total size of the transfer in bytes.
  public let totalBytes: Int64

  /// Transfer completion as a value between `0.0` and `1.0`.
  /// Returns `0` when `totalBytes` is zero.
  public var fractionCompleted: Double {
    guard totalBytes > 0 else { return 0 }
    return Double(bytesTransferred) / Double(totalBytes)
  }
}

/// A handle for an upload.
///
/// The success value is a ``FileUploadResponse`` containing the storage path and object ID.
/// TUS uploads support ``StorageTransferTask/pause()``, ``StorageTransferTask/resume()``, and
/// ``StorageTransferTask/cancel()``; multipart uploads only support cancel.
public typealias StorageUploadTask = StorageTransferTask<FileUploadResponse>

/// A handle for a download to disk.
///
/// The success value is a `URL` pointing to a temporary file on disk.
/// Move or read the file before the app exits — it is not guaranteed to persist across launches.
public typealias StorageDownloadTask = StorageTransferTask<URL>

1	//
2	// StorageTransferTask.swift
3	// Storage
4	//
5	// Created by Guilherme Souza on 04/05/26.
6	//
7
8	import Foundation
9
10	/// A handle to an in-flight upload or download.
11	///
12	/// ## Usage patterns
13	///
14	/// Fire and forget
15	/// ```swift
16	/// storage.from("avatars").upload("user.jpg", data: imageData)
17	/// ```
18	///
19	/// Await the result
20	/// ```swift
21	/// let response = try await storage.from("avatars")
22	/// .upload("user.jpg", data: imageData)
23	/// .value
24	/// ```
25	///
26	/// Observe progress
27	/// ```swift
28	/// let task = storage.from("avatars").upload("user.jpg", data: imageData)
29	/// for await event in task.events {
30	/// switch event {
31	/// case .progress(let p):
32	/// updateProgressBar(p.fractionCompleted)
33	/// case .completed(let response):
34	/// print("Uploaded to \(response.fullPath)")
35	/// case .failed(let error):
36	/// print("Upload failed: \(error.message)")
37	/// }
38	/// }
39	/// ```
40	///
41	/// Pause, resume, and cancel (TUS uploads only)
42	/// ```swift
43	/// // TUS (resumable) upload — use method: .resumable to force TUS
44	/// let upload = storage.from("videos").upload("clip.mp4", fileURL: fileURL, method: .resumable)
45	/// await upload.pause() // suspend mid-upload
46	/// await upload.resume() // continue from where it left off
47	/// await upload.cancel() // abort entirely
48	///
49	/// // Download
50	/// let download = storage.from("videos").download(path: "clip.mp4")
51	/// await download.pause() // suspend; captures resume data if server supports range requests
52	/// await download.resume() // continue from last byte, or restart if no resume data
53	/// await download.cancel() // abort entirely
54	/// ```
55	///
56	/// Both ``events`` and ``value`` are independent: consuming one does not affect the other.
57	public final class StorageTransferTask<Success: Sendable>: Sendable {
58
59	/// A stream of transfer lifecycle events.
60	///
61	/// Emits ``TransferEvent/progress(_:)`` events while the transfer is active, then terminates
62	/// with either ``TransferEvent/completed(_:)`` or ``TransferEvent/failed(_:)``.
63	public let events: AsyncStream<TransferEvent<Success>>
64
65	private let _resultTask: Task<Success, any Error>
66	private let _pause: @Sendable () async -> Void
67	private let _resume: @Sendable () async -> Void
68	private let _cancel: @Sendable () async -> Void
69
70	init(
71	events: AsyncStream<TransferEvent<Success>>,
72	resultTask: Task<Success, any Error>,
73	pause: @Sendable @escaping () async -> Void,
74	resume: @Sendable @escaping () async -> Void,
75	cancel: @Sendable @escaping () async -> Void
76	) {	26✔
77	self.events = events	26✔
78	self._resultTask = resultTask	26✔
79	self._pause = pause	26✔
80	self._resume = resume	26✔
81	self._cancel = cancel	26✔
82	}	26✔
83
84	/// The transfer outcome as a `Result`. Never throws — inspect `.success` / `.failure` directly.
85	public var result: Result<Success, any Error> {
86	get async { await _resultTask.result }	18✔
87	}
88
89	/// Awaits the success value. Throws `StorageError` on failure or cancellation.
90	public var value: Success {
91	get async throws { try await result.get() }	18✔
92	}
93
94	/// Suspends the transfer.
95	///
96	/// Only supported for TUS (resumable) uploads. For multipart uploads this is a no-op —
97	/// use ``cancel()`` and re-upload from scratch if you need to stop a multipart transfer.
98	/// For TUS uploads the current in-flight chunk is drained before the task suspends.
NEW 99	public func pause() async { await _pause() }	×
100
101	/// Resumes a previously paused transfer.
102	///
103	/// Only supported for TUS (resumable) uploads. For multipart uploads this is a no-op.
104	/// For TUS uploads the server is HEAD-queried to re-sync the byte offset before uploading resumes.
NEW 105	public func resume() async { await _resume() }	×
106
107	/// Cancels the transfer immediately.
108	///
109	/// Cancelling a transfer that has already completed or failed is a no-op.
110	/// After cancellation, ``value`` throws a ``StorageError`` with code ``StorageErrorCode/cancelled``,
111	/// and the ``events`` stream ends with a ``TransferEvent/failed(_:)`` event.
112	public func cancel() async {	2✔
113	// Order is load-bearing: _cancel() must run first.	2✔
114	// It sets the engine's state to .cancelled and finishes the continuations before	2✔
115	// Swift's structured cancellation propagates. If _resultTask.cancel() fired first,	2✔
116	// the resulting CancellationError would reach handleRunError while the engine is	2✔
117	// still in .uploading/.creating, causing it to call cancel() a second time and	2✔
118	// race with this explicit cancellation path.	2✔
119	await _cancel()	2✔
120	_resultTask.cancel()	2✔
121	}	2✔
122	}
123
124	extension StorageTransferTask {
125	/// Returns a new task that applies `transform` to the success value.
126	/// Progress events pass through unchanged. Pause/resume/cancel delegate to `self`.
127	func mapResult<NewSuccess: Sendable>(
128	_ transform: @Sendable @escaping (Success) throws -> NewSuccess
129	) -> StorageTransferTask<NewSuccess> {	4✔
130	let (newStream, newContinuation) = AsyncStream<TransferEvent<NewSuccess>>.makeStream()	4✔
131	let (resultStream, resultContinuation) = AsyncStream<Result<NewSuccess, any Error>>.makeStream(	4✔
132	bufferingPolicy: .bufferingNewest(1))	4✔
133		4✔
134	let bridgeTask = Task {	4✔
135	for await event in self.events {	7✔
136	switch event {	7✔
137	case .progress(let p):	7✔
138	newContinuation.yield(.progress(p))	3✔
139	case .completed(let value):	7✔
140	do {	3✔
141	let mapped = try transform(value)	3✔
142	newContinuation.yield(.completed(mapped))	2✔
143	newContinuation.finish()	2✔
144	resultContinuation.yield(.success(mapped))	2✔
145	resultContinuation.finish()	2✔
146	} catch {	2✔
147	let storageError = StorageError.fileSystemError(underlying: error)	1✔
148	newContinuation.yield(.failed(storageError))	1✔
149	newContinuation.finish()	1✔
150	resultContinuation.yield(.failure(storageError))	1✔
151	resultContinuation.finish()	1✔
152	}	1✔
153	case .failed(let error):	7✔
154	newContinuation.yield(.failed(error))	1✔
155	newContinuation.finish()	1✔
156	resultContinuation.yield(.failure(error))	1✔
157	resultContinuation.finish()	1✔
158	}	7✔
159	}	7✔
160	newContinuation.finish()	4✔
161	}	4✔
162		4✔
163	let newResultTask = Task<NewSuccess, any Error> {	4✔
164	for await r in resultStream { return try r.get() }	4✔
NEW 165	throw StorageError.cancelled	×
166	}	4✔
167		4✔
168	return StorageTransferTask<NewSuccess>(	4✔
169	events: newStream,	4✔
170	resultTask: newResultTask,	4✔
171	pause: self._pause,	4✔
172	resume: self._resume,	4✔
173	cancel: {	4✔
NEW 174	await self._cancel()	×
NEW 175	bridgeTask.cancel()	×
NEW 176	// Ensure resultContinuation is finished so newResultTask exits via the stream	×
NEW 177	// (yielding StorageError.cancelled) rather than via Task cancellation	×
NEW 178	// (which would throw CancellationError), keeping the error type deterministic.	×
NEW 179	resultContinuation.finish()	×
NEW 180	}	×
181	)	4✔
182	}	4✔
183	}
184
185	/// An event emitted during a transfer.
186	public enum TransferEvent<Success: Sendable>: Sendable {
187	/// Periodic progress update during an active transfer.
188	case progress(TransferProgress)
189	/// The transfer finished successfully. Terminal — the stream ends after this event.
190	case completed(Success)
191	/// The transfer failed or was cancelled. Terminal — the stream ends after this event.
192	case failed(StorageError)
193	}
194
195	/// Byte-level progress for a transfer.
196	public struct TransferProgress: Sendable {
197	/// Number of bytes sent or received so far.
198	public let bytesTransferred: Int64
199
200	/// Total size of the transfer in bytes.
201	public let totalBytes: Int64
202
203	/// Transfer completion as a value between `0.0` and `1.0`.
204	/// Returns `0` when `totalBytes` is zero.
205	public var fractionCompleted: Double {	6✔
206	guard totalBytes > 0 else { return 0 }	6✔
207	return Double(bytesTransferred) / Double(totalBytes)	4✔
208	}	6✔
209	}
210
211	/// A handle for an upload.
212	///
213	/// The success value is a ``FileUploadResponse`` containing the storage path and object ID.
214	/// TUS uploads support ``StorageTransferTask/pause()``, ``StorageTransferTask/resume()``, and
215	/// ``StorageTransferTask/cancel()``; multipart uploads only support cancel.
216	public typealias StorageUploadTask = StorageTransferTask<FileUploadResponse>
217
218	/// A handle for a download to disk.
219	///
220	/// The success value is a `URL` pointing to a temporary file on disk.
221	/// Move or read the file before the app exits — it is not guaranteed to persist across launches.
222	public typealias StorageDownloadTask = StorageTransferTask<URL>

supabase / supabase-swift / 25488993349

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