[PAY-3307] Upgrade chat blasts to chats in sdk

.changeset/chatty-cameras-yawn.md

@@ -0,0 +1,5 @@
+---
+'@audius/sdk': patch
+---
+
+Upgrade chat blasts to real chats internally

comms/discovery/rpcz/apply.go

@@ -449,7 +449,9 @@ func websocketNotify(rpcJson json.RawMessage, userId int32, timestamp time.Time)
 			Metadata schema.Metadata `json:"metadata"`
 		}{
 			rpcJson,
-			schema.Metadata{Timestamp: timestamp.Format(time.RFC3339Nano), UserID: encodedUserId},
+			schema.Metadata{Timestamp: timestamp.Format(time.RFC3339Nano),
+				// Note this is the userId of the user sending the message
+				UserID: encodedUserId},
 		}
 
 		j, err := json.Marshal(data)
@@ -462,6 +464,8 @@ func websocketNotify(rpcJson json.RawMessage, userId int32, timestamp time.Time)
 			websocketPush(subscribedUserId, j)
 		}
 
+	} else if gjson.GetBytes(rpcJson, "method").String() == "chat.blast" {
+		websocketPushAll(rpcJson, timestamp)
 	}
 }
 

comms/discovery/rpcz/websocket.go

@@ -1,10 +1,13 @@
 package rpcz
 
 import (
+	"encoding/json"
 	"net"
 	"sync"
 	"time"
 
+	"comms.audius.co/discovery/misc"
+	"comms.audius.co/discovery/schema"
 	"github.com/gobwas/ws"
 	"github.com/gobwas/ws/wsutil"
 	"golang.org/x/exp/slog"
@@ -96,3 +99,36 @@ func websocketPush(userId int32, payload []byte) {
 	}
 
 }
+
+func websocketPushAll(rpcJson json.RawMessage, timestamp time.Time) {
+	mu.Lock()
+	defer mu.Unlock()
+
+	for _, s := range websockets {
+		encodedUserId, _ := misc.EncodeHashId(int(s.userId))
+
+		data := struct {
+			RPC      json.RawMessage `json:"rpc"`
+			Metadata schema.Metadata `json:"metadata"`
+		}{
+			rpcJson,
+			schema.Metadata{Timestamp: timestamp.Format(time.RFC3339Nano),
+				// Note this is the userId of the user receiving the message
+				UserID: encodedUserId},
+		}
+
+		payload, err := json.Marshal(data)
+		if err != nil {
+			logger.Warn("invalid websocket json " + err.Error())
+			return
+		}
+
+		err = wsutil.WriteServerMessage(s.conn, ws.OpText, payload)
+		if err != nil {
+			logger.Info("websocket push failed: " + err.Error())
+			removeWebsocket(s)
+		} else {
+			logger.Debug("websocket push all", "payload", string(payload))
+		}
+	}
+}

packages/common/src/store/pages/chat/sagas.ts

@@ -138,8 +138,13 @@ function* doFetchLatestChats() {
     let hasMoreChats = true
     let data: UserChat[] = []
     let firstResponse: TypedCommsResponse<UserChat[]> | undefined
+    const currentUserId = yield* select(getUserId)
+    if (!currentUserId) {
+      throw new Error('User not found')
+    }
     while (hasMoreChats) {
       const response = yield* call([sdk.chats, sdk.chats.getAll], {
+        userId: encodeHashId(currentUserId)!,
         before,
         after: summary?.next_cursor,
         limit: CHAT_PAGE_SIZE
@@ -175,7 +180,12 @@ function* doFetchMoreChats() {
     const sdk = yield* call(audiusSdk)
     const summary = yield* select(getChatsSummary)
     const before = summary?.prev_cursor
+    const currentUserId = yield* select(getUserId)
+    if (!currentUserId) {
+      throw new Error('User not found')
+    }
     const response = yield* call([sdk.chats, sdk.chats.getAll], {
+      userId: encodeHashId(currentUserId)!,
       before,
       limit: CHAT_PAGE_SIZE
     })

packages/libs/src/sdk/api/chats/ChatsApi.ts

@@ -11,6 +11,7 @@ import type { AuthService } from '../../services/Auth'
 import type { DiscoveryNodeSelectorService } from '../../services/DiscoveryNodeSelector/types'
 import type { LoggerService } from '../../services/Logger'
 import type { EventEmitterTarget } from '../../utils/EventEmitterTarget'
+import { encodeHashId } from '../../utils/hashId'
 import { parseParams } from '../../utils/parseParams'
 import {
   BaseAPI,
@@ -64,10 +65,12 @@ import type {
   ChatMessage,
   ChatWebsocketEventData,
   RPCPayloadRequest,
-  ValidatedChatPermissions
+  ValidatedChatPermissions,
+  ChatBlast,
+  ChatCreateRPC
 } from './serverTypes'
 
-const GENERIC_MESSAGE_ERROR = 'Error: this message can not be displayed'
+const GENERIC_MESSAGE_ERROR = 'Error: this message cannot be displayed'
 
 export class ChatsApi
   extends BaseAPI
@@ -173,14 +176,18 @@ export class ChatsApi
    * @param params.limit the max number of chats to get
    * @param params.before a timestamp cursor for pagination
    * @param params.after a timestamp cursor for pagination
-   * @param params.currentUserId the user to act on behalf of
+   * @param params.userId the user to act on behalf of
    * @returns the chat list response
    */
   public async getAll(params?: ChatGetAllRequest) {
-    const { currentUserId, limit, before, after } = await parseParams(
+    const { userId, limit, before, after } = await parseParams(
       'getAll',
       ChatGetAllRequestSchema
     )(params)
+
+    // Get new blasts and upgrade them to chats
+    this.upgradeBlasts(userId)
+
     const path = `/comms/chats`
     const query: HTTPQuery = {
       timestamp: new Date().getTime()
@@ -194,8 +201,8 @@ export class ChatsApi
     if (after) {
       query.after = after
     }
-    if (currentUserId) {
-      query.current_user_id = currentUserId
+    if (userId) {
+      query.current_user_id = userId
     }
     const response = await this.signAndSendRequest({
       method: 'GET',
@@ -282,6 +289,23 @@ export class ChatsApi
     }
   }
 
+  /**
+   * Gets a list of chat blasts for which chats haven't been created yet
+   * @returns the blast messages list response
+   */
+  public async getBlasts(): Promise<TypedCommsResponse<ChatBlast[]>> {
+    const query: HTTPQuery = {
+      timestamp: new Date().getTime()
+    }
+    const res = await this.signAndSendRequest({
+      method: 'GET',
+      path: `/comms/blasts`,
+      headers: {},
+      query
+    })
+    return (await res.json()) as TypedCommsResponse<ChatBlast[]>
+  }
+
   /**
    * Gets the total unread message count for a user
    * @param params.currentUserId the user to act on behalf of
@@ -418,7 +442,7 @@ export class ChatsApi
    * @param params.currentUserId the user to act on behalf of
    * @returns the rpc object
    */
-  public async create(params: ChatCreateRequest) {
+  public async create(params: ChatCreateRequest): Promise<ChatCreateRPC> {
     const { currentUserId, userId, invitedUserIds } = await parseParams(
       'create',
       ChatCreateRequestSchema
@@ -428,14 +452,14 @@ export class ChatsApi
     const chatSecret = secp.utils.randomPrivateKey()
     const invites = await this.createInvites(userId, invitedUserIds, chatSecret)
 
-    return await this.sendRpc({
+    return (await this.sendRpc({
       current_user_id: currentUserId,
       method: 'chat.create',
       params: {
         chat_id: chatId,
         invites
       }
-    })
+    })) as ChatCreateRPC
   }
 
   /**
@@ -767,6 +791,31 @@ export class ChatsApi
     return base64.decode(json.data)
   }
 
+  private async upgradeBlasts(userId: string) {
+    const blasts = await this.getBlasts()
+    Promise.all(
+      blasts.data.map(async (blast) => {
+        const encodedSenderId = encodeHashId(blast.from_user_id)
+        if (encodedSenderId) {
+          await this.create({
+            userId,
+            invitedUserIds: [encodedSenderId]
+          })
+          this.eventEmitter.emit('message', {
+            chatId: blast.pending_chat_id,
+            message: {
+              message_id: blast.pending_chat_id + blast.chat_id,
+              message: blast.plaintext,
+              sender_user_id: encodedSenderId,
+              created_at: blast.created_at,
+              reactions: []
+            }
+          })
+        }
+      })
+    )
+  }
+
   private async getSignatureHeader(payload: string) {
     const [allSignatureBytes, recoveryByte] = await this.auth.sign(payload)
     const signatureBytes = new Uint8Array(65)
@@ -853,6 +902,9 @@ export class ChatsApi
               created_at: data.metadata.timestamp
             }
           })
+        } else if (data.rpc.method === 'chat.blast') {
+          const userId = data.metadata.userId
+          await this.upgradeBlasts(userId)
         }
       }
       handleAsync()

packages/libs/src/sdk/api/chats/clientTypes.ts

@@ -19,7 +19,7 @@ export const ChatListenRequestSchema = z.optional(
 export type ChatListenRequest = z.infer<typeof ChatListenRequestSchema>
 
 export const ChatGetAllRequestSchema = z.object({
-  currentUserId: z.optional(z.string()),
+  userId: z.string(),
   limit: z.optional(z.number()),
   before: z.optional(z.string()),
   after: z.optional(z.string())

packages/libs/src/sdk/api/chats/serverTypes.ts

@@ -156,6 +156,16 @@ export type ChatInvite = {
   invite_code: string
 }
 
+export type ChatBlast = {
+  chat_id: string
+  pending_chat_id: string
+  from_user_id: number
+  plaintext: string
+  created_at: string
+  audience: ChatBlastAudience
+  audience_track_id?: number
+}
+
 export type ValidatedChatPermissions = {
   user_id: string
   permits: ChatPermission