feat: global upload manager to track file uploads outside of AttachmentManager (#1715)

## CLA

- [ ] I have signed the [Stream
CLA](https://docs.google.com/forms/d/e/1FAIpQLScFKsKkAJI7mhCr7K9rEIOpqIDThrWxuvxnwUq2XkHyG154vQ/viewform)
(required).
- [ ] Code changes are tested

## Description of the changes, What, Why and How?

RN PR for reference:
GetStream/stream-chat-react-native#3527

React Native allows hitting the send button before attachment uploads
are completed. This means they need a way to track attachment upload
status outside of `AttachmentManager`.

To solve this we have an `UploadManager` (can be accessed via
`client.uploadManager`) that uses the `AttachmentManager` of a given
channel to do attachment upload, and has a state to track
`uploadProgress` for attachments.

Attachments are tracked by `id` in `UploadManager`.

Long-term solution: we talked with @isekovanic that `messageComposer`
should have support for this feature, but that is a bigger task, not the
scope of this PR.

## Changelog

-

Author: szuperaz

docs/fileUpload.md

@@ -122,6 +122,9 @@ const imageResponse = await channel.sendImage(file, file.name, file.type, undefi
 
 When using the message composer’s attachment manager, upload progress is tracked when `config.attachments.trackUploadProgress` is `true` (the default). Progress is stored on each attachment’s `localMetadata.uploadProgress` (0–100 for the default upload path, from the axios progress event; the initial state is 0% when the upload starts).
 
-With a custom `doUploadRequest`, the function receives an optional second argument `options` with `onProgress?: (percent: number | undefined) => void`. Call `onProgress` from your upload implementation to drive the same `localMetadata.uploadProgress` updates. If you do not call it, `uploadProgress` stays at 0 until the upload finishes.
+With a custom `doUploadRequest`, the function receives an optional second argument `options` with:
 
-Set `trackUploadProgress` to `false` to skip setting `uploadProgress` (will be `undefined` in this case) and to omit progress callbacks to both the default channel upload and custom `doUploadRequest`.
+- `onProgress?: (percent: number | undefined) => void` — call this from your upload implementation to drive the same `localMetadata.uploadProgress` updates. If you do not call it, `uploadProgress` stays at 0 until the upload finishes.
+- `abortSignal?: AbortSignal` — the SDK aborts this signal when the upload is cancelled (for example the user removes the attachment, or `client.uploadManager.reset()` runs on disconnect). Forward it to your transport (axios `signal`, `fetch` `signal`, etc.) if you want to cancel upload request.
+
+Set `trackUploadProgress` to `false` to skip setting `uploadProgress` (will be `undefined` in this case) and to omit `onProgress` to both the default channel upload and custom `doUploadRequest`.

package.json

@@ -51,7 +51,7 @@
   "dependencies": {
     "@types/jsonwebtoken": "^9.0.8",
     "@types/ws": "^8.5.14",
-    "axios": "^1.12.2",
+    "axios": "^1.15.1",
     "base64-js": "^1.5.1",
     "form-data": "^4.0.4",
     "isomorphic-ws": "^5.0.0",

src/client.ts

@@ -9,6 +9,7 @@ import type WebSocket from 'isomorphic-ws';
 import { Channel } from './channel';
 import { ClientState } from './client_state';
 import { StableWSConnection } from './connection';
+import { UploadManager } from './uploadManager';
 import { CheckSignature, DevToken, JWTUserToken } from './signing';
 import { TokenManager } from './token_manager';
 import { WSConnectionFallback } from './connection_fallback';
@@ -296,6 +297,10 @@ export type MessageComposerSetupState = {
 export class StreamChat {
   private static _instance?: unknown | StreamChat; // type is undefined|StreamChat, unknown is due to TS limitations with statics
   messageDeliveryReporter: MessageDeliveryReporter;
+  /**
+   * @internal
+   */
+  uploadManager: UploadManager;
   _user?: OwnUserResponse | UserResponse;
   appSettingsPromise?: Promise<AppSettingsAPIResponse>;
   activeChannels: {
@@ -401,6 +406,7 @@ export class StreamChat {
     this.moderation = new Moderation(this);
 
     this.notifications = options?.notifications ?? new NotificationManager();
+    this.uploadManager = new UploadManager(this);
 
     // set the secret
     if (secretOrOptions && isString(secretOrOptions)) {
@@ -1020,6 +1026,7 @@ export class StreamChat {
     this.state = new ClientState({ client: this });
     // reset thread manager
     this.threads.resetState();
+    this.uploadManager.reset();
 
     // Since we wipe all user data already, we should reset token manager as well
     closePromise

src/index.ts

@@ -33,6 +33,7 @@ export type {
 export * from './thread_manager';
 export * from './token_manager';
 export * from './types';
+export * from './uploadManager';
 export * from './channel_manager';
 export * from './offline-support';
 export * from './LiveLocationManager';

src/messageComposer/attachmentManager.ts

@@ -237,12 +237,12 @@ export class AttachmentManager {
         return attachments;
       }
     }
-    return null;
+    return stateAttachments;
   };
 
   updateAttachment = (attachmentToUpdate: LocalAttachment) => {
     const updatedAttachments = this.prepareAttachmentUpdate(attachmentToUpdate);
-    if (updatedAttachments) {
+    if (updatedAttachments && updatedAttachments !== this.attachments) {
       this.state.partialNext({ attachments: updatedAttachments });
     }
   };
@@ -253,28 +253,35 @@ export class AttachmentManager {
     let hasUpdates = false;
     attachmentsToUpsert.forEach((attachment) => {
       const updatedAttachments = this.prepareAttachmentUpdate(attachment);
-      if (updatedAttachments) {
-        attachments = updatedAttachments;
-        hasUpdates = true;
-      } else {
+      if (updatedAttachments === null) {
         const localAttachment = ensureIsLocalAttachment(attachment);
         if (localAttachment) {
           attachments.push(localAttachment);
           hasUpdates = true;
         }
+      } else if (updatedAttachments !== this.attachments) {
+        attachments = updatedAttachments;
+        hasUpdates = true;
       }
+      // else: id exists and merge was a no-op (`prepareAttachmentUpdate` returns current state)
     });
     if (hasUpdates) {
       this.state.partialNext({ attachments });
     }
   };
 
   removeAttachments = (localAttachmentIds: string[]) => {
+    if (!localAttachmentIds.length) return;
+
     this.state.partialNext({
       attachments: this.attachments.filter(
         (attachment) => !localAttachmentIds.includes(attachment.localMetadata?.id),
       ),
     });
+
+    for (const id of localAttachmentIds) {
+      this.client.uploadManager.deleteUploadRecord(id);
+    }
   };
 
   getUploadConfigCheck = async (
@@ -464,13 +471,21 @@ export class AttachmentManager {
         }
       : undefined;
 
+    const axiosUploadConfig =
+      progressHandler || options?.abortSignal
+        ? {
+            ...(progressHandler ? { onUploadProgress: progressHandler } : {}),
+            ...(options?.abortSignal ? { signal: options.abortSignal } : {}),
+          }
+        : undefined;
+
     if (isFileReference(fileLike)) {
       return this.channel[isImageFile(fileLike) ? 'sendImage' : 'sendFile'](
         fileLike.uri,
         fileLike.name,
         fileLike.type,
         undefined,
-        progressHandler ? { onUploadProgress: progressHandler } : undefined,
+        axiosUploadConfig,
       );
     }
 
@@ -485,13 +500,7 @@ export class AttachmentManager {
     // eslint-disable-next-line @typescript-eslint/no-unused-vars
     const { duration, ...result } = await this.channel[
       isImageFile(fileLike) ? 'sendImage' : 'sendFile'
-    ](
-      file,
-      undefined,
-      undefined,
-      undefined,
-      progressHandler ? { onUploadProgress: progressHandler } : undefined,
-    );
+    ](file, undefined, undefined, undefined, axiosUploadConfig);
     return result;
   };
 
@@ -537,37 +546,9 @@ export class AttachmentManager {
       return localAttachment;
     }
 
-    const shouldTrackProgress = this.config.trackUploadProgress;
-    const uploadingAttachment: LocalUploadAttachment = {
-      ...attachment,
-      localMetadata: {
-        ...attachment.localMetadata,
-        uploadState: 'uploading',
-        ...(shouldTrackProgress && { uploadProgress: 0 }),
-      },
-    };
-    this.upsertAttachments([uploadingAttachment]);
-
-    const uploadOptions = shouldTrackProgress
-      ? {
-          onProgress: (percent: number | undefined) => {
-            this.updateAttachment({
-              ...uploadingAttachment,
-              localMetadata: {
-                ...uploadingAttachment.localMetadata,
-                uploadProgress: percent,
-              },
-            });
-          },
-        }
-      : undefined;
-
     let response: MinimumUploadRequestResult;
     try {
-      response = await this.doUploadRequest(
-        localAttachment.localMetadata.file,
-        uploadOptions,
-      );
+      response = await this.upload(attachment);
     } catch (error) {
       const reason = error instanceof Error ? error.message : 'unknown error';
       const failedAttachment: LocalUploadAttachment = {
@@ -655,35 +636,10 @@ export class AttachmentManager {
       return preUpload.state.attachment;
     }
 
-    const shouldTrackProgress = this.config.trackUploadProgress;
-    attachment = {
-      ...attachment,
-      localMetadata: {
-        ...attachment.localMetadata,
-        uploadState: 'uploading',
-        ...(shouldTrackProgress && { uploadProgress: 0 }),
-      },
-    };
-    this.upsertAttachments([attachment]);
-
-    const uploadOptions = shouldTrackProgress
-      ? {
-          onProgress: (percent: number | undefined) => {
-            this.updateAttachment({
-              ...attachment,
-              localMetadata: {
-                ...attachment.localMetadata,
-                uploadProgress: percent,
-              },
-            });
-          },
-        }
-      : undefined;
-
     let response: MinimumUploadRequestResult | undefined;
     let error: Error | undefined;
     try {
-      response = await this.doUploadRequest(file, uploadOptions);
+      response = await this.upload(attachment);
     } catch (err) {
       error = err instanceof Error ? err : undefined;
     }
@@ -725,4 +681,44 @@ export class AttachmentManager {
       iterableFiles.slice(0, this.availableUploadSlots).map(this.uploadFile),
     );
   };
+
+  private upload(attachment: LocalUploadAttachment) {
+    const localId = attachment.localMetadata.id;
+
+    this.upsertAttachments([
+      {
+        ...attachment,
+        localMetadata: {
+          ...attachment.localMetadata,
+          uploadState: 'uploading',
+          uploadProgress: this.config.trackUploadProgress ? 0 : undefined,
+        },
+      },
+    ]);
+
+    const unsubscribe = this.client.uploadManager.state.subscribeWithSelector(
+      (s) => ({ upload: s.uploads[localId] }),
+      ({ upload: nextUpload }) => {
+        if (!nextUpload) return;
+        this.updateAttachment({
+          ...attachment,
+          localMetadata: {
+            ...attachment.localMetadata,
+            uploadState: 'uploading',
+            uploadProgress: nextUpload.uploadProgress,
+          },
+        });
+      },
+    );
+
+    return this.client.uploadManager
+      .upload({
+        id: localId,
+        channelCid: this.channel.cid,
+        file: attachment.localMetadata.file,
+      })
+      .finally(() => {
+        unsubscribe();
+      });
+  }
 }

src/messageComposer/configuration/types.ts

@@ -6,9 +6,14 @@ export type MinimumUploadRequestResult = { file: string; thumb_url?: string } &
   Record<string, unknown>
 >;
 
-/** Optional second argument to `UploadRequestFn`; integrators may call `onProgress` to report 0–100 or `undefined` when indeterminate. */
+/**
+ * Optional second argument to `UploadRequestFn`.
+ * - Call `onProgress` to report 0–100 or `undefined` when indeterminate.
+ * - Forward `abortSignal` to your upload implementation to cancel the upload if the user deletes the attachment while it's uploading.
+ */
 export type UploadRequestOptions = {
   onProgress?: (percent: number | undefined) => void;
+  abortSignal?: AbortSignal;
 };
 
 export type UploadRequestFn = (