opti1337/chad
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects 1 Releases 35 Wiki Activity

cringe sfx

Browse Source
This commit is contained in:
opti1337
2026-04-12 22:35:47 +06:00
parent 7b2cfb4b56
commit 92f7bae50d
15 changed files with 700 additions and 597 deletions
Download Patch File Download Diff File Expand all files Collapse all files

569
server/socket/channel.ts
View File

@@ -1,449 +1,150 @@
import type { types } from 'mediasoup'
import type { Server as SocketServer } from 'socket.io'
import type {
Namespace,
SomeSocket,
ChadSocket,
ClientToServerEvents,
ExtractCallbackPayload,
SocketServer,
} from '../types/socket.ts'
import { consola } from 'consola'
import { channelPublicSelect } from '../dto/channel.dto.ts'
import prisma from '../prisma/client.ts'
import { socketToClient } from '../utils/socket-to-client.ts'
export default function (io: SocketServer<Namespace>) {
io.on('connection', async (socket) => {
consola.info('[WebRtc]', 'Client connected', socket.id)
export default async function (io: SocketServer, socket: ChadSocket) {
// io.on('channel:join', async (cb) => {
// if (socket.data.joined) {
// consola.error('[WebRtc]', 'Already joined')
// cb({ error: 'Already joined' })
// return
// }
//
// socket.data.joined = true
// socket.data.rtpCapabilities = rtpCapabilities
//
// // Get current channel from Socket.IO rooms
// const currentChannelId = Array.from(socket.rooms).find(room => room !== socket.id) || 'default'
// const joinedSockets = await getJoinedSockets(socket.id, currentChannelId)
//
// cb(joinedSockets.map(socketToClient))
//
// for (const joinedSocket of joinedSockets) {
// for (const producer of joinedSocket.data.producers.values()) {
// createConsumer(
// socket,
// joinedSocket,
// producer,
// )
// }
// }
//
// // Broadcast only to same channel using Socket.IO room
// socket.to(currentChannelId).emit('newPeer', socketToClient(socket))
// })
socket.data.joined = false
socket.data.inputMuted = false
socket.data.outputMuted = false
socket.data.transports = new Map()
socket.data.producers = new Map()
socket.data.consumers = new Map()
const user = await prisma.user.findUnique({
where: {
id: socket.handshake.auth.userId,
},
select: {
id: true,
username: true,
displayName: true,
},
})
if (!user) {
socket.disconnect()
return
socket.on('channel:join', async ({ channelId }, cb) => {
try {
cb(await handleJoin(channelId))
}
catch (e) {
cb(e)
}
const { id, username, displayName } = user
socket.data.userId = id
socket.data.username = username
socket.data.displayName = displayName
socket.emit('authenticated', { channels })
socket.on('join', async ({ rtpCapabilities }, cb) => {
if (socket.data.joined) {
consola.error('[WebRtc]', 'Already joined')
cb({ error: 'Already joined' })
}
socket.data.joined = true
socket.data.rtpCapabilities = rtpCapabilities
const joinedSockets = await getJoinedSockets()
cb(joinedSockets.map(socketToClient))
for (const joinedSocket of joinedSockets.filter(joinedSocket => joinedSocket.id !== socket.id)) {
for (const producer of joinedSocket.data.producers.values()) {
createConsumer(
socket,
joinedSocket,
producer,
)
}
}
socket.broadcast.emit('newPeer', socketToClient(socket))
})
socket.on('getRtpCapabilities', (cb) => {
cb(router.rtpCapabilities)
})
socket.on('createTransport', async ({ producing, consuming }, cb) => {
try {
const transport = await router.createWebRtcTransport({
listenInfos: [
{
protocol: 'udp',
ip: '0.0.0.0',
announcedAddress: process.env.ANNOUNCED_ADDRESS || '127.0.0.1',
portRange: {
min: 40000,
max: 40100,
},
},
],
enableUdp: true,
preferUdp: true,
appData: {
producing,
consuming,
},
})
socket.data.transports.set(transport.id, transport)
cb({
id: transport.id,
iceParameters: transport.iceParameters,
iceCandidates: transport.iceCandidates,
dtlsParameters: transport.dtlsParameters,
})
transport.on('icestatechange', (iceState: types.IceState) => {
if (iceState === 'disconnected' || iceState === 'closed') {
consola.info('[WebRtc]', '[WebRtcTransport]', `"icestatechange" event [iceState:${iceState}], closing peer`, transport.id)
socket.disconnect()
}
})
transport.on('dtlsstatechange', (dtlsState: types.DtlsState) => {
if (dtlsState === 'failed' || dtlsState === 'closed') {
consola.warn('WebRtcTransport "dtlsstatechange" event [dtlsState:%s], closing peer', dtlsState)
socket.disconnect()
}
})
}
catch (error) {
if (error instanceof Error) {
consola.error('[WebRtc]', '[createTransport]', error.message)
cb({ error: error.message })
}
}
})
socket.on('connectTransport', async ({ transportId, dtlsParameters }, cb) => {
const transport = socket.data.transports.get(transportId)
if (!transport) {
consola.error('[WebRtc]', '[connectTransport]', `Transport with id ${transportId} not found`)
cb({ error: 'Transport not found' })
return
}
try {
await transport.connect({ dtlsParameters })
cb({ ok: true })
}
catch (error) {
if (error instanceof Error) {
consola.error('[WebRtc]', '[connectTransport]', error.message)
cb({ error: error.message })
}
}
})
socket.on('produce', async ({ transportId, kind, rtpParameters, appData }, cb) => {
if (!socket.data.joined) {
consola.error('Peer not joined yet')
cb({ error: 'Peer not joined yet' })
return
}
const transport = socket.data.transports.get(transportId)
if (!transport) {
consola.error('[WebRtc]', '[produce]', `Transport with id ${transportId} not found`)
cb({ error: 'Transport not found' })
return
}
try {
const producer = await transport.produce({ kind, rtpParameters, appData: { ...appData, socketId: socket.id } })
socket.data.producers.set(producer.id, producer)
cb({ id: producer.id })
const otherSockets = await getJoinedSockets(socket.id)
for (const otherSocket of otherSockets) {
createConsumer(
otherSocket,
socket,
producer,
)
}
// TODO: Add into the AudioLevelObserver and ActiveSpeakerObserver.
// https://github.com/versatica/mediasoup-demo/blob/v3/server/lib/Room.js#L1276
}
catch (error) {
if (error instanceof Error) {
consola.error('[WebRtc]', '[produce]', error.message)
cb({ error: error.message })
}
}
})
socket.on('closeProducer', async ({ producerId }, cb) => {
if (!socket.data.joined) {
consola.error('Peer not joined yet')
cb({ error: 'Peer not joined yet' })
return
}
const producer = socket.data.producers.get(producerId)
if (!producer) {
consola.error(`producer with id "${producerId}" not found`)
cb({ error: `producer with id "${producerId}" not found` })
return
}
producer.close()
socket.data.producers.delete(producerId)
cb({ ok: true })
})
socket.on('pauseProducer', async ({ producerId }, cb) => {
if (!socket.data.joined) {
consola.error('Peer not joined yet')
cb({ error: 'Peer not joined yet' })
return
}
const producer = socket.data.producers.get(producerId)
if (!producer) {
consola.error(`producer with id "${producerId}" not found`)
cb({ error: `producer with id "${producerId}" not found` })
return
}
if (producer.paused)
return
await producer.pause()
cb({ ok: true })
})
socket.on('resumeProducer', async ({ producerId }, cb) => {
if (!socket.data.joined) {
consola.error('Peer not joined yet')
cb({ error: 'Peer not joined yet' })
return
}
const producer = socket.data.producers.get(producerId)
if (!producer) {
consola.error(`producer with id "${producerId}" not found`)
cb({ error: `producer with id "${producerId}" not found` })
return
}
await producer.resume()
cb({ ok: true })
})
socket.on('pauseConsumer', async ({ consumerId }, cb) => {
if (!socket.data.joined) {
consola.error('Peer not joined yet')
cb({ error: 'Peer not joined yet' })
return
}
const consumer = socket.data.consumers.get(consumerId)
if (!consumer) {
consola.error(`consumer with id "${consumerId}" not found`)
cb({ error: `consumer with id "${consumerId}" not found` })
return
}
await consumer.pause()
cb({ ok: true })
})
socket.on('resumeConsumer', async ({ consumerId }, cb) => {
if (!socket.data.joined) {
consola.error('Peer not joined yet')
cb({ error: 'Peer not joined yet' })
return
}
const consumer = socket.data.consumers.get(consumerId)
if (!consumer) {
consola.error(`consumer with id "${consumerId}" not found`)
cb({ error: `consumer with id "${consumerId}" not found` })
return
}
await consumer.resume()
cb({ ok: true })
})
socket.on('updateClient', async (updatedClient, cb) => {
if (typeof updatedClient.inputMuted === 'boolean') {
socket.data.inputMuted = updatedClient.inputMuted
}
if (typeof updatedClient.outputMuted === 'boolean') {
socket.data.outputMuted = updatedClient.outputMuted
}
cb(socketToClient(socket))
namespace.emit('clientChanged', socket.id, socketToClient(socket))
})
socket.on('disconnect', () => {
consola.info('Client disconnected:', socket.id)
if (socket.data.joined) {
socket.broadcast.emit('peerClosed', socket.id)
}
for (const transport of socket.data.transports.values()) {
transport.close()
}
})
})
async function getJoinedSockets(excludeId?: string) {
const sockets = await namespace.fetchSockets()
await handleJoin('default')
return sockets.filter(socket => socket.data.joined && (excludeId ? excludeId !== socket.id : true))
const channels = await prisma.channel.findMany({
select: channelPublicSelect,
orderBy: { name: 'asc' },
})
type ChannelJoinCallback = ExtractCallbackPayload<ClientToServerEvents['channel:join']>
async function handleJoin(channelId: string): Promise<ChannelJoinCallback> {
try {
const channel = await prisma.channel.findUnique({
where: { id: channelId },
select: channelPublicSelect,
})
if (!channel) {
return { error: 'Channel not found' }
}
if (channel.maxClients) {
const socketsInChannel = await io.in(channelId).fetchSockets()
if (socketsInChannel.length >= channel.maxClients) {
return { error: 'Channel is full' }
}
}
const oldChannelId = Array.from(socket.rooms).find(room => room !== socket.id)
// for (const producer of socket.data.producers.values()) {
// producer.close()
// }
// socket.data.producers.clear()
//
// for (const consumer of socket.data.consumers.values()) {
// consumer.close()
// }
// socket.data.consumers.clear()
if (oldChannelId) {
await socket.leave(oldChannelId)
io.emit('channel:user-left', { channelId: oldChannelId, clientId: socket.id })
// // Auto-delete non-persistent empty channels
// if (isLeavingNonPersistentChannel) {
// const oldChannelSockets = await io.in(oldChannelId).fetchSockets()
//
// if (oldChannelSockets.length === 0) {
// const oldChannel = await prisma.channel.findUnique({
// where: { id: oldChannelId },
// select: { persistent: true, id: true },
// })
//
// if (oldChannel && !oldChannel.persistent) {
// await prisma.channel.delete({ where: { id: oldChannelId } })
// io.emit('channelDeleted', { channelId: oldChannelId })
// consola.info('[Channel]', `Auto-deleted empty non-persistent channel: ${oldChannelId}`)
// }
// }
// }
}
await socket.join(channelId)
// Get new channel members
// const newChannelSockets = await getJoinedSockets(socket.id, channelId)
//
// // Create consumers for existing producers in new channel
// for (const peer of newChannelSockets) {
// for (const producer of peer.data.producers.values()) {
// createConsumer(socket, peer, producer)
// }
// }
io.emit('channel:user-joined', {
channelId,
client: socketToClient(socket),
})
return {
channel,
clients: newChannelSockets.map(socketToClient),
}
}
catch (error) {
consola.error('[channel:join]', error)
if (error instanceof Error) {
return { error: error.message }
}
else {
return { error: 'Something went wrong' }
}
}
}
async function createConsumer(
consumerSocket: SomeSocket,
producerSocket: SomeSocket,
producer: types.Producer,
) {
if (
!consumerSocket.data.rtpCapabilities
|| !router.canConsume(
{
producerId: producer.id,
rtpCapabilities: consumerSocket.data.rtpCapabilities,
},
)
) {
return
}
const transport = Array.from(consumerSocket.data.transports.values())
.find(t => t.appData.consuming)
if (!transport) {
consola.error('createConsumer() | Transport for consuming not found')
return
}
let consumer: types.Consumer
try {
consumer = await transport.consume(
{
producerId: producer.id,
rtpCapabilities: consumerSocket.data.rtpCapabilities,
// Enable NACK for OPUS.
enableRtx: true,
paused: true,
ignoreDtx: true,
},
)
}
catch (error) {
consola.error('_createConsumer() | transport.consume():%o', error)
return
}
consumerSocket.data.consumers.set(consumer.id, consumer)
consumer.on('transportclose', () => {
consumerSocket.data.consumers.delete(consumer.id)
})
consumer.on('producerclose', () => {
consumerSocket.data.consumers.delete(consumer.id)
consumerSocket.emit('consumerClosed', { consumerId: consumer.id })
})
consumer.on('producerpause', () => {
consumerSocket.emit('consumerPaused', { consumerId: consumer.id })
})
consumer.on('producerresume', () => {
consumerSocket.emit('consumerResumed', { consumerId: consumer.id })
})
consumer.on('score', (score: types.ConsumerScore) => {
consumerSocket.emit('consumerScore', { consumerId: consumer.id, score })
})
try {
await consumerSocket.emitWithAck(
'newConsumer',
{
socketId: producerSocket.id,
producerId: producer.id,
id: consumer.id,
kind: consumer.kind,
rtpParameters: consumer.rtpParameters,
type: consumer.type,
appData: producer.appData,
producerPaused: consumer.producerPaused,
},
)
await consumer.resume()
consumerSocket.emit(
'consumerScore',
{
consumerId: consumer.id,
score: consumer.score,
},
)
}
catch (error) {
consola.error('_createConsumer() | failed:%o', error)
}
return {
channels,
}
}

168
server/socket/webrtc.ts
View File

@@ -1,49 +1,21 @@
import type { types } from 'mediasoup'
import type { ActiveSpeakerObserver, AudioLevelObserver } from 'mediasoup/types'
import type { Server as SocketServer } from 'socket.io'
import type {
ChadClient,
ChadSocket,
SomeSocket,
} from '../types/socket.ts'
import { consola } from 'consola'
import prisma from '../prisma/client.ts'
import { socketToClient } from '../utils/socket-to-client.ts'
export default async function (io: SocketServer, router: types.Router) {
const audioLevelObserver = await router.createAudioLevelObserver({
maxEntries: 10,
threshold: -80,
interval: 800,
})
const activeSpeakerObserver = await router.createActiveSpeakerObserver()
audioLevelObserver.on('volumes', async (volumes: types.AudioLevelObserverVolume[]) => {
io.emit('speakingPeers', volumes.map(({ producer, volume }) => {
const { socketId } = producer.appData as { socketId: ChadClient['socketId'] }
return {
clientId: socketId,
volume,
}
}))
})
audioLevelObserver.on('silence', () => {
io.emit('speakingPeers', [])
io.emit('activeSpeaker', undefined)
})
activeSpeakerObserver.on('dominantspeaker', ({ producer }) => {
const { socketId } = producer.appData as { socketId: ChadClient['socketId'] }
io.emit('activeSpeaker', socketId)
})
export default async function (
io: SocketServer,
socket: ChadSocket,
router: types.Router,
audioLevelObserver: AudioLevelObserver,
activeSpeakerObserver: ActiveSpeakerObserver,
) {
io.on('connection', async (socket) => {
consola.info('[WebRtc]', 'Client connected', socket.id)
socket.data.joined = false
socket.data.inputMuted = false
socket.data.outputMuted = false
@@ -51,62 +23,11 @@ export default async function (io: SocketServer, router: types.Router) {
socket.data.producers = new Map()
socket.data.consumers = new Map()
const user = await prisma.user.findUnique({
where: {
id: socket.handshake.auth.userId,
},
select: {
id: true,
username: true,
displayName: true,
},
})
if (!user) {
socket.disconnect()
return
}
const { id, username, displayName } = user
socket.data.userId = id
socket.data.username = username
socket.data.displayName = displayName
socket.emit('authenticated', { channels })
socket.on('join', async ({ rtpCapabilities }, cb) => {
if (socket.data.joined) {
consola.error('[WebRtc]', 'Already joined')
cb({ error: 'Already joined' })
}
socket.data.joined = true
socket.data.rtpCapabilities = rtpCapabilities
const joinedSockets = await getJoinedSockets()
cb(joinedSockets.map(socketToClient))
for (const joinedSocket of joinedSockets.filter(joinedSocket => joinedSocket.id !== socket.id)) {
for (const producer of joinedSocket.data.producers.values()) {
createConsumer(
socket,
joinedSocket,
producer,
)
}
}
socket.broadcast.emit('newPeer', socketToClient(socket))
})
socket.on('getRtpCapabilities', (cb) => {
socket.on('webrtc:get-rtp-capabilities', (cb) => {
cb(router.rtpCapabilities)
})
socket.on('createTransport', async ({ producing, consuming }, cb) => {
socket.on('webrtc:create-transport', async ({ producing, consuming }, cb) => {
try {
const transport = await router.createWebRtcTransport({
listenInfos: [
@@ -161,7 +82,7 @@ export default async function (io: SocketServer, router: types.Router) {
}
})
socket.on('connectTransport', async ({ transportId, dtlsParameters }, cb) => {
socket.on('webrtc:connect-transport', async ({ transportId, dtlsParameters }, cb) => {
const transport = socket.data.transports.get(transportId)
if (!transport) {
@@ -184,7 +105,7 @@ export default async function (io: SocketServer, router: types.Router) {
}
})
socket.on('produce', async ({ transportId, kind, rtpParameters, appData }, cb) => {
socket.on('webrtc:produce', async ({ transportId, kind, rtpParameters, appData }, cb) => {
if (!socket.data.joined) {
consola.error('Peer not joined yet')
cb({ error: 'Peer not joined yet' })
@@ -192,6 +113,14 @@ export default async function (io: SocketServer, router: types.Router) {
return
}
// Block production in default channel
const currentChannelId = Array.from(socket.rooms).find(room => room !== socket.id) || 'default'
if (currentChannelId === 'default') {
consola.error('Cannot produce in default channel')
cb({ error: 'Cannot produce media in default channel' })
return
}
const transport = socket.data.transports.get(transportId)
if (!transport) {
@@ -208,7 +137,8 @@ export default async function (io: SocketServer, router: types.Router) {
cb({ id: producer.id })
const otherSockets = await getJoinedSockets(socket.id)
// Filter by channel when creating consumers
const otherSockets = await getJoinedSockets(socket.id, currentChannelId)
for (const otherSocket of otherSockets) {
createConsumer(
@@ -229,7 +159,7 @@ export default async function (io: SocketServer, router: types.Router) {
}
})
socket.on('closeProducer', async ({ producerId }, cb) => {
socket.on('webrtc:close-producer', async ({ producerId }, cb) => {
if (!socket.data.joined) {
consola.error('Peer not joined yet')
cb({ error: 'Peer not joined yet' })
@@ -253,7 +183,7 @@ export default async function (io: SocketServer, router: types.Router) {
cb({ ok: true })
})
socket.on('pauseProducer', async ({ producerId }, cb) => {
socket.on('webrtc:pause-producer', async ({ producerId }, cb) => {
if (!socket.data.joined) {
consola.error('Peer not joined yet')
cb({ error: 'Peer not joined yet' })
@@ -278,7 +208,7 @@ export default async function (io: SocketServer, router: types.Router) {
cb({ ok: true })
})
socket.on('resumeProducer', async ({ producerId }, cb) => {
socket.on('webrtc:resume-producer', async ({ producerId }, cb) => {
if (!socket.data.joined) {
consola.error('Peer not joined yet')
cb({ error: 'Peer not joined yet' })
@@ -300,7 +230,7 @@ export default async function (io: SocketServer, router: types.Router) {
cb({ ok: true })
})
socket.on('pauseConsumer', async ({ consumerId }, cb) => {
socket.on('webrtc:pause-consumer', async ({ consumerId }, cb) => {
if (!socket.data.joined) {
consola.error('Peer not joined yet')
cb({ error: 'Peer not joined yet' })
@@ -322,7 +252,7 @@ export default async function (io: SocketServer, router: types.Router) {
cb({ ok: true })
})
socket.on('resumeConsumer', async ({ consumerId }, cb) => {
socket.on('webrtc:resume-consumer', async ({ consumerId }, cb) => {
if (!socket.data.joined) {
consola.error('Peer not joined yet')
cb({ error: 'Peer not joined yet' })
@@ -344,7 +274,7 @@ export default async function (io: SocketServer, router: types.Router) {
cb({ ok: true })
})
socket.on('updateClient', async (updatedClient, cb) => {
socket.on('webrtc:update-client', async (updatedClient, cb) => {
if (typeof updatedClient.inputMuted === 'boolean') {
socket.data.inputMuted = updatedClient.inputMuted
}
@@ -355,14 +285,21 @@ export default async function (io: SocketServer, router: types.Router) {
cb(socketToClient(socket))
io.emit('clientChanged', socket.id, socketToClient(socket))
io.emit('webrtc:client-changed', socket.id, socketToClient(socket))
})
socket.on('disconnect', () => {
consola.info('Client disconnected:', socket.id)
// Get current channel from Socket.IO rooms
const channelId = Array.from(socket.rooms).find(room => room !== socket.id)
if (socket.data.joined) {
socket.broadcast.emit('peerClosed', socket.id)
// Notify only same channel using Socket.IO room
if (channelId) {
socket.to(channelId).emit('webrtc:peer-closed', socket.id)
io.emit('channelUserLeft', { channelId, clientId: socket.id })
}
}
for (const transport of socket.data.transports.values()) {
@@ -371,10 +308,21 @@ export default async function (io: SocketServer, router: types.Router) {
})
})
async function getJoinedSockets(excludeId?: string) {
const sockets = await io.fetchSockets()
async function getJoinedSockets(excludeId?: string, channelId?: string) {
let sockets = await io.fetchSockets()
return sockets.filter(socket => socket.data.joined && (excludeId ? excludeId !== socket.id : true))
// Filter by channel using Socket.IO rooms
if (channelId) {
sockets = await io.in(channelId).fetchSockets()
}
return sockets.filter((socket) => {
if (!socket.data.joined)
return false
if (excludeId && socket.id === excludeId)
return false
return true
})
}
async function createConsumer(
@@ -432,24 +380,24 @@ export default async function (io: SocketServer, router: types.Router) {
consumer.on('producerclose', () => {
consumerSocket.data.consumers.delete(consumer.id)
consumerSocket.emit('consumerClosed', { consumerId: consumer.id })
consumerSocket.emit('webrtc:consumer-closed', { consumerId: consumer.id })
})
consumer.on('producerpause', () => {
consumerSocket.emit('consumerPaused', { consumerId: consumer.id })
consumerSocket.emit('webrtc:consumer-paused', { consumerId: consumer.id })
})
consumer.on('producerresume', () => {
consumerSocket.emit('consumerResumed', { consumerId: consumer.id })
consumerSocket.emit('webrtc:consumer-resumed', { consumerId: consumer.id })
})
consumer.on('score', (score: types.ConsumerScore) => {
consumerSocket.emit('consumerScore', { consumerId: consumer.id, score })
consumerSocket.emit('webrtc:consumer-score', { consumerId: consumer.id, score })
})
try {
await consumerSocket.emitWithAck(
'newConsumer',
'webrtc:new-consumer',
{
socketId: producerSocket.id,
producerId: producer.id,
@@ -465,7 +413,7 @@ export default async function (io: SocketServer, router: types.Router) {
await consumer.resume()
consumerSocket.emit(
'consumerScore',
'webrtc:consumer-score',
{
consumerId: consumer.id,
score: consumer.score,