send direct message

src/app/module/chat/domain/chat-service.service.ts

@@ -25,6 +25,7 @@ import { SendLocalMessagesUseCaseService } from './use-case/messages-send-offlin
 import { RemoveMemberUseCaseService } from './use-case/member/-use-case.service'
 import { AddMemberUseCaseService } from './use-case/member-add-use-case.service'
 import { UpdateRoomByIdUseCaseService } from './use-case/room-update-by-id-use-case.service'
+import { MessageMarkAsReadUseCaseService } from './use-case/message/message-mark-as-read-use-case.service'
 import { GetMessageAttachmentLocallyUseCaseService } from 'src/app/module/chat/domain/use-case/message-get-attachment-localy-use-case.service';
 import { GetRoomListUseCaseService } from 'src/app/module/chat/domain/use-case/room-get-list-use-case.service';
 import { filter } from 'rxjs/operators';
@@ -36,6 +37,7 @@ import { UserRemoveListInputDTO } from '../data/dto/room/userRemoveListInputDTO'
 import { AddMemberToRoomInputDTO } from '../data/dto/room/addMemberToRoomInputDto';
 import { RoomUpdateInputDTO } from '../data/dto/room/roomUpdateInputDTO';
 import { RoomType } from "src/app/module/chat/domain/entity/group";
+import { sendReadAt } from "src/app/module/chat/data/repository/message/message-live-signalr-data-source.service";
 
 export const InstanceId = uuidv4();
 
@@ -73,7 +75,8 @@ export class ChatServiceService {
     private UpdateRoomByIdUseCaseService: UpdateRoomByIdUseCaseService,
     private RemoveMemberUseCaseService: RemoveMemberUseCaseService,
     private MessageReadAtByIdUseCaseService: MessageReadAtByIdUseCaseService,
-    private SendLocalMessagesUseCaseService: SendLocalMessagesUseCaseService
+    private SendLocalMessagesUseCaseService: SendLocalMessagesUseCaseService,
+    private MessageMarkAsReadUseCaseService: MessageMarkAsReadUseCaseService
   ) {
     this.messageLiveSignalRDataSourceService.getMessageDelete()
     .pipe()
@@ -102,7 +105,6 @@ export class ChatServiceService {
       })
     ).subscribe(async (message) => {
       if(message?.id) {
-        console.log('create message')
         this.SocketMessageCreateUseCaseService.execute(message)
       }
     })
@@ -115,7 +117,6 @@ export class ChatServiceService {
         }
       })
     ).subscribe(async (message) => {
-      console.log('123', message)
       this.MemberListUpdateStatusUseCaseService.execute(message.data as any)
     })
 
@@ -213,6 +214,10 @@ export class ChatServiceService {
 
   }
 
+  sendReadAt(sendReadAt: sendReadAt) {
+    return this.MessageMarkAsReadUseCaseService.execute(sendReadAt)
+  }
+
   listenToIncomingMessage(roomId:string) {
     return this.ListenMessageByRoomIdNewUseCase.execute({roomId})
   }

src/app/module/chat/domain/entity/message.ts

@@ -14,12 +14,18 @@ export const MessageEntitySchema = z.object({
   oneShot: z.boolean(),
   sentAt: z.string().optional(),
   requireUnlock: z.boolean(),
+  editedAt: z.string().nullable().optional(),
   sender: z.object({
     wxUserId: z.number(),
     wxFullName: z.string(),
     wxeMail: z.string(),
     userPhoto: z.string(),
   }),
+  info: z.array(z.object({
+    memberId: z.number(),
+    readAt: z.string().nullable(),
+    deliverAt: z.string().nullable()
+  })).optional(),
   sending: z.boolean().optional(),
   attachments: z.array(z.object({
     fileType: z.nativeEnum(MessageAttachmentFileType),
@@ -47,6 +53,11 @@ export class MessageEntity implements Message  {
   oneShot: boolean = false
   sentAt: string
   requireUnlock: boolean = false
+  info: {
+    memberId?: number
+    readAt?: string,
+    deliverAt?: string
+  }[] = []
   sender: {
     wxUserId: number,
     wxFullName: string,

src/app/module/chat/domain/service/room-expiration.service.ts

@@ -0,0 +1,9 @@
+import { Injectable } from '@angular/core';
+
+@Injectable({
+  providedIn: 'root'
+})
+export class RoomExpirationService {
+
+  constructor() { }
+}

src/app/module/chat/domain/use-case/listen-send-message.service.ts

@@ -14,8 +14,12 @@ export class ListenSendMessageUseCase {
 
   execute({roomId}: {roomId: string}) {
 
+    console.log('reciee message')
     return this.MessageSocketRepositoryService.listenToMessages().pipe(
-      filter((message) => message?.requestId?.startsWith(InstanceId) && message?.roomId == roomId),
+      filter((message) => {
+        console.log(message, roomId)
+        return message?.requestId?.startsWith(InstanceId) && message?.roomId == roomId
+      }),
       map(message => message)
     )
 

src/app/module/chat/domain/use-case/message-create-use-case.service.ts

@@ -15,8 +15,9 @@ import { MessageTable } from '../../infra/database/dexie/schema/message';
 import { MessageMapper } from '../mapper/messageMapper';
 import { SignalRService } from '../../infra/socket/signal-r.service';
 import { RoomType } from "src/app/module/chat/domain/entity/group";
-
-
+import { TracingType, XTracerAsync } from 'src/app/services/monitoring/opentelemetry/tracer';
+import { MemberListLocalRepository } from 'src/app/module/chat/data/repository/member-list-local-repository.service'
+import { SessionStore } from 'src/app/store/session.service';
 const MessageInputUseCaseSchema = z.object({
   memberId: z.number(),
   roomId: z.string(),
@@ -39,11 +40,13 @@ export class MessageCreateUseCaseService {
     private AttachmentLocalRepositoryService: AttachmentLocalDataSource,
     private messageLocalDataSourceService: MessageLocalDataSourceService,
     private messageLiveSignalRDataSourceService: SignalRService,
-    private messageSocketRepositoryService: MessageSocketRepositoryService
+    private messageSocketRepositoryService: MessageSocketRepositoryService,
+    private MemberListLocalRepository: MemberListLocalRepository
   ) { }
 
 
-  async execute(message: MessageEntity, messageEnum: RoomType) {
+  @XTracerAsync({name:'MessageCreateUseCaseService', module:'chat',  bugPrint: true})
+  async execute(message: MessageEntity, messageEnum: RoomType, tracing?: TracingType) {
 
     const validation = zodSafeValidation<MessageEntity>(MessageEntitySchema, message)
 
@@ -93,15 +96,32 @@ export class MessageCreateUseCaseService {
         //====================
         message.sending = true
 
-        const DTO = MessageMapper.fromDomain(message, message.requestId)
-
         let sendMessageResult: Result<MessageOutPutDataDTO, any>
         if(messageEnum == RoomType.Group) {
+          const DTO = MessageMapper.fromDomain(message, message.requestId)
           sendMessageResult = await this.messageLiveSignalRDataSourceService.sendMessage<MessageOutPutDataDTO>(DTO)
         } else {
-          sendMessageResult = await this.messageSocketRepositoryService.sendDirectMessage(DTO)
+
+          if(message.receiverId) {
+            const DTO = MessageMapper.fromDomain(message, message.requestId)
+            sendMessageResult = await this.messageSocketRepositoryService.sendDirectMessage(DTO)
+          } else {
+            const getRoomMembers = await this.MemberListLocalRepository.directMember({
+              roomId:message.roomId,
+              currentUserId: SessionStore.user.UserId
+            })
+            if(getRoomMembers.isOk()) {
+              message.receiverId = getRoomMembers.value.wxUserId
+              const DTO = MessageMapper.fromDomain(message, message.requestId)
+              sendMessageResult = await this.messageSocketRepositoryService.sendDirectMessage(DTO)
+            } else {
+              console.log('not found direct users', getRoomMembers.error)
+            }
+          }
+
+
         }
-        
+
         // return this sendMessageResult
 
         if(sendMessageResult.isOk()) {
@@ -127,7 +147,7 @@ export class MessageCreateUseCaseService {
           await this.messageLocalDataSourceService.update(message.$id, {sending: false, $id: message.$id})
           return err('no connection')
         }
-        
+
       }
     } else {
 
@@ -136,6 +156,11 @@ export class MessageCreateUseCaseService {
           zodErrorList: validation.error.errors,
           data: message.attachments
         })
+      } else {
+        Logger.error('failed to send message, validation failed', {
+          zodErrorList: validation.error.errors,
+          data: message
+        })
       }
 
     }

src/app/module/chat/domain/use-case/message-delete-by-id-live-use-case.service.ts

@@ -3,8 +3,7 @@ import { z } from 'zod';
 import { SafeValidateSchema, ValidateSchema } from 'src/app/services/decorators/validate-schema.decorator';
 import { MessageRemoteDataSourceService } from '../../data/repository/message/message-remote-data-source.service';
 import { MessageSocketRepositoryService } from '../../data/repository/message/message-live-signalr-data-source.service';
-
-
+import { XTracerAsync, TracingType } from 'src/app/services/monitoring/opentelemetry/tracer';
 export const MessageDeleteInputDTOSchema = z.object({
   requestId: z.string().optional(),
   roomId: z.string(),
@@ -22,8 +21,11 @@ export class MessageDeleteLiveUseCaseService {
     public repository: MessageSocketRepositoryService
   ) { }
 
-  @SafeValidateSchema(MessageDeleteInputDTOSchema, 'MessageDeleteUseCaseService')
-  async execute(data: MessageDeleteInputDTO) {
+  @XTracerAsync({name:'MessageDeleteLiveUseCaseService', module:'chat',  bugPrint: true})
+  async execute(data: MessageDeleteInputDTO, tracing?: TracingType) {
+    tracing.log('MessageDeleteLiveUseCaseService payload', {
+      data: data
+    })
     return this.repository.sendMessageDelete(data)
   }
 }

src/app/module/chat/domain/use-case/message/message-mark-as-read-use-case.service.ts

@@ -0,0 +1,18 @@
+import { Injectable } from '@angular/core';
+import { MessageSocketRepositoryService, sendReadAt } from '../../../data/repository/message/message-live-signalr-data-source.service';
+import { XTracerAsync, TracingType } from 'src/app/services/monitoring/opentelemetry/tracer';
+@Injectable({
+  providedIn: 'root'
+})
+export class MessageMarkAsReadUseCaseService {
+
+  constructor(
+    private MessageSocketRepositoryService: MessageSocketRepositoryService,
+  ) { }
+
+
+  @XTracerAsync({name:'MessageMarkAsReadUseCaseService', module:'chat',  bugPrint: true})
+  async execute(sendReadAt: sendReadAt, tracing?: TracingType) {
+    return this.MessageSocketRepositoryService.sendReadAt(sendReadAt)
+  }
+}

src/app/module/chat/domain/use-case/room-get-list-use-case.service.ts

@@ -4,7 +4,7 @@ import { RoomRemoteDataSourceService } from '../../data/repository/room-remote-r
 import { RoomLocalRepository } from '../../data/repository/room-local-repository.service';
 import { captureAndReraiseAsync } from 'src/app/services/decorators/captureAndReraiseAsync';
 import { isHttpResponse } from 'src/app/services/http.service';
-
+import { CronJobService } from 'src/app/utils/task-scheduler'
 @Injectable({
   providedIn: 'root'
 })
@@ -14,6 +14,7 @@ export class GetRoomListUseCaseService {
     private roomRemoteDataSourceService: RoomRemoteDataSourceService,
     // private roomMemoryDataSourceService: Store<RoomRemoteDataSourceState>,
     private roomLocalDataSourceService: RoomLocalRepository,
+    private CronJobService: CronJobService
   ) { }
 
   @captureAndReraiseAsync('RoomRepositoryService/list')
@@ -28,6 +29,12 @@ export class GetRoomListUseCaseService {
 
           for( const roomData of  roomsToInsert) {
             this.roomLocalDataSourceService.createRoom(roomData.chatRoom)
+
+            if(roomData.chatRoom.expirationDate) {
+              console.log('room expiration date schedule')
+              this.CronJobService.createCronJob('remove expired room', new Date(roomData.chatRoom.expirationDate), this.execute)
+            }
+
           }
 
           for( const roomData of  roomsToUpdate) {

src/app/module/chat/domain/use-case/socket/socket-message-create-use-case.service.ts

@@ -13,7 +13,7 @@ export class SocketMessageCreateUseCaseService {
     private messageLocalDataSourceService: MessageLocalDataSourceService,
   ) { }
 
-  @XTracerAsync({name:'Socket-Message-Create-UseCase', bugPrint: true})
+  @XTracerAsync({name:'Socket-Message-Create-UseCase', module:'chat',  bugPrint: true})
   async execute(input: any, tracing?: TracingType) {
 
     ParamsValidation(MessageOutPutDataDTOSchema, input, tracing)

src/app/module/chat/domain/use-case/socket/socket-message-update-use-case.service.ts

@@ -15,7 +15,7 @@ export class SocketMessageUpdateUseCaseService {
   ) { }
 
 
-  @XTracerAsync({name:'Socket-Message-Update-UseCase', bugPrint: true})
+  @XTracerAsync({name:'Socket-Message-Update-UseCase', bugPrint: true, module:'chat',})
   async execute(data: MessageOutPutDataDTO, tracing?: TracingType) {
 
     ParamsValidation(MessageOutPutDataDTOSchema, data, tracing)
@@ -30,11 +30,13 @@ export class SocketMessageUpdateUseCaseService {
 
     if(result.isOk()) {
       tracing?.addEvent("Message found")
-      return await this.messageLocalDataSourceService.update(result.value.$id, incomingMessage)
+      const updateResult = await this.messageLocalDataSourceService.update(result.value.$id, incomingMessage)
+      tracing.setAttribute('outcome', 'success')
+      return updateResult
     } else {
       tracing?.addEvent("Message not found")
     }
 
-    tracing.setAttribute('outcome', 'success')
+
   }
 }

src/app/module/chat/domain/use-case/sync-all-room-messages.service.ts

@@ -3,8 +3,11 @@ import { MessageLocalDataSourceService } from '../../data/repository/message/mes
 import { messageListDetermineChanges } from '../../data/async/list/rooms/messageListChangedetector';
 import { MessageTable } from '../../infra/database/dexie/schema/message';
 import { MessageRemoteDataSourceService } from '../../data/repository/message/message-remote-data-source.service';
+import { MessageSocketRepositoryService } from '../../data/repository/message/message-live-signalr-data-source.service';
 import { ok } from 'neverthrow';
 import { RoomLocalRepository } from '../../data/repository/room-local-repository.service';
+import { SessionStore } from 'src/app/store/session.service';
+import { Logger } from 'src/app/services/logger/main/service';
 
 @Injectable({
   providedIn: 'root'
@@ -14,6 +17,7 @@ export class SyncAllRoomMessagesService {
   constructor(
     private messageLocalDataSourceService: MessageLocalDataSourceService,
     private messageRemoteDataSourceService: MessageRemoteDataSourceService,
+    private MessageSocketRepositoryService: MessageSocketRepositoryService,
     private roomLocalDataSourceService: RoomLocalRepository,
   ) { }
 
@@ -35,15 +39,38 @@ export class SyncAllRoomMessagesService {
 
           const { addedItems, changedItems, deletedItems } = messageListDetermineChanges(result.value.data, localResult)
 
+
           for(const message of changedItems) {
             let clone: MessageTable = message
             clone.roomId = room.id
+
             this.messageLocalDataSourceService.findOrUpdate(clone)
+
+            const me = message.info.find(e => e.memberId == SessionStore.user.UserId)
+
+            if(!me) {
+              this.MessageSocketRepositoryService.sendDeliverAt({
+                memberId: SessionStore.user.UserId,
+                messageId: message.id,
+                roomId: message.roomId,
+                requestId: "string"
+              })
+            }
           }
 
           for(const message of addedItems) {
             let clone: MessageTable = message
             clone.roomId = room.id
+
+            const me = message.info.find(e => e.memberId == SessionStore.user.UserId)
+            if(!me) {
+              this.MessageSocketRepositoryService.sendDeliverAt({
+                memberId: SessionStore.user.UserId,
+                messageId: message.id,
+                roomId: message.roomId,
+                requestId: "string"
+              })
+            }
           }
 
 
@@ -54,6 +81,8 @@ export class SyncAllRoomMessagesService {
             this.messageLocalDataSourceService.deleteByMessageId(message.id)
           }
 
+        } else {
+          Logger.error('failed to get room message '+room.id)
         }
       }