tf-events.ts

const gruSdk = createGruSdk({
    serviceName: 'demo-backend',
    endpoint: 'https://vector.tf.eulerstream.com',
    headers: {
        authorization: `Basic ${Buffer.from('vector:' + process.env.VECTOR_AUTH_PASSWORD).toString('base64')}`,
    },
});

        tiktokConnectionWrapper.connection.on('decodedData', (protoName, decodedData, payload) => {
            if (!payload) return;
            if (!roomId) {
                pendingEvents.push({protoName, payload});
                return;
            }
            gruSdk.webcastEvents.emit({
                body: new Uint8Array(payload),
                attributes: { protoName, roomId }
            });
        });

        // Forward errors to the Gru Events SDK
        tiktokConnectionWrapper.connection.on('error', (err) => {
            gruSdk.webcastInfo.emit({
                body: {error: {message: err?.message || (typeof err === 'string' ? err : JSON.stringify(err)) || 'Unknown error'}},
                attributes: {
                    objectType: WebcastInfoObjectType.WEBCAST_ROOM_ERROR,
                    uniqueId,
                },
            });
        });

 // Redirect wrapper control events once
        tiktokConnectionWrapper.once('connected', state => {
            socket.emit('tiktokConnected', state);
            roomId = roomId || state.roomId;
            const resolvedRoomId = roomId;
            gruSdk.webcastInfo.emit({
                body: {roomId: resolvedRoomId, roomInfo: state.roomInfo || {}},
                attributes: {
                    objectType: WebcastInfoObjectType.WEBCAST_ROOM_CONNECT,
                    roomId: resolvedRoomId,
                },
            });

            // Flush any events that arrived before roomId was set
            for (const evt of pendingEvents) {
                gruSdk.webcastEvents.emit({
                    body: new Uint8Array(evt.payload),
                    attributes: { protoName: evt.protoName, roomId: resolvedRoomId }
                });
            }
            pendingEvents = [];

        });
        tiktokConnectionWrapper.on('disconnected', reason => {
            socket.emit('tiktokDisconnected', reason);
            gruSdk.webcastInfo.emit({
                body: {code: reason?.code ?? 0, reason: typeof reason === 'string' ? reason : reason?.reason ?? ''},
                attributes: {
                    objectType: WebcastInfoObjectType.WEBCAST_ROOM_DISCONNECT,
                    uniqueId,
                    roomId,
                },
            });
        });

        // Now that all listeners are registered, start the connection
        tiktokConnectionWrapper.connect();
    });



                const wrapper = new TikTokConnectionWrapper(username, options, true);
                bulkConnections.push(wrapper);

                wrapper.once('connected', (state) => {
                    socket.emit('bulkConnected', username);
                    console.log(`[Bulk] Connected to ${username}`);

                    const roomId = wrapper.connection.roomId || state.roomId;

                    // Emit connect event to Gru SDK
                    gruSdk.webcastInfo.emit({
                        body: {roomId, roomInfo: state.roomInfo || {}},
                        attributes: {
                            objectType: WebcastInfoObjectType.WEBCAST_ROOM_CONNECT,
                            roomId,
                        },
                    });

                    // Forward all events to Gru SDK
                    wrapper.connection.on('decodedData', (protoName, decodedData, payload) => {
                        if (roomId && payload) {
                            gruSdk.webcastEvents.emit({
                                body: new Uint8Array(payload),
                                attributes: { protoName, roomId }
                            });
                        }
                    });

                    // Notify browser of events for stats
                    for (const eventType of eventTypes) {
                        wrapper.connection.on(eventType, () => {
                            socket.emit('bulkEvent', username, eventType);
                        });
                    }
                });

                wrapper.on('disconnected', (reason) => {
                    const reasonStr = typeof reason === 'string' ? reason : reason?.reason || JSON.stringify(reason) || 'Unknown';
                    socket.emit('bulkDisconnected', username, reasonStr);
                    console.log(`[Bulk] ${username} disconnected: ${reasonStr}`);

                    const roomId = wrapper.connection.roomId;
                    gruSdk.webcastInfo.emit({
                        body: {code: reason?.code ?? 0, reason: typeof reason === 'string' ? reason : reason?.reason ?? ''},
                        attributes: {
                            objectType: WebcastInfoObjectType.WEBCAST_ROOM_DISCONNECT,
                            uniqueId: username,
                            roomId,
                        },
                    });
                });

                wrapper.connection.on('error', (err) => {
                    gruSdk.webcastInfo.emit({
                        body: {error: {message: err?.message || (typeof err === 'string' ? err : JSON.stringify(err)) || 'Unknown error'}},
                        attributes: {
                            objectType: WebcastInfoObjectType.WEBCAST_ROOM_ERROR,
                            uniqueId: username,
                        },
                    });
                });

                wrapper.connect();



// Graceful shutdown
process.on('SIGTERM', async () => {
    await gruSdk.shutdown();
    process.exit(0);
});