sync message on recoonect on ui

src/app/core/chat/usecase/message/http-listen-to-message-load-history-use-case.ts

@@ -0,0 +1,34 @@
+import { HttpErrorResponse } from "@angular/common/http";
+import { Injectable } from "@angular/core";
+import { Result } from "neverthrow";
+import { Observable } from "rxjs";
+import { filter, map } from "rxjs/operators";
+import { HttpAdapter } from "src/app/infra/http/adapter";
+import { MessageOutPutDTO } from "src/app/module/chat/data/dto/message/messageOutputDTO";
+import { z } from "zod";
+
+const HttpListenToMessageLoadHistoryUseCaseInputSchema = z.object({
+  roomId: z.string()
+})
+
+export type HttpListenToMessageLoadHistoryUseCaseInput = z.infer<typeof HttpListenToMessageLoadHistoryUseCaseInputSchema> 
+
+@Injectable({
+  providedIn: 'root'
+})
+export class HttpListenToMessageLoadHistoryUseCase{
+
+  constructor(
+    private http: HttpAdapter
+  ) {}
+
+  execute(input: HttpListenToMessageLoadHistoryUseCaseInput): Observable<MessageOutPutDTO>  {
+    return this.http.listen().pipe(
+      filter((response)=> response.isOk()),
+      filter((response: any)=> {
+        return response.value.url.includes(`Room/${input.roomId}/Messages`)
+      }),
+      map((response: any) => response.value.data as MessageOutPutDTO)
+    )
+  }
+}

src/app/core/chat/usecase/socket/socket-on-reconnect-use-case.ts

@@ -1,9 +1,20 @@
+import { Injectable } from "@angular/core";
+import { Observable } from "rxjs";
+import { ISignalRService } from "src/app/infra/socket/adapter";
 import { z } from "zod"
 
 const e = z.object({})
 
-export class SocketOnConnectUseCase {
+@Injectable({
+  providedIn: 'root'
+})
+export class SocketOnReconnectUseCase {
 
-  constructor() {}
-  execute() {}
+  constructor(private SignalRService: ISignalRService) {}
+  
+  execute(): Observable<boolean> {
+
+    const connection = this.SignalRService.onReconnect()
+    return connection
+  }
 }

src/app/infra/database/dexie/instance/chat/schema/message.ts

@@ -18,7 +18,7 @@ export const MessageTableSchema = z.object({
     wxUserId: z.number(),
     wxFullName: z.string(),
     wxeMail: z.string(),
-    userPhoto: z.string(),
+    userPhoto: z.string().nullable().optional(),
   }),
   receiverId: z.number().optional(),
   sending: z.boolean().optional(),

src/app/infra/http/adapter.ts

@@ -0,0 +1,14 @@
+import { HttpClient, HttpErrorResponse, HttpHeaders } from '@angular/common/http';
+import { HttpResult } from './type';
+import { Result } from 'neverthrow';
+import { Observable } from 'rxjs';
+
+
+export abstract  class HttpAdapter {
+  abstract post<T>(url: string, body: any): Promise<Result<HttpResult<T>, HttpErrorResponse>>
+  abstract get<T>(url: string, options?: Object): Promise<Result<HttpResult<T>, HttpErrorResponse>>
+  abstract put<T>(url: string, body: any): Promise<Result<HttpResult<T>, HttpErrorResponse>>
+  abstract patch<T>(url: string, body?: Object): Promise<Result<HttpResult<T>, HttpErrorResponse>> 
+  abstract delete<T>(url: string, body?: Object): Promise<Result<HttpResult<T>, HttpErrorResponse>>
+  abstract listen():Observable<Result<HttpResult<any>, HttpErrorResponse>>
+}

src/app/infra/http/http.module.ts

@@ -0,0 +1,17 @@
+import { NgModule, CUSTOM_ELEMENTS_SCHEMA } from '@angular/core';
+import { HttpAdapter } from './adapter';
+import { HttpService } from './http.service';
+
+@NgModule({
+  imports: [],
+  declarations: [],
+  schemas: [],
+  providers: [
+    {
+      provide: HttpAdapter,
+      useClass: HttpService, // or MockDataService
+    }
+  ],
+  entryComponents: []
+})
+export class HttpModule {}

src/app/infra/http/http.service.ts

@@ -0,0 +1,123 @@
+import { HttpErrorResponse, HttpHeaders, HttpParams, HttpResponse } from '@angular/common/http';
+import { Injectable } from '@angular/core';
+import { HttpClient } from '@angular/common/http';
+import { ok, err, Result } from 'neverthrow';
+import { HttpResult } from './type';
+import { BehaviorSubject } from 'rxjs';
+
+@Injectable({
+  providedIn: 'root'
+})
+export class HttpService {
+
+  private responseSubject = new BehaviorSubject<Result<HttpResult<any>, HttpErrorResponse>>(null);
+  
+  constructor(private http: HttpClient) { }
+
+  async post<T>(url: string, body: any): Promise<Result<HttpResult<T>, HttpErrorResponse>> {
+    try {
+      const response = await this.http.post<T>(url, body, { observe: 'response' }).toPromise();
+      const data = {
+        data: response.body,
+        status: response.status,
+        headers: response.headers,
+        url: response.url || url
+      }
+      this.responseSubject.next(ok(data))
+      return ok(data);
+    } catch (e) {
+      this.responseSubject.next(err(e))
+      return err(e as HttpErrorResponse);
+    }
+  }
+
+  async get<T>(url: string, options = {}): Promise<Result<HttpResult<T>, HttpErrorResponse>> {
+    try {
+      const response = await this.http.get<T>(url, { ...options, observe: 'response' }).toPromise();
+
+      const data = {
+        data: response.body,
+        status: response.status,
+        headers: response.headers,
+        url: response.url || url
+      }
+
+      this.responseSubject.next(ok(data))
+      return ok(data);
+    } catch (e) {
+      this.responseSubject.next(err(e))
+      return err(e as HttpErrorResponse);
+    }
+  }
+
+  async put<T>(url: string, body: any): Promise<Result<HttpResult<T>, HttpErrorResponse>> {
+    try {
+      const response = await this.http.put<T>(url, body, { observe: 'response' }).toPromise();
+
+      const data = {
+        data: response.body,
+        status: response.status,
+        headers: response.headers,
+        url: response.url || url
+      }
+
+      this.responseSubject.next(ok(data))
+      return ok(data);
+    } catch (e) {
+      this.responseSubject.next(err(e))
+      return err(e as HttpErrorResponse);
+    }
+  }
+
+  async patch<T>(url: string, body: any = {}): Promise<Result<HttpResult<T>, HttpErrorResponse>> {
+    try {
+      const response = await this.http.patch<T>(url, body, { observe: 'response' }).toPromise();
+
+      const data = {
+        data: response.body,
+        status: response.status,
+        headers: response.headers,
+        url: response.url || url
+      }
+
+      this.responseSubject.next(ok(data))
+      return ok(data);
+    } catch (e) {
+      this.responseSubject.next(err(e))
+      return err(e as HttpErrorResponse);
+    }
+  }
+
+  async delete<T>(url: string, body = {}): Promise<Result<HttpResult<T>, HttpErrorResponse>> {
+    const options = {
+      body: body, // Pass payload as the body of the request
+      observe: 'response' as 'body'
+    };
+
+    try {
+      const response: any = await this.http.delete<T>(url, options).toPromise();
+
+      const data = {
+        data: response?.body,
+        status: response?.status,
+        headers: response?.headers,
+        url: response?.url || url
+      }
+
+      this.responseSubject.next(ok(data))
+      return ok(data as any);
+    } catch (e) {
+      this.responseSubject.next(err(e))
+      return err(e as HttpErrorResponse);
+    }
+  }
+
+  listen() {
+    return this.responseSubject.asObservable()
+  }
+}
+
+export function isHttpResponse(data: any): data is HttpResponse<any> {
+  return typeof data.status == 'number';
+}
+

src/app/infra/http/type.ts

@@ -0,0 +1,13 @@
+import { HttpHeaders, HttpParams } from '@angular/common/http';
+
+export interface Options {
+  headers?: HttpHeaders
+  params?: HttpParams
+}
+
+export interface HttpResult<T> {
+  data: T | null;
+  status: number;
+  headers: HttpHeaders;
+  url: string;
+}

src/app/infra/socket/adapter.ts

@@ -9,5 +9,6 @@ export abstract class ISignalRService {
   abstract join(): void;
   abstract getData<T>(): Observable<{ method: string; data: T }>;
   abstract getConnectionState(): Observable<boolean>;
+  abstract onReconnect(): Observable<boolean>;
   abstract newConnection(): void;
 }

src/app/infra/socket/signalR/signal-r.service.ts

@@ -3,8 +3,7 @@ import { BehaviorSubject, Observable, Subject, timer } from 'rxjs';
 import { Platform } from '@ionic/angular';
 import { SignalRConnection, SocketMessage } from './signalR';
 import { Plugins } from '@capacitor/core';
-import { z } from 'zod';
-import { filter, map, switchMap } from 'rxjs/operators';
+import { filter, map, skip, switchMap } from 'rxjs/operators';
 import { Result } from 'neverthrow';
 import { HubConnection } from '@microsoft/signalr';
 import { MessageOutPutDataDTO } from 'src/app/module/chat/data/dto/message/messageOutputDTO';
@@ -127,9 +126,17 @@ export class SignalRService {
     return this.connection.join()
   }
 
+  // onReconnect() {
+  //   const connection = this.getConnectionState()
+  //   return connection.pipe(
+  //     skip(1) // Skip the first value
+  //   )
+  // }
+
   getData<T>() {
     return this.sendDataSubject.asObservable() as BehaviorSubject<{method: string, data: T}>
   }
+
   public getConnectionState(): Observable<boolean> {
     return this.connectingSubject.asObservable();
   }

src/app/infra/socket/signalR/signalR.ts

@@ -58,8 +58,6 @@ export class SignalRConnection {
       })
       .catch(error => {
         console.log('Error while starting connection: ' + error);
-        this.connectionStateSubject.next(false);
-
         if(this.hasConnectOnce) {
           setTimeout(()=> {
             this.attempReconnect();

src/app/module/chat/chat.module.ts

@@ -7,9 +7,22 @@ import { Subject, timer } from 'rxjs';
 import { UserTypingLocalRepository } from './data/repository/typing/user-typing-local-data-source.service';
 import { UserTypingRemoteRepositoryService } from './data/repository/typing/user-typing-live-data-source.service';
 import { RoomService } from 'src/app/module/chat/domain/service/room.service'
+import { HttpListenToMessageLoadHistoryAdapter, SocketOnReconnectAdapter } from './domain/adapter';
+import { ISignalRService } from 'src/app/infra/socket/adapter';
+import { HttpModule } from 'src/app/infra/http/http.module';
+import { HttpListenToMessageLoadHistoryUseCase } from 'src/app/core/chat/usecase/message/http-listen-to-message-load-history-use-case';
 @NgModule({
-  imports: [],
-  providers: [],
+  imports: [HttpModule],
+  providers: [
+    {
+      provide: ISignalRService,
+      useClass: SignalRService, // or MockDataService
+    },
+    {
+      provide: HttpListenToMessageLoadHistoryAdapter,
+      useClass: HttpListenToMessageLoadHistoryUseCase, // or MockDataService
+    },
+  ],
   declarations: [],
   schemas: [],
   entryComponents: []

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

@@ -11,6 +11,7 @@ import { SessionStore } from 'src/app/store/session.service';
 import { MessageDeleteInputDTO } from '../../dto/message/messageDeleteInputDTO';
 import { InstanceId } from '../../../domain/chat-service.service';
 import { v4 as uuidv4 } from 'uuid'
+import { HttpAdapter } from 'src/app/infra/http/adapter';
 
 @Injectable({
   providedIn: 'root'
@@ -22,6 +23,7 @@ export class MessageRemoteDataSourceService {
   constructor(
     private httpService: HttpService,
     private socket: SignalRService,
+    private http: HttpAdapter
   ) {}
 
   @APIReturn(MessageOutPutDTOSchema, 'post/Messages')
@@ -37,6 +39,13 @@ export class MessageRemoteDataSourceService {
 
   // @APIReturn(MessageOutPutDTOSchema, 'get/Messages')
   async getMessagesFromRoom(id: string): DataSourceReturn<MessageOutPutDTO>  {
+
+    var a = await this.http.get<MessageOutPutDTO>(`${this.baseUrl}/Room/${id}/Messages`)
+
+    return a.map((e) => {
+      return e.data
+    })
+    
     return await this.httpService.get(`${this.baseUrl}/Room/${id}/Messages`);
   }
 

src/app/module/chat/domain/adapter.ts

@@ -1,8 +1,22 @@
+import { HttpErrorResponse } from "@angular/common/http";
+import { Result } from "neverthrow";
+import { Observable } from "rxjs";
 import { MessageEntity } from "src/app/core/chat/entity/message";
+import { HttpListenToMessageLoadHistoryUseCaseInput } from "src/app/core/chat/usecase/message/http-listen-to-message-load-history-use-case";
+import { HttpResult } from "src/app/infra/http/type";
 import { UseCase } from "src/app/utils/use-case-interface";
+import { MessageOutPutDTO } from "../data/dto/message/messageOutputDTO";
 
 export abstract class ITemplateCreateAdapter implements UseCase<MessageEntity, any> {
   execute(input: MessageEntity): Promise<any> {
     throw new Error("Method not implemented.");
   }
-}
+}
+
+export abstract class SocketOnReconnectAdapter {
+  abstract execute(): Observable<boolean>
+}
+
+export abstract class HttpListenToMessageLoadHistoryAdapter {
+  abstract execute(input: HttpListenToMessageLoadHistoryUseCaseInput): Observable<MessageOutPutDTO>
+}

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

@@ -4,7 +4,7 @@ import { SessionStore } from 'src/app/store/session.service';
 import { MessageReactionInput, MessageReactionUseCaseService } from 'src/app/module/chat/domain/use-case/message/message-reaction-by-id-use-case.service';
 import { MessageUpdateInput, MessageUpdateUseCaseService } from 'src/app/module/chat/domain/use-case/message/message-update-by-id-use-case.service';
 import { MemberAdminUseCaseService, MemberSetAdminDTO } from 'src/app/module/chat/domain/use-case/member/member-admin-use-case.service';
-import { MessageCreateUseCaseService, MessageEnum } from 'src/app/module/chat/domain/use-case/message/message-create-use-case.service';
+import { MessageCreateUseCaseService } from 'src/app/module/chat/domain/use-case/message/message-create-use-case.service';
 import { SignalRService } from 'src/app/infra/socket/signalR/signal-r.service';
 import { SocketMessageDeleteUseCaseService } from 'src/app/module/chat/domain/use-case/socket/socket-message-delete-use-case.service';
 import { SocketMessageUpdateUseCaseService } from 'src/app/module/chat/domain/use-case/socket/socket-message-update-use-case.service';
@@ -20,7 +20,6 @@ import { CreateRoomUseCaseService } from './use-case/room/room-create-use-case.s
 import { RoomLeaveUseCase } from './use-case/room/room-leave-by-id-use-case.service';
 import { SyncAllRoomMessagesService } from './use-case/message/sync-all-room-messages.service';
 import { ListenSendMessageUseCase } from './use-case/message/listen-send-message.service'
-import { MessageReadAtByIdUseCaseService } from './use-case/message/message-read-at-by-id-use-case.service'
 import { SendLocalMessagesUseCaseService } from './use-case/message/messages-send-offline-use-case.service'
 import { RemoveMemberUseCaseService } from './use-case/member/-use-case.service'
 import { AddMemberUseCaseService } from './use-case/member/member-add-use-case.service'
@@ -40,10 +39,10 @@ import { AddMemberToRoomInputDTO } from '../data/dto/room/addMemberToRoomInputDt
 import { RoomUpdateInputDTO } from '../data/dto/room/roomUpdateInputDTO';
 import { RoomType } from "src/app/core/chat/entity/group";
 import { sendReadAt } from "src/app/module/chat/data/repository/message/message-live-signalr-data-source.service";
-
+import { HttpListenToMessageLoadHistoryAdapter } from './adapter'
+import { HttpListenToMessageLoadHistoryUseCaseInput } from 'src/app/core/chat/usecase/message/http-listen-to-message-load-history-use-case';
 export const InstanceId = uuidv4();
 
-
 @Injectable({
   providedIn: 'root'
 })
@@ -76,11 +75,11 @@ export class ChatServiceService {
     private AddMemberUseCaseService: AddMemberUseCaseService,
     private UpdateRoomByIdUseCaseService: UpdateRoomByIdUseCaseService,
     private RemoveMemberUseCaseService: RemoveMemberUseCaseService,
-    private MessageReadAtByIdUseCaseService: MessageReadAtByIdUseCaseService,
     private SendLocalMessagesUseCaseService: SendLocalMessagesUseCaseService,
     private MessageMarkAsReadUseCaseService: MessageMarkAsReadUseCaseService,
     private SocketConnectUseCaseService: SocketConnectUseCaseService,
-    private MessageMarkAllMessageAsReadByRoomIdService: MessageMarkAllMessageAsReadByRoomIdService
+    private MessageMarkAllMessageAsReadByRoomIdService: MessageMarkAllMessageAsReadByRoomIdService,
+    private HttpListenToMessageLoadHistory: HttpListenToMessageLoadHistoryAdapter,
   ) {
     this.messageLiveSignalRDataSourceService.getMessageDelete()
     .pipe()
@@ -184,6 +183,9 @@ export class ChatServiceService {
   }
 
 
+  listenToMessageLoadHistory(input: HttpListenToMessageLoadHistoryUseCaseInput) {
+    return this.HttpListenToMessageLoadHistory.execute(input)
+  }
 
   getRoomList() {
     return this.GetRoomListUseCaseService.execute()

src/app/services/http/http-service.service.spec.ts

@@ -1,16 +0,0 @@
-import { TestBed } from '@angular/core/testing';
-
-import { HttpServiceService } from './http-service.service';
-
-describe('HttpServiceService', () => {
-  let service: HttpServiceService;
-
-  beforeEach(() => {
-    TestBed.configureTestingModule({});
-    service = TestBed.inject(HttpServiceService);
-  });
-
-  it('should be created', () => {
-    expect(service).toBeTruthy();
-  });
-});

src/app/ui/chat/component/messages/messages.page.ts

@@ -140,6 +140,7 @@ export class MessagesPage implements OnInit, OnChanges, AfterViewInit, OnDestroy
   messageUpdateSubject: Subscription
   messageSendSubject: Subscription
   messageTypingSubject: Subscription
+  messageOnReconnectSubject: Subscription
 
   messages1: {[key: string]: MessageEntity[]} = {}
   MessageAttachmentFileType = MessageAttachmentFileType
@@ -228,6 +229,8 @@ export class MessagesPage implements OnInit, OnChanges, AfterViewInit, OnDestroy
 
     }) as any
 
+
+
   }
 
   messageStatus(message: MessageEntity) {
@@ -268,6 +271,22 @@ export class MessagesPage implements OnInit, OnChanges, AfterViewInit, OnDestroy
       this.sendReadMessage()
     }, 1000)
 
+    this.messageOnReconnectSubject?.unsubscribe()
+    this.messageOnReconnectSubject = this.chatServiceService.listenToMessageLoadHistory({roomId: this.roomId}).subscribe((messages) => {
+      
+      console.log('message', messages);
+      
+      for(const message of messages.data) {
+        const found = this.messages1[this.roomId].find((e) => e.id == message.id)
+
+        if(!found) {
+          const msg = new MessageEntity()
+          Object.assign(msg, message)
+          this.messages1[this.roomId].push(msg)
+        }
+      }
+    })
+
   }
 
   async loadAttachment() {
@@ -424,6 +443,7 @@ export class MessagesPage implements OnInit, OnChanges, AfterViewInit, OnDestroy
     });
   }
 
+  onDisConnect() {}
 
   toggleEmojiPicker(message: MessageEntity) {
     if (this.selectedMessage === message) {