add readAt and delete message

src/app/module/chat/data/async/socket/message-async.service.ts

@@ -6,6 +6,7 @@ import { SignalRService } from '../../../infra/socket/signal-r.service';
 import { filter } from 'rxjs/operators';
 import { InstanceId } from '../../repository/message-respository.service';
 import { SafeValidateSchema } from 'src/app/services/decorators/validate-schema.decorator';
+import { MessageOutPutDataDTO } from '../../dto/message/messageOutputDTO';
 
 @Injectable({
   providedIn: 'root'
@@ -20,27 +21,67 @@ export class MessageAsyncService {
   ) {
 
     this.messageLiveSignalRDataSourceService.getMessage().pipe(
-      filter((message: any) => {
-        return !message?.requestId?.startsWith(InstanceId) && message?.requestId
+      filter((message) => {
+        return !message?.requestId?.startsWith(InstanceId)
       })
     ).subscribe(async (message) => {
 
-      console.log('message async ', message)
+      if(message?.id) {
 
-      const id = message.id + ''
-      delete message.id;
+        console.log('message async ', message)
 
-      const incomingMessage =  {
-        ...message,
-        messageId: id,
-        sending: false,
-        roomId:message.chatRoomId
+        const id = message.id + ''
+        delete message.id;
+  
+        const incomingMessage =  {
+          ...message,
+          messageId: id,
+          sending: false,
+          roomId:message.chatRoomId
+        }
+  
+        this.incomingMessage(incomingMessage)
       }
 
-      this.incomingMessage(incomingMessage)
 
     })
 
+
+    this.messageLiveSignalRDataSourceService.getMessageUpdate().pipe(
+      filter((message) => {
+        return !message?.requestId?.startsWith(InstanceId)
+      })
+    ).subscribe(async (message) => {
+
+      if(message?.id) {
+
+        console.log('message async ', message)
+
+        const id = message.id + ''
+        delete message.id;
+  
+        const incomingMessage =  {
+          ...message,
+          messageId: id,
+          sending: false,
+          roomId:message.chatRoomId
+        }
+  
+        this.incomingMessage(incomingMessage)
+      }
+
+
+    })
+
+    this.messageLiveSignalRDataSourceService.getMessageDelete()
+    .pipe()
+    .subscribe(async (message) => {
+      if(message.id) {
+        this.incomingDeleted(message)
+      }
+      
+    })
+
   }
 
   @SafeValidateSchema(IncomingMessageSchema, 'socket/incomingMessage')
@@ -54,4 +95,15 @@ export class MessageAsyncService {
       console.log(result.error)
     }
   }
+
+
+  async incomingDeleted(data: MessageOutPutDataDTO) {
+    const result = await this.messageLocalDataSourceService.deleteByMessageId(data.id)
+
+    if(result.isOk()) {
+
+    } else {
+      console.log(result.error)
+    }
+  }
 }

src/app/module/chat/data/data-source/message/message-local-data-source.service.ts

@@ -88,6 +88,19 @@ export class MessageLocalDataSourceService {
     }
   }
 
+  async deleteByMessageId(messageId: string): Promise<Result<undefined|TableMessage, any>> {
+    try {
+      console.log(messageId)
+      const lastMessage = await messageDataSource.message
+        .where('messageId')
+        .equals(messageId).delete()
+
+      return ok(lastMessage[0]); // Get the last message
+    } catch (error) {
+      return err(error);
+    }
+  }
+
 
   async sendMessage(data: MessageInputDTO) {
 
@@ -121,6 +134,25 @@ export class MessageLocalDataSourceService {
   }
 
 
+  async messageExist({messageId}) {
+    try {
+      const existingMessage = await messageDataSource.message
+      .where('messageId')
+      .equals(messageId)
+      .first();
+
+      if (existingMessage) {
+        return ok(true)
+      } else {
+        return err(false)
+      }
+
+    } catch (error) {
+      return err(false);
+    }
+
+  }
+
   async update(data: TableMessage ) {
 
     try {

src/app/module/chat/data/dto/message/messageDeleteInputDTO.ts

@@ -0,0 +1,10 @@
+import { z } from "zod";
+
+export const MessageDeleteInputDTOSchema = z.object({
+  requestId: z.string(),
+  roomId: z.string(),
+  messageId: z.string(),
+  senderId: z.number(),
+});
+
+export type MessageDeleteInputDTO = z.infer<typeof MessageDeleteInputDTOSchema>

src/app/module/chat/data/dto/message/messageOutputDTO.ts

@@ -16,7 +16,8 @@ const DataSchema = z.object({
   deliverAt: z.string().datetime().nullable(),
   canEdit: z.boolean(),
   oneShot: z.boolean(),
-  requireUnlock: z.boolean()
+  requireUnlock: z.boolean(),
+  requestId: z.string()
 });
 
 

src/app/module/chat/data/repository/message-respository.service.ts

@@ -8,6 +8,7 @@ import { SignalRService } from '../../infra/socket/signal-r.service';
 import { v4 as uuidv4 } from 'uuid'
 import { filter } from 'rxjs/operators';
 import { err, ok } from 'neverthrow';
+import { MessageDeleteInputDTO } from '../../domain/use-case/message-delete-live-use-case.service';
 
 export const InstanceId = uuidv4();
 
@@ -81,12 +82,20 @@ export class MessageRepositoryService {
     }
   }
 
+
+  sendMessageDelete(data: MessageDeleteInputDTO) {
+    
+    data['requestId'] = InstanceId +'@'+ uuidv4();
+
+    return this.messageLiveSignalRDataSourceService.sendMessageDelete(data)
+  }
+
   async sendReadAt({roomId}) {
     const result = await this.messageLocalDataSourceService.getLastMessageByRoomId(roomId)
     if(result.isOk()) {
       if(result.value) {
-
-        // return await this.messageLiveSignalRDataSourceService.sendReadAt({roomId, memberId: SessionStore.user.UserId, chatMessageId: result.value.messageId})
+        
+        return await this.messageLiveSignalRDataSourceService.sendReadAt({roomId, memberId: SessionStore.user.UserId, chatMessageId: result.value.messageId})
       }
       return ok(true)
     }