PxlLoewe/var-monorepo
1
0
Fork 0
Code Issues 22 Pull Requests Actions Packages Projects Releases 2 Wiki Activity

completed Audio frontend

Browse Source
This commit is contained in:
PxlLoewe
2025-03-17 00:41:07 -07:00
parent e722adaf8e
commit 9a7049ec44
28 changed files with 252 additions and 889 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
apps/dispatch-server/routes/livekit.ts
View File

@@ -5,12 +5,12 @@ if (!process.env.LIVEKIT_API_KEY) throw new Error("LIVEKIT_API_KEY not set");
if (!process.env.LIVEKIT_API_SECRET)
throw new Error("LIVEKIT_API_SECRET not set");
const createToken = async () => {
const createToken = async (roomName: string) => {
// If this room doesn't exist, it'll be automatically created when the first
// participant joins
const roomName = "quickstart-room";
// Identifier to be used for participant.
// It's available as LocalParticipant.identity with livekit-client SDK
// TODO: Move function to dispatch nextjs app as API route to use authentication of nextAuth
const participantName =
"quickstart-username" + Math.random().toString(36).substring(7);
@@ -31,8 +31,10 @@ const createToken = async () => {
const router = Router();
router.get("/token", async (req, res) => {
const roomName = req.query.roomName as string;
console.log("roomName", roomName);
res.send({
token: await createToken(),
token: await createToken(roomName),
});
});

2
apps/dispatch-server/socket-events/connect-dispatch.ts
View File

@@ -14,6 +14,8 @@ export const handle =
console.log("User connected to dispatch server");
await pubClient.json.set(`dispatchers:${socket.id}`, "$", {
logoffTime,
loginTime: new Date().toISOString(),
lastSeen: new Date().toISOString(),
selectedZone,
userId,
});

32
apps/dispatch/app/(dispatch)/_components/ToggleTalkButton.tsx
View File

@@ -1,32 +0,0 @@
"use client";
import { useTalkStore } from "../../_store/useTalkStore";
export const ToggleTalkButton = () => {
const { isTalking, toggleTalking } = useTalkStore();
return (
<button
onClick={toggleTalking}
className={`${
isTalking
? "bg-red-500 hover:bg-red-600"
: "bg-transparent hover:bg-neutral-300"
}`}
>
<svg
xmlns="http://www.w3.org/2000/svg"
className="h-5 w-5"
fill="none"
viewBox="0 0 24 24"
stroke="currentColor"
>
<path
strokeLinecap="round"
strokeLinejoin="round"
strokeWidth="2"
d="M12 18.75a6 6 0 0 0 6-6v-1.5m-6 7.5a6 6 0 0 1-6-6v-1.5m6 7.5v3.75m-3.75 0h7.5M12 15.75a3 3 0 0 1-3-3V4.5a3 3 0 1 1 6 0v8.25a3 3 0 0 1-3 3Z"
/>
</svg>
</button>
);
};

12
apps/dispatch/app/(dispatch)/_components/navbar/Navbar.tsx
View File

@@ -1,11 +1,10 @@
"use client";
import { ToggleTalkButton } from "../ToggleTalkButton";
import Link from "next/link";
import { Connection } from "../Connection";
import { Connection } from "./_components/Connection";
import { ThemeSwap } from "./_components/ThemeSwap";
import { useState } from "react";
import { Audio } from "./_components/Audio";
import { useState } from "react";
export default function Navbar() {
const [isDark, setIsDark] = useState(false);
@@ -25,14 +24,7 @@ export default function Navbar() {
<a className="btn btn-ghost text-xl">VAR Leitstelle V2</a>
</div>
<div className="flex gap-2">
<ul className="menu menu-horizontal bg-base-200 rounded-box flex items-center gap-2">
<li>
<ToggleTalkButton />
</li>
<li>
<Audio />
</li>
</ul>
</div>
<div className="flex gap-6">
<ThemeSwap isDark={isDark} toggleTheme={toggleTheme} />

216
apps/dispatch/app/(dispatch)/_components/navbar/_components/Audio.tsx
View File

@@ -1,159 +1,111 @@
"use client";
import { useEffect, useState } from "react";
import { useConnectionStore } from "_store/connectionStore";
import {
LocalParticipant,
LocalTrackPublication,
Participant,
RemoteParticipant,
RemoteTrack,
RemoteTrackPublication,
Room,
RoomEvent,
Track,
VideoPresets,
} from "livekit-client";
import { connectionStore } from "../../../../_store/connectionStore";
Circle,
Disc,
Mic,
PlugZap,
ShieldQuestion,
Signal,
SignalLow,
SignalMedium,
WifiOff,
ZapOff,
} from "lucide-react";
import { useAudioStore } from "_store/audioStore";
import { cn } from "helpers/cn";
import { ConnectionQuality } from "livekit-client";
const ROOMS = ["LST_01", "LST_02", "LST_03", "LST_04", "LST_05"];
export const Audio = () => {
const connection = connectionStore();
const [token, setToken] = useState("");
const [room, setRoom] = useState<Room | null>(null);
useEffect(() => {
const fetchToken = async () => {
const response = await fetch(
`${process.env.NEXT_PUBLIC_DISPATCH_SERVER_URL}/livekit/token`,
);
const data = await response.json();
setToken(data.token);
};
fetchToken();
}, []);
const connection = useConnectionStore();
const {
isTalking,
toggleTalking,
connect,
state,
connectionQuality,
disconnect,
remoteParticipants,
room,
} = useAudioStore();
const [selectedRoom, setSelectedRoom] = useState<string>("LST_01");
useEffect(() => {
const joinRoom = async () => {
if (!connection.isConnected) return;
if (!token) return;
if (!process.env.NEXT_PUBLIC_LIVEKIT_URL)
return console.error("NEXT_PUBLIC_LIVEKIT_URL not set");
console.log("Connecting to room", {
token,
url: process.env.NEXT_PUBLIC_LIVEKIT_URL,
});
const url = "ws://localhost:7880";
/* const token =
"eyJhbGciOiJIUzI1NiJ9.eyJ2aWRlbyI6eyJyb29tSm9pbiI6dHJ1ZSwicm9vbSI6InF1aWNrc3RhcnQtcm9vbSJ9LCJpc3MiOiJBUElBbnNHZHRkWXAySG8iLCJleHAiOjE3NDIxNjAwOTIsIm5iZiI6MCwic3ViIjoicXVpY2tzdGFydC11c2VybmFtZSJ9.ih7my6oXby6yYBfzt5LXHKN5WZU9exIqS8CdpRJRLQI"; */
console.log("Connecting to room", { token, url });
const room = new Room({
// automatically manage subscribed video quality
adaptiveStream: true,
// optimize publishing bandwidth and CPU for published tracks
dynacast: true,
// default capture settings
videoCaptureDefaults: {
resolution: VideoPresets.h720.resolution,
},
});
// pre-warm connection, this can be called as early as your page is loaded
room.prepareConnection(url, token);
// set up event listeners
room
.on(RoomEvent.TrackSubscribed, handleTrackSubscribed)
.on(RoomEvent.TrackUnsubscribed, handleTrackUnsubscribed)
.on(RoomEvent.ActiveSpeakersChanged, handleActiveSpeakerChange)
.on(RoomEvent.Disconnected, handleDisconnect)
.on(RoomEvent.LocalTrackUnpublished, handleLocalTrackUnpublished);
// connect to room
await room.connect(url, token);
console.log("connected to room", room.name);
// publish local camera and mic tracks
await room.localParticipant.enableCameraAndMicrophone();
function handleTrackSubscribed(
track: RemoteTrack,
publication: RemoteTrackPublication,
participant: RemoteParticipant,
) {
if (
track.kind === Track.Kind.Video ||
track.kind === Track.Kind.Audio
) {
// attach it to a new HTMLVideoElement or HTMLAudioElement
const element = track.attach();
}
}
function handleTrackUnsubscribed(
track: RemoteTrack,
publication: RemoteTrackPublication,
participant: RemoteParticipant,
) {
// remove tracks from all attached elements
track.detach();
}
function handleLocalTrackUnpublished(
publication: LocalTrackPublication,
participant: LocalParticipant,
) {
// when local tracks are ended, update UI to remove them from rendering
publication.track?.detach();
}
function handleActiveSpeakerChange(speakers: Participant[]) {
// show UI indicators when participant is speaking
}
function handleDisconnect() {
console.log("disconnected from room");
}
setRoom(room);
if (state === "connected") return;
connect(selectedRoom);
};
joinRoom();
return () => {
room?.disconnect();
disconnect();
};
}, [token, connection.isConnected]);
}, [connection.isConnected]);
return (
<>
<details className="dropdown">
<summary className="dropdown flex items-center gap-1">
<svg
xmlns="http://www.w3.org/2000/svg"
className="h-5 w-5"
fill="none"
viewBox="0 0 24 24"
stroke="currentColor"
<div className="bg-base-200 rounded-box flex items-center gap-2 p-3">
<button
onClick={() => {
if (state === "connected") toggleTalking();
if (state === "disconnected") connect(selectedRoom);
}}
className={cn(
"btn btn-sm btn-soft border-none hover:bg-inherit",
!isTalking && "bg-transparent hover:bg-sky-400/20",
isTalking && "bg-red-500 hover:bg-red-600",
state === "disconnected" && "bg-red-500 hover:bg-red-500",
state === "connecting" &&
"bg-yellow-500 hover:bg-yellow-500 cursor-default",
)}
>
<path
strokeLinecap="round"
strokeLinejoin="round"
strokeWidth="2"
d="M9 19v-6a2 2 0 00-2-2H5a2 2 0 00-2 2v6a2 2 0 002 2h2a2 2 0 002-2zm0 0V9a2 2 0 012-2h2a2 2 0 012 2v10m-6 0a2 2 0 002 2h2a2 2 0 002-2m0 0V5a2 2 0 012-2h2a2 2 0 012 2v14a2 2 0 01-2 2h-2a2 2 0 01-2-2z"
/>
</svg>
<div className="badge badge-soft badge-success">1</div>
{state === "connected" && <Mic />}
{state === "disconnected" && <WifiOff />}
{state === "connecting" && <PlugZap />}
</button>
{state === "connected" && (
<details className="dropdown">
<summary className="dropdown btn btn-ghost flex items-center gap-1">
{connectionQuality === ConnectionQuality.Excellent && <Signal />}
{connectionQuality === ConnectionQuality.Good && <SignalMedium />}
{connectionQuality === ConnectionQuality.Poor && <SignalLow />}
{connectionQuality === ConnectionQuality.Lost && <ZapOff />}
{connectionQuality === ConnectionQuality.Unknown && (
<ShieldQuestion />
)}
<div className="badge badge-soft badge-success">
{remoteParticipants}
</div>
</summary>
<ul className="menu dropdown-content bg-base-100 rounded-box z-1 w-52 p-2 shadow-sm">
<li>
<a>Rufgruppe 1</a>
</li>
<li>
<a>Rufgruppe 2</a>
<ul className="menu dropdown-content bg-base-100 rounded-box z-[99999999] w-52 p-2 shadow-sm">
{ROOMS.map((r) => (
<li key={r}>
<button
className="btn btn-sm btn-ghost text-left"
onClick={() => {
if (selectedRoom === r) return;
setSelectedRoom(r);
connect(r);
}}
>
{room?.name === r && (
<Disc className="text-success text-sm" width={15} />
)}
{r}
</button>
</li>
))}
</ul>
</details>
)}
</div>
</>
);
};

4
apps/dispatch/app/(dispatch)/_components/Connection.tsx → apps/dispatch/app/(dispatch)/_components/navbar/_components/Connection.tsx
View File

@@ -1,12 +1,12 @@
"use client";
import { useSession } from "next-auth/react";
import { connectionStore } from "../../_store/connectionStore";
import { useConnectionStore } from "../../../../_store/connectionStore";
import { useEffect, useRef, useState } from "react";
import { useForm } from "react-hook-form";
export const ConnectionBtn = () => {
const modalRef = useRef<HTMLDialogElement>(null);
const connection = connectionStore((state) => state);
const connection = useConnectionStore((state) => state);
const [form, setForm] = useState({
logoffTime: "",
selectedZone: "LST_01",

82
apps/dispatch/app/_store/audioStore.ts Normal file
View File

@@ -0,0 +1,82 @@
import {
handleActiveSpeakerChange,
handleDisconnect,
handleLocalTrackUnpublished,
handleTrackSubscribed,
handleTrackUnsubscribed,
} from "helpers/liveKitEventHandler";
import { ConnectionQuality, Room, RoomEvent } from "livekit-client";
import { create } from "zustand";
let interval: NodeJS.Timeout;
type TalkState = {
isTalking: boolean;
state: "connecting" | "connected" | "disconnected";
connectionQuality: ConnectionQuality;
remoteParticipants: number;
toggleTalking: () => void;
connect: (roomName: string) => void;
disconnect: () => void;
room: Room | null;
};
const getToken = async (roomName: string) => {
const response = await fetch(
`${process.env.NEXT_PUBLIC_DISPATCH_SERVER_URL}/livekit/token?roomName=${roomName}`,
);
const data = await response.json();
return data.token;
};
export const useAudioStore = create<TalkState>((set, get) => ({
isTalking: false,
state: "disconnected",
remoteParticipants: 0,
connectionQuality: ConnectionQuality.Unknown,
room: null,
toggleTalking: () => set((state) => ({ isTalking: !state.isTalking })),
connect: async (roomName) => {
const connectedRoom = get().room;
if (interval) clearInterval(interval);
if (connectedRoom) {
connectedRoom.disconnect();
connectedRoom.removeAllListeners();
}
set({ state: "connecting" });
const url = process.env.NEXT_PUBLIC_LIVEKIT_URL;
if (!url) return console.error("NEXT_PUBLIC_LIVEKIT_URL not set");
const token = await getToken(roomName);
const room = new Room();
await room.prepareConnection(url, token);
room
// Connection events
.on(RoomEvent.Connected, () => {
set({ state: "connected", room });
})
.on(RoomEvent.Disconnected, () => {
set({ state: "disconnected" });
handleDisconnect();
})
.on(RoomEvent.ConnectionQualityChanged, (connectionQuality) =>
set({ connectionQuality }),
)
// Track events
.on(RoomEvent.TrackSubscribed, handleTrackSubscribed)
.on(RoomEvent.TrackUnsubscribed, handleTrackUnsubscribed)
.on(RoomEvent.ActiveSpeakersChanged, handleActiveSpeakerChange)
.on(RoomEvent.LocalTrackUnpublished, handleLocalTrackUnpublished);
await room.connect(url, token);
interval = setInterval(() => {
set({
remoteParticipants:
room.numParticipants === 0 ? 0 : room.numParticipants - 1, // Unreliable and delayed
});
}, 500);
},
disconnect: () => {
get().room?.disconnect();
},
}));

6
apps/dispatch/app/_store/connectionStore.ts
View File

@@ -12,7 +12,7 @@ interface ConnectionStore {
disconnect: () => void;
}
export const connectionStore = create<ConnectionStore>((set) => ({
export const useConnectionStore = create<ConnectionStore>((set) => ({
isConnected: false,
selectedZone: "LST_01",
connect: async (uid, selectedZone, logoffTime) =>
@@ -34,8 +34,8 @@ export const connectionStore = create<ConnectionStore>((set) => ({
}));
socket.on("connect", () => {
connectionStore.setState({ isConnected: true });
useConnectionStore.setState({ isConnected: true });
});
socket.on("disconnect", () => {
connectionStore.setState({ isConnected: false });
useConnectionStore.setState({ isConnected: false });
});

11
apps/dispatch/app/_store/useTalkStore.ts
View File

@@ -1,11 +0,0 @@
import { create } from "zustand";
type TalkState = {
isTalking: boolean;
toggleTalking: () => void;
};
export const useTalkStore = create<TalkState>((set) => ({
isTalking: false,
toggleTalking: () => set((state) => ({ isTalking: !state.isTalking })),
}));

6
apps/dispatch/app/helpers/cn.ts Normal file
View File

@@ -0,0 +1,6 @@
import clsx, { ClassValue } from "clsx";
import { twMerge } from "tailwind-merge";
export const cn = (...inputs: ClassValue[]) => {
return twMerge(clsx(inputs));
};

46
apps/dispatch/app/helpers/liveKitEventHandler.ts Normal file
View File

@@ -0,0 +1,46 @@
import {
LocalParticipant,
LocalTrackPublication,
Participant,
RemoteParticipant,
RemoteTrack,
RemoteTrackPublication,
Track,
} from "livekit-client";
export const handleTrackSubscribed = (
track: RemoteTrack,
publication: RemoteTrackPublication,
participant: RemoteParticipant,
) => {
if (track.kind === Track.Kind.Video || track.kind === Track.Kind.Audio) {
// attach it to a new HTMLVideoElement or HTMLAudioElement
const element = track.attach();
element.play();
}
};
export const handleTrackUnsubscribed = (
track: RemoteTrack,
publication: RemoteTrackPublication,
participant: RemoteParticipant,
) => {
// remove tracks from all attached elements
track.detach();
};
export const handleLocalTrackUnpublished = (
publication: LocalTrackPublication,
participant: LocalParticipant,
) => {
// when local tracks are ended, update UI to remove them from rendering
publication.track?.detach();
};
export const handleActiveSpeakerChange = (speakers: Participant[]) => {
// show UI indicators when participant is speaking
};
export const handleDisconnect = () => {
console.log("disconnected from room");
};

1
apps/dispatch/package.json
View File

@@ -18,6 +18,7 @@
"@tailwindcss/postcss": "^4.0.14",
"leaflet": "^1.9.4",
"livekit-client": "^2.9.7",
"lucide-react": "^0.482.0",
"next": "^15.1.0",
"next-auth": "^4.24.11",
"postcss": "^8.5.1",

1
apps/dispatch/tsconfig.json
View File

@@ -1,6 +1,7 @@
{
"extends": "@repo/typescript-config/nextjs.json",
"compilerOptions": {
"baseUrl": "./app",
"plugins": [
{
"name": "next"

1
apps/hub/tsconfig.json
View File

@@ -1,6 +1,7 @@
{
"extends": "@repo/typescript-config/nextjs.json",
"compilerOptions": {
"baseUrl": "./app",
"plugins": [
{
"name": "next"

8
apps/mediasoup-server/.d.ts vendored
View File

@@ -1,8 +0,0 @@
declare module "next-auth/jwt" {
interface JWT {
uid: string;
firstname: string;
lastname: string;
email: string;
}
}

4
apps/mediasoup-server/.env.example
View File

@@ -1,4 +0,0 @@
PORT=3002
REDIS_HOST=localhost
REDIS_PORT=6379
DISPATCH_APP_TOKEN=

31
apps/mediasoup-server/index.ts
View File

@@ -1,31 +0,0 @@
import "dotenv/config";
import express from "express";
import { createServer } from "http";
import { Server } from "socket.io";
import { createAdapter } from "@socket.io/redis-adapter";
import { jwtMiddleware } from "modules/socketJWTmiddleware";
import { pubClient, subClient } from "modules/redis";
const app = express();
const server = createServer(app);
pubClient.keys("dispatchers*").then((keys) => {
if (!keys) return;
keys.forEach(async (key) => {
await pubClient.json.del(key);
});
});
const io = new Server(server, {
adapter: createAdapter(pubClient, subClient),
cors: {},
});
io.use(jwtMiddleware);
io.on("connection", (socket) => {
console.log("User conencted to mediasoup server");
});
server.listen(process.env.PORT, () => {
console.log(`Server running on port ${process.env.PORT}`);
});

317
apps/mediasoup-server/modules/mediasoup/Channel.ts
View File

@@ -1,317 +0,0 @@
import { Worker } from 'mediasoup/node/lib/Worker';
import { types as MediasoupTypes } from 'mediasoup';
import logger from 'modules/winston/logger';
import { Router } from 'mediasoup/node/lib/Router';
import { DtlsParameters, WebRtcTransport } from 'mediasoup/node/lib/WebRtcTransport';
import { Socket } from 'socket.io';
import { UserDocument } from 'models/user';
import { MediaKind, RtpCapabilities, RtpParameters } from 'mediasoup/node/lib/RtpParameters';
import { EventEmitter } from 'stream';
import { ServerTransportParams } from '@common/types/mediasoup';
import { ConsumerOptions } from 'mediasoup-client/lib/Consumer';
import { routerOptions, webRtcTransportConfig } from './config';
/* *
* Represents a Voice-Channel: all transports, Producers and Consumers are managed using this Class
* Does NOT manages events for when a user joins/ leaves the channel
*/
export class MediasoupChannel extends EventEmitter {
worker: Worker;
router: Router | undefined;
transports: WebRtcTransport[] = [];
producers: MediasoupTypes.Producer[] = [];
consumers: MediasoupTypes.Consumer[] = [];
constructor(worker: Worker) {
super();
this.worker = worker;
this.init();
}
async init() {
this.router = await this.worker.createRouter(/* routerOptions */ routerOptions);
}
handleSocket(socket: Socket, user: UserDocument) {
if (!this.router || !this.worker) {
logger.warn('Rejected user connection: channel not yet initialized', { system: 'mediasoup' });
return socket.emit('error-message', { error: 'channel not yet initialized' });
}
if (this.producers.length >= Number(process.env.MAX_VOICE_CONNECTIONS)) {
logger.warn('Rejected user connection: to many connections', { system: 'mediasoup' });
return socket.emit('error-message', { error: 'to many connections', system: 'mediasoup' });
}
socket.emit('sfu-ready');
socket.on('get-rtp-capabilities', (callback) => {
const { rtpCapabilities } = this.router!;
callback(rtpCapabilities);
});
socket.on('create-webrtc-transport', async (callback) => {
try {
const transport = await this.createWebRtcTransport();
this.transports.push(transport);
// how to delete unused transports which never connected to a client?
// For now:
socket.once('disconnect', () => {
transport.close();
this.transports = this.transports.filter((t) => t.id !== transport.id);
});
// send the parameters for the created transport back to the client
// https://mediasoup.org/documentation/v3/mediasoup-client/api/#TransportOptions
callback({
id: transport.id,
iceParameters: transport.iceParameters,
iceCandidates: transport.iceCandidates,
dtlsParameters: transport.dtlsParameters
} as ServerTransportParams);
} catch (err) {
const error = err as Error;
callback({
error: error.message
});
}
});
socket.on(
'transport-connect',
async ({ dtlsParameters, transportId }: { dtlsParameters: DtlsParameters; transportId: string }) => {
try {
const transport = this.transports.find((t) => t.id === transportId);
transport?.on('@close', () => {
this.transports = this.transports.filter((t) => t.id !== transportId);
});
if (!transport) return;
await transport.connect({ dtlsParameters });
} catch (error) {
logger.warn('cannot connect transport', { service: 'mediasoup' });
socket.emit('error-message', { error: (error as Error).message });
}
}
);
socket.on(
'transport-produce',
async (
{
kind,
rtpParameters,
transportId
}: { kind: MediaKind; rtpParameters: RtpParameters; transportId: string },
callback
) => {
try {
const transport = this.transports.find((t) => t.id === transportId);
if (!transport) throw Error('transport not found');
const producer = await transport.produce({
kind,
rtpParameters,
paused: true
});
// DEBUG
this.producers.push(producer);
this.emit('new-producer', { producerId: producer.id, user: user.getPublicUser() });
producer.appData.user = user.getPublicUser();
producer.observer.on('pause', () => {
this.emit('producer-paused', { producerId: producer.id });
});
producer.observer.on('resume', () => {
this.emit('producer-resumed', { producerId: producer.id });
});
producer.on('transportclose', () => {
this.producers = this.producers.filter((p) => p.id !== producer.id);
this.emit('producer-closed', { producerId: producer.id });
producer.close();
});
// Send back to the client the Producer's id
callback({
id: producer.id
});
} catch (err) {
logger.warn(`Error while creating Producer on Transport! ${err}`, { service: 'mediasoup' });
const error = err as Error;
callback({ error: error.message });
}
}
);
socket.on(
'transport-consume',
async (
{
rtpCapabilities,
producerId,
transportId
}: {
rtpCapabilities: RtpCapabilities;
producerId: string;
transportId: string;
},
callback
) => {
try {
const transport = this.transports.find((t) => t.id === transportId);
if (!transport) {
socket.disconnect();
throw Error('transport not found');
}
// check if the router can consume the specified producer
if (
this.router!.canConsume({
producerId,
rtpCapabilities
})
) {
// transport can now consume and return a consumer
const producer = this.producers.find((p) => p.id === producerId);
if (!producer) {
socket.disconnect();
throw Error(`producer ${producerId} not found`);
}
const consumer = await transport.consume({
producerId,
rtpCapabilities,
appData: { user, producerId },
paused: producer.paused // Important: otherwise remote video stays black, no audio
});
this.consumers.push(consumer);
// Cannot detect when consumer is closed dirrectly
// Assuming that:
// 1. when producer is closed, consumer is also closed
// 2. when transport is closed, consumer is also closed
// 2. and when socket is closed, consumer is also closed
consumer.on('transportclose', () => {
consumer.close();
});
consumer.observer.on('close', () => {
this.consumers = this.consumers.filter((c) => c.id !== consumer.id);
});
// Bad, because it adds a listener for each consumer
/* socket.once('disconnect', () => {
consumer.close();
this.consumers = this.consumers.filter((c) => c.id !== consumer.id);
}); */
consumer.on('producerclose', () => {
consumer.close();
});
// from the consumer extract the following params
// to send back to the Client
const consumerOptions: ConsumerOptions = {
id: consumer.id,
producerId,
kind: consumer.kind,
rtpParameters: consumer.rtpParameters,
appData: { user }
};
// send the parameters to the client
callback({ consumerOptions });
} else {
logger.warn(`Cannot consume producer with id: ${producerId}`, { service: 'mediasoup' });
callback({
params: {
error: true
}
});
}
} catch (err) {
const error = err as Error;
logger.error(`error while creating consumer ${error}`, { service: 'Mediasoup' });
callback({
error
});
}
}
);
socket.on('consumer-resume', async ({ consumerId }: { consumerId: string }, resumeCallback) => {
logger.silly(`Resuming consumer with id ${consumerId}`, { service: 'mediasoup' });
const consumer = this.consumers.find((c) => c.id === consumerId);
if (!consumer) return resumeCallback({ error: 'consumer not found' });
await consumer.resume();
if (typeof resumeCallback === 'function') {
resumeCallback();
}
return true;
});
socket.on('consumer-pause', async ({ consumerId }: { consumerId: string }, pauseCallback) => {
logger.silly(`Pausing consumer with id ${consumerId}`, { service: 'mediasoup' });
const consumer = this.consumers.find((c) => c.id === consumerId);
if (!consumer) return pauseCallback({ error: 'consumer not found' });
await consumer.pause();
if (typeof pauseCallback === 'function') {
pauseCallback();
}
return true;
});
socket.on(
'producer-resume',
async ({ producerId, source }: { producerId: string; source: string }, resumeCallback) => {
logger.silly(`Resuming producer with id ${producerId}`, { service: 'mediasoup' });
const producer = this.producers.find((p) => p.id === producerId);
if (!producer) return resumeCallback({ error: 'producer not found' });
if (source === 'admin') {
this.producers.forEach((p) => p.pause());
this.emit('producer-pausing-forced', { user: user.getPublicUser() });
}
await producer.resume();
if (resumeCallback) {
resumeCallback();
}
return true;
}
);
socket.on('producer-pause', async ({ producerId }: { producerId: string }, pauseCallback) => {
logger.silly(`Pausing producer with id ${producerId}`, { service: 'mediasoup' });
const producer = this.producers.find((p) => p.id === producerId);
if (!producer) return pauseCallback({ error: 'producer not found' });
await producer.pause();
if (typeof pauseCallback === 'function') {
pauseCallback();
}
return true;
});
return true;
}
async createWebRtcTransport(): Promise<WebRtcTransport> {
// https://mediasoup.org/documentation/v3/mediasoup/api/#router-createWebRtcTransport
const transport = await this.router!.createWebRtcTransport(webRtcTransportConfig);
transport.on('dtlsstatechange', (dtlsState) => {
// Not reliable
logger.silly('transport dtlsstatechange', { dtlsState, service: 'mediasoup' });
if (dtlsState === 'closed') {
transport.close();
this.transports = this.transports.filter((t) => t.id !== transport.id);
}
});
return transport;
}
}

51
apps/mediasoup-server/modules/mediasoup/config.ts
View File

@@ -1,51 +0,0 @@
import os from 'os';
import { types as mediasoupTypes } from 'mediasoup';
export const workerSettings: mediasoupTypes.WorkerSettings = {
rtcMinPort: 2000,
rtcMaxPort: 2300,
logLevel: 'debug',
logTags: ['info', 'ice', 'dtls', 'rtp', 'srtp', 'rtcp', 'message']
};
export const routerOptions: mediasoupTypes.RouterOptions = {
mediaCodecs: [
{
kind: 'audio',
mimeType: 'audio/opus',
clockRate: 48000,
channels: 2
},
{
kind: 'video',
mimeType: 'video/VP8',
clockRate: 90000,
parameters: {
'x-google-start-bitrate': 1000
}
}
]
};
export const webRtcTransportConfig: mediasoupTypes.WebRtcTransportOptions = {
// https://mediasoup.org/documentation/v3/mediasoup/api/#WebRtcTransportOptions
listenInfos: [
{
protocol: 'tcp',
ip: '0.0.0.0',
announcedIp: process.env.MEDIASOUP_ANOUNCE_IP // public ip
},
{
protocol: 'udp',
ip: '0.0.0.0',
announcedIp: process.env.MEDIASOUP_ANOUNCE_IP // public ip
}
],
enableUdp: true,
enableTcp: true,
preferUdp: true
};
export default {
numWorkers: Object.keys(os.cpus()).length
};

21
apps/mediasoup-server/modules/mediasoup/worker.ts
View File

@@ -1,21 +0,0 @@
import { createWorker as mediasoupCreateWorker } from 'mediasoup';
import logger from 'modules/winston/logger';
import { workerSettings } from './config';
/**
* * For now, each channel uses its own worker
* ! Do not create more Workers than the number of CPU-Cores
*/
export const createWorker = async () => {
const worker = await mediasoupCreateWorker(workerSettings);
logger.info(`Mediasoup worker created`, { service: 'mediasoup' });
worker.on('died', (error) => {
// This implies something serious happened, so kill the application
logger.error(`Mediasoup worker crashed! ${error}`, { error, service: 'mediasoup' });
setTimeout(() => process.exit(1), 2000); // exit in 2 seconds
});
return worker;
};

8
apps/mediasoup-server/modules/redis.ts
View File

@@ -1,8 +0,0 @@
import { createClient } from "redis";
export const pubClient = createClient();
export const subClient = pubClient.duplicate();
Promise.all([pubClient.connect(), subClient.connect()]).then(() => {
console.log("Redis connected");
});

28
apps/mediasoup-server/modules/socketJWTmiddleware.ts
View File

@@ -1,28 +0,0 @@
import { ExtendedError, Server, Socket } from "socket.io";
import { prisma } from "@repo/db";
if (!process.env.DISPATCH_APP_TOKEN)
throw new Error("DISPATCH_APP_TOKEN is not defined");
export const jwtMiddleware = async (
socket: Socket,
next: (err?: ExtendedError) => void,
) => {
try {
const { uid } = socket.handshake.auth;
if (!uid) return new Error("Authentication error");
/* const token = socket.handshake.auth?.token;
if (!token) return new Error("Authentication error");
const decoded = jwt.verify(token, process.env.DISPATCH_APP_TOKEN!); */
// socket.data.userId = decoded.; // User ID lokal speichern
const user = await prisma.user.findUniqueOrThrow({
where: { id: uid },
});
socket.data.user = user;
next();
} catch (err) {
console.error(err);
next(new Error("Authentication error"));
}
};

5
apps/mediasoup-server/nodemon.json
View File

@@ -1,5 +0,0 @@
{
"watch": ["."],
"ext": "ts",
"exec": "tsx index.ts"
}

33
apps/mediasoup-server/package.json
View File

@@ -1,33 +0,0 @@
{
"name": "mediasoup-server",
"exports": {
"helpers": "./helper"
},
"scripts": {
"dev": "nodemon",
"build": "tsc"
},
"devDependencies": {
"@repo/db": "*",
"@repo/typescript-config": "*",
"@types/express": "^5.0.0",
"@types/node": "^22.13.5",
"@types/nodemailer": "^6.4.17",
"concurrently": "^9.1.2",
"typescript": "latest"
},
"dependencies": {
"@react-email/components": "^0.0.33",
"@redis/json": "^1.0.7",
"@socket.io/redis-adapter": "^8.3.0",
"axios": "^1.7.9",
"cron": "^4.1.0",
"dotenv": "^16.4.7",
"express": "^4.21.2",
"jsonwebtoken": "^9.0.2",
"nodemailer": "^6.10.0",
"react": "^19.0.0",
"redis": "^4.7.0",
"socket.io": "^4.8.1"
}
}

158
apps/mediasoup-server/socket-events/sfu.ts
View File

@@ -1,158 +0,0 @@
/* eslint-disable @typescript-eslint/no-unused-vars */
/* eslint-disable @typescript-eslint/no-explicit-any */
import { Worker } from 'mediasoup/node/lib/types';
import { UserDocument } from 'models/user';
import { MediasoupChannel } from 'modules/mediasoup/Channel';
import { createWorker } from 'modules/mediasoup/worker';
import { Socket } from 'socket.io';
import { User } from '@common/types/user';
import { EventEmitter } from 'stream';
import { setMemberNickname } from 'modules/bot/bot';
import PilotController from './pilot';
interface SfuClient {
socket: Socket;
channelId: string;
user: UserDocument;
}
interface ISfuController {
worker: Worker[];
observer: EventEmitter;
clients: Map<string, SfuClient>;
channel: Map<string, MediasoupChannel>;
init: () => void;
handle: (channelId: string, socket: Socket, user: UserDocument) => void;
disconnectUser: (userId: string) => void;
}
const SfuController: ISfuController = {
clients: new Map(),
channel: new Map(),
observer: new EventEmitter(),
worker: [],
disconnectUser: async (userId: string) => {
SfuController.clients.forEach((client) => {
if (client.user._id.toString() === userId) {
client.socket.emit('error-message', { error: 'DISCONNECTED_BY_ADMIN' });
client.socket.disconnect();
SfuController.clients.delete(client.socket.id);
}
});
},
init: async () => {
// Setup Worker and Channel
// Scalable!
SfuController.worker = [
await createWorker(),
await createWorker(),
await createWorker(),
await createWorker(),
await createWorker()
];
SfuController.channel.set('1', new MediasoupChannel(SfuController.worker[0])); // LST_VAR_RD_01
SfuController.channel.set('2', new MediasoupChannel(SfuController.worker[1])); // LST_VAR_RD_02
SfuController.channel.set('3', new MediasoupChannel(SfuController.worker[2])); // LST_VAR_RD_03
SfuController.channel.set('4', new MediasoupChannel(SfuController.worker[3])); // LST_VAR_RD_04
SfuController.channel.set('x', new MediasoupChannel(SfuController.worker[4])); // LST_VAR_RESERVE
// setup listends for new connections, producer-close event is handled by Channel
SfuController.channel.forEach((channel, channelId) => {
channel.on('producer-pausing-forced', ({ user }: { user: User.PublicUser }) => {
SfuController.clients.forEach((client) => {
if (client.channelId === channelId && client.user._id.toString() !== user.id) {
client.socket.emit('producer-pausing-forced', { user });
}
});
});
channel.on('new-producer', ({ producerId, user }) => {
SfuController.clients.forEach((client) => {
if (client.channelId === channelId) {
client.socket.emit('new-producer', { producerId, user });
}
});
});
channel.on('producer-closed', ({ producerId }) => {
SfuController.clients.forEach((client) => {
if (client.channelId === channelId) {
client.socket.emit('producer-closed', { producerId });
}
});
});
channel.on('producer-paused', ({ producerId }) => {
SfuController.clients.forEach((client) => {
if (client.channelId === channelId) {
client.socket.emit('producer-paused', { producerId });
}
});
});
channel.on('producer-resumed', ({ producerId }) => {
SfuController.clients.forEach((client) => {
if (client.channelId === channelId) {
client.socket.emit('producer-resumed', { producerId });
}
});
});
});
},
handle: (channelId, socket, user) => {
const channel = SfuController.channel.get(channelId);
if (!channel) {
socket.emit('error-message', { error: 'invalid channel id' });
return;
}
// check for double connections
if (
Array.from(SfuController.clients.values()).find(
(client) => client.user._id.toString() === user._id.toString()
) &&
process.env.ALLOW_DOUBLE_CONNECTION === 'false'
) {
socket.emit('error-message', { error: 'DOUBLE_CONNECTION' });
return;
}
SfuController.clients.set(socket.id, { socket, channelId, user });
// Update Discord username for dispatcher
SfuController.observer.emit('channel-changed', channelId, user);
const userInPilot = PilotController.clients.get(user._id.toString());
if (userInPilot) {
userInPilot.voiceChannel = channelId;
PilotController.clients.set(user._id.toString(), userInPilot);
PilotController.observer.emit('stations-changed');
}
channel.handleSocket(socket, user);
socket.on('get-producers', (getPcb) => {
getPcb(channel.producers.map((p) => ({ producerId: p.id, user: p.appData.user, paused: p.paused })));
});
socket.on('disconnect', () => {
socket.removeAllListeners();
SfuController.clients.delete(socket.id);
const userInPilotDisconnect = PilotController.clients.get(user._id.toString());
if (userInPilotDisconnect) {
userInPilotDisconnect.voiceChannel = undefined;
PilotController.clients.set(user._id.toString(), userInPilotDisconnect);
PilotController.observer.emit('stations-changed');
}
});
socket.on('set-should-transmit', (eventData: { shouldTransmit: boolean; source: string }) => {
SfuController.clients.forEach((client) => {
if (client.user._id.toString() === user._id.toString()) {
client.socket.emit('set-should-transmit', eventData);
}
});
});
}
};
SfuController.init();
export default SfuController;

11
apps/mediasoup-server/tsconfig.json
View File

@@ -1,11 +0,0 @@
{
"extends": "@repo/typescript-config/base.json",
"compilerOptions": {
"outDir": "dist",
"allowImportingTsExtensions": false,
"baseUrl": ".",
"jsx": "react"
},
"include": ["**/*.ts", "./index.ts"],
"exclude": ["node_modules", "dist"]
}

15
package-lock.json generated
View File

@@ -28,6 +28,7 @@
"@tailwindcss/postcss": "^4.0.14",
"leaflet": "^1.9.4",
"livekit-client": "^2.9.7",
"lucide-react": "^0.482.0",
"next": "^15.1.0",
"next-auth": "^4.24.11",
"postcss": "^8.5.1",
@@ -76,6 +77,15 @@
"typescript": "latest"
}
},
"apps/dispatch/node_modules/lucide-react": {
"version": "0.482.0",
"resolved": "https://registry.npmjs.org/lucide-react/-/lucide-react-0.482.0.tgz",
"integrity": "sha512-XM8PzHzSrg8ATmmO+fzf+JyYlVVdQnJjuyLDj2p4V2zEtcKeBNAqAoJIGFv1x2HSBa7kT8gpYUxwdQ0g7nypfw==",
"license": "ISC",
"peerDependencies": {
"react": "^16.5.1 || ^17.0.0 || ^18.0.0 || ^19.0.0"
}
},
"apps/docs": {
"version": "0.1.0",
"dependencies": {
@@ -162,6 +172,7 @@
}
},
"apps/mediasoup-server": {
"extraneous": true,
"dependencies": {
"@react-email/components": "^0.0.33",
"@redis/json": "^1.0.7",
@@ -10159,10 +10170,6 @@
"node": ">= 0.6"
}
},
"node_modules/mediasoup-server": {
"resolved": "apps/mediasoup-server",
"link": true
},
"node_modules/memoize-one": {
"version": "6.0.0",
"resolved": "https://registry.npmjs.org/memoize-one/-/memoize-one-6.0.0.tgz",

11
test/livekit.yaml
View File

@@ -1,11 +0,0 @@
port: 7880
rtc:
udp_port: 7882
tcp_port: 7881
use_external_ip: false
enable_loopback_candidate: false
keys:
APIAnsGdtdYp2Ho: tdPjVsYUx8ddC7K9NvdmVAeLRF9GeADD6Fedm1x63fWC
logging:
json: false
level: info