Integrate LiveKit SFU and harden booking workflow
This commit is contained in:
@@ -1,22 +0,0 @@
|
||||
/**
|
||||
* Builds the WebRTC ICE server list from environment variables so a TURN
|
||||
* server can be added later without any code changes (see DEPLOYMENT.md).
|
||||
* P2P works fine with just STUN for most networks; TURN becomes necessary
|
||||
* behind restrictive NATs/firewalls.
|
||||
*/
|
||||
export function getIceServers(): RTCIceServer[] {
|
||||
const servers: RTCIceServer[] = [
|
||||
{ urls: process.env.STUN_SERVER || "stun:stun.l.google.com:19302" },
|
||||
];
|
||||
|
||||
const turnServer = process.env.TURN_SERVER;
|
||||
if (turnServer) {
|
||||
servers.push({
|
||||
urls: turnServer,
|
||||
username: process.env.TURN_USERNAME || undefined,
|
||||
credential: process.env.TURN_PASSWORD || undefined,
|
||||
});
|
||||
}
|
||||
|
||||
return servers;
|
||||
}
|
||||
@@ -0,0 +1,33 @@
|
||||
const KICK_RETENTION_MS = 24 * 60 * 60_000;
|
||||
type KickList = { ids: Set<string>; expiresAt: number };
|
||||
const kickedParticipantIds = new Map<string, KickList>();
|
||||
const kickedClientIds = new Map<string, KickList>();
|
||||
|
||||
function activeList(map: Map<string, KickList>, eventSlug: string): Set<string> | undefined {
|
||||
const entry = map.get(eventSlug);
|
||||
if (!entry) return undefined;
|
||||
if (entry.expiresAt <= Date.now()) {
|
||||
map.delete(eventSlug);
|
||||
return undefined;
|
||||
}
|
||||
return entry.ids;
|
||||
}
|
||||
|
||||
export function isMeetingParticipantKicked(eventSlug: string, participantId: string, clientId?: string): boolean {
|
||||
return Boolean(
|
||||
activeList(kickedParticipantIds, eventSlug)?.has(participantId) ||
|
||||
(clientId && activeList(kickedClientIds, eventSlug)?.has(clientId)),
|
||||
);
|
||||
}
|
||||
|
||||
export function markMeetingParticipantKicked(eventSlug: string, participantId: string, clientId?: string): void {
|
||||
const expiresAt = Date.now() + KICK_RETENTION_MS;
|
||||
const participantIds = activeList(kickedParticipantIds, eventSlug) ?? new Set<string>();
|
||||
participantIds.add(participantId);
|
||||
kickedParticipantIds.set(eventSlug, { ids: participantIds, expiresAt });
|
||||
if (clientId) {
|
||||
const clientIds = activeList(kickedClientIds, eventSlug) ?? new Set<string>();
|
||||
clientIds.add(clientId);
|
||||
kickedClientIds.set(eventSlug, { ids: clientIds, expiresAt });
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,54 @@
|
||||
import { createHmac } from "node:crypto";
|
||||
import { AccessToken, RoomServiceClient, TrackSource } from "livekit-server-sdk";
|
||||
import type { MeetingRole } from "./token";
|
||||
|
||||
function requiredEnv(name: "LIVEKIT_URL" | "LIVEKIT_API_KEY" | "LIVEKIT_API_SECRET"): string {
|
||||
const value = process.env[name];
|
||||
if (!value) throw new Error(`${name} is not set`);
|
||||
return value;
|
||||
}
|
||||
|
||||
export function getLiveKitClientUrl(): string {
|
||||
return requiredEnv("LIVEKIT_URL").replace(/^http:/, "ws:").replace(/^https:/, "wss:").replace(/\/$/, "");
|
||||
}
|
||||
|
||||
function getLiveKitApiUrl(): string {
|
||||
return requiredEnv("LIVEKIT_URL").replace(/^ws:/, "http:").replace(/^wss:/, "https:").replace(/\/$/, "");
|
||||
}
|
||||
|
||||
/** Stable opaque room name; the public event slug is never sent to LiveKit logs. */
|
||||
export function getLiveKitRoomName(eventSlug: string): string {
|
||||
const secret = requiredEnv("LIVEKIT_API_SECRET");
|
||||
return `anouma-${createHmac("sha256", secret).update(eventSlug).digest("hex").slice(0, 32)}`;
|
||||
}
|
||||
|
||||
export async function createLiveKitToken(args: {
|
||||
eventSlug: string;
|
||||
participantId: string;
|
||||
name: string;
|
||||
role: MeetingRole;
|
||||
clientId?: string;
|
||||
}): Promise<string> {
|
||||
const room = getLiveKitRoomName(args.eventSlug);
|
||||
const token = new AccessToken(requiredEnv("LIVEKIT_API_KEY"), requiredEnv("LIVEKIT_API_SECRET"), {
|
||||
identity: args.participantId,
|
||||
name: args.name,
|
||||
ttl: "2h",
|
||||
metadata: JSON.stringify({ role: args.role, clientId: args.clientId }),
|
||||
});
|
||||
token.addGrant({
|
||||
roomJoin: true,
|
||||
room,
|
||||
canPublish: true,
|
||||
canSubscribe: true,
|
||||
canPublishData: true,
|
||||
canPublishSources: args.role === "host"
|
||||
? [TrackSource.CAMERA, TrackSource.MICROPHONE, TrackSource.SCREEN_SHARE, TrackSource.SCREEN_SHARE_AUDIO]
|
||||
: [TrackSource.CAMERA, TrackSource.MICROPHONE],
|
||||
});
|
||||
return token.toJwt();
|
||||
}
|
||||
|
||||
export function getLiveKitRoomService(): RoomServiceClient {
|
||||
return new RoomServiceClient(getLiveKitApiUrl(), requiredEnv("LIVEKIT_API_KEY"), requiredEnv("LIVEKIT_API_SECRET"));
|
||||
}
|
||||
@@ -1,24 +0,0 @@
|
||||
import type { MeetingRole } from "./token";
|
||||
|
||||
export type ParticipantSummary = {
|
||||
participantId: string;
|
||||
name: string;
|
||||
role: MeetingRole;
|
||||
};
|
||||
|
||||
export type ClientToServerMessage =
|
||||
| { type: "auth"; token: string }
|
||||
| { type: "offer"; to: string; payload: RTCSessionDescriptionInit }
|
||||
| { type: "answer"; to: string; payload: RTCSessionDescriptionInit }
|
||||
| { type: "ice-candidate"; to: string; payload: RTCIceCandidateInit }
|
||||
| { type: "kick"; targetParticipantId: string };
|
||||
|
||||
export type ServerToClientMessage =
|
||||
| { type: "welcome"; participantId: string; role: MeetingRole; participants: ParticipantSummary[] }
|
||||
| ({ type: "peer-joined" } & ParticipantSummary)
|
||||
| { type: "peer-left"; participantId: string }
|
||||
| { type: "offer"; from: string; payload: RTCSessionDescriptionInit }
|
||||
| { type: "answer"; from: string; payload: RTCSessionDescriptionInit }
|
||||
| { type: "ice-candidate"; from: string; payload: RTCIceCandidateInit }
|
||||
| { type: "kicked" }
|
||||
| { type: "auth-error"; reason: string };
|
||||
@@ -2,6 +2,7 @@ import { getPayload } from "payload";
|
||||
import config from "@payload-config";
|
||||
import { combineDateAndTime } from "./status";
|
||||
import { sendReminderEmail } from "@/lib/email/sendReminderEmail";
|
||||
import { formatBerlinDate, formatBerlinTime } from "@/lib/calendar/timezone";
|
||||
|
||||
const REMINDER_TOLERANCE_MS = 10 * 60 * 1000; // cron may run every few minutes
|
||||
|
||||
@@ -10,11 +11,11 @@ function isDue(reminderAt: number, now: number): boolean {
|
||||
}
|
||||
|
||||
function formatDate(date: Date): string {
|
||||
return date.toLocaleDateString("de-DE", { day: "2-digit", month: "long", year: "numeric" });
|
||||
return formatBerlinDate(date, { day: "2-digit", month: "long", year: "numeric" });
|
||||
}
|
||||
|
||||
function formatTime(date: Date): string {
|
||||
return date.toLocaleTimeString("de-DE", { hour: "2-digit", minute: "2-digit" });
|
||||
return formatBerlinTime(date, { hour: "2-digit", minute: "2-digit" });
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -1,92 +0,0 @@
|
||||
import type { WebSocket } from "ws";
|
||||
import type { MeetingRole } from "./token";
|
||||
|
||||
export type Participant = {
|
||||
ws: WebSocket;
|
||||
participantId: string;
|
||||
clientId?: string;
|
||||
name: string;
|
||||
role: MeetingRole;
|
||||
};
|
||||
|
||||
type Room = {
|
||||
participants: Map<string, Participant>;
|
||||
kickedParticipantIds: Set<string>;
|
||||
// Keyed by the client-declared, localStorage-persisted clientId (see
|
||||
// token.ts) rather than the per-connection participantId, so a kick
|
||||
// survives the participant simply reconnecting (which always mints a
|
||||
// fresh participantId) — see isKicked below.
|
||||
kickedClientIds: Set<string>;
|
||||
};
|
||||
|
||||
// In-memory only — this server process is the single source of truth for
|
||||
// signaling (no media is ever stored or relayed through it). If this app is
|
||||
// ever scaled to multiple instances, this state needs to move to a shared
|
||||
// store (e.g. Redis pub/sub) alongside a real SFU.
|
||||
const rooms = new Map<string, Room>();
|
||||
|
||||
function getOrCreateRoom(eventSlug: string): Room {
|
||||
let room = rooms.get(eventSlug);
|
||||
if (!room) {
|
||||
room = { participants: new Map(), kickedParticipantIds: new Set(), kickedClientIds: new Set() };
|
||||
rooms.set(eventSlug, room);
|
||||
}
|
||||
return room;
|
||||
}
|
||||
|
||||
export function isKicked(eventSlug: string, participantId: string, clientId?: string): boolean {
|
||||
const room = rooms.get(eventSlug);
|
||||
if (!room) return false;
|
||||
if (room.kickedParticipantIds.has(participantId)) return true;
|
||||
return clientId ? room.kickedClientIds.has(clientId) : false;
|
||||
}
|
||||
|
||||
export function joinRoom(eventSlug: string, participant: Participant): Participant[] {
|
||||
const room = getOrCreateRoom(eventSlug);
|
||||
const existing = [...room.participants.values()];
|
||||
// Two connections claiming the same participantId (e.g. two tabs replaying
|
||||
// the same stored session) would otherwise silently overwrite the map
|
||||
// entry here and orphan the first socket — close it first so it doesn't
|
||||
// leak as an unreachable broadcast target.
|
||||
room.participants.get(participant.participantId)?.ws.close(4002, "replaced-by-new-connection");
|
||||
room.participants.set(participant.participantId, participant);
|
||||
return existing;
|
||||
}
|
||||
|
||||
export function leaveRoom(eventSlug: string, participantId: string) {
|
||||
const room = rooms.get(eventSlug);
|
||||
if (!room) return;
|
||||
room.participants.delete(participantId);
|
||||
if (room.participants.size === 0) rooms.delete(eventSlug);
|
||||
}
|
||||
|
||||
export function kickFromRoom(eventSlug: string, participantId: string) {
|
||||
const room = rooms.get(eventSlug);
|
||||
if (!room) return;
|
||||
room.kickedParticipantIds.add(participantId);
|
||||
const target = room.participants.get(participantId);
|
||||
if (target?.clientId) room.kickedClientIds.add(target.clientId);
|
||||
room.participants.delete(participantId);
|
||||
}
|
||||
|
||||
export function getParticipant(eventSlug: string, participantId: string): Participant | undefined {
|
||||
return rooms.get(eventSlug)?.participants.get(participantId);
|
||||
}
|
||||
|
||||
export function getRoomParticipants(eventSlug: string): Participant[] {
|
||||
return [...(rooms.get(eventSlug)?.participants.values() ?? [])];
|
||||
}
|
||||
|
||||
export function broadcast(
|
||||
eventSlug: string,
|
||||
message: unknown,
|
||||
opts: { exclude?: string } = {},
|
||||
) {
|
||||
const room = rooms.get(eventSlug);
|
||||
if (!room) return;
|
||||
const data = JSON.stringify(message);
|
||||
for (const p of room.participants.values()) {
|
||||
if (opts.exclude && p.participantId === opts.exclude) continue;
|
||||
if (p.ws.readyState === p.ws.OPEN) p.ws.send(data);
|
||||
}
|
||||
}
|
||||
@@ -1,115 +0,0 @@
|
||||
import type { WebSocket, WebSocketServer } from "ws";
|
||||
import { verifyMeetingToken } from "./token";
|
||||
import {
|
||||
broadcast,
|
||||
getParticipant,
|
||||
isKicked,
|
||||
joinRoom,
|
||||
kickFromRoom,
|
||||
leaveRoom,
|
||||
type Participant,
|
||||
} from "./rooms";
|
||||
import type { ClientToServerMessage, ParticipantSummary } from "./protocol";
|
||||
|
||||
// A connection has this long to send its "auth" message before being
|
||||
// dropped. The token travels as the first WS message rather than a
|
||||
// `?token=` query parameter so it doesn't end up persisted in reverse-proxy
|
||||
// or access logs; it's still HMAC-signed and short-lived either way.
|
||||
const AUTH_TIMEOUT_MS = 10_000;
|
||||
|
||||
function participantSummary(p: Participant): ParticipantSummary {
|
||||
return { participantId: p.participantId, name: p.name, role: p.role };
|
||||
}
|
||||
|
||||
export function attachSignalingServer(wss: WebSocketServer) {
|
||||
wss.on("connection", (ws: WebSocket) => {
|
||||
let self: Participant | null = null;
|
||||
let eventSlug: string | null = null;
|
||||
|
||||
const authTimeout = setTimeout(() => {
|
||||
if (!self) ws.close(4000, "auth-timeout");
|
||||
}, AUTH_TIMEOUT_MS);
|
||||
|
||||
function handleAuth(token: string) {
|
||||
const payload = verifyMeetingToken(token);
|
||||
if (!payload) {
|
||||
ws.send(JSON.stringify({ type: "auth-error", reason: "invalid-token" }));
|
||||
ws.close(4000, "invalid-token");
|
||||
return;
|
||||
}
|
||||
if (isKicked(payload.eventSlug, payload.participantId, payload.clientId)) {
|
||||
ws.close(4001, "removed-by-host");
|
||||
return;
|
||||
}
|
||||
|
||||
eventSlug = payload.eventSlug;
|
||||
self = {
|
||||
ws,
|
||||
participantId: payload.participantId,
|
||||
clientId: payload.clientId,
|
||||
name: payload.name,
|
||||
role: payload.role,
|
||||
};
|
||||
|
||||
const existing = joinRoom(eventSlug, self);
|
||||
|
||||
ws.send(
|
||||
JSON.stringify({
|
||||
type: "welcome",
|
||||
participantId: self.participantId,
|
||||
role: self.role,
|
||||
participants: existing.map(participantSummary),
|
||||
}),
|
||||
);
|
||||
|
||||
broadcast(eventSlug, { type: "peer-joined", ...participantSummary(self) }, {
|
||||
exclude: self.participantId,
|
||||
});
|
||||
}
|
||||
|
||||
ws.on("message", (raw) => {
|
||||
let message: ClientToServerMessage;
|
||||
try {
|
||||
message = JSON.parse(raw.toString());
|
||||
} catch {
|
||||
return;
|
||||
}
|
||||
|
||||
if (message.type === "auth") {
|
||||
clearTimeout(authTimeout);
|
||||
if (self) return; // already authenticated on this connection
|
||||
handleAuth(message.token);
|
||||
return;
|
||||
}
|
||||
|
||||
if (!self || !eventSlug) return; // no messages accepted before auth
|
||||
|
||||
if (message.type === "offer" || message.type === "answer" || message.type === "ice-candidate") {
|
||||
const target = getParticipant(eventSlug, message.to);
|
||||
if (target && target.ws.readyState === target.ws.OPEN) {
|
||||
target.ws.send(
|
||||
JSON.stringify({ type: message.type, from: self.participantId, payload: message.payload }),
|
||||
);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
if (message.type === "kick") {
|
||||
if (self.role !== "host") return; // server-side authority, not client-claimed
|
||||
const target = getParticipant(eventSlug, message.targetParticipantId);
|
||||
if (!target) return;
|
||||
kickFromRoom(eventSlug, target.participantId);
|
||||
target.ws.send(JSON.stringify({ type: "kicked" }));
|
||||
target.ws.close(4001, "removed-by-host");
|
||||
broadcast(eventSlug, { type: "peer-left", participantId: target.participantId });
|
||||
}
|
||||
});
|
||||
|
||||
ws.on("close", () => {
|
||||
clearTimeout(authTimeout);
|
||||
if (!self || !eventSlug) return;
|
||||
leaveRoom(eventSlug, self.participantId);
|
||||
broadcast(eventSlug, { type: "peer-left", participantId: self.participantId });
|
||||
});
|
||||
});
|
||||
}
|
||||
@@ -1,3 +1,5 @@
|
||||
import { berlinParts, berlinWallTimeToUTC } from "@/lib/calendar/timezone";
|
||||
|
||||
export type MeetingStatus = "scheduled" | "open" | "live" | "ended";
|
||||
|
||||
/**
|
||||
@@ -9,9 +11,10 @@ export function combineDateAndTime(dateISO: string, timeISO?: string | null): Da
|
||||
const date = new Date(dateISO);
|
||||
if (!timeISO) return date;
|
||||
const time = new Date(timeISO);
|
||||
const combined = new Date(date);
|
||||
combined.setHours(time.getHours(), time.getMinutes(), 0, 0);
|
||||
return combined;
|
||||
if (Number.isNaN(date.getTime()) || Number.isNaN(time.getTime())) return new Date(NaN);
|
||||
const dateParts = berlinParts(date);
|
||||
const timeParts = berlinParts(time);
|
||||
return berlinWallTimeToUTC(dateParts, timeParts.hour, timeParts.minute);
|
||||
}
|
||||
|
||||
export type MeetingWindow = {
|
||||
|
||||
+13
-3
@@ -61,7 +61,8 @@ export function createMeetingToken(args: {
|
||||
}
|
||||
|
||||
/** Verifies signature + expiry. Returns null if invalid/expired/tampered. */
|
||||
export function verifyMeetingToken(token: string): MeetingTokenPayload | null {
|
||||
export function verifyMeetingToken(token: unknown): MeetingTokenPayload | null {
|
||||
if (typeof token !== "string" || token.length > 4096) return null;
|
||||
const parts = token.split(".");
|
||||
if (parts.length !== 2) return null;
|
||||
const [body, signature] = parts;
|
||||
@@ -73,8 +74,17 @@ export function verifyMeetingToken(token: string): MeetingTokenPayload | null {
|
||||
|
||||
try {
|
||||
const payload = JSON.parse(Buffer.from(body, "base64url").toString("utf8")) as MeetingTokenPayload;
|
||||
if (typeof payload.exp !== "number" || payload.exp < Date.now()) return null;
|
||||
if (!payload.eventSlug || !payload.participantId || !payload.name || !payload.role) return null;
|
||||
if (
|
||||
!Number.isFinite(payload.exp) ||
|
||||
payload.exp < Date.now() ||
|
||||
typeof payload.eventSlug !== "string" ||
|
||||
payload.eventSlug.length > 200 ||
|
||||
typeof payload.participantId !== "string" ||
|
||||
payload.participantId.length > 100 ||
|
||||
typeof payload.name !== "string" ||
|
||||
payload.name.length > 120 ||
|
||||
(payload.role !== "host" && payload.role !== "participant")
|
||||
) return null;
|
||||
return payload;
|
||||
} catch {
|
||||
return null;
|
||||
|
||||
Reference in New Issue
Block a user