feat(videochat): implement LiveKit integration for video chat functionality

- Add LiveKit client setup and token generation for room access.
- Implement room management functions including CreateRoom, OpenRoom, ListRooms, and DeleteRoom.
- Introduce participant management with ListParticipants and RemoveParticipant functions.
- Create tests for room and participant management, ensuring proper authentication and authorization.
- Add error handling for LiveKit API interactions.
- Define data structures for room metadata and token responses.
- Implement status endpoint for service availability check.
This commit is contained in:
fabio
2026-09-29 18:45:21 +02:00
parent a051eef991
commit 0fe1bc2292
21 changed files with 1904 additions and 12 deletions

61
videochat/README.md Normal file
View File

@@ -0,0 +1,61 @@
# Broadcast LiveKit (uno a molti)
Configurare i secret Encore `LiveKitURL`, `LiveKitAPIKey` e `LiveKitAPISecret` sul server.
Il frontend usa solo il token partecipante, mai la API secret.
## Accesso
- Solo gli amministratori possono creare stanze (`POST /videochat/rooms`).
- Il creatore è l'unico proprietario; anche un altro amministratore entra come spettatore.
- Tutti gli utenti autenticati possono elencare i broadcast e richiedere un token.
- Il proprietario pubblica audio, videocamera e schermo. Gli spettatori hanno
`canPublish=false`, `canPublishData=false`, `canSubscribe=true` nel JWT firmato.
- Nessun token frontend ha permessi amministrativi o modifica metadati.
- Elenco partecipanti, disconnessione partecipanti e chiusura richiedono il proprietario.
## API
| Metodo | Percorso | Accesso |
| --- | --- | --- |
| GET | `/videochat/status` | Pubblico |
| POST | `/videochat/rooms` | Admin (diventa proprietario) |
| GET | `/videochat/rooms` | Autenticato |
| POST | `/videochat/token` | Autenticato |
| DELETE | `/videochat/rooms/:roomName` | Proprietario |
| GET | `/videochat/rooms/:roomName/participants` | Proprietario |
| DELETE | `/videochat/rooms/:roomName/participants/:identity` | Proprietario |
La lista include i broadcast presenti su LiveKit (`active: true`) e le stanze
di `personal_setup` con `can_broadcast: true` non ancora presenti (`active: false`).
Il confronto usa il nome tecnico oppure proprietario e nome visualizzato.
Le stanze inattive riportano il nome configurato, il proprietario e `mode: "broadcast"`;
SID, partecipanti e data di creazione restano vuoti o zero. Non vengono create
su LiveKit e non possono ricevere un token finché non sono avviate.
Le richieste usano `Authorization: Bearer <session-token>` o il cookie `session`.
Creazione: `{"name":"Titolo della diretta","max_participants":0,"empty_timeout":300}`.
Zero nei campi numerici usa i valori predefiniti di LiveKit; il massimo comprende
anche il proprietario. Il timeout è in secondi.
La risposta stanza aggiunge `display_name`, `owner_id`, `mode: "broadcast"`.
`name` è un identificativo univoco generato dal server: usarlo nei percorsi API e
in `POST /videochat/token` con `{"room_name":"<name restituito>"}`.
La risposta token include `role: "owner" | "viewer"`, `owner_id`, `server_url`,
`participant_token`, `identity`, `room_name`, `expires_in`.
## Ciclo di vita
Le stanze sono effimere. Titolo e proprietario sono conservati nei metadati
LiveKit per la durata della stanza, anche se Encore viene riavviato. Quando
LiveKit elimina una stanza vuota, occorre crearne una nuova; non è un archivio
persistente delle trasmissioni. I nomi interni non vengono riutilizzati.
Le vecchie stanze senza metadati broadcast non compaiono nella lista e non
ricevono nuovi token tramite questa API; non vengono eliminate automaticamente.
I vecchi token meeting già emessi mantengono i permessi fino alla loro scadenza.
La disconnessione di uno spettatore non è un ban: può rientrare. La chiusura
elimina la stanza e disconnette tutti. Su LiveKit self-hosted i JWT già emessi
non sono revocabili da queste API: fino alla scadenza (10 minuti) possono ancora
consentire accesso o ricreazione della vecchia stanza interna. Tale stanza non
avrà metadati broadcast, non apparirà nella lista e non riceverà nuovi token.
Per un blocco irrevocabile servono controllo dell'ammissione/revoca aggiuntivi.

145
videochat/livekit.go Normal file
View File

@@ -0,0 +1,145 @@
package videochat
import (
"context"
"encoding/json"
"errors"
"net/url"
"strings"
"time"
"unicode"
authsvc "encore.app/auth"
"encore.dev/beta/auth"
"encore.dev/beta/errs"
lkauth "github.com/livekit/protocol/auth"
"github.com/livekit/protocol/livekit"
lksdk "github.com/livekit/server-sdk-go/v2"
"github.com/twitchtv/twirp"
)
var secrets struct {
LiveKitURL string
LiveKitAPIKey string
LiveKitAPISecret string
}
const tokenTTL = 10 * time.Minute
const apiTimeout = 10 * time.Second
func liveKitClient() (*lksdk.RoomServiceClient, string, error) {
serverURL, err := websocketURL(secrets.LiveKitURL)
if err != nil || strings.TrimSpace(secrets.LiveKitAPIKey) == "" || strings.TrimSpace(secrets.LiveKitAPISecret) == "" {
return nil, "", &errs.Error{Code: errs.FailedPrecondition, Message: "LiveKit is not configured"}
}
return lksdk.NewRoomServiceClient(serverURL, secrets.LiveKitAPIKey, secrets.LiveKitAPISecret), serverURL, nil
}
func websocketURL(raw string) (string, error) {
u, err := url.Parse(strings.TrimSpace(raw))
if err != nil || u.Hostname() == "" || u.User != nil || u.RawQuery != "" || u.Fragment != "" {
return "", errors.New("invalid LiveKit URL")
}
switch u.Scheme {
case "http", "ws":
u.Scheme = "ws"
case "https", "wss":
u.Scheme = "wss"
default:
return "", errors.New("invalid LiveKit URL scheme")
}
return strings.TrimRight(u.String(), "/"), nil
}
func requireAdmin() error {
if _, ok := auth.UserID(); !ok {
return &errs.Error{Code: errs.Unauthenticated, Message: "must be logged in"}
}
data, _ := auth.Data().(*authsvc.AuthData)
if data == nil || !data.Role.IsAdmin() {
return &errs.Error{Code: errs.PermissionDenied, Message: "admin role required"}
}
return nil
}
func validateRoomName(name string) error {
if strings.TrimSpace(name) == "" || name != strings.TrimSpace(name) || len(name) > 128 || strings.ContainsAny(name, "/\\") || strings.ContainsFunc(name, unicode.IsControl) {
return &errs.Error{Code: errs.InvalidArgument, Message: "room name must contain 1-128 bytes, without slashes, surrounding whitespace or control characters"}
}
return nil
}
func liveKitError(err error) error {
if err == nil {
return nil
}
code := errs.Unavailable
var rpcErr twirp.Error
if errors.As(err, &rpcErr) {
switch rpcErr.Code() {
case twirp.NotFound:
code = errs.NotFound
case twirp.InvalidArgument:
code = errs.InvalidArgument
case twirp.AlreadyExists:
code = errs.AlreadyExists
}
}
return errs.WrapCode(err, code, "LiveKit request failed")
}
// Ownership is written only by this server when creating a fresh LiveKit room.
type broadcastMetadata struct {
Mode string `json:"mode"`
OwnerID string `json:"owner_id"`
DisplayName string `json:"display_name"`
}
func readBroadcast(room *livekit.Room) (broadcastMetadata, bool) {
var metadata broadcastMetadata
err := json.Unmarshal([]byte(room.Metadata), &metadata)
return metadata, err == nil && metadata.Mode == "broadcast" && metadata.OwnerID != "" && metadata.DisplayName != ""
}
func findBroadcast(ctx context.Context, client *lksdk.RoomServiceClient, name string) (*livekit.Room, broadcastMetadata, error) {
result, err := client.ListRooms(ctx, &livekit.ListRoomsRequest{Names: []string{name}})
if err != nil {
return nil, broadcastMetadata{}, liveKitError(err)
}
for _, room := range result.Rooms {
if room.Name == name {
if metadata, ok := readBroadcast(room); ok {
return room, metadata, nil
}
}
}
return nil, broadcastMetadata{}, &errs.Error{Code: errs.NotFound, Message: "broadcast room not found"}
}
func requireRoomOwner(ctx context.Context, client *lksdk.RoomServiceClient, name string) error {
uid, ok := auth.UserID()
if !ok {
return &errs.Error{Code: errs.Unauthenticated, Message: "must be logged in"}
}
_, metadata, err := findBroadcast(ctx, client, name)
if err != nil {
return err
}
if metadata.OwnerID != string(uid) {
return &errs.Error{Code: errs.PermissionDenied, Message: "room owner required"}
}
return nil
}
func signJoinToken(key, secret, identity, room string, owner bool) (string, error) {
grant := &lkauth.VideoGrant{RoomJoin: true, Room: room}
grant.SetCanPublish(owner)
grant.SetCanSubscribe(true)
grant.SetCanPublishData(false)
grant.SetCanUpdateOwnMetadata(false)
return lkauth.NewAccessToken(key, secret).
SetIdentity(identity).
SetValidFor(tokenTTL).
SetVideoGrant(grant).
ToJWT()
}

325
videochat/livekit_test.go Normal file
View File

@@ -0,0 +1,325 @@
package videochat
import (
"context"
"io"
"net/http"
"net/http/httptest"
"strings"
"testing"
"time"
authsvc "encore.app/auth"
"encore.dev/beta/errs"
"encore.dev/et"
lkauth "github.com/livekit/protocol/auth"
"github.com/livekit/protocol/livekit"
lksdk "github.com/livekit/server-sdk-go/v2"
"github.com/twitchtv/twirp"
"google.golang.org/protobuf/proto"
)
func TestJoinTokenPermissions(t *testing.T) {
token, err := signJoinToken("test-key", "test-secret", "user-123", "demo", true)
if err != nil {
t.Fatal(err)
}
verifier, err := lkauth.ParseAPIToken(token)
if err != nil {
t.Fatal(err)
}
claims, grants, err := verifier.Verify("test-secret")
if err != nil {
t.Fatal(err)
}
if verifier.APIKey() != "test-key" || grants.Identity != "user-123" {
t.Fatal("unexpected token issuer or identity")
}
g := grants.Video
if g == nil || g.Room != "demo" || !g.RoomJoin || !g.GetCanPublish() || !g.GetCanSubscribe() || g.GetCanPublishData() || g.GetCanUpdateOwnMetadata() {
t.Fatalf("unexpected participant grants: %+v", g)
}
if g.RoomAdmin || g.RoomCreate || g.RoomList || g.RoomRecord {
t.Fatal("participant must not receive administrative grants")
}
if remaining := time.Until(claims.ExpiresAt.Time); remaining < tokenTTL-5*time.Second || remaining > tokenTTL {
t.Fatalf("unexpected token lifetime: %v", remaining)
}
if _, _, err := verifier.Verify("wrong-secret"); err == nil {
t.Fatal("token must reject an incorrect signing secret")
}
}
func TestValidation(t *testing.T) {
for _, name := range []string{"", " ", " demo", "demo ", "a/b", "a\\b", "a\x00b", strings.Repeat("a", 129)} {
if err := validateRoomName(name); errs.Code(err) != errs.InvalidArgument {
t.Errorf("expected invalid room name %q, got %v", name, err)
}
}
if err := validateRoomName("demo-123"); err != nil {
t.Fatal(err)
}
for _, raw := range []string{"", "example.com", "ftp://example.com", "https://user:pass@example.com", "https://example.com?key=secret"} {
if _, err := websocketURL(raw); err == nil {
t.Errorf("accepted invalid URL %q", raw)
}
}
for raw, want := range map[string]string{"https://example.com/": "wss://example.com", "http://localhost:7880": "ws://localhost:7880", "wss://example.com": "wss://example.com"} {
if got, err := websocketURL(raw); err != nil || got != want {
t.Errorf("websocketURL(%q) = %q, %v; want %q", raw, got, err, want)
}
}
}
func TestManagementRequiresAuthentication(t *testing.T) {
defer et.OverrideAuthInfo("", nil)
et.OverrideAuthInfo("", nil)
ctx := context.Background()
_, createErr := CreateRoom(ctx, &CreateRoomParams{Name: "demo"})
_, listErr := ListRooms(ctx)
deleteErr := DeleteRoom(ctx, "demo")
_, participantsErr := ListParticipants(ctx, "demo")
removeErr := RemoveParticipant(ctx, "demo", "user-456")
for _, err := range []error{createErr, listErr, deleteErr, participantsErr, removeErr} {
if errs.Code(err) != errs.Unauthenticated {
t.Fatalf("expected unauthenticated, got %v", err)
}
}
et.OverrideAuthInfo("viewer", &authsvc.AuthData{Role: "user"})
if _, err := CreateRoom(ctx, &CreateRoomParams{Name: "demo"}); errs.Code(err) != errs.PermissionDenied {
t.Fatalf("only administrators can create rooms: %v", err)
}
}
func TestViewerTokenCannotPublish(t *testing.T) {
token, err := signJoinToken("test-key", "test-secret", "viewer", "demo", false)
if err != nil {
t.Fatal(err)
}
verifier, err := lkauth.ParseAPIToken(token)
if err != nil {
t.Fatal(err)
}
_, grants, err := verifier.Verify("test-secret")
if err != nil {
t.Fatal(err)
}
g := grants.Video
if g.GetCanPublish() || g.GetCanPublishData() || g.GetCanUpdateOwnMetadata() || !g.GetCanSubscribe() || g.RoomAdmin || g.RoomCreate {
t.Fatalf("unsafe viewer permissions: %+v", g)
}
}
func TestCreateTokenChecksRoomAndIdentity(t *testing.T) {
original := secrets
defer func() { secrets = original; et.OverrideAuthInfo("", nil) }()
roomExists := true
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/twirp/livekit.RoomService/ListRooms" {
t.Errorf("unexpected LiveKit operation: %s", r.URL.Path)
}
body, err := io.ReadAll(r.Body)
var request livekit.ListRoomsRequest
if err != nil || proto.Unmarshal(body, &request) != nil || len(request.Names) != 1 || request.Names[0] != "demo" {
t.Error("room lookup must filter by the requested name")
}
verifier, err := lkauth.ParseAPIToken(strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer "))
if err != nil {
t.Error(err)
w.WriteHeader(http.StatusUnauthorized)
return
}
_, grants, err := verifier.Verify("test-secret")
if err != nil || grants.Video == nil || !grants.Video.RoomList {
t.Error("missing signed room-list permission")
}
w.Header().Set("Content-Type", "application/protobuf")
result := &livekit.ListRoomsResponse{}
if roomExists {
result.Rooms = []*livekit.Room{{Name: "demo", Metadata: `{"mode":"broadcast","owner_id":"owner-123","display_name":"Demo"}`}}
}
payload, err := proto.Marshal(result)
if err != nil {
t.Error(err)
}
_, _ = w.Write(payload)
}))
defer server.Close()
secrets.LiveKitURL = server.URL
secrets.LiveKitAPIKey = "test-key"
secrets.LiveKitAPISecret = "test-secret"
ctx := context.Background()
et.OverrideAuthInfo("", nil)
if _, err := CreateToken(ctx, &TokenParams{RoomName: "demo"}); errs.Code(err) != errs.Unauthenticated {
t.Fatalf("expected unauthenticated, got %v", err)
}
et.OverrideAuthInfo("user-123", &authsvc.AuthData{Role: "user"})
response, err := CreateToken(ctx, &TokenParams{RoomName: "demo"})
if err != nil {
t.Fatal(err)
}
if response.Role != "viewer" || response.OwnerID != "owner-123" || response.Identity != "user-123" || response.ExpiresIn != 600 || !strings.HasPrefix(response.ServerURL, "ws://") {
t.Fatalf("unexpected token response: %+v", response)
}
verifier, err := lkauth.ParseAPIToken(response.ParticipantToken)
if err != nil {
t.Fatal(err)
}
_, grants, err := verifier.Verify("test-secret")
if err != nil || grants.Identity != "user-123" || grants.Video.Room != "demo" || grants.Video.GetCanPublish() {
t.Fatal("token must be bound to the authenticated user and requested room")
}
for _, role := range []string{"user", "admin"} {
if role == "admin" {
et.OverrideAuthInfo("other-admin", &authsvc.AuthData{Role: "admin"})
}
if err := requireRoomOwner(ctx, mustTestClient(t), "demo"); errs.Code(err) != errs.PermissionDenied {
t.Fatalf("non-owner must not manage rooms: %v", err)
}
if err := DeleteRoom(ctx, "demo"); errs.Code(err) != errs.PermissionDenied {
t.Fatalf("non-owner delete: %v", err)
}
if _, err := ListParticipants(ctx, "demo"); errs.Code(err) != errs.PermissionDenied {
t.Fatalf("non-owner participants: %v", err)
}
if err := RemoveParticipant(ctx, "demo", "viewer"); errs.Code(err) != errs.PermissionDenied {
t.Fatalf("non-owner removal: %v", err)
}
}
et.OverrideAuthInfo("owner-123", &authsvc.AuthData{Role: "admin"})
if err := requireRoomOwner(ctx, mustTestClient(t), "demo"); err != nil {
t.Fatal(err)
}
ownerResponse, err := CreateToken(ctx, &TokenParams{RoomName: "demo"})
if err != nil || ownerResponse.Role != "owner" {
t.Fatalf("owner token: %v", err)
}
ownerVerifier, _ := lkauth.ParseAPIToken(ownerResponse.ParticipantToken)
_, ownerGrants, err := ownerVerifier.Verify("test-secret")
if err != nil || !ownerGrants.Video.GetCanPublish() {
t.Fatal("owner must be able to publish")
}
roomExists = false
if _, err := CreateToken(ctx, &TokenParams{RoomName: "demo"}); errs.Code(err) != errs.NotFound {
t.Fatalf("expected missing room, got %v", err)
}
}
func TestLiveKitErrorMapping(t *testing.T) {
for rpc, want := range map[twirp.ErrorCode]errs.ErrCode{twirp.NotFound: errs.NotFound, twirp.InvalidArgument: errs.InvalidArgument, twirp.Unavailable: errs.Unavailable} {
if err := liveKitError(twirp.NewError(rpc, "upstream error")); errs.Code(err) != want {
t.Errorf("expected %v, got %v", want, err)
}
}
}
func mustTestClient(t *testing.T) *lksdk.RoomServiceClient {
t.Helper()
client, _, err := liveKitClient()
if err != nil {
t.Fatal(err)
}
return client
}
func TestLegacyRoomsHaveNoInferredOwner(t *testing.T) {
for _, metadata := range []string{"", "{}", `{"mode":"broadcast"}`, `{"mode":"meeting","owner_id":"admin","display_name":"demo"}`} {
if _, ok := readBroadcast(&livekit.Room{Metadata: metadata}); ok {
t.Fatalf("accepted unmanaged room: %s", metadata)
}
}
}
// Exercise creation, listing, ownership and moderation against the real SDK's HTTP transport.
func TestBroadcastLifecycle(t *testing.T) {
useTestProfilesDB(t)
original := secrets
defer func() { secrets = original; et.OverrideAuthInfo("", nil) }()
var stored *livekit.Room
removed := ""
deleted := false
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
data, _ := io.ReadAll(r.Body)
var response proto.Message
switch r.URL.Path {
case "/twirp/livekit.RoomService/CreateRoom":
var request livekit.CreateRoomRequest
if err := proto.Unmarshal(data, &request); err != nil {
t.Error(err)
}
stored = &livekit.Room{Sid: "sid-1", Name: request.Name, Metadata: request.Metadata, MaxParticipants: request.MaxParticipants}
response = stored
case "/twirp/livekit.RoomService/ListRooms":
result := &livekit.ListRoomsResponse{}
if stored != nil {
result.Rooms = []*livekit.Room{stored, {Name: "legacy-room"}}
}
response = result
case "/twirp/livekit.RoomService/ListParticipants":
response = &livekit.ListParticipantsResponse{Participants: []*livekit.ParticipantInfo{{Sid: "viewer-sid", Identity: "viewer"}}}
case "/twirp/livekit.RoomService/RemoveParticipant":
var request livekit.RoomParticipantIdentity
if err := proto.Unmarshal(data, &request); err != nil {
t.Error(err)
}
removed = request.Identity
response = &livekit.RemoveParticipantResponse{}
case "/twirp/livekit.RoomService/DeleteRoom":
stored = nil
deleted = true
response = &livekit.DeleteRoomResponse{}
default:
t.Errorf("unexpected operation: %s", r.URL.Path)
w.WriteHeader(404)
return
}
w.Header().Set("Content-Type", "application/protobuf")
payload, _ := proto.Marshal(response)
_, _ = w.Write(payload)
}))
defer server.Close()
secrets.LiveKitURL, secrets.LiveKitAPIKey, secrets.LiveKitAPISecret = server.URL, "test-key", "test-secret"
ctx := context.Background()
et.OverrideAuthInfo("owner", &authsvc.AuthData{Role: "admin"})
created, err := CreateRoom(ctx, &CreateRoomParams{Name: "My broadcast", MaxParticipants: 20})
if err != nil {
t.Fatal(err)
}
if created.OwnerID != "owner" || created.DisplayName != "My broadcast" || created.Mode != "broadcast" || !strings.HasPrefix(created.Name, "broadcast-") || created.MaxParticipants != 20 {
t.Fatalf("invalid created room: %+v", created)
}
firstName := created.Name
created, err = CreateRoom(ctx, &CreateRoomParams{Name: "My broadcast"})
if err != nil || created.Name == firstName {
t.Fatal("duplicate titles must have distinct room identities")
}
et.OverrideAuthInfo("viewer", &authsvc.AuthData{Role: "user"})
listed, err := ListRooms(ctx)
if err != nil || len(listed.Rooms) != 1 || listed.Rooms[0].OwnerID != "owner" {
t.Fatalf("viewer must see broadcasts only: %+v, %v", listed, err)
}
if err := DeleteRoom(ctx, created.Name); errs.Code(err) != errs.PermissionDenied || deleted {
t.Fatal("viewer deleted the room")
}
if err := RemoveParticipant(ctx, created.Name, "owner"); errs.Code(err) != errs.PermissionDenied || removed != "" {
t.Fatal("viewer removed the owner")
}
et.OverrideAuthInfo("owner", &authsvc.AuthData{Role: "admin"})
members, err := ListParticipants(ctx, created.Name)
if err != nil || len(members.Participants) != 1 {
t.Fatalf("owner list participants: %v", err)
}
if err := RemoveParticipant(ctx, created.Name, "owner"); errs.Code(err) != errs.InvalidArgument {
t.Fatal("owner self-removal must be rejected")
}
if err := RemoveParticipant(ctx, created.Name, "viewer"); err != nil || removed != "viewer" {
t.Fatalf("owner remove viewer: %v", err)
}
if err := DeleteRoom(ctx, created.Name); err != nil || !deleted {
t.Fatalf("owner close: %v", err)
}
if _, err := CreateToken(ctx, &TokenParams{RoomName: created.Name}); errs.Code(err) != errs.NotFound {
t.Fatal("closed room must not receive new tokens")
}
}

79
videochat/open_room.go Normal file
View File

@@ -0,0 +1,79 @@
package videochat
import (
"context"
"encoding/json"
"encore.dev/beta/auth"
"encore.dev/beta/errs"
"github.com/livekit/protocol/livekit"
)
// OpenRoom returns an existing broadcast or provisions an enabled personal setup.
// The owner comes exclusively from administrator-managed settings, never the caller.
//
//encore:api auth method=POST path=/videochat/rooms/:roomName/open
func OpenRoom(ctx context.Context, roomName string) (*Room, error) {
if _, ok := auth.UserID(); !ok {
return nil, &errs.Error{Code: errs.Unauthenticated, Message: "must be logged in"}
}
if err := validateRoomName(roomName); err != nil {
return nil, err
}
client, _, err := liveKitClient()
if err != nil {
return nil, err
}
ctx, cancel := context.WithTimeout(ctx, apiTimeout)
defer cancel()
// Serialize lookup and creation across application instances for this name.
tx, err := profilesDB.Begin(ctx)
if err != nil {
return nil, errs.WrapCode(err, errs.Internal, "failed to open room")
}
defer tx.Rollback()
if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1, 0))`, "videochat/open/"+roomName); err != nil {
return nil, errs.WrapCode(err, errs.Internal, "failed to lock room creation")
}
result, err := client.ListRooms(ctx, &livekit.ListRoomsRequest{})
if err != nil {
return nil, liveKitError(err)
}
for _, room := range result.Rooms {
if room.Name == roomName {
if _, ok := readBroadcast(room); !ok {
return nil, &errs.Error{Code: errs.FailedPrecondition, Message: "room name is already used by an unmanaged room"}
}
return roomResponse(room), nil
}
}
// Multiple enabled owners with the same name are ambiguous: do not pick one.
var owner string
var count int
err = tx.QueryRow(ctx, `SELECT COUNT(*), COALESCE(MIN(user_id::text), '') FROM personal_setup WHERE can_broadcast = TRUE AND room_name = $1`, roomName).Scan(&count, &owner)
if err != nil {
return nil, errs.WrapCode(err, errs.Internal, "failed to fetch configured room")
}
if count == 0 {
return nil, &errs.Error{Code: errs.NotFound, Message: "no enabled broadcast setup for this room"}
}
if count != 1 {
return nil, &errs.Error{Code: errs.FailedPrecondition, Message: "room name is configured for more than one owner"}
}
for _, room := range result.Rooms {
if metadata, ok := readBroadcast(room); ok && metadata.OwnerID == owner && metadata.DisplayName == roomName {
return roomResponse(room), nil
}
}
metadata, err := json.Marshal(broadcastMetadata{Mode: "broadcast", OwnerID: owner, DisplayName: roomName})
if err != nil {
return nil, errs.WrapCode(err, errs.Internal, "failed to encode room metadata")
}
room, err := client.CreateRoom(ctx, &livekit.CreateRoomRequest{Name: roomName, Metadata: string(metadata)})
if err != nil {
return nil, liveKitError(err)
}
return roomResponse(room), nil
}

129
videochat/open_room_test.go Normal file
View File

@@ -0,0 +1,129 @@
package videochat
import (
"context"
"io"
"net/http"
"net/http/httptest"
"sync"
"testing"
authsvc "encore.app/auth"
"encore.dev/beta/errs"
"encore.dev/et"
"github.com/livekit/protocol/livekit"
"google.golang.org/protobuf/proto"
)
func TestOpenRoomCreatesAndReusesConfiguredRoom(t *testing.T) {
useTestProfilesDB(t)
ctx := context.Background()
const owner = "00000000-0000-4000-8000-000000000001"
if _, err := profilesDB.Exec(ctx, `INSERT INTO user_profiles (user_id, display_name, email) VALUES ($1, 'Owner', 'open@example.invalid')`, owner); err != nil {
t.Fatal(err)
}
if _, err := profilesDB.Exec(ctx, `INSERT INTO personal_setup (user_id, can_broadcast, room_name) VALUES ($1, TRUE, 'Configured room')`, owner); err != nil {
t.Fatal(err)
}
var mu sync.Mutex
var rooms []*livekit.Room
creates := 0
failList := false
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
mu.Lock()
defer mu.Unlock()
var response proto.Message
switch r.URL.Path {
case "/twirp/livekit.RoomService/ListRooms":
if failList {
w.WriteHeader(503)
return
}
response = &livekit.ListRoomsResponse{Rooms: rooms}
case "/twirp/livekit.RoomService/CreateRoom":
body, _ := io.ReadAll(r.Body)
var request livekit.CreateRoomRequest
if err := proto.Unmarshal(body, &request); err != nil {
t.Error(err)
}
creates++
room := &livekit.Room{Sid: "created-sid", Name: request.Name, Metadata: request.Metadata}
rooms = append(rooms, room)
response = room
default:
t.Errorf("unexpected operation: %s", r.URL.Path)
w.WriteHeader(404)
return
}
payload, _ := proto.Marshal(response)
w.Header().Set("Content-Type", "application/protobuf")
_, _ = w.Write(payload)
}))
defer server.Close()
original := secrets
defer func() { secrets = original; et.OverrideAuthInfo("", nil) }()
secrets.LiveKitURL, secrets.LiveKitAPIKey, secrets.LiveKitAPISecret = server.URL, "test-key", "test-secret"
et.OverrideAuthInfo("", nil)
if _, err := OpenRoom(ctx, "Configured room"); errs.Code(err) != errs.Unauthenticated {
t.Fatalf("anonymous open: %v", err)
}
et.OverrideAuthInfo("viewer", &authsvc.AuthData{Role: "user"})
if _, err := OpenRoom(ctx, "unknown"); errs.Code(err) != errs.NotFound {
t.Fatalf("unconfigured room: %v", err)
}
if _, err := OpenRoom(ctx, "bad/name"); errs.Code(err) != errs.InvalidArgument {
t.Fatalf("invalid name: %v", err)
}
// Concurrent viewers may provision the configured room, but never become its owner.
var wg sync.WaitGroup
for i := 0; i < 4; i++ {
wg.Add(1)
go func() {
defer wg.Done()
et.OverrideAuthInfo("viewer", &authsvc.AuthData{Role: "user"})
room, err := OpenRoom(ctx, "Configured room")
if err != nil {
t.Error(err)
return
}
if !room.Active || room.OwnerID != owner || room.Name != "Configured room" || room.SID != "created-sid" {
t.Errorf("invalid room: %+v", room)
}
}()
}
wg.Wait()
if creates != 1 {
t.Fatalf("expected a single creation, got %d", creates)
}
token, err := CreateToken(ctx, &TokenParams{RoomName: "Configured room"})
if err != nil || token.Role != "viewer" || token.OwnerID != owner {
t.Fatalf("opener must remain viewer: %+v, %v", token, err)
}
mu.Lock()
failList = true
mu.Unlock()
if _, err := OpenRoom(ctx, "Configured room"); err == nil {
t.Fatal("upstream errors must not be treated as missing rooms")
}
if creates != 1 {
t.Fatal("upstream failure created a room")
}
mu.Lock()
failList = false
rooms = []*livekit.Room{{Name: "Configured room"}}
mu.Unlock()
if _, err := OpenRoom(ctx, "Configured room"); errs.Code(err) != errs.FailedPrecondition {
t.Fatalf("unmanaged collision: %v", err)
}
mu.Lock()
rooms = nil
mu.Unlock()
if _, err := profilesDB.Exec(ctx, `UPDATE personal_setup SET can_broadcast = FALSE, room_name = '' WHERE user_id = $1`, owner); err != nil {
t.Fatal(err)
}
if _, err := OpenRoom(ctx, "Configured room"); errs.Code(err) != errs.NotFound {
t.Fatalf("disabled room was created: %v", err)
}
}

84
videochat/participants.go Normal file
View File

@@ -0,0 +1,84 @@
package videochat
import (
"context"
"strings"
"encore.dev/beta/auth"
"encore.dev/beta/errs"
"github.com/livekit/protocol/livekit"
)
type Participant struct {
SID string `json:"sid"`
Identity string `json:"identity"`
Name string `json:"name"`
State string `json:"state"`
}
type ListParticipantsResponse struct {
Participants []*Participant `json:"participants"`
}
// ListParticipants lists participants in a room. Requires the room owner.
//
//encore:api auth method=GET path=/videochat/rooms/:roomName/participants
func ListParticipants(ctx context.Context, roomName string) (*ListParticipantsResponse, error) {
if _, ok := auth.UserID(); !ok {
return nil, &errs.Error{Code: errs.Unauthenticated, Message: "must be logged in"}
}
if err := validateRoomName(roomName); err != nil {
return nil, err
}
client, _, err := liveKitClient()
if err != nil {
return nil, err
}
ctx, cancel := context.WithTimeout(ctx, apiTimeout)
defer cancel()
if err := requireRoomOwner(ctx, client, roomName); err != nil {
return nil, err
}
result, err := client.ListParticipants(ctx, &livekit.ListParticipantsRequest{Room: roomName})
if err != nil {
return nil, liveKitError(err)
}
response := &ListParticipantsResponse{Participants: make([]*Participant, 0, len(result.Participants))}
for _, p := range result.Participants {
response.Participants = append(response.Participants, &Participant{
SID: p.Sid, Identity: p.Identity, Name: p.Name, State: p.State.String(),
})
}
return response, nil
}
// RemoveParticipant disconnects a participant. Requires the room owner.
// It does not ban the user from requesting another token.
//
//encore:api auth method=DELETE path=/videochat/rooms/:roomName/participants/:identity
func RemoveParticipant(ctx context.Context, roomName, identity string) error {
if _, ok := auth.UserID(); !ok {
return &errs.Error{Code: errs.Unauthenticated, Message: "must be logged in"}
}
if err := validateRoomName(roomName); err != nil {
return err
}
if strings.TrimSpace(identity) == "" {
return &errs.Error{Code: errs.InvalidArgument, Message: "participant identity is required"}
}
client, _, err := liveKitClient()
if err != nil {
return err
}
ctx, cancel := context.WithTimeout(ctx, apiTimeout)
defer cancel()
if err := requireRoomOwner(ctx, client, roomName); err != nil {
return err
}
uid, _ := auth.UserID()
if identity == string(uid) {
return &errs.Error{Code: errs.InvalidArgument, Message: "owner cannot remove themselves; leave or close the room instead"}
}
_, err = client.RemoveParticipant(ctx, &livekit.RoomParticipantIdentity{Room: roomName, Identity: identity})
return liveKitError(err)
}

173
videochat/rooms.go Normal file
View File

@@ -0,0 +1,173 @@
package videochat
import (
"context"
"encoding/json"
"encore.dev/beta/auth"
"encore.dev/beta/errs"
"encore.dev/storage/sqldb"
"encore.dev/types/uuid"
"github.com/livekit/protocol/livekit"
)
var profilesDB = sqldb.Named("profiles")
type CreateRoomParams struct {
Name string `json:"name"`
MaxParticipants uint32 `json:"max_participants"`
EmptyTimeout uint32 `json:"empty_timeout"`
}
type Room struct {
Active bool `json:"active"`
OwnerID string `json:"owner_id"`
DisplayName string `json:"display_name"`
Mode string `json:"mode"`
SID string `json:"sid"`
Name string `json:"name"`
NumParticipants uint32 `json:"num_participants"`
MaxParticipants uint32 `json:"max_participants"`
CreationTime int64 `json:"creation_time"`
}
type ListRoomsResponse struct {
Rooms []*Room `json:"rooms"`
}
func roomResponse(room *livekit.Room) *Room {
metadata, _ := readBroadcast(room)
return &Room{Active: true, OwnerID: metadata.OwnerID, DisplayName: metadata.DisplayName, Mode: metadata.Mode, SID: room.Sid, Name: room.Name, NumParticipants: room.NumParticipants,
MaxParticipants: room.MaxParticipants, CreationTime: room.CreationTime}
}
// CreateRoom creates a LiveKit room. Requires an administrator.
//
//encore:api auth method=POST path=/videochat/rooms
func CreateRoom(ctx context.Context, p *CreateRoomParams) (*Room, error) {
if err := requireAdmin(); err != nil {
return nil, err
}
if err := validateRoomName(p.Name); err != nil {
return nil, err
}
client, _, err := liveKitClient()
if err != nil {
return nil, err
}
ctx, cancel := context.WithTimeout(ctx, apiTimeout)
defer cancel()
uid, _ := auth.UserID()
id, err := uuid.NewV4()
if err != nil {
return nil, errs.WrapCode(err, errs.Internal, "failed to create room identifier")
}
metadata, err := json.Marshal(broadcastMetadata{Mode: "broadcast", OwnerID: string(uid), DisplayName: p.Name})
if err != nil {
return nil, errs.WrapCode(err, errs.Internal, "failed to encode room metadata")
}
room, err := client.CreateRoom(ctx, &livekit.CreateRoomRequest{
Name: "broadcast-" + id.String(), Metadata: string(metadata), MaxParticipants: p.MaxParticipants, EmptyTimeout: p.EmptyTimeout,
})
if err != nil {
return nil, liveKitError(err)
}
return roomResponse(room), nil
}
// ListRooms lists live broadcasts and enabled personal setups for authenticated viewers.
// Configured rooms absent from LiveKit are returned with active=false.
//
//encore:api auth method=GET path=/videochat/rooms
func ListRooms(ctx context.Context) (*ListRoomsResponse, error) {
if _, ok := auth.UserID(); !ok {
return nil, &errs.Error{Code: errs.Unauthenticated, Message: "must be logged in"}
}
client, _, err := liveKitClient()
if err != nil {
return nil, err
}
ctx, cancel := context.WithTimeout(ctx, apiTimeout)
defer cancel()
result, err := client.ListRooms(ctx, &livekit.ListRoomsRequest{})
if err != nil {
return nil, liveKitError(err)
}
response := &ListRoomsResponse{Rooms: make([]*Room, 0, len(result.Rooms))}
for _, room := range result.Rooms {
if _, ok := readBroadcast(room); ok {
response.Rooms = append(response.Rooms, roomResponse(room))
}
}
configured, err := configuredBroadcastRooms(ctx)
if err != nil {
return nil, err
}
response.Rooms = mergeBroadcastRooms(response.Rooms, configured)
return response, nil
}
func configuredBroadcastRooms(ctx context.Context) ([]*Room, error) {
rows, err := profilesDB.Query(ctx, `
SELECT user_id::text, room_name
FROM personal_setup
WHERE can_broadcast = TRUE AND room_name <> ''
ORDER BY room_name, user_id
`)
if err != nil {
return nil, errs.WrapCode(err, errs.Internal, "failed to list configured broadcast rooms")
}
defer rows.Close()
rooms := make([]*Room, 0)
for rows.Next() {
room := &Room{Mode: "broadcast"}
if err := rows.Scan(&room.OwnerID, &room.Name); err != nil {
return nil, errs.WrapCode(err, errs.Internal, "failed to read configured broadcast room")
}
room.DisplayName = room.Name
rooms = append(rooms, room)
}
if err := rows.Err(); err != nil {
return nil, errs.WrapCode(err, errs.Internal, "failed to list configured broadcast rooms")
}
return rooms, nil
}
func mergeBroadcastRooms(active, configured []*Room) []*Room {
type ownerTitle struct{ owner, title string }
names := make(map[string]bool, len(active))
titles := make(map[ownerTitle]bool, len(active))
for _, room := range active {
names[room.Name] = true
titles[ownerTitle{room.OwnerID, room.DisplayName}] = true
}
for _, room := range configured {
if !names[room.Name] && !titles[ownerTitle{room.OwnerID, room.Name}] {
active = append(active, room)
}
}
return active
}
// DeleteRoom closes a room and disconnects its participants. Requires the room owner.
//
//encore:api auth method=DELETE path=/videochat/rooms/:roomName
func DeleteRoom(ctx context.Context, roomName string) error {
if _, ok := auth.UserID(); !ok {
return &errs.Error{Code: errs.Unauthenticated, Message: "must be logged in"}
}
if err := validateRoomName(roomName); err != nil {
return err
}
client, _, err := liveKitClient()
if err != nil {
return err
}
ctx, cancel := context.WithTimeout(ctx, apiTimeout)
defer cancel()
if err := requireRoomOwner(ctx, client, roomName); err != nil {
return err
}
_, err = client.DeleteRoom(ctx, &livekit.DeleteRoomRequest{Room: roomName})
return liveKitError(err)
}

99
videochat/rooms_test.go Normal file
View File

@@ -0,0 +1,99 @@
package videochat
import (
"context"
"net/http"
"net/http/httptest"
"testing"
authsvc "encore.app/auth"
"encore.dev/et"
"github.com/livekit/protocol/livekit"
"google.golang.org/protobuf/proto"
)
func useTestProfilesDB(t *testing.T) {
t.Helper()
testDB, err := et.NewTestDatabase(context.Background(), "profiles")
if err != nil {
t.Fatal(err)
}
original := profilesDB
profilesDB = testDB
t.Cleanup(func() { profilesDB = original })
}
func TestListRoomsIncludesInactiveSetups(t *testing.T) {
useTestProfilesDB(t)
ctx := context.Background()
const owner = "00000000-0000-4000-8000-000000000001"
const offline = "00000000-0000-4000-8000-000000000002"
const disabled = "00000000-0000-4000-8000-000000000003"
const sameTitle = "00000000-0000-4000-8000-000000000004"
for _, id := range []string{owner, offline, disabled, sameTitle} {
if _, err := profilesDB.Exec(ctx, `INSERT INTO user_profiles (user_id, display_name, email) VALUES ($1, 'Test', $2)`, id, id+"@example.invalid"); err != nil {
t.Fatal(err)
}
}
for _, setup := range []struct {
id, name string
enabled bool
}{
{owner, "Live show", true}, {offline, "Offline show", true},
{disabled, "", false}, {sameTitle, "Live show", true},
} {
if _, err := profilesDB.Exec(ctx, `INSERT INTO personal_setup (user_id, can_broadcast, room_name) VALUES ($1, $2, $3)`, setup.id, setup.enabled, setup.name); err != nil {
t.Fatal(err)
}
}
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/twirp/livekit.RoomService/ListRooms" {
t.Errorf("unexpected operation: %s", r.URL.Path)
}
payload, err := proto.Marshal(&livekit.ListRoomsResponse{Rooms: []*livekit.Room{
{Sid: "sid-live", Name: "broadcast-generated", NumParticipants: 3,
Metadata: `{"mode":"broadcast","owner_id":"` + owner + `","display_name":"Live show"}`},
{Name: "unmanaged"},
}})
if err != nil {
t.Error(err)
}
w.Header().Set("Content-Type", "application/protobuf")
_, _ = w.Write(payload)
}))
defer server.Close()
original := secrets
defer func() { secrets = original; et.OverrideAuthInfo("", nil) }()
secrets.LiveKitURL, secrets.LiveKitAPIKey, secrets.LiveKitAPISecret = server.URL, "test-key", "test-secret"
et.OverrideAuthInfo("viewer", &authsvc.AuthData{Role: "user"})
result, err := ListRooms(ctx)
if err != nil {
t.Fatal(err)
}
if len(result.Rooms) != 3 {
t.Fatalf("expected one live and two inactive rooms: %+v", result.Rooms)
}
byOwner := make(map[string]*Room)
for _, room := range result.Rooms {
byOwner[room.OwnerID] = room
}
if room := byOwner[owner]; room == nil || !room.Active || room.Name != "broadcast-generated" || room.SID != "sid-live" || room.NumParticipants != 3 {
t.Fatalf("live room must retain its data without duplication: %+v", room)
}
for _, id := range []string{offline, sameTitle} {
if room := byOwner[id]; room == nil || room.Active || room.SID != "" || room.Mode != "broadcast" || room.Name != room.DisplayName {
t.Fatalf("expected inactive configured room: %+v", room)
}
}
}
func TestMergeBroadcastRoomsMatchesTechnicalName(t *testing.T) {
live := &Room{Active: true, Name: "broadcast-123", OwnerID: "owner", DisplayName: "Title"}
result := mergeBroadcastRooms([]*Room{live}, []*Room{{Name: "broadcast-123", OwnerID: "owner"}})
if len(result) != 1 || result[0] != live {
t.Fatal("technical room name must not produce a duplicate")
}
if got := mergeBroadcastRooms([]*Room{}, nil); got == nil || len(got) != 0 {
t.Fatal("empty list must stay an empty JSON array")
}
}

57
videochat/tokens.go Normal file
View File

@@ -0,0 +1,57 @@
package videochat
import (
"context"
"encore.dev/beta/auth"
"encore.dev/beta/errs"
)
type TokenParams struct {
RoomName string `json:"room_name"`
}
type TokenResponse struct {
OwnerID string `json:"owner_id"`
Role string `json:"role"`
ServerURL string `json:"server_url"`
ParticipantToken string `json:"participant_token" encore:"sensitive"`
RoomName string `json:"room_name"`
Identity string `json:"identity"`
ExpiresIn int64 `json:"expires_in"`
}
// CreateToken grants the authenticated user access to an existing room by name.
//
//encore:api auth method=POST path=/videochat/token
func CreateToken(ctx context.Context, p *TokenParams) (*TokenResponse, error) {
uid, ok := auth.UserID()
if !ok {
return nil, &errs.Error{Code: errs.Unauthenticated, Message: "must be logged in"}
}
if err := validateRoomName(p.RoomName); err != nil {
return nil, err
}
client, serverURL, err := liveKitClient()
if err != nil {
return nil, err
}
ctx, cancel := context.WithTimeout(ctx, apiTimeout)
defer cancel()
_, metadata, err := findBroadcast(ctx, client, p.RoomName)
if err != nil {
return nil, err
}
token, err := signJoinToken(secrets.LiveKitAPIKey, secrets.LiveKitAPISecret, string(uid), p.RoomName, metadata.OwnerID == string(uid))
if err != nil {
return nil, errs.WrapCode(err, errs.Internal, "failed to generate LiveKit token")
}
role := "viewer"
if metadata.OwnerID == string(uid) {
role = "owner"
}
return &TokenResponse{OwnerID: metadata.OwnerID, Role: role,
ServerURL: serverURL, ParticipantToken: token, RoomName: p.RoomName,
Identity: string(uid), ExpiresIn: int64(tokenTTL.Seconds()),
}, nil
}

15
videochat/videochat.go Normal file
View File

@@ -0,0 +1,15 @@
// Service videochat provides video chat APIs.
package videochat
import "context"
type StatusResponse struct {
Status string `json:"status"`
}
// Status reports whether the videochat service is available.
//
//encore:api public method=GET path=/videochat/status
func Status(ctx context.Context) (*StatusResponse, error) {
return &StatusResponse{Status: "ok"}, nil
}