feat(core): abort uploads

packages/core/src/highlevel/methods/files/normalize-input-file.ts

@@ -17,6 +17,7 @@ export async function _normalizeInputFile(
     input: InputFileLike,
     params: {
         progressCallback?: (uploaded: number, total: number) => void
+        abortSignal?: AbortSignal
         fileName?: string
         fileSize?: number
         fileMime?: string

packages/core/src/highlevel/methods/files/normalize-input-media.ts

@@ -27,6 +27,7 @@ export async function _normalizeInputMedia(
     params: {
         progressCallback?: (uploaded: number, total: number) => void
         uploadPeer?: tl.TypeInputPeer
+        abortSignal?: AbortSignal
         businessConnectionId?: string
     } = {},
     uploadMedia = false,
@@ -258,6 +259,7 @@ export async function _normalizeInputMedia(
             fileSize: media.fileSize,
             requireFileSize: media.type === 'photo',
             requireExtension: media.type === 'photo',
+            abortSignal: params.abortSignal,
         })
         inputFile = uploaded.inputFile
         mime = uploaded.mime
@@ -265,7 +267,7 @@ export async function _normalizeInputMedia(
 
     let videoCover: tl.TypeInputPhoto | undefined
     if (media.type === 'video' && media.cover) {
-        const inputMedia = await _normalizeInputMedia(client, media.cover)
+        const inputMedia = await _normalizeInputMedia(client, media.cover, params)
         assertTypeIs('uploadMediaIfNeeded', inputMedia, 'inputMediaPhoto')
 
         videoCover = inputMedia.id
@@ -281,7 +283,7 @@ export async function _normalizeInputMedia(
             peer: uploadPeer,
             media: inputMedia,
             businessConnectionId: params.businessConnectionId,
-        })
+        }, { abortSignal: params.abortSignal })
 
         if (photo) {
             assertTypeIs('normalizeInputMedia (@ messages.uploadMedia)', res, 'messageMediaPhoto')
@@ -382,7 +384,7 @@ export async function _normalizeInputMedia(
     }
 
     if ('thumb' in media && media.thumb) {
-        thumb = await _normalizeInputFile(client, media.thumb, {})
+        thumb = await _normalizeInputFile(client, media.thumb, { abortSignal: params.abortSignal })
     }
 
     const attributes: tl.TypeDocumentAttribute[] = []

packages/core/src/highlevel/methods/files/upload-file.ts

@@ -110,6 +110,8 @@ export async function uploadFile(
          * or throw an error if it cannot be guessed.
          */
         requireExtension?: boolean
+
+        abortSignal?: AbortSignal
     },
 ): Promise<UploadedFile> {
     // normalize params
@@ -117,6 +119,7 @@ export async function uploadFile(
     let fileSize = -1 // unknown
     let fileName = params.fileName
     let fileMime = params.fileMime
+    const abortSignal = params.abortSignal
 
     if (client.platform.normalizeFile) {
         const res = await client.platform.normalizeFile(file)
@@ -247,6 +250,7 @@ export async function uploadFile(
     let readableEnded = false
 
     const uploadNextPart = async (): Promise<void> => {
+        if (abortSignal?.aborted) return
         const thisIdx = idx++
 
         await lock.acquire()
@@ -263,6 +267,8 @@ export async function uploadFile(
             lock.release()
         }
 
+        if (abortSignal?.aborted) return
+
         const ended = part.length < partSize
         if (ended) {
             readableEnded = true
@@ -308,8 +314,9 @@ export async function uploadFile(
                 bytes: part,
             } satisfies tl.upload.RawSaveFilePartRequest)
 
-        const result = await client.call(request, { kind: connectionKind })
+        const result = await client.call(request, { kind: connectionKind, abortSignal })
         if (!result) throw new Error(`Failed to upload part ${idx}`)
+        if (abortSignal?.aborted) return
 
         pos += part.length
 
@@ -325,6 +332,8 @@ export async function uploadFile(
 
     await Promise.all(Array.from({ length: poolSize }, uploadNextPart))
 
+    abortSignal?.throwIfAborted()
+
     if (fileName === undefined) {
         // infer file extension from mime type. for some media types,
         // telegram requires us to specify the file extension

packages/core/src/highlevel/methods/messages/send-common.ts

@@ -132,6 +132,8 @@ export interface CommonSendParams {
      */
     allowPaidMessages?: tl.Long
 
+    abortSignal?: AbortSignal
+
     /**
      * ID of a message effect to use when sending the message
      * (see {@link TelegramClient.getAvailableMessageEffects})

packages/core/src/highlevel/methods/messages/send-media-group.ts

@@ -81,6 +81,7 @@ export async function sendMediaGroup(
                 // fuck my life
                 uploadPeer: peer,
                 businessConnectionId: params.businessConnectionId,
+                abortSignal: params.abortSignal,
             },
             true,
         )
@@ -120,7 +121,7 @@ export async function sendMediaGroup(
             allowPaidFloodskip: params.allowPaidFloodskip,
             allowPaidStars: params.allowPaidMessages,
         },
-        { chainId },
+        { chainId, abortSignal: params.abortSignal },
     )
 
     assertIsUpdatesGroup('sendMediaGroup', res)

packages/core/src/highlevel/methods/messages/send-media.ts

@@ -81,6 +81,7 @@ export async function sendMedia(
     const inputMedia = await _normalizeInputMedia(client, media, {
         progressCallback: params.progressCallback,
         uploadPeer: peer,
+        abortSignal: params.abortSignal,
     })
 
     const [message, entities] = await _normalizeInputText(
@@ -116,7 +117,7 @@ export async function sendMedia(
             allowPaidFloodskip: params.allowPaidFloodskip,
             allowPaidStars: params.allowPaidMessages,
         },
-        { chainId },
+        { chainId, abortSignal: params.abortSignal },
     )
 
     const msg = _findMessageInUpdate(client, res, false, !params.shouldDispatch, false, randomId)

packages/core/src/highlevel/methods/messages/send-text.ts

@@ -88,7 +88,7 @@ export async function sendText(
             allowPaidFloodskip: params.allowPaidFloodskip,
             allowPaidStars: params.allowPaidMessages,
         },
-        { chainId },
+        { chainId, abortSignal: params.abortSignal },
     )
 
     if (res._ === 'updateShortSentMessage') {