doneit-web/src/app/module/chat/infra/socket/signalR.ts

import * as signalR from '@microsoft/signalr';
import { BehaviorSubject, Observable, race, timer } from 'rxjs';
import { ok, Result, err } from 'neverthrow';
import { SessionStore } from 'src/app/store/session.service';
import { filter, first } from 'rxjs/operators';
import { v4 as uuidv4 } from 'uuid'
import { UserTypingDTO } from '../../data/dto/typing/typingInputDTO';
import { MessageOutPutDataDTO } from '../../data/dto/message/messageOutputDTO';
import { MessageDeleteInputDTO } from '../../data/dto/message/messageDeleteInputDTO';
import { MessageReactionInput } from '../../domain/use-case/message-reaction-by-id-use-case.service';
import { ISignalRInput } from './signal-r.service';

export class SignalRConnection {

  private hubConnection: signalR.HubConnection;
  private messageSubject: BehaviorSubject<MessageOutPutDataDTO> = new BehaviorSubject<MessageOutPutDataDTO>(null);
  private messageDelete: BehaviorSubject<MessageOutPutDataDTO> = new BehaviorSubject<MessageOutPutDataDTO>(null);
  private messageUPdateSubject: BehaviorSubject<MessageOutPutDataDTO> = new BehaviorSubject<MessageOutPutDataDTO>(null);
  private typingSubject: BehaviorSubject<UserTypingDTO> = new BehaviorSubject<UserTypingDTO>(null);
  private readAtSubject: BehaviorSubject<string> = new BehaviorSubject<any>(null);
  private connectionStateSubject: BehaviorSubject<boolean> = new BehaviorSubject<boolean>(false);
  private disconnectSubject: BehaviorSubject<boolean> = new BehaviorSubject<boolean>(false);
  private reconnectSubject: BehaviorSubject<boolean> = new BehaviorSubject<boolean>(false);
  private sendLaterSubject: BehaviorSubject<Object> = new BehaviorSubject<Object>(false);
  private reconnect = true

  private sendDataSubject: BehaviorSubject<{method: string, data: any}> = new BehaviorSubject<{method: string, data: any}>(null);
  private pendingRequests: Map<string, { resolve: Function; reject: Function }> = new Map();
  url: string
  private hasConnectOnce = false

  constructor({url}) {
    this.url = url
  }

  establishConnection(): Promise<Result<signalR.HubConnection, false>> {
    return new Promise((resolve, reject) => {

      const hubConnection = new signalR.HubConnectionBuilder()
      .withUrl(this.url)
      .build();

      this.hubConnection = hubConnection

      hubConnection
      .start()
      .then(() => {
        this.hubConnection = hubConnection
        this.hasConnectOnce = true
        console.log('Connection started');
        this.connectionStateSubject.next(true);
        this.join()
        this.addMessageListener()
        resolve(ok(hubConnection))
      })
      .catch(error => {
        console.log('Error while starting connection: ' + error);
        this.connectionStateSubject.next(false);

        if(this.hasConnectOnce) {
          setTimeout(()=> {
            this.attempReconnect();
          }, 2000)
        }
        resolve(err(false))
      });

      hubConnection.onclose(() => {
        console.log('Connection closed');
        this.connectionStateSubject.next(false);
        this.disconnectSubject.next(true)

        this.pendingRequests.forEach((_, requestId) => {
          const { reject } = this.pendingRequests.get(requestId);
          reject(err(false));
          this.pendingRequests.delete(requestId);
        });

        if(this.reconnect) {
          this.attempReconnect();
        }

      });
    })
  }


  async attempReconnect() {
    const attempConnection = await this.establishConnection()

    if(attempConnection.isOk()) {
      this.reconnectSubject.next(true)
    }

  }

  public join() {
    if(this.connectionStateSubject.value == true) {

      console.log('join=================')
      this.hubConnection.invoke("Join", SessionStore.user.UserId, SessionStore.user.FullName);
      //this.hubConnection.invoke("Join", 105, "UserFirefox");
    } else {
      this.sendLaterSubject.next({method: 'SendMessage', args:["Join", 312, "Daniel"]})
    }

  }

  public async sendMessage(data: Object & { requestId}):Promise<Result<any, any>>  {
    return new Promise((resolve, reject) => {

      if(this.connectionStateSubject.value == true) {
        console.log('sendMessage', data)
        this.hubConnection.invoke("SendMessage", data)

        //
        this.pendingRequests.set(data.requestId, { resolve, reject: (data) => resolve(data) });

        this.messageSubject.pipe(
          filter((message: any) => data.requestId == message?.requestId),
          first()
        ).subscribe(value => {
          resolve(ok(value))
          console.log('Received valid value:', value);
          //
          this.pendingRequests.delete(data.requestId);
        });

      } else {
        this.sendLaterSubject.next({method: 'SendMessage', args: data})
        return reject(err(false))
      }


    })
  }

  public async deleteMessage(data: MessageDeleteInputDTO) {
    return new Promise((resolve, reject) => {
      if(this.connectionStateSubject.value == true) {
        console.log('delete message', data)
        this.hubConnection.invoke("DeleteMessage", data)

        this.messageSubject.pipe(
          filter((message: any) => data.requestId == message?.requestId),
          first()
        ).subscribe((value) => {
          resolve(ok(value))
        })
      }

    })
  }

  public async typing(data: Object & { roomId, UserName, userId }):Promise<Result<any, any>>  {
    return new Promise((resolve, reject) => {

      const requestId = uuidv4()
      if(this.connectionStateSubject.value == true) {
        // console.log('send typing', data)

        try {
          this.hubConnection.invoke("Typing", data)

        } catch (error) {}

        this.typingSubject.pipe(
          filter((message: any) => {
            return  requestId == message?.requestId
          }),
          first()
        ).subscribe(value => {
          resolve(ok(value));
        });

      } else {
        this.sendLaterSubject.next({method: 'SendMessage', args: data})
        return reject(err(false))
      }


    })
  }

  public async sendReadAt(data: Object & { roomId, memberId, chatMessageId}):Promise<Result<any, any>>  {
    return new Promise((resolve, reject) => {

      const requestId = uuidv4()
      if(this.connectionStateSubject.value == true) {

        try {
          this.hubConnection.invoke("ReadAt", { roomId: data.roomId, memberId: data.memberId, requestId, messageId: data.chatMessageId} as any)

        } catch (error) {}

        this.readAtSubject.pipe(
          filter((message: any) => {
            return  requestId == message?.requestId
          }),
          first()
        ).subscribe(value => {
          resolve(ok(value));
        });

      } else {
        this.sendLaterSubject.next({method: 'SendMessage', args: data})
        return reject(err(false))
      }


    })
  }


  public async sendReactMessage(data: MessageReactionInput):Promise<Result<any, any>>  {
    return new Promise((resolve, reject) => {

      const requestId = uuidv4()
      if(this.connectionStateSubject.value == true) {

        try {
          this.hubConnection.invoke("ReactMessage", { roomId: data.roomId, memberId: data.memberId, requestId, messageId: data.messageId} as any)

        } catch (error) {}

        this.messageUPdateSubject.pipe(
          filter((message: any) => {
            return  requestId == message?.requestId
          }),
          first()
        ).subscribe(value => {
          resolve(ok(value));
        });

        // // Create an observable for the timeout (8 seconds)
        // const timeout$ = timer(8000);

        // // Create an observable for the message
        // const message$ = this.messageUPdateSubject.pipe(
        //   filter((message: MessageOutPutDataDTO) => requestId === message?.requestId),
        //   first()
        // );

        // // Use race to choose whichever completes first (either the message or the timeout)
        // race(message$, timeout$).subscribe({
        //   next: (value) => {
        //     if (typeof value === 'number') {
        //       // If the value is a number, it means the timeout fired
        //       reject(err('No response, timeout'));
        //     } else {
        //       resolve(ok(value));
        //     }
        //   },
        //   error: (error) => reject(err('Error: ' + error))
        // });

      } else {
        this.sendLaterSubject.next({method: 'SendMessage', args: data})
        return reject(err(false))
      }

    })
  }


  sendData<T>(input: ISignalRInput): Promise<Result<T, any>> {
    return new Promise((resolve, reject) => {

      if(this.connectionStateSubject.value == true) {

        this.hubConnection.invoke(input.method, input.data)

        this.sendDataSubject.pipe(
          filter((message) => input.data.requestId == message?.data.requestId),
          first()
        ).subscribe(value => {
          resolve(ok(value.data as unknown  as T))
          console.log('Received valid value:', value);
        });

      } else {
        this.sendLaterSubject.next({method: 'SendMessage', args: input})
        return reject(err(false))
      }

    })
  }

  private addMessageListener(): void {
    console.log('listening')
    this.hubConnection.on('ReceiveMessage', (message: MessageOutPutDataDTO) => {
      console.log('ReceiveMessage', message)
      this.messageSubject.next(message);
      this.sendDataSubject.next({
        method: 'ReceiveMessage',
        data: message
      })
    });

    this.hubConnection.on('TypingMessage', (_typing: UserTypingDTO) => {
      this.typingSubject.next(_typing);
      this.sendDataSubject.next({
        method: 'ReceiveMessage',
        data: _typing
      })
    });

    this.hubConnection.on('AvailableUsers', (data: any) => {
      this.typingSubject.next(data);
      this.sendDataSubject.next({
        method: 'AvailableUsers',
        data: data
      })
    });

    this.hubConnection.on('ReadAt', (_message) => {
      console.log('ReadAt', _message)
      this.readAtSubject.next(_message);
      this.sendDataSubject.next({
        method: 'ReceiveMessage',
        data: _message
      })
    });

    this.hubConnection.on('DeleteMessage', (_message) => {
      console.log('DeleteMessage', _message)
      this.messageDelete.next(_message);
      this.sendDataSubject.next({
        method: 'DeleteMessage',
        data: _message
      })
    });

    this.hubConnection.on('UpdateMessage', (_message) => {
      console.log('UpdateMessage', _message)
      this.messageUPdateSubject.next(_message);
      this.sendDataSubject.next({
        method: 'ReceiveMessage',
        data: _message
      })
    })


    this.hubConnection.on('GroupAddedMembers', (_message) => {
      console.log('GroupAddedMembers', _message)
      this.sendDataSubject.next({
        method: 'GroupAddedMembers',
        data: _message
      })
    })

    this.hubConnection.on('GroupDeletedMembers', (_message) => {
      console.log('GroupDeletedMembers', _message)
      this.sendDataSubject.next({
        method: 'GroupDeletedMembers',
        data: _message
      })
    })
  }

  public getMessageUpdateSubject() {
    return this.messageUPdateSubject.asObservable()
  }

  public getMessages() {
    return this.messageSubject.asObservable()
  }

  public getTyping() {
    return this.typingSubject.asObservable()
  }

  public getConnectionState(): Observable<boolean> {
    return this.connectionStateSubject.asObservable();
  }

  public getDisconnectTrigger(): Observable<boolean> {
    return this.disconnectSubject.asObservable();
  }

  public getSendLater() {
    return this.sendLaterSubject.asObservable();
  }

  public getMessageDelete() {
    return this.messageDelete.asObservable()
  }

  public getData() {
    return this.sendDataSubject.asObservable()
  }

  public closeConnection(): void {
    this.reconnect = false
    if (this.hubConnection) {
      this.hubConnection
        .stop()
        .then(() => {
          console.log('Connection closed by user');
          this.connectionStateSubject.next(false);
        })
        .catch(err => console.log('Error while closing connection: ' + err));
    }
  }

}