Fis-MessageTransmission/src/utils/socket.utils.ts

import { BehaviorSubject, Observable, Observer, Subject } from 'rxjs';
import { createServer } from 'http';
import { Server, Socket as SocketForConnectedClient } from 'socket.io';
import { io, Socket as ClientSocket } from 'socket.io-client';
import * as fs from 'fs'
import { v4 as uuidv4 } from 'uuid'
import { ConnectionState, Transport, TransportEvent, TransportMessage } from '../interface/connector.interface';
import { ConnectedClientSocket, ConnectedServerSocket } from '../transport/websocket';
import { EventMessage } from '../interface/transport.interface';

export function startSocketServer(port: number): Observable<SocketForConnectedClient> {
    return new Observable((observer) => {
        try {
            console.log(`Socket Server ${port} Started....`)
            let httpServer = createServer();
            let socketServer = new Server(httpServer)

            // something wrong here
            socketServer.on('connection', (socket) => {
                observer.next(socket)
            })

            socketServer.engine.on("connection_error", (err) => {
                console.log(err.req);      // the request object
                console.log(err.code);     // the error code, for example 1
                console.log(err.message);  // the error message, for example "Session ID unknown"
                console.log(err.context);  // some additional error context
            });

            // Start the socketServer
            httpServer.listen(port)
        } catch (error) {
            observer.error(error)
        }
    })
}

export async function startClientSocketConnection(serverUrl: string): Promise<ClientSocket> {
    return new Promise((resolve, reject) => {
        try {
            // let clientSocket = io(serverUrl)
            let clientSocket: ClientSocket = io(serverUrl, {
                reconnection: true,              // Enable automatic reconnections
                reconnectionAttempts: 1000,       // Retry up to 10 times
                reconnectionDelay: 500,          // Start with a 500ms delay
                reconnectionDelayMax: 10000,     // Delay can grow to a max of 10 seconds
                randomizationFactor: 0.3,
            })
            resolve(clientSocket)
        }
        catch (error) {
            reject(error)
        }
    })
}

// After establishing connection to the server, set up the credentials, confirm whether or not if there's any credentials, if not ask for one from the server
export function handleClientSocketConnection(socket: ClientSocket, serversConnected: ConnectedServerSocket[]): Observable<TransportEvent> {
    return new Observable((eventNotification: Observer<TransportEvent>) => {
        let clientName!: string
        let buffer: any[] = []
        let receiverProfileInfo!: ConnectedServerSocket

        // Listen for a connection event
        socket.on('connect', () => {
            console.log('Connected to the server:', socket.id)
            if (clientName) {
                checkOwnClientInfo(clientName).then((profile: ConnectedServerSocket) => {
                    receiverProfileInfo = profile
                    socket.emit('profile', {
                        name: 'Old Client',
                        data: profile
                    })
                }).catch((error) => {
                    socket.emit('profile', {
                        name: 'New Client',
                        data: null
                    })
                })
            } else {
                socket.emit('profile', {
                    name: 'New Client',
                    data: null
                })
            }
        });

        // Listen for messages from the server. Generally here's the responses
        socket.on('message', (msg: any) => {
            console.log(`Websocket Client Transport Receieve Msg`, msg.id)
            if (receiverProfileInfo) {
                // publish to event
                eventNotification.next({
                    id: uuidv4(),
                    event: 'New Message',
                    data: msg
                })
            } else {
                // Do nothing. just store in local array first. Cannot process without information. but then again, don['t need information if acting as client
                // but for consistency sake, will impose the standard 
                buffer.push(msg) // store locally for now
            }
        })

        socket.on('profile', (data: { name: string, message: any }) => {
            console.log(data)
            if (data.name == 'New Profile') {
                console.log(`Assigned client Name: ${(data.message as ConnectedServerSocket).id}`)
                receiverProfileInfo = data.message as ConnectedServerSocket
                writeFile(data.message as ConnectedServerSocket, (data.message as ConnectedServerSocket).id).then(() => {
                    clientName = receiverProfileInfo.id
                    // broadcast event to allow retransmission to release buffer
                    eventNotification.next({
                        id: uuidv4(),
                        event: `New Server`,
                        data: {
                            clientId: (data.message as ConnectedServerSocket).id,
                            message: `New Websocket Channel ${(data.message as ConnectedServerSocket).id} established.`
                        } as EventMessage
                    })

                }).catch((error) => { }) // do nothing at the moment. 
                // Update websocket instance record
                receiverProfileInfo = {
                    id: (data.message as ConnectedServerSocket).id,
                    dateCreated: new Date(),
                    socketInstance: socket,
                    connectionState: new BehaviorSubject<ConnectionState>(`ONLINE`)
                }
                serversConnected.push(receiverProfileInfo)
            }
            if (data.name == 'Adjusted Profile') {
                console.log(`Assigned client Name: ${(data.message as ConnectedServerSocket).id}`)
                receiverProfileInfo = data.message as ConnectedServerSocket
                writeFile(data.message as ConnectedServerSocket, (data.message as ConnectedServerSocket).id).then(() => {
                    // broadcast event to allow retransmission to release buffer
                    eventNotification.next({
                        id: uuidv4(),
                        event: 'Client Reconnected',
                        data: {
                            clientId: (data.message as ConnectedServerSocket).id,
                            message: `Existing Websocket Channel ${(data.message as ConnectedServerSocket).id} re-established.`
                        } as EventMessage
                    })
                }).catch((error) => { }) // do nothing at the moment. 
                // Update websocket instance record
                let clientObj: ConnectedServerSocket | undefined = serversConnected.find(obj => obj.id === (data.message as ConnectedServerSocket).id)
                if (clientObj) {
                    clientObj.socketInstance = receiverProfileInfo.socketInstance
                    clientObj.connectionState.next('ONLINE')
                }
            }
            if (data.name == 'Error') {
                console.log(`Server cannot find credentials`, data.message)
                // logic to request for new credentials
                setTimeout(() => {
                    socket.emit('profile', {
                        name: 'New Client',
                        data: null
                    })
                }, 2000)
            }
        })

        // Handle disconnection
        socket.on('disconnect', () => {
            console.log('Websocket Client disconnected from the server');
            if (receiverProfileInfo) {
                eventNotification.next({
                    id: uuidv4(),
                    event: `Client Disconnected`,
                    data: {
                        clientId: receiverProfileInfo.id,
                        message: `Server for Channel ${receiverProfileInfo.id} disconnected.`
                    } as EventMessage
                })
                receiverProfileInfo.connectionState.next(`OFFLINE`)
            }
        });
    })
}

// For SERVER Usage: set up socket listeners to start listening for different events
export function handleNewSocketClient(socket: SocketForConnectedClient, connectedClientSocket: ConnectedClientSocket[]): Observable<TransportEvent> {
    return new Observable((event: Observer<TransportEvent>) => {
        console.log(`Setting up listeners for socket:${socket.id}`)
        // returns the socket client instance 
        // listen to receiver's initiotion first before assigning 'credentials'
        socket.on(`profile`, (message: { name: string, data: any }) => {
            if (message.name == 'New Client') {
                let clientInstance: ConnectedClientSocket = {
                    id: uuidv4(), // client should only be assigned at this level. And is passed around for reference pointing
                    dateCreated: new Date(),
                    socketInstance: socket,
                    connectionState: new BehaviorSubject<ConnectionState>(`ONLINE`)
                }
                // send to receiver for reference
                socket.emit('profile', {
                    name: `New Profile`, message: { id: clientInstance.id }
                })
                // publish first event notification
                event.next({
                    id: uuidv4(),
                    event: `New Client`,
                    data: {
                        clientId: clientInstance.id,
                        message: `New Client Connected. Adapter ID assigned: ${clientInstance.id}`,
                        payload: clientInstance
                    } as EventMessage
                })
                // Update connected clientInstance info to adapter
                connectedClientSocket.push(clientInstance)
                startListening(socket, clientInstance, event)
            } else {
                // update first
                let clientInstance: ConnectedClientSocket | undefined = connectedClientSocket.find(obj => obj.id === message.data.id)
                if (clientInstance) {
                    console.log(`Socket Client ${clientInstance.id} Found`)
                    socket.emit('profile', { name: 'Adjusted Profile', message: { id: clientInstance.id } })
                    // replace socket instance since the previous has been terminated
                    clientInstance.socketInstance = socket
                    // need to start listening again, because it's assigned a different socket instance this time round
                    startListening(socket, clientInstance, event)
                    event.next({
                        id: uuidv4(),
                        event: 'Client Reconnected',
                        data: {
                            clientId: clientInstance.id,
                            message: `Client ${clientInstance.id} connection re-established`,
                            payload: clientInstance
                        } as EventMessage
                    })
                    // Resume operation
                    if (clientInstance.connectionState.getValue() == 'OFFLINE') {
                        clientInstance.connectionState.next(`ONLINE`)
                    }
                } else {
                    console.log(`Profile Not Found`)
                    socket.emit('profile', { name: 'Error', message: 'Receiver Profile Not found' })
                }
            }
        })
    })
}


// Specifically to write receiver profile information
export async function writeFile(data: ConnectedServerSocket, filename: string): Promise<boolean> {
    return new Promise((resolve, reject) => {
        // Write JSON data to a file
        fs.writeFile(`${filename}.json`, JSON.stringify(data, null, 2), (err) => {
            if (err) {
                console.error('Error writing file', err);
                reject(false)
            } else {
                console.log('File has been written');
                resolve(true)
            }
        });
    })
}



// Check if filename exists. Return profile information if there's any
export async function checkOwnClientInfo(filename?: string): Promise<ConnectedServerSocket> {
    return new Promise((resolve, reject) => {
        // Check if the file exists
        if (fs.existsSync(`${filename}.json`)) {
            try {
                // Read the file contents
                const fileData = fs.readFileSync(`${filename}.json`, 'utf8');

                // If the file is empty, return an error
                if (fileData.trim() === "") {
                    throw new Error("File is empty");
                }

                // Parse and return the data if present
                const jsonData = JSON.parse(fileData);
                resolve(jsonData)

            } catch (err) {
                // Handle parsing errors or other file-related errors
                console.error("Error reading or parsing file:", err);
                reject('');
            }
        } else {
            console.error("File does not exist");
            reject('');
        }
    })
}

export function startListening(socket: SocketForConnectedClient, client: ConnectedClientSocket, eventListener: Observer<TransportEvent>): void {
    /* Generally, we don't need this unless in the case of being the receiver */
    socket.on('message', (message: any) => {
        eventListener.next({
            id: uuidv4(),
            event: 'New Message',
            data: {
                id: uuidv4(),
                dateCreated: new Date(),
                transport: Transport.Websocket,
                target: client.id, // this ref to be associated with the client/channel
                payload: message
            } as TransportMessage
        })
    })

    socket.on('disconnect', () => {
        eventListener.next({
            id: uuidv4(),
            event: 'Server Disconnected',
            data: {
                clientID: client.id,
                time: new Date()
            }
        })
        eventListener.error(`Client ${client.id} disconnected. Terminating this observable event for this client socket...`)
        eventListener.complete()
    })
}