From adc0b236a35ae9590867a1052065c6758616e904 Mon Sep 17 00:00:00 2001 From: fabio Date: Sat, 10 Oct 2026 14:17:59 +0200 Subject: [PATCH] feat: add broadcast session tracking and chatlog service with database migrations --- broadcasters/chatlog.go | 132 ++++++++++++++ .../migrations/2_create_chatlog.up.sql | 9 + ...l => 3_broadcaster_uuid_and_status.up.sql} | 6 +- .../migrations/4_add_enabled_used.up.sql | 11 ++ .../migrations/5_chatlog_room_text.up.sql | 4 + videochat/livekit.go | 48 ++++-- .../1_create_broadcast_sessions.up.sql | 16 ++ videochat/open_room.go | 12 +- videochat/participants.go | 10 +- videochat/rooms.go | 26 ++- videochat/sessions.go | 163 ++++++++++++++++++ videochat/sessions_test.go | 61 +++++++ videochat/tokens.go | 3 +- 13 files changed, 473 insertions(+), 28 deletions(-) create mode 100644 broadcasters/chatlog.go create mode 100644 broadcasters/migrations/2_create_chatlog.up.sql rename broadcasters/migrations/{2_broadcaster_uuid_and_status.up.sql => 3_broadcaster_uuid_and_status.up.sql} (66%) create mode 100644 broadcasters/migrations/4_add_enabled_used.up.sql create mode 100644 broadcasters/migrations/5_chatlog_room_text.up.sql create mode 100644 videochat/migrations/1_create_broadcast_sessions.up.sql create mode 100644 videochat/sessions.go create mode 100644 videochat/sessions_test.go diff --git a/broadcasters/chatlog.go b/broadcasters/chatlog.go new file mode 100644 index 0000000..a639851 --- /dev/null +++ b/broadcasters/chatlog.go @@ -0,0 +1,132 @@ +package broadcasters + +import ( + "context" + "strings" + "time" + + "encore.dev/beta/errs" +) + +type ChatLogEntry struct { + ID int64 `json:"id"` + Timestamp time.Time `json:"timestamp"` + Room string `json:"room"` + Type string `json:"type"` + Data string `json:"data"` +} + +type AddChatLogParams struct { + Type string `json:"type"` + Data string `json:"data"` +} + +type ListChatLogParams struct { + Limit int `json:"limit,omitempty" query:"limit"` + Type string `json:"type,omitempty" query:"type"` +} + +type ListChatLogResponse struct { + Messages []*ChatLogEntry `json:"messages"` +} + +// AddChatLog inserisce un nuovo evento o messaggio nella tabella chatlog. +// +//encore:api public method=POST path=/broadcasters/id/:id/chatlog +func AddChatLog(ctx context.Context, id string, p *AddChatLogParams) (*ChatLogEntry, error) { + if strings.TrimSpace(id) == "" { + return nil, &errs.Error{Code: errs.InvalidArgument, Message: "id stanza obbligatorio"} + } + if p == nil || strings.TrimSpace(p.Data) == "" { + return nil, &errs.Error{Code: errs.InvalidArgument, Message: "data obbligatorio"} + } + + msgType := strings.ToUpper(strings.TrimSpace(p.Type)) + if msgType == "" { + msgType = "MESSAGE" + } else if msgType == "ALERT" { + msgType = "ALLERT" + } + switch msgType { + case "ACTION", "MESSAGE", "ALLERT", "CMD": + default: + msgType = "MESSAGE" + } + + entry := &ChatLogEntry{ + Room: id, + Type: msgType, + Data: p.Data, + } + + err := db.QueryRow(ctx, ` + INSERT INTO chatlog (room, type, data, timestamp) + VALUES ($1, $2, $3, NOW()) + RETURNING id, timestamp + `, entry.Room, entry.Type, entry.Data).Scan(&entry.ID, &entry.Timestamp) + if err != nil { + return nil, errs.WrapCode(err, errs.Internal, "impossibile salvare il messaggio di chat") + } + + return entry, nil +} + +// ListChatLog preleva gli ultimi messaggi per la stanza (default 30) e li ordina +// in ordine cronologico crescente (dal più vecchio al più recente), pronti per l'append nella UI. +// +//encore:api public method=GET path=/broadcasters/id/:id/chatlog +func ListChatLog(ctx context.Context, id string, p *ListChatLogParams) (*ListChatLogResponse, error) { + if strings.TrimSpace(id) == "" { + return nil, &errs.Error{Code: errs.InvalidArgument, Message: "id stanza obbligatorio"} + } + + limit := 30 + filterType := "" + if p != nil { + if p.Limit > 0 { + limit = p.Limit + } + if t := strings.TrimSpace(p.Type); t != "" { + filterType = strings.ToUpper(t) + } + } + if limit > 100 { + limit = 100 + } + + query := ` + SELECT id, timestamp, room, type, data + FROM ( + SELECT id, timestamp, room, type, data + FROM chatlog + WHERE room = $1 AND ($3 = '' OR type = $3) + ORDER BY timestamp DESC, id DESC + LIMIT $2 + ) sub + ORDER BY timestamp ASC, id ASC + ` + + rows, err := db.Query(ctx, query, id, limit, filterType) + if err != nil { + return nil, errs.WrapCode(err, errs.Internal, "impossibile recuperare i messaggi di chat") + } + defer rows.Close() + + var list []*ChatLogEntry + for rows.Next() { + var e ChatLogEntry + if err := rows.Scan(&e.ID, &e.Timestamp, &e.Room, &e.Type, &e.Data); err != nil { + return nil, errs.WrapCode(err, errs.Internal, "errore nella lettura dei messaggi") + } + list = append(list, &e) + } + if err := rows.Err(); err != nil { + return nil, errs.WrapCode(err, errs.Internal, "errore iterazione messaggi") + } + + if list == nil { + list = []*ChatLogEntry{} + } + + return &ListChatLogResponse{Messages: list}, nil +} diff --git a/broadcasters/migrations/2_create_chatlog.up.sql b/broadcasters/migrations/2_create_chatlog.up.sql new file mode 100644 index 0000000..9cf7d5f --- /dev/null +++ b/broadcasters/migrations/2_create_chatlog.up.sql @@ -0,0 +1,9 @@ +CREATE TABLE chatlog ( + id BIGSERIAL PRIMARY KEY, + timestamp TIMESTAMPTZ NOT NULL DEFAULT NOW(), + room UUID NOT NULL, + type TEXT NOT NULL CHECK (type IN ('ACTION','MESSAGE','ALLERT','CMD')), + data TEXT NOT NULL +); + +CREATE INDEX idx_chatlog_room ON chatlog(room); diff --git a/broadcasters/migrations/2_broadcaster_uuid_and_status.up.sql b/broadcasters/migrations/3_broadcaster_uuid_and_status.up.sql similarity index 66% rename from broadcasters/migrations/2_broadcaster_uuid_and_status.up.sql rename to broadcasters/migrations/3_broadcaster_uuid_and_status.up.sql index ca70567..f2f163f 100644 --- a/broadcasters/migrations/2_broadcaster_uuid_and_status.up.sql +++ b/broadcasters/migrations/3_broadcaster_uuid_and_status.up.sql @@ -6,6 +6,6 @@ ALTER TABLE broadcasters RENAME COLUMN new_id TO id; ALTER TABLE broadcasters ALTER COLUMN id SET NOT NULL; ALTER TABLE broadcasters ADD PRIMARY KEY (id); -ALTER TABLE broadcasters ADD COLUMN enabled BOOLEAN NOT NULL DEFAULT FALSE; -ALTER TABLE broadcasters ADD COLUMN used BOOLEAN NOT NULL DEFAULT FALSE; -UPDATE broadcasters SET used = TRUE, enabled = TRUE; +-- enabled column already added in previous migration +-- used column already added in previous migration +-- data update for enabled/used columns moved to previous migration diff --git a/broadcasters/migrations/4_add_enabled_used.up.sql b/broadcasters/migrations/4_add_enabled_used.up.sql new file mode 100644 index 0000000..5dc3192 --- /dev/null +++ b/broadcasters/migrations/4_add_enabled_used.up.sql @@ -0,0 +1,11 @@ +-- 4_add_enabled_used.up.sql +-- Aggiunge le colonne "enabled" e "used" alla tabella broadcasters. +-- Queste colonne sono usate dal codice Go per indicare lo stato del profilo. + +ALTER TABLE broadcasters + ADD COLUMN enabled BOOLEAN NOT NULL DEFAULT FALSE, + ADD COLUMN used BOOLEAN NOT NULL DEFAULT FALSE; + +-- Indici facoltativi per velocizzare le query su queste colonne +CREATE INDEX idx_broadcasters_enabled ON broadcasters(enabled); +CREATE INDEX idx_broadcasters_used ON broadcasters(used); diff --git a/broadcasters/migrations/5_chatlog_room_text.up.sql b/broadcasters/migrations/5_chatlog_room_text.up.sql new file mode 100644 index 0000000..8ed4e3f --- /dev/null +++ b/broadcasters/migrations/5_chatlog_room_text.up.sql @@ -0,0 +1,4 @@ +-- 5_chatlog_room_text.up.sql +-- Modifica la colonna "room" da UUID a TEXT per supportare qualsiasi identificatore di stanza (es. nomi LiveKit, slug, broadcast-uuid). + +ALTER TABLE chatlog ALTER COLUMN room TYPE TEXT; diff --git a/videochat/livekit.go b/videochat/livekit.go index 1a6ce94..756797c 100644 --- a/videochat/livekit.go +++ b/videochat/livekit.go @@ -101,34 +101,54 @@ func readBroadcast(room *livekit.Room) (broadcastMetadata, bool) { 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) +func findBroadcast(ctx context.Context, client *lksdk.RoomServiceClient, nameOrID string) (*livekit.Room, broadcastMetadata, error) { + // Prima prova la ricerca filtrata per nome esatto + result, err := client.ListRooms(ctx, &livekit.ListRoomsRequest{Names: []string{nameOrID}}) + if err == nil { + for _, room := range result.Rooms { + if room.Name == nameOrID || room.Sid == nameOrID { + if metadata, ok := readBroadcast(room); ok { + return room, metadata, nil + } + } + } } - for _, room := range result.Rooms { - if room.Name == name { - if metadata, ok := readBroadcast(room); ok { - return room, metadata, nil + + // Se nameOrID è un SID LiveKit (RM_...), cerca su tutte le room + if strings.HasPrefix(nameOrID, "RM_") { + result, err = client.ListRooms(ctx, &livekit.ListRoomsRequest{}) + if err != nil { + return nil, broadcastMetadata{}, liveKitError(err) + } + for _, room := range result.Rooms { + if room.Sid == nameOrID || room.Name == nameOrID { + 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 { +func getOwnedBroadcast(ctx context.Context, client *lksdk.RoomServiceClient, nameOrID string) (*livekit.Room, error) { uid, ok := auth.UserID() if !ok { - return &errs.Error{Code: errs.Unauthenticated, Message: "must be logged in"} + return nil, &errs.Error{Code: errs.Unauthenticated, Message: "must be logged in"} } - _, metadata, err := findBroadcast(ctx, client, name) + room, metadata, err := findBroadcast(ctx, client, nameOrID) if err != nil { - return err + return nil, err } if metadata.OwnerID != string(uid) { - return &errs.Error{Code: errs.PermissionDenied, Message: "room owner required"} + return nil, &errs.Error{Code: errs.PermissionDenied, Message: "room owner required"} } - return nil + return room, nil +} + +func requireRoomOwner(ctx context.Context, client *lksdk.RoomServiceClient, nameOrID string) error { + _, err := getOwnedBroadcast(ctx, client, nameOrID) + return err } func signJoinToken(key, secret, identity, room string, owner bool) (string, error) { diff --git a/videochat/migrations/1_create_broadcast_sessions.up.sql b/videochat/migrations/1_create_broadcast_sessions.up.sql new file mode 100644 index 0000000..35b7029 --- /dev/null +++ b/videochat/migrations/1_create_broadcast_sessions.up.sql @@ -0,0 +1,16 @@ +-- 1_create_broadcast_sessions.up.sql +-- Tabella per registrare ogni apertura e chiusura delle trasmissioni nel servizio videochat. + +CREATE TABLE broadcast_sessions ( + id BIGSERIAL PRIMARY KEY, + owner_id TEXT NOT NULL, + sid TEXT NOT NULL, + name TEXT NOT NULL, + action TEXT NOT NULL CHECK (action IN ('OPEN', 'CLOSE')), + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +CREATE INDEX idx_broadcast_sessions_owner ON broadcast_sessions(owner_id); +CREATE INDEX idx_broadcast_sessions_sid ON broadcast_sessions(sid); +CREATE INDEX idx_broadcast_sessions_name ON broadcast_sessions(name); +CREATE INDEX idx_broadcast_sessions_created_at ON broadcast_sessions(created_at); diff --git a/videochat/open_room.go b/videochat/open_room.go index 2db052a..d7f9827 100644 --- a/videochat/open_room.go +++ b/videochat/open_room.go @@ -42,9 +42,11 @@ func OpenRoom(ctx context.Context, roomName string) (*Room, error) { } for _, room := range result.Rooms { if room.Name == roomName { - if _, ok := readBroadcast(room); !ok { + metadata, ok := readBroadcast(room) + if !ok { return nil, &errs.Error{Code: errs.FailedPrecondition, Message: "room name is already used by an unmanaged room"} } + _ = logBroadcastSession(ctx, metadata.OwnerID, room.Sid, room.Name, "OPEN") return roomResponse(room), nil } } @@ -88,6 +90,7 @@ func OpenRoom(ctx context.Context, roomName string) (*Room, error) { owner := owners[0] for _, room := range result.Rooms { if metadata, ok := readBroadcast(room); ok && metadata.OwnerID == owner && metadata.DisplayName == roomName { + _ = logBroadcastSession(ctx, metadata.OwnerID, room.Sid, room.Name, "OPEN") return roomResponse(room), nil } } @@ -95,9 +98,14 @@ func OpenRoom(ctx context.Context, roomName string) (*Room, error) { 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)}) + room, err := client.CreateRoom(ctx, &livekit.CreateRoomRequest{ + Name: roomName, + Metadata: string(metadata), + EmptyTimeout: 300, + }) if err != nil { return nil, liveKitError(err) } + _ = logBroadcastSession(ctx, owner, room.Sid, room.Name, "OPEN") return roomResponse(room), nil } diff --git a/videochat/participants.go b/videochat/participants.go index 347f2fc..e13dbd9 100644 --- a/videochat/participants.go +++ b/videochat/participants.go @@ -37,10 +37,11 @@ func ListParticipants(ctx context.Context, roomName string) (*ListParticipantsRe } ctx, cancel := context.WithTimeout(ctx, apiTimeout) defer cancel() - if err := requireRoomOwner(ctx, client, roomName); err != nil { + room, err := getOwnedBroadcast(ctx, client, roomName) + if err != nil { return nil, err } - result, err := client.ListParticipants(ctx, &livekit.ListParticipantsRequest{Room: roomName}) + result, err := client.ListParticipants(ctx, &livekit.ListParticipantsRequest{Room: room.Name}) if err != nil { return nil, liveKitError(err) } @@ -107,13 +108,14 @@ func RemoveParticipant(ctx context.Context, roomName, identity string) error { } ctx, cancel := context.WithTimeout(ctx, apiTimeout) defer cancel() - if err := requireRoomOwner(ctx, client, roomName); err != nil { + room, err := getOwnedBroadcast(ctx, client, roomName) + if 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}) + _, err = client.RemoveParticipant(ctx, &livekit.RoomParticipantIdentity{Room: room.Name, Identity: identity}) return liveKitError(err) } diff --git a/videochat/rooms.go b/videochat/rooms.go index a510a9e..bab28d4 100644 --- a/videochat/rooms.go +++ b/videochat/rooms.go @@ -73,6 +73,7 @@ func CreateRoom(ctx context.Context, p *CreateRoomParams) (*Room, error) { if err != nil { return nil, liveKitError(err) } + _ = logBroadcastSession(ctx, string(uid), room.Sid, room.Name, "OPEN") return roomResponse(room), nil } @@ -95,11 +96,15 @@ func ListRooms(ctx context.Context) (*ListRoomsResponse, error) { return nil, liveKitError(err) } response := &ListRoomsResponse{Rooms: make([]*Room, 0, len(result.Rooms))} + activeSIDs := make(map[string]bool, len(result.Rooms)) for _, room := range result.Rooms { - if _, ok := readBroadcast(room); ok { + if metadata, ok := readBroadcast(room); ok { + activeSIDs[room.Sid] = true + _ = logBroadcastSession(ctx, metadata.OwnerID, room.Sid, room.Name, "OPEN") response.Rooms = append(response.Rooms, roomResponse(room)) } } + syncClosedSessions(ctx, activeSIDs) configured, err := configuredBroadcastRooms(ctx) if err != nil { return nil, err @@ -223,9 +228,22 @@ func DeleteRoom(ctx context.Context, roomName string) error { } ctx, cancel := context.WithTimeout(ctx, apiTimeout) defer cancel() - if err := requireRoomOwner(ctx, client, roomName); err != nil { + room, err := getOwnedBroadcast(ctx, client, roomName) + if err != nil { return err } - _, err = client.DeleteRoom(ctx, &livekit.DeleteRoomRequest{Room: roomName}) - return liveKitError(err) + metadata, _ := readBroadcast(room) + ownerID := metadata.OwnerID + if ownerID == "" { + uid, _ := auth.UserID() + ownerID = string(uid) + } + + _, err = client.DeleteRoom(ctx, &livekit.DeleteRoomRequest{Room: room.Name}) + if err != nil { + return liveKitError(err) + } + + _ = logBroadcastSession(ctx, ownerID, room.Sid, room.Name, "CLOSE") + return nil } diff --git a/videochat/sessions.go b/videochat/sessions.go new file mode 100644 index 0000000..e7b4748 --- /dev/null +++ b/videochat/sessions.go @@ -0,0 +1,163 @@ +package videochat + +import ( + "context" + "time" + + "encore.dev/beta/auth" + "encore.dev/beta/errs" + "encore.dev/rlog" + "encore.dev/storage/sqldb" +) + +var db = sqldb.NewDatabase("videochat", sqldb.DatabaseConfig{ + Migrations: "./migrations", +}) + +type BroadcastSession struct { + ID int64 `json:"id"` + OwnerID string `json:"owner_id"` + SID string `json:"sid"` + Name string `json:"name"` + Action string `json:"action"` // "OPEN" o "CLOSE" + CreatedAt time.Time `json:"created_at"` +} + +type ListSessionsParams struct { + OwnerID string `json:"owner_id,omitempty" query:"owner_id"` + Limit int `json:"limit,omitempty" query:"limit"` +} + +type ListSessionsResponse struct { + Sessions []*BroadcastSession `json:"sessions"` +} + +// logBroadcastSession inserisce un record di apertura o chiusura trasmissione in modo idempotente. +func logBroadcastSession(ctx context.Context, ownerID, sid, name, action string) error { + if ownerID == "" || sid == "" || name == "" { + return nil + } + + if action == "OPEN" { + var exists bool + _ = db.QueryRow(ctx, ` + SELECT EXISTS( + SELECT 1 FROM broadcast_sessions + WHERE sid = $1 AND action = 'OPEN' + ) + `, sid).Scan(&exists) + if exists { + return nil + } + } else if action == "CLOSE" { + var exists bool + _ = db.QueryRow(ctx, ` + SELECT EXISTS( + SELECT 1 FROM broadcast_sessions + WHERE sid = $1 AND action = 'CLOSE' + ) + `, sid).Scan(&exists) + if exists { + return nil + } + } + + _, err := db.Exec(ctx, ` + INSERT INTO broadcast_sessions (owner_id, sid, name, action, created_at) + VALUES ($1, $2, $3, $4, NOW()) + `, ownerID, sid, name, action) + if err != nil { + rlog.Error("failed to insert broadcast session", "err", err, "owner_id", ownerID, "sid", sid, "action", action) + return err + } + rlog.Info("logged broadcast session", "owner_id", ownerID, "sid", sid, "name", name, "action", action) + return nil +} + +// syncClosedSessions rileva le sessioni che risultavano aperte ma la cui stanza LiveKit non è più attiva, +// e registra automaticamente la chiusura (CLOSE). +func syncClosedSessions(ctx context.Context, activeSIDs map[string]bool) { + rows, err := db.Query(ctx, ` + SELECT o.owner_id, o.sid, o.name + FROM broadcast_sessions o + WHERE o.action = 'OPEN' + AND NOT EXISTS ( + SELECT 1 FROM broadcast_sessions c + WHERE c.sid = o.sid AND c.action = 'CLOSE' + ) + `) + if err != nil { + return + } + defer rows.Close() + + type toClose struct { + ownerID, sid, name string + } + var list []toClose + for rows.Next() { + var c toClose + if err := rows.Scan(&c.ownerID, &c.sid, &c.name); err == nil { + if !activeSIDs[c.sid] { + list = append(list, c) + } + } + } + for _, item := range list { + _ = logBroadcastSession(ctx, item.ownerID, item.sid, item.name, "CLOSE") + } +} + +// ListSessions restituisce lo storico delle sessioni di trasmissione (aperture/chiusure). +// +//encore:api auth method=GET path=/videochat/sessions +func ListSessions(ctx context.Context, p *ListSessionsParams) (*ListSessionsResponse, error) { + uid, ok := auth.UserID() + if !ok { + return nil, &errs.Error{Code: errs.Unauthenticated, Message: "must be logged in"} + } + + limit := 50 + if p != nil && p.Limit > 0 { + limit = p.Limit + } + if limit > 100 { + limit = 100 + } + + ownerFilter := string(uid) + // Se admin, può filtrare su un altro owner se specificato + if err := requireAdmin(); err == nil && p != nil && p.OwnerID != "" { + ownerFilter = p.OwnerID + } + + rows, err := db.Query(ctx, ` + SELECT id, owner_id, sid, name, action, created_at + FROM broadcast_sessions + WHERE ($1 = '' OR owner_id = $1) + ORDER BY created_at DESC, id DESC + LIMIT $2 + `, ownerFilter, limit) + if err != nil { + return nil, errs.WrapCode(err, errs.Internal, "failed to list broadcast sessions") + } + defer rows.Close() + + var sessions []*BroadcastSession + for rows.Next() { + var s BroadcastSession + if err := rows.Scan(&s.ID, &s.OwnerID, &s.SID, &s.Name, &s.Action, &s.CreatedAt); err != nil { + return nil, errs.WrapCode(err, errs.Internal, "failed to scan broadcast session") + } + sessions = append(sessions, &s) + } + if err := rows.Err(); err != nil { + return nil, errs.WrapCode(err, errs.Internal, "error iterating broadcast sessions") + } + + if sessions == nil { + sessions = []*BroadcastSession{} + } + + return &ListSessionsResponse{Sessions: sessions}, nil +} diff --git a/videochat/sessions_test.go b/videochat/sessions_test.go new file mode 100644 index 0000000..6dbb404 --- /dev/null +++ b/videochat/sessions_test.go @@ -0,0 +1,61 @@ +package videochat + +import ( + "context" + "testing" +) + +func TestBroadcastSessionLifecycle(t *testing.T) { + ctx := context.Background() + + // 1. Log OPEN + err := logBroadcastSession(ctx, "test-owner", "test-sid-1", "test-room", "OPEN") + if err != nil { + t.Fatalf("failed to log OPEN: %v", err) + } + + // 2. Duplicate OPEN should be ignored (idempotent) + err = logBroadcastSession(ctx, "test-owner", "test-sid-1", "test-room", "OPEN") + if err != nil { + t.Fatalf("failed on duplicate OPEN: %v", err) + } + + var openCount int + err = db.QueryRow(ctx, "SELECT count(*) FROM broadcast_sessions WHERE sid = $1 AND action = 'OPEN'", "test-sid-1").Scan(&openCount) + if err != nil || openCount != 1 { + t.Fatalf("expected exactly 1 OPEN record, got %d (err: %v)", openCount, err) + } + + // 3. Log CLOSE + err = logBroadcastSession(ctx, "test-owner", "test-sid-1", "test-room", "CLOSE") + if err != nil { + t.Fatalf("failed to log CLOSE: %v", err) + } + + // 4. Duplicate CLOSE should be ignored + err = logBroadcastSession(ctx, "test-owner", "test-sid-1", "test-room", "CLOSE") + if err != nil { + t.Fatalf("failed on duplicate CLOSE: %v", err) + } + + var closeCount int + err = db.QueryRow(ctx, "SELECT count(*) FROM broadcast_sessions WHERE sid = $1 AND action = 'CLOSE'", "test-sid-1").Scan(&closeCount) + if err != nil || closeCount != 1 { + t.Fatalf("expected exactly 1 CLOSE record, got %d (err: %v)", closeCount, err) + } + + // 5. Test syncClosedSessions + err = logBroadcastSession(ctx, "test-owner-2", "test-sid-orphan", "test-room-2", "OPEN") + if err != nil { + t.Fatalf("failed to log orphan OPEN: %v", err) + } + + // Active SIDs does not contain test-sid-orphan -> should be automatically closed + syncClosedSessions(ctx, map[string]bool{"other-sid": true}) + + var orphanCloseCount int + err = db.QueryRow(ctx, "SELECT count(*) FROM broadcast_sessions WHERE sid = $1 AND action = 'CLOSE'", "test-sid-orphan").Scan(&orphanCloseCount) + if err != nil || orphanCloseCount != 1 { + t.Fatalf("expected syncClosedSessions to create CLOSE record, got %d (err: %v)", orphanCloseCount, err) + } +} diff --git a/videochat/tokens.go b/videochat/tokens.go index 6f67279..505dfaa 100644 --- a/videochat/tokens.go +++ b/videochat/tokens.go @@ -38,10 +38,11 @@ func CreateToken(ctx context.Context, p *TokenParams) (*TokenResponse, error) { } ctx, cancel := context.WithTimeout(ctx, apiTimeout) defer cancel() - _, metadata, err := findBroadcast(ctx, client, p.RoomName) + room, metadata, err := findBroadcast(ctx, client, p.RoomName) if err != nil { return nil, err } + _ = logBroadcastSession(ctx, metadata.OwnerID, room.Sid, room.Name, "OPEN") 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")