sdk: 2.0.0-beta.6 - batch sends for unicasts

Author: tsmbl

packages/sdk/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@dialectlabs/sdk",
-  "version": "2.0.0-beta.5",
+  "version": "2.0.0-beta.6",
   "type": "module",
   "repository": "[email protected]:dialectlabs/sdk.git",
   "author": "dialectlabs",

packages/sdk/src/dapp/dapp.interface.ts

@@ -53,6 +53,8 @@ export interface DappAddresses {
 
 export interface DappMessages {
   send(command: SendDappMessageCommand): Promise<void>;
+
+  sendBatch(commands: UnicastDappMessageCommand[]): Promise<void>;
 }
 
 export interface CreateDappCommand {

packages/sdk/src/dialect-cloud-api/data-service-errors.ts

@@ -6,6 +6,7 @@ import {
   IllegalArgumentError,
   ResourceAlreadyExistsError,
   ResourceNotFoundError,
+  TooManyRequestsError,
   UnknownError,
 } from '../sdk/errors';
 import { DialectCloudUnreachableError } from '../internal/errors';
@@ -38,6 +39,9 @@ export async function withErrorParsing<T>(
       if (e.statusCode === 409) {
         throw onResourceAlreadyExists(e);
       }
+      if (e.statusCode === 429) {
+        throw new TooManyRequestsError(createMessage(e));
+      }
       if (e.statusCode === 422) {
         throw new BusinessConstraintViolationError(createMessage(e));
       }

packages/sdk/src/index.ts

@@ -20,6 +20,7 @@ export * from './messaging/messaging.interface';
 export * from './messaging/errors';
 export * from './internal/messaging/data-service-messaging';
 export * from './utils/bytes-utils';
+export * from './utils/collection-utils';
 export * from './sdk/errors';
 export * from './sdk/sdk.interface';
 export * from './sdk/constants';

packages/sdk/src/internal/dapp/alerts-v2-dapp-messages.ts

@@ -16,6 +16,7 @@ import {
   type SendAlertRequest,
 } from '../../dialect-cloud-api/v2/application-api';
 import { withErrorParsing } from '../../dialect-cloud-api/data-service-errors';
+import { chunk } from '../../utils/collection-utils';
 
 export const AddressTypeToChannel = {
   [AddressType.Email]: Channel.Email,
@@ -43,6 +44,59 @@ export class AlertsV2DappMessages implements DappMessages {
     return this.broadcast(command);
   }
 
+  async sendBatch(commands: UnicastDappMessageCommand[]): Promise<void> {
+    if (commands.length === 0) {
+      return;
+    }
+
+    // Convert commands to SendAlertRequest using map and filter, applying same business logic as send method
+    const alerts = commands
+      .filter((command) => command.addressTypes?.length !== 0)
+      .map((command) => this.unicastCommandToAlert(command));
+
+    // If no valid alerts after filtering, return early
+    if (alerts.length === 0) {
+      return;
+    }
+
+    // Split into chunks of max 500 items
+    const chunks = chunk(alerts, 500);
+
+    // Send chunks sequentially
+    for (const alertChunk of chunks) {
+      try {
+        await withErrorParsing(
+          this.api.sendBatch(this.dappId, {
+            alerts: alertChunk,
+          }),
+        );
+      } catch (e) {
+        console.error(
+          `Error during sending batch dapp messages: ${JSON.stringify(e)}`,
+        );
+      }
+    }
+  }
+
+  private unicastCommandToAlert(
+    command: UnicastDappMessageCommand,
+  ): SendAlertRequest {
+    return {
+      recipient: {
+        type: 'subscriber',
+        walletAddress: command.recipient.toString(),
+      },
+      channels: AlertsV2DappMessages.mapChannels(command.addressTypes),
+      topicId: command.notificationTypeId,
+      message: {
+        title: command.title,
+        body: command.message,
+        image: command.imageUrl,
+        actions: AlertsV2DappMessages.mapActions(command.actionsV2),
+      },
+    };
+  }
+
   static mapChannels(addressTypes?: AddressType[]) {
     if (!addressTypes) {
       return [Channel.InApp, Channel.Email, Channel.Telegram];
@@ -73,22 +127,8 @@ export class AlertsV2DappMessages implements DappMessages {
   }
 
   private async unicast(command: UnicastDappMessageCommand) {
-    await withErrorParsing(
-      this.api.sendAlert(this.dappId, {
-        recipient: {
-          type: 'subscriber',
-          walletAddress: command.recipient.toString(),
-        },
-        channels: AlertsV2DappMessages.mapChannels(command.addressTypes),
-        topicId: command.notificationTypeId,
-        message: {
-          title: command.title,
-          body: command.message,
-          image: command.imageUrl,
-          actions: AlertsV2DappMessages.mapActions(command.actionsV2),
-        },
-      }),
-    );
+    const alertRequest = this.unicastCommandToAlert(command);
+    await withErrorParsing(this.api.sendAlert(this.dappId, alertRequest));
   }
 
   private async multicast(command: MulticastDappMessageCommand) {

packages/sdk/src/internal/dapp/dapp-messages-facade.ts

@@ -1,6 +1,7 @@
 import type {
   DappMessages,
   SendDappMessageCommand,
+  UnicastDappMessageCommand,
 } from '../../dapp/dapp.interface';
 import { DialectSdkError, IllegalArgumentError } from '../../sdk/errors';
 
@@ -37,4 +38,29 @@ export class DappMessagesFacade implements DappMessages {
       throw error;
     }
   }
+
+  async sendBatch(commands: UnicastDappMessageCommand[]): Promise<void> {
+    const allSettled = await Promise.allSettled(
+      this.dappMessageBackends.map((it) => it.sendBatch(commands)),
+    );
+    const errors = allSettled
+      .filter((it) => it.status === 'rejected')
+      .map((it) => it as PromiseRejectedResult)
+      .map((it) => it.reason as DialectSdkError);
+    if (errors.length > 0) {
+      console.error(
+        `Error during sending batch dapp messages: ${errors.map((it) =>
+          JSON.stringify(it),
+        )}`,
+      );
+    }
+    const fulfilled = allSettled.filter((it) => it.status === 'fulfilled');
+    if (errors.length > 0 && fulfilled.length === 0) {
+      const error: DialectSdkError = {
+        ...errors[0]!,
+        details: errors,
+      };
+      throw error;
+    }
+  }
 }

packages/sdk/src/internal/dapp/data-service-dapp-messages.ts

@@ -41,22 +41,27 @@ export class DataServiceDappMessages implements DappMessages {
       this.addressTypePredicate,
     );
 
-    if ('recipient' in command) {
-      return this.unicast({
-        ...command,
-        addressTypes: addressTypesFiltered,
-      });
-    }
-    if ('recipients' in command) {
-      return this.multicast({
-        ...command,
-        addressTypes: addressTypesFiltered,
-      });
+    if (addressTypesFiltered.length === 0) {
+      return;
     }
-    return this.broadcast({
+
+    const commandWithAddressTypesFiltered = {
       ...command,
       addressTypes: addressTypesFiltered,
-    });
+    };
+
+    if ('recipient' in commandWithAddressTypesFiltered) {
+      return this.unicast(commandWithAddressTypesFiltered);
+    }
+    if ('recipients' in commandWithAddressTypesFiltered) {
+      return this.multicast(commandWithAddressTypesFiltered);
+    }
+    return this.broadcast(commandWithAddressTypesFiltered);
+  }
+
+  async sendBatch(commands: UnicastDappMessageCommand[]): Promise<void> {
+    console.warn('DataServiceDappMessages.sendBatch - not supported');
+    return;
   }
 
   private async unicast(command: UnicastDappMessageCommand) {

packages/sdk/src/sdk/errors.ts

@@ -72,4 +72,10 @@ export class ResourceNotFoundError extends DialectSdkError {
   }
 }
 
+export class TooManyRequestsError extends DialectSdkError {
+  constructor(msg?: string) {
+    super(TooManyRequestsError.name, 'Error', msg);
+  }
+}
+
 export abstract class IdentityError extends DialectSdkError {}

packages/sdk/src/utils/collection-utils.ts

@@ -4,3 +4,15 @@ export function groupBy<T, K extends keyof any>(arr: T[], key: (i: T) => K) {
     return groups;
   }, {} as Record<K, T[]>);
 }
+
+export function chunk<T>(array: T[], size: number): T[][] {
+  if (size <= 0) {
+    throw new Error('Chunk size must be greater than 0');
+  }
+
+  const chunks: T[][] = [];
+  for (let i = 0; i < array.length; i += size) {
+    chunks.push(array.slice(i, i + size));
+  }
+  return chunks;
+}