create direct message

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

@@ -0,0 +1,210 @@
+import { Injectable } from '@angular/core';
+import { liveQuery } from 'Dexie';
+import { err, ok, Result } from 'neverthrow';
+import { Observable, Subject } from 'rxjs';
+import { filter } from 'rxjs/operators';
+import { MessageEntity } from '../../../domain/entity/message';
+import { DexieRepository } from 'src/app/infra/repository/dexie/dexie-repository.service';
+import { MessageTable, MessageTableSchema } from 'src/app/module/chat/infra/database/dexie/schema/message';
+import { chatDatabase } from '../../../infra/database/dexie/service';
+import { Observable as  DexieObservable, PromiseExtended } from 'Dexie';
+
+@Injectable({
+  providedIn: 'root'
+})
+export class MessageLocalDataSourceService extends DexieRepository<MessageTable>  {
+
+  messageSubject = new Subject();
+
+  constructor() {
+    super(chatDatabase.message, MessageTableSchema)
+
+    this.setAllSenderToFalse();
+  }
+
+  async setAllSenderToFalse() {
+    try {
+      await chatDatabase.transaction('rw', chatDatabase.message, async () => {
+        // Perform the update operation within the transaction
+        await chatDatabase.message.toCollection().modify({ sending: false });
+      });
+      console.log('All messages updated successfully.');
+    } catch (error) {
+      console.error('Error updating messages:', error);
+    }
+  }
+
+  async getLastMessageByRoomId(roomId: string): Promise<Result<undefined|MessageTable, any>> {
+    try {
+      console.log({roomId})
+      const lastMessage = await chatDatabase.message
+        .where('roomId')
+        .equals(roomId)
+        .reverse()
+        .sortBy('id');
+
+      return ok(lastMessage[0]); // Get the last message
+    } catch (error) {
+      return err(error);
+    }
+  }
+
+  async deleteByMessageId(id: string): Promise<Result<undefined|MessageTable, any>> {
+    try {
+      console.log(id)
+      const lastMessage = await chatDatabase.message
+        .where('id')
+        .equals(id).delete()
+
+      return ok(lastMessage[0]); // Get the last message
+    } catch (error) {
+      return err(error);
+    }
+  }
+
+
+  async sendMessage(data: MessageTable) {
+
+    const dataValidation = MessageTableSchema.safeParse(data)
+    if(dataValidation.success) {
+
+      const safeData = dataValidation.data
+      safeData.sending = true
+
+      try {
+        const result = await chatDatabase.message.add(safeData)
+        this.messageSubject.next({roomId: safeData.roomId});
+        return ok(result as number)
+      } catch (e) {
+        return err(false)
+      }
+
+    } else {
+      console.log(dataValidation)
+      return err(dataValidation)
+    }
+  }
+
+
+ //  @ValidateSchema(tableSchema)
+  async createMessage(data: MessageTable) {
+
+    try {
+      const result = await chatDatabase.message.add(data)
+      this.messageSubject.next({roomId: data.roomId});
+      return ok(result)
+    } catch (e) {
+      return err(false)
+    }
+
+  }
+
+  async createManyMessage(data: MessageTable[]) {
+
+    try {
+      const result = await chatDatabase.message.bulkAdd(data)
+      this.messageSubject.next({roomId: data[0].roomId});
+      return ok(result)
+    } catch (e) {
+      return err(false)
+    }
+
+  }
+
+  async messageExist({id}) {
+    try {
+
+      console.log({id});
+
+      const existingMessage = await chatDatabase.message
+      .where('id')
+      .equals(id)
+      .first();
+
+      if (existingMessage) {
+        return ok(existingMessage)
+      } else {
+        return err(false)
+      }
+
+    } catch (error) {
+      return err(false);
+    }
+
+  }
+
+  // not used
+  async updateByMessageId(data: MessageTable ) {
+
+    try {
+      const result = await chatDatabase.message.update(data.id as any, data)
+      return ok(result)
+    } catch (e) {
+      return err(false)
+    }
+
+  }
+
+
+  async findOrUpdate(data: MessageTable) {
+    const findResult = await this.findMessageById(data.id)
+
+    if(findResult.isOk()) {
+      return this.update(findResult.value.$id, data)
+    } else  {
+      return this.createMessage(data)
+    }
+  }
+
+  getItems(roomId: string): PromiseExtended<MessageEntity[]> {
+    return chatDatabase.message.where('roomId').equals(roomId).sortBy('$id') as any
+  }
+
+  getItemsLive(roomId: string): DexieObservable<MessageEntity[]> {
+    return liveQuery(() => chatDatabase.message.where('roomId').equals(roomId).sortBy('$id') as any)
+  }
+
+
+  async findMessageById(id: string) {
+    try {
+      const a = await chatDatabase.message.where('id').equals(id).first()
+
+      if(a) {
+        return ok(a)
+      } else {
+        return err('not found')
+      }
+
+    } catch (e) {
+      return err('DB error')
+    }
+
+  }
+
+
+  subscribeToNewMessage(roomId: string): Observable<MessageTable>  {
+    return this.messageSubject.pipe(
+      filter((message: MessageTable) =>
+        message.roomId === roomId
+      )
+    )
+  }
+
+
+  async getOfflineMessages () {
+    try {
+      const allMessages = await chatDatabase.message
+        .filter(msg => typeof msg.id !== 'string' && msg.sending == false)
+        .toArray();
+
+
+        console.log("======================================================================")
+        console.log({localMessage:allMessages})
+
+      return allMessages as MessageEntity[];
+    } catch (error) {
+      console.error('Error fetching messages:', error);
+    }
+  }
+}
+