Fis-Retransmission/services/buffer.service.ts

// bufferService.ts
import { BehaviorSubject, Observable, Subject, from } from "rxjs";
import mongoose, { Connection, Model } from "mongoose";
import { ConnectionState, Message, MessageLog, } from "../interfaces/general.interface";

export class BufferService {
  private bufferIdentifier!: string
  private messageStream: Subject<Message>;
  private connectionState: BehaviorSubject<ConnectionState>;
  private messageBuffered: Message[] = [];
  private messageModel!: Model<Message>
  private readonly dbUrl!: string
  private bufferLimit!: number
  constructor(
    messageFromApp: Subject<Message>,
    connectionStateSubject: BehaviorSubject<ConnectionState>,
    dbName: string,
    bufferLimit?: number
  ) {
    if (bufferLimit) {
      this.bufferLimit = bufferLimit
    } else {
      this.bufferLimit = 10000 // default buffer limit
    }
    this.bufferIdentifier = dbName
    this.messageStream = messageFromApp;
    this.connectionState = connectionStateSubject;
    this.setupSubscriptions(); // Note: The handle buffer will push the data in local array before pushing to mongo via initial check up model

    /* Disabled for now due to missing data not during transmision. The issue was suspected to be it's async nature of confusing timing
    when it was queued into the event queue. Sometimes the messages will be late to be saved */
    // this.initializeDatabaseConnection(dbName)
    //   .then(async (connection: mongoose.Connection) => {
    //     const grpcMessageSchema = require("../models/message.schema");
    //     this.messageModel = connection.model<Message>(
    //       "Message",
    //       grpcMessageSchema
    //     );
    //     await this.transferLocalBufferToMongoDB(); // transfer all data from local array into mongodb after the mongo setup is complete
    //   })
    //   .catch((error) => {
    //     console.error("Database initialization failed:", error);
    //     // Implement retry logic or additional error handling here. Perhaps retry logic in the future...
    //   });
  }

  // to be exposed to acquire the messages
  public getMessages(): Observable<Message> {
    return this.messageStream as Observable<Message>;
  }

  public getStateObservable(): BehaviorSubject<ConnectionState> {
    return this.connectionState;
  }

  // To subscrcibe for the message stream as well as the connection state
  private setupSubscriptions(): void {
    this.messageStream.subscribe({
      next: (message: Message) => this.handleIncomingMessage(message),
      error: (err) =>
        console.error("Error in messageToBePublished subject:", err),
      complete: () =>
        console.log("messageToBePublished subscription completed"),
    });

    this.connectionState.subscribe({
      next: (state: ConnectionState) => this.handleConnectionStateChanges(state),
      error: (err) => console.error("Error in connectionState subject:", err),
      complete: () => console.log("connectionState subscription completed"),
    });
  }

  private async initializeDatabaseConnection(
    dbName: string
  ): Promise<Connection> {
    try {
      console.log(`${this.dbUrl}${dbName}`);
      const connection: mongoose.Connection = await mongoose.createConnection(
        `${this.dbUrl}${dbName}`
      );
      console.log(`Connected to ${this.dbUrl}${dbName}`);
      return connection;
    } catch (error) {
      console.error("Error connecting to MongoDB:", error);
      throw error;
    }
  }

  private handleIncomingMessage(message: any): void {
    if (this.connectionState.getValue().status === `LIMIT_EXCEEDED`) {
      // do nothing... Let handleConnectionStateChanges deal with this state
    }
    if (this.connectionState.getValue().status === "BUFFER") {
      this.bufferMessage(message);
    }
    if (this.connectionState.getValue().status === "DIRECT_PUBLISH") {
      /* Note: Since the main outGoingMessage is being published irregardless
      of the connection state, so there's no need to do anything aside from
      releasing buffered messages which will be handled by handleConnectionStateChange */
      // additional logic here
    }
  }

  private handleConnectionStateChanges(state: ConnectionState): void {
    console.log(`${this.bufferIdentifier}: ${this.connectionState.getValue().status}`);
    if (state.status === `LIMIT_EXCEEDED`) {
      console.log(`Limit exceed. Clearing buffered messages...`)
      let message: Message = {
        id: `test`,
        message: `Limit exceed. Please take care. Buffer Service Out!`
      }
      this.messageStream.next(message)
      // this.messageStream.unsubscribe() //destroy existing subscription
      this.messageBuffered = []
    }
    if (state.status === "BUFFER") {
      if (state.payload && typeof state.payload !== "string") {
        this.bufferMessage(state.payload); // Buffer the last message immediately
      }
    }
    if (state.status === "DIRECT_PUBLISH") {
      // Relese the messages by inserting them into the outgoing Messages together.
      this.releaseBufferedMessages(this.messageStream);
    }
  }

  private async bufferMessage(message: any): Promise<void> {
    if (this.messageModel) {
      try {
        // const newMessage = new this.messageModel(message);
        await this.messageModel.create(message);
        this.messageModel.countDocuments({}).then((count) => {
          console.log(`Message${(message.message as MessageLog).appData.msgId} saved to MongoDB buffer. There is ${count} messages in datatbase under ${this.bufferIdentifier} at the moment.`);
          // if connection status okay 
          // if(this.connectionState.getValue().status == "DIRECT_PUBLISH")
          // {
          //   console.log("Message count release " + count);
          //   // Then release back to message stream
          //   this.releaseBufferedMessages(this.messageStream);
          // }
        });
      } catch (error) {
        console.error("Error saving message to MongoDB:", error);
        // Implement retry logic or additional error handling here 
      }
    } else {
      if (this.bufferLimit > this.messageBuffered.length) {
        this.messageBuffered.push(message);
        console.log(this.messageBuffered) // Fallback to local buffer if model is not defined
        console.log(`pushing into local array buffer under ${this.bufferIdentifier}.... There is now ${this.messageBuffered.length} messages`);
      } else {
        let reportState: ConnectionState = {
          status: `LIMIT_EXCEEDED`,
          reason: `${this.bufferLimit} Limit exceeded. Buffer Service will be terminated...`
        }
        this.connectionState.next(reportState)
      }
    }
  }

  private releaseBufferedMessages(
    messageFromBuffer: Subject<Message>
  ): Promise<boolean> {
    return new Promise(async (resolve, reject) => {
      if (this.messageModel) {
        try {
          // use then 
          let countPromise = checkMessageCount(this.messageModel, this.bufferIdentifier);
          countPromise.then(async (amount) => {
            console.log("Amount1:" + amount);

            // let countPromise = checkMessageCount(this.messageModel, this.bufferIdentifier);
            // countPromise.then(async (amount)=>{

            //   console.log("Amount2:"+amount);
            // })

            while (amount > 0) {
              console.log("AmountInLoop1:" + amount)
              try {
                await extractData(messageFromBuffer, this.messageModel); // New function to process a batch
              } catch (processError) {
                console.error('Error processing batch:', processError);
              }
              console.log('Checking again...');
              amount = await checkMessageCount(this.messageModel, this.bufferIdentifier);
              console.log("AmountInLoop:" + amount)
            }
            console.log('All messages extracted.');
          })
          let amount: number = await countPromise

          resolve(true);
        } catch (error) {
          console.error('Error in releaseBufferedMessages:', error);
          reject(false);
        }

        async function checkMessageCount(messageModel: Model<Message>, bufferIdentifier: string): Promise<any> {
          return new Promise((resolve, reject) => {
            messageModel.countDocuments({}).then((count) => {
              console.log(`There is ${count} messages in database under ${bufferIdentifier} at the moment. Releasing them....`);
              resolve(count)
            }).catch((error) => {
              console.error(error)
              reject(error)
            })
          })
        }

        // Stream all the data inside the database out and deleting them
        async function extractData(messageFromBuffer: Subject<Message>, messageModel: Model<Message>): Promise<any> {
          return new Promise((resolve, reject) => {
            const stream = messageModel.find().cursor();

            stream.on("data", async (message) => {
              // Process each message individually`
              messageFromBuffer.next(message);
            });

            stream.on("error", (error) => {
              console.error("Error streaming messages from MongoDB:", error);
              reject(error);
            });

            stream.on("end", async () => {
              // Delete the data once it has been streamed
              try {
                if (messageModel) {
                  await messageModel.deleteMany({});
                  console.log("Data in Mongo deleted successfully.");
                } else {
                  console.log(`Message Mongoose Model is not intiated properly...`);
                }
              } catch (err) {
                console.error("Error deleting data:", err);
              }
              resolve(true);
            });
          })
        }
      }
      if (!this.messageModel) {
        // If MongoDB model is not defined, use the local buffer
        console.log(`Releasing buffer Message under ${this.bufferIdentifier}: currently there is ${this.messageBuffered.length} messages to be released`);
        this.messageBuffered.forEach((message) =>
          this.messageStream.next(message)
        );
        this.messageBuffered.length = 0; // Clear the local buffer after transferring
        if (this.messageBuffered.length < 1) {
          resolve(true);
        } else {
          reject(`Somehow the array is not emptied. This should not happen`);
        }
      }
    });
  }

  private async transferLocalBufferToMongoDB(): Promise<void> {
    return new Promise((resolve, reject) => {
      console.log(`Transferring local array buffered Message under ${this.bufferIdentifier}: currently there is ${this.messageBuffered.length}. Transferring to database...`);
      if (this.messageModel) {
        let locallyBufferedMessage: Observable<Message> = from(this.messageBuffered);
        locallyBufferedMessage.subscribe({
          next: async (message: Message) => {
            try {
              if (this.messageModel) {
                await this.messageModel.create(message);
                console.log(`Transferring ${(message.message as MessageLog).appData.msgId} into database.`);
              }
            } catch (error) {
              console.error("Error transferring message to MongoDB:", error);
            }
          },
          error: (err) => console.error(err),
          complete: () => {
            if (this.messageModel) {
              this.messageModel.countDocuments({}).then((count) => {
                console.log(`Local buffered message transfer completed. There is a total of ${count} messages in database ${this.bufferIdentifier} at the moment.`)
                this.messageBuffered = [] // Clear local buffer after transferring
              });
            }
          },
        });
      }
    });
  }

  // Additional methods as required...
}