feat: update message pending tasks (#1716)

## 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?

This PR introduces support for adding message updates as a pending task
whenever we are offline and persisting it. It goes hand in hand with an
extended pending task API through `updatePendingTask` (living in the UI
SDK implementation currently).

The flow is as follows:

- If we're online, we simply update the message
- If we're offline and the message is an already successful one, we
simply queue up the pending task
- If we're offline and the message is a failed one (for example, we send
a message while offline and then immediately edit it), we update the
`send-message` pending task with the new data
- When this happens, we make sure to not change the
`message_text_changed_at` property so that the newly sent message does
not appear as edited

This comes with a couple of caveats:

- Attachments are out of scope, since currently our async uploads
feature still lives on the UI SDK (once that gets moved to the LLC we
can support that too)
  - Offline updates containing attachments will simply be skipped
- The optimistic updates still live in the UI SDK, their move towards
the LLC will come in the next 2 quarters gradually

## Changelog

-

Author: isekovanic

src/client.ts

@@ -265,6 +265,7 @@ import { ReminderManager } from './reminders';
 import { StateStore } from './store';
 import type { MessageComposer } from './messageComposer';
 import type { AbstractOfflineDB } from './offline-support';
+import { getPendingTaskChannelData } from './offline-support/util';
 
 function isString(x: unknown): x is string {
   return typeof x === 'string' || x instanceof String;
@@ -3114,6 +3115,38 @@ export class StreamChat {
       throw Error('Please specify the message.id when calling updateMessage');
     }
 
+    const messageId = message.id as string;
+
+    try {
+      if (this.offlineDb) {
+        return await this.offlineDb.queueTask<UpdateMessageAPIResponse>({
+          task: {
+            ...getPendingTaskChannelData(message.cid),
+            messageId,
+            payload: [message, partialUserOrUserId, options],
+            type: 'update-message',
+          },
+        });
+      }
+    } catch (error) {
+      this.logger('error', `offlineDb:updateMessage`, {
+        tags: ['channel', 'offlineDb'],
+        error,
+      });
+    }
+
+    return await this._updateMessage(message, partialUserOrUserId, options);
+  }
+
+  async _updateMessage(
+    message: LocalMessage | Partial<MessageResponse>,
+    partialUserOrUserId?: string | { id: string },
+    options?: UpdateMessageOptions,
+  ) {
+    if (!message.id) {
+      throw Error('Please specify the message.id when calling updateMessage');
+    }
+
     // should not include user object
     const payload = toUpdatedMessagePayload(message);
 

src/offline-support/offline_support_api.ts

@@ -1,4 +1,11 @@
-import type { APIErrorResponse, ChannelResponse, Event } from '../types';
+import type {
+  APIErrorResponse,
+  ChannelResponse,
+  Event,
+  LocalMessage,
+  Message,
+  MessageResponse,
+} from '../types';
 
 import type {
   OfflineDBApi,
@@ -11,7 +18,8 @@ import type { StreamChat } from '../client';
 import type { AxiosError } from 'axios';
 import { OfflineDBSyncManager } from './offline_sync_manager';
 import { StateStore } from '../store';
-import { runDetached } from '../utils';
+import { localMessageToNewMessagePayload, runDetached } from '../utils';
+import { isMessageUpdateReplayable } from './util';
 
 /**
  * Abstract base class for an offline database implementation used with StreamChat.
@@ -310,6 +318,16 @@ export abstract class AbstractOfflineDB implements OfflineDBApi {
    */
   abstract addPendingTask: OfflineDBApi['addPendingTask'];
 
+  /**
+   * @abstract
+   * Updates a pending task in the DB, given its ID.
+   * Will return the prepared queries for delayed execution (even if they are
+   * already executed).
+   * @param {DBUpdatePendingTaskType} options
+   * @returns {Promise<ExecuteBatchDBQueriesType>}
+   */
+  abstract updatePendingTask: OfflineDBApi['updatePendingTask'];
+
   /**
    * @abstract
    * Deletes a pending task from the DB, given its ID.
@@ -1076,7 +1094,7 @@ export abstract class AbstractOfflineDB implements OfflineDBApi {
       return await attemptTaskExecution();
     } catch (e) {
       if (!this.shouldSkipQueueingTask(e as AxiosError<APIErrorResponse>)) {
-        await this.addPendingTask(task);
+        await this.handleAddPendingTask({ task });
       }
       throw e;
     }
@@ -1092,13 +1110,112 @@ export abstract class AbstractOfflineDB implements OfflineDBApi {
   private shouldSkipQueueingTask = (error: AxiosError<APIErrorResponse>) =>
     error?.response?.data?.code === 4 || error?.response?.data?.code === 17;
 
+  private mergeFailedMessageUpdateIntoPendingSendMessage = ({
+    editedMessage,
+    pendingMessage,
+  }: {
+    editedMessage: LocalMessage | Partial<MessageResponse>;
+    pendingMessage: Message;
+  }) => {
+    const normalizedEditedMessageSource = {
+      ...editedMessage,
+    } as LocalMessage & { message_text_updated_at?: string };
+
+    if (editedMessage.status === 'failed') {
+      delete normalizedEditedMessageSource.message_text_updated_at;
+    }
+
+    const normalizedEditedMessage = localMessageToNewMessagePayload(
+      normalizedEditedMessageSource,
+    );
+    const pendingMessageStatus = (pendingMessage as { status?: string }).status;
+
+    return {
+      ...pendingMessage,
+      ...normalizedEditedMessage,
+      ...(typeof pendingMessageStatus !== 'undefined'
+        ? { status: pendingMessageStatus }
+        : {}),
+    } as Message;
+  };
+
+  private isPendingSendMessageTask = (
+    task: PendingTask,
+  ): task is Extract<PendingTask, { type: 'send-message' }> =>
+    task.type === 'send-message';
+
+  private handleOfflineFailedUpdateMessagePendingTask = async (
+    task: Extract<PendingTask, { type: 'update-message' }>,
+  ) => {
+    const [message] = task.payload;
+    if (!message.id) {
+      return;
+    }
+
+    const pendingTasks = await this.getPendingTasks({ messageId: message.id });
+    const pendingSendMessageTask = pendingTasks.find(this.isPendingSendMessageTask);
+
+    if (!pendingSendMessageTask) {
+      return;
+    }
+
+    const updatedPendingSendMessage = this.mergeFailedMessageUpdateIntoPendingSendMessage(
+      {
+        editedMessage: message,
+        pendingMessage: pendingSendMessageTask.payload[0],
+      },
+    );
+
+    const updatedPendingTask: Extract<PendingTask, { type: 'send-message' }> = {
+      ...pendingSendMessageTask,
+      payload: [updatedPendingSendMessage, pendingSendMessageTask.payload[1]],
+    };
+
+    if (pendingSendMessageTask.id) {
+      await this.updatePendingTask({
+        id: pendingSendMessageTask.id,
+        task: updatedPendingTask,
+      });
+      return;
+    }
+
+    await this.addPendingTask({
+      ...updatedPendingTask,
+      id: undefined,
+    });
+  };
+
+  /**
+   * Central ingress for persisting pending tasks. It either stores the task as-is
+   * or rewrites an existing pending `send-message` task for offline edits of failed messages.
+   */
+  public handleAddPendingTask = async ({ task }: { task: PendingTask }) => {
+    if (task.type === 'update-message' && !isMessageUpdateReplayable(task.payload[0])) {
+      return;
+    }
+
+    if (
+      task.type === 'update-message' &&
+      !this.client.wsConnection?.isHealthy &&
+      task.payload[0].status === 'failed'
+    ) {
+      await this.handleOfflineFailedUpdateMessagePendingTask(task);
+      return;
+    }
+
+    await this.addPendingTask(task);
+  };
+
   /**
    * Executes a task from the list of supported pending tasks. Currently supported pending tasks
    * are:
+   * - Updating a message
    * - Deleting a message
    * - Sending a reaction
    * - Removing a reaction
    * - Sending a message
+   * - Creating a draft
+   * - Deleting a draft
    * It will throw if we try to execute a pending task that is not supported.
    * @param task - The task we want to execute
    * @param isPendingTask - a control value telling us if it's an actual pending task being executed
@@ -1108,6 +1225,10 @@ export abstract class AbstractOfflineDB implements OfflineDBApi {
     { task }: { task: PendingTask },
     isPendingTask = false,
   ) => {
+    if (task.type === 'update-message') {
+      return await this.client._updateMessage(...task.payload);
+    }
+
     if (task.type === 'delete-message') {
       return await this.client._deleteMessage(...task.payload);
     }

src/offline-support/types.ts

@@ -227,6 +227,16 @@ export type DBDeletePendingTaskType = {
   id: number;
 };
 
+/**
+ * Update a pending task by ID.
+ */
+export type DBUpdatePendingTaskType = {
+  /** ID of the pending task. */
+  id: number;
+  /** The next task payload to persist. */
+  task: PendingTask;
+};
+
 /**
  * Options to delete a reaction from a message.
  */
@@ -372,6 +382,9 @@ export interface OfflineDBApi {
   addPendingTask: (task: PendingTask) => Promise<() => Promise<void>>;
   getPendingTasks: (conditions?: DBGetPendingTasksType) => Promise<PendingTask[]>;
   deleteDraft: (options: DBDeleteDraftType) => Promise<ExecuteBatchDBQueriesType>;
+  updatePendingTask: (
+    options: DBUpdatePendingTaskType,
+  ) => Promise<ExecuteBatchDBQueriesType>;
   deletePendingTask: (
     options: DBDeletePendingTaskType,
   ) => Promise<ExecuteBatchDBQueriesType>;
@@ -397,6 +410,7 @@ export type OfflineDBState = {
 };
 
 export type PendingTaskTypes = {
+  updateMessage: 'update-message';
   deleteMessage: 'delete-message';
   deleteReaction: 'delete-reaction';
   sendReaction: 'send-reaction';
@@ -417,6 +431,10 @@ export type PendingTask = {
       payload: Parameters<Channel['sendReaction']>;
       type: PendingTaskTypes['sendReaction'];
     }
+  | {
+      payload: Parameters<StreamChat['updateMessage']>;
+      type: PendingTaskTypes['updateMessage'];
+    }
   | {
       payload: Parameters<StreamChat['deleteMessage']>;
       type: PendingTaskTypes['deleteMessage'];

src/offline-support/util.ts

@@ -0,0 +1,32 @@
+import type { Attachment, LocalMessage, MessageResponse } from '../types';
+
+export const isLocalUrl = (value: string | undefined) =>
+  !!value && !value.startsWith('http');
+
+export const isAttachmentReplayable = (attachment: Attachment) => {
+  if (!attachment || typeof attachment !== 'object') {
+    return true;
+  }
+
+  return !isLocalUrl(attachment.asset_url) && !isLocalUrl(attachment.image_url);
+};
+
+export const isMessageUpdateReplayable = (
+  message: LocalMessage | Partial<MessageResponse>,
+) => !message.attachments?.some((attachment) => !isAttachmentReplayable(attachment));
+
+export const getPendingTaskChannelData = (cid?: string) => {
+  if (!cid) {
+    return {};
+  }
+
+  const separatorIndex = cid.indexOf(':');
+  if (separatorIndex <= 0 || separatorIndex === cid.length - 1) {
+    return {};
+  }
+
+  return {
+    channelId: cid.slice(separatorIndex + 1),
+    channelType: cid.slice(0, separatorIndex),
+  };
+};