support huddles (#106)
* add token endpoint for livekit * fix: invalid type passed to hook * fix: inject livekit env variables for orion * fix: show controls indicator above stream # shortcut * return livekit server url from api * simple huddle implementation with streams * set human name in livekit room context * simplify deployment tooling * join huddle with audio automatically * feat: show when there is an active huddle This introduces a webhook which listens to events from livekit and updates our firestore stream particle. It keeps the client simple, reacting to changes to firestore docs. * use headphones icon for huddles * fix: screenshare not working in electron The default VideoConference component from livekit doesn't support screenshare in electron. This attempts to compose our own layout with livekit components ourselves and introduces our own flow for screenshare.
This commit was merged in pull request #106.
This commit is contained in:
Arjun Patel
+145
-13
@@ -4,37 +4,46 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"cloud.google.com/go/firestore"
|
||||
"github.com/flowy-live/llink/internal/auth"
|
||||
"github.com/flowy-live/llink/internal/depot"
|
||||
"github.com/flowy-live/llink/internal/human"
|
||||
"github.com/flowy-live/llink/internal/livekit"
|
||||
"github.com/flowy-live/llink/internal/middleware"
|
||||
"github.com/flowy-live/llink/internal/network"
|
||||
"github.com/flowy-live/llink/internal/particle"
|
||||
"github.com/flowy-live/llink/internal/utils"
|
||||
"github.com/flowy-live/llink/internal/waitlist"
|
||||
"github.com/livekit/protocol/webhook"
|
||||
)
|
||||
|
||||
type Handler struct {
|
||||
authSvc auth.AuthService
|
||||
humanSvc human.Service
|
||||
networkSvc network.Service
|
||||
particleSvc particle.Service
|
||||
depotSvc depot.Service
|
||||
waitlistSvc waitlist.Service
|
||||
authSvc auth.AuthService
|
||||
humanSvc human.Service
|
||||
networkSvc network.Service
|
||||
particleSvc particle.Service
|
||||
depotSvc depot.Service
|
||||
waitlistSvc waitlist.Service
|
||||
livekitClient livekit.Client
|
||||
firestoreClient *firestore.Client
|
||||
}
|
||||
|
||||
func NewHandler(authSvc auth.AuthService, humanSvc human.Service, networkSvc network.Service, particleSvc particle.Service, depotSvc depot.Service, waitlistSvc waitlist.Service) *Handler {
|
||||
func NewHandler(authSvc auth.AuthService, humanSvc human.Service, networkSvc network.Service, particleSvc particle.Service, depotSvc depot.Service, waitlistSvc waitlist.Service, livekitClient livekit.Client, firestoreClient *firestore.Client) *Handler {
|
||||
return &Handler{
|
||||
authSvc: authSvc,
|
||||
humanSvc: humanSvc,
|
||||
networkSvc: networkSvc,
|
||||
particleSvc: particleSvc,
|
||||
depotSvc: depotSvc,
|
||||
waitlistSvc: waitlistSvc,
|
||||
authSvc: authSvc,
|
||||
humanSvc: humanSvc,
|
||||
networkSvc: networkSvc,
|
||||
particleSvc: particleSvc,
|
||||
depotSvc: depotSvc,
|
||||
waitlistSvc: waitlistSvc,
|
||||
livekitClient: livekitClient,
|
||||
firestoreClient: firestoreClient,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -109,6 +118,18 @@ type RevokeInvitationRequest struct {
|
||||
Email string `json:"email"`
|
||||
}
|
||||
|
||||
// LiveKit DTOs
|
||||
|
||||
type GetLivekitTokenRequest struct {
|
||||
NetworkId string `json:"network_id"`
|
||||
StreamId string `json:"stream_id"`
|
||||
}
|
||||
|
||||
type GetLivekitTokenResponse struct {
|
||||
Token string `json:"token"`
|
||||
ServerUrl string `json:"server_url"`
|
||||
}
|
||||
|
||||
// Depot DTOs
|
||||
|
||||
type PrepareUploadRequest struct {
|
||||
@@ -1040,6 +1061,117 @@ func (h *Handler) networkToDTO(ctx context.Context, n *network.Network) (Network
|
||||
}, nil
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// LiveKit Handlers
|
||||
// ============================================================================
|
||||
|
||||
func (h *Handler) GetLivekitToken(w http.ResponseWriter, r *http.Request) {
|
||||
humanId, ok := middleware.HumanIdFromContext(r.Context())
|
||||
if !ok {
|
||||
http.Error(w, "unauthorized", http.StatusUnauthorized)
|
||||
return
|
||||
}
|
||||
humanEmail, ok := middleware.EmailFromContext(r.Context())
|
||||
if !ok {
|
||||
http.Error(w, "unauthorized", http.StatusUnauthorized)
|
||||
return
|
||||
}
|
||||
|
||||
var req GetLivekitTokenRequest
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
http.Error(w, "invalid request body", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
if req.NetworkId == "" {
|
||||
http.Error(w, "network_id is required", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
if req.StreamId == "" {
|
||||
http.Error(w, "stream_id is required", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
// Compose room name encoding both network and stream IDs for webhook resolution
|
||||
roomName := req.NetworkId + "/" + req.StreamId
|
||||
|
||||
token, err := h.livekitClient.GetJoinToken(roomName, humanId, humanEmail)
|
||||
if err != nil {
|
||||
slog.Error("failed to generate livekit token", "error", err, "humanId", humanId, "roomName", roomName)
|
||||
http.Error(w, "failed to generate token", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(GetLivekitTokenResponse{Token: token, ServerUrl: h.livekitClient.ServerUrl()})
|
||||
}
|
||||
|
||||
// HandleLivekitWebhook processes LiveKit webhook events for huddle presence.
|
||||
// It verifies the webhook signature (not user auth), then updates the stream
|
||||
// particle's huddle_active_participants field in Firestore.
|
||||
func (h *Handler) HandleLivekitWebhook(w http.ResponseWriter, r *http.Request) {
|
||||
event, err := webhook.ReceiveWebhookEvent(r, h.livekitClient.KeyProvider())
|
||||
if err != nil {
|
||||
slog.Error("failed to verify livekit webhook", "error", err)
|
||||
http.Error(w, "unauthorized", http.StatusUnauthorized)
|
||||
return
|
||||
}
|
||||
|
||||
eventType := event.GetEvent()
|
||||
slog.Info("received livekit webhook", "event", eventType, "room", event.GetRoom().GetName())
|
||||
|
||||
switch eventType {
|
||||
case "participant_joined", "participant_left", "room_finished":
|
||||
// Handle these events
|
||||
default:
|
||||
w.WriteHeader(http.StatusOK)
|
||||
return
|
||||
}
|
||||
|
||||
// Parse room name to extract networkId and streamId
|
||||
roomName := event.GetRoom().GetName()
|
||||
parts := strings.SplitN(roomName, "/", 2)
|
||||
if len(parts) != 2 {
|
||||
slog.Error("invalid room name format", "room", roomName)
|
||||
http.Error(w, "invalid room name", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
networkId, streamId := parts[0], parts[1]
|
||||
docPath := fmt.Sprintf("networks/%s/children/%s", networkId, streamId)
|
||||
docRef := h.firestoreClient.Doc(docPath)
|
||||
ctx := r.Context()
|
||||
|
||||
var participantIds []string
|
||||
|
||||
if eventType == "room_finished" {
|
||||
// Room is done — clear the participants
|
||||
participantIds = []string{}
|
||||
} else {
|
||||
// Use ListParticipants for authoritative state (avoids drift from missed webhooks)
|
||||
participants, err := h.livekitClient.ListParticipants(ctx, roomName)
|
||||
if err != nil {
|
||||
slog.Error("failed to list participants", "error", err, "room", roomName)
|
||||
// Return 200 so LiveKit doesn't retry
|
||||
w.WriteHeader(http.StatusOK)
|
||||
return
|
||||
}
|
||||
|
||||
participantIds = make([]string, 0, len(participants))
|
||||
for _, p := range participants {
|
||||
participantIds = append(participantIds, p.Identity)
|
||||
}
|
||||
}
|
||||
|
||||
_, err = docRef.Update(ctx, []firestore.Update{
|
||||
{Path: "huddle_active_participants", Value: participantIds},
|
||||
})
|
||||
if err != nil {
|
||||
slog.Error("failed to update huddle participants in firestore", "error", err, "path", docPath)
|
||||
}
|
||||
|
||||
w.WriteHeader(http.StatusOK)
|
||||
}
|
||||
|
||||
func extractBearerToken(r *http.Request) string {
|
||||
authHeader := r.Header.Get("Authorization")
|
||||
if authHeader == "" {
|
||||
|
||||
@@ -0,0 +1,78 @@
|
||||
package livekit
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"github.com/flowy-live/llink/internal/utils"
|
||||
"github.com/livekit/protocol/auth"
|
||||
"github.com/livekit/protocol/livekit"
|
||||
lksdk "github.com/livekit/server-sdk-go/v2"
|
||||
)
|
||||
|
||||
type Client interface {
|
||||
// GetJoinToken generates a JWT for a participant to join a room.
|
||||
// Name will show up in the participant data.
|
||||
GetJoinToken(roomId string, humanId string, name string) (string, error)
|
||||
ServerUrl() string
|
||||
// ListParticipants returns the current participants in a room.
|
||||
ListParticipants(ctx context.Context, roomName string) ([]*livekit.ParticipantInfo, error)
|
||||
// KeyProvider returns the key provider for verifying webhook signatures.
|
||||
KeyProvider() auth.KeyProvider
|
||||
}
|
||||
|
||||
type clientImpl struct {
|
||||
apiSecret string
|
||||
apiKey string
|
||||
hostUrl string
|
||||
roomService *lksdk.RoomServiceClient
|
||||
keyProvider auth.KeyProvider
|
||||
}
|
||||
|
||||
func NewClient() Client {
|
||||
apiKey := utils.MustGetEnv("LIVEKIT_API_KEY")
|
||||
apiSecret := utils.MustGetEnv("LIVEKIT_API_SECRET")
|
||||
hostUrl := utils.MustGetEnv("LIVEKIT_URL")
|
||||
|
||||
roomService := lksdk.NewRoomServiceClient(hostUrl, apiKey, apiSecret)
|
||||
|
||||
return &clientImpl{
|
||||
apiSecret: apiSecret,
|
||||
apiKey: apiKey,
|
||||
hostUrl: hostUrl,
|
||||
roomService: roomService,
|
||||
keyProvider: auth.NewSimpleKeyProvider(apiKey, apiSecret),
|
||||
}
|
||||
}
|
||||
|
||||
func (c *clientImpl) ServerUrl() string {
|
||||
return c.hostUrl
|
||||
}
|
||||
|
||||
func (c *clientImpl) GetJoinToken(room, humanId, name string) (string, error) {
|
||||
at := auth.NewAccessToken(c.apiKey, c.apiSecret)
|
||||
grant := &auth.VideoGrant{
|
||||
RoomJoin: true,
|
||||
Room: room,
|
||||
}
|
||||
at.SetVideoGrant(grant).
|
||||
SetIdentity(humanId).
|
||||
SetName(name).
|
||||
SetValidFor(time.Hour)
|
||||
|
||||
return at.ToJWT()
|
||||
}
|
||||
|
||||
func (c *clientImpl) ListParticipants(ctx context.Context, roomName string) ([]*livekit.ParticipantInfo, error) {
|
||||
resp, err := c.roomService.ListParticipants(ctx, &livekit.ListParticipantsRequest{
|
||||
Room: roomName,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return resp.Participants, nil
|
||||
}
|
||||
|
||||
func (c *clientImpl) KeyProvider() auth.KeyProvider {
|
||||
return c.keyProvider
|
||||
}
|
||||
@@ -46,8 +46,9 @@ type FirestoreStreamParticle struct {
|
||||
CreatedByHumanId string `firestore:"created_by_human_id"`
|
||||
Type string `firestore:"type"`
|
||||
// Properties FirestoreStreamParticleProperties `firestore:"properties"`
|
||||
CreatedAt time.Time `firestore:"created_at,serverTimestamp"`
|
||||
LastChildCreatedAt *time.Time `firestore:"last_child_created_at,omitempty"`
|
||||
VisibleTo []string `firestore:"visible_to"`
|
||||
UpdatedAt *time.Time `firestore:"updated_at,omitempty"`
|
||||
CreatedAt time.Time `firestore:"created_at,serverTimestamp"`
|
||||
LastChildCreatedAt *time.Time `firestore:"last_child_created_at,omitempty"`
|
||||
VisibleTo []string `firestore:"visible_to"`
|
||||
UpdatedAt *time.Time `firestore:"updated_at,omitempty"`
|
||||
HuddleActiveParticipants []string `firestore:"huddle_active_participants,omitempty"`
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user