fabio/encore-test
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add broadcast session tracking and chatlog service with database migrations

Browse Source
This commit is contained in:
fabio
2026-10-10 14:17:59 +02:00
parent 8e91572ec9
commit adc0b236a3
13 changed files with 473 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
videochat/livekit.go
View File

@@ -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) {

16
videochat/migrations/1_create_broadcast_sessions.up.sql Normal file
View File

@@ -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);

12
videochat/open_room.go
View File

@@ -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
}

10
videochat/participants.go
View File

@@ -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)
}

26
videochat/rooms.go
View File

@@ -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
}

163
videochat/sessions.go Normal file
View File

@@ -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
}

61
videochat/sessions_test.go Normal file
View File

@@ -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)
}
}

3
videochat/tokens.go
View File

@@ -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")