Fis-Retransmission/services/buffer.service.ts

// bufferService.ts
import { BehaviorSubject, Observable, Subject, from, map, switchMap } from 'rxjs';
import mongoose, { Connection, Model, Document } from 'mongoose';
import { ConnectionState, Message, MessageLog } from '../interfaces/general.interface';
import { resolve } from 'path';

export class BufferService {
    private messageFromApplication: Subject<Message>;
    private messageFromBuffer: Subject<Message>
    private connectionState: BehaviorSubject<ConnectionState>
    private currentSource: Subject<Message>
    private messageBuffer: Message[] = [];
    private messageModel: Model<Message> | undefined;
    private readonly dbUrl: string = process.env.MONGO as string;


    constructor(
        messageFromApp: Subject<Message>,
        connectionStateSubject: BehaviorSubject<ConnectionState>,
        dbName: string
    ) {
        this.messageFromBuffer = new Subject<Message>();
        this.messageFromApplication = messageFromApp;
        this.currentSource = this.messageFromBuffer
        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

        /* Disable for now. Use local array first */
        // this.initializeDatabaseConnection(dbName).then((connection: mongoose.Connection) => {
        //     const grpcMessageSchema = require('../models/message.schema');
        //     this.messageModel = connection.model<Message>('Message', grpcMessageSchema)
        //     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
        // });
    }

    public getMessages(): Observable<Message> {
        return this.currentSource as Observable<Message>
    }

    private setupSubscriptions(): void {
        this.messageFromApplication.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: Message): void {
        if (this.connectionState.getValue().status === 'BUFFER') {
            this.bufferMessage(message);
        }
        if (this.connectionState.getValue().status === 'DIRECT_PUBLISH') {
            // console.log(`There is ${this.messageBuffer.length} messages in the buffer`) // somehw this is called repeatedly
            // do nothing for now
            // Start releasing buffered messages, but don't wait for it to complete
            // this.isBufferNotEmpty().then(isNotEmpty => {
            //     if (isNotEmpty) {
            //         // this.releaseBufferedMessages(this.messageFromBuffer);
            //         // Continue to buffer new incoming message during the release process
            //         this.bufferMessage(message);
            //     } else {
            //         // If buffer is empty, switch source and handle the message directly
            //         this.messageToBePublished.next(message); // Handle the message directly
            //     }
            // });
        }
    }

    private handleConnectionStateChanges(state: ConnectionState): void {
        console.log(this.connectionState.getValue().status)
        if (state.status === 'BUFFER') {
            this.currentSource = this.messageFromBuffer
            if (state.payload && typeof state.payload !== 'string') {
                this.bufferMessage(state.payload); // Buffer the last message immediately
            }
        }
        if (state.status === 'DIRECT_PUBLISH') {
            if (this.messageBuffer.length > 0) { // temporary only since i am not using mongoDB for buffering atm
                this.releaseBufferedMessages(this.messageFromBuffer).then(() => {
                    console.log(`Switching to main publisher from source`)
                    this.currentSource = this.messageFromApplication
                }).catch((err) => {
                    console.error(err)
                })
            }
            if (this.messageBuffer.length < 1) {
                this.currentSource = this.messageFromApplication
            }
            /* This is for mongo  */
            // this.isBufferNotEmpty().then(isNotEmpty => {
            //     if (isNotEmpty) {
            //         this.currentMessageSource = this.messageFromBuffer
            //     } else {
            //         this.currentMessageSource = this.messageToBePublished
            //     }
            // })
        }
    }

    private async isBufferNotEmpty(): Promise<boolean> {
        if (this.messageModel) {
            // Check the count in MongoDB
            const count = await this.messageModel.estimatedDocumentCount().exec();
            return count > 0;
        } else {
            // Check the local buffer
            return this.messageBuffer.length > 0;
        }
    }

    private async bufferMessage(message: Message): Promise<void> {
        if (this.messageModel) {
            try {
                // const newMessage = new this.messageModel(message);
                await this.messageModel.create(message);
                console.log(`Message${(message.message as MessageLog).appData.msgId} saved to MongoDB buffer`);
            } catch (error) {
                console.error('Error saving message to MongoDB:', error);
                // Implement retry logic or additional error handling here
            }
        } else {
            this.messageBuffer.push(message); // Fallback to local buffer if model is not defined
            console.log(`pushing ${(message.message as MessageLog).appData.msgId} into local array buffer.... There is now ${this.messageBuffer.length} messages`)
        }
    }


    private releaseBufferedMessages(messageFromBuffer: Subject<Message>): Promise<boolean> {
        return new Promise((resolve, reject) => {
            if (this.messageModel) {
                const stream = this.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 (this.messageModel) {
                            await this.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: currently there is ${this.messageBuffer.length} messages to be released`)
                this.messageBuffer.forEach(message => this.messageFromBuffer.next(message));
                this.messageBuffer.length = 0 // Clear the local buffer after transferring
                if (this.messageBuffer.length < 1) {
                    resolve(true)
                } else {
                    reject(`Somehow the array is not emptied. This should not happen`)
                }
            }
        })
    }

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

    private async transferLocalBufferToMongoDB(): Promise<void> {
        if (this.messageModel) {
            this.messageBuffer.forEach(async message => {
                try {
                    if (this.messageModel) {
                        await this.messageModel.create(message);
                    }
                } catch (error) {
                    console.error('Error transferring message to MongoDB:', error);
                }
            })
            this.messageBuffer = []; // Clear local buffer after transferring
        }
    }

    // Additional methods as required...
}