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.
This commit is contained in:
talksik
+100
-20
@@ -4,10 +4,13 @@ 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"
|
||||
@@ -17,27 +20,30 @@ import (
|
||||
"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
|
||||
livekitClient livekit.Client
|
||||
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, livekitClient livekit.Client) *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,
|
||||
livekitClient: livekitClient,
|
||||
authSvc: authSvc,
|
||||
humanSvc: humanSvc,
|
||||
networkSvc: networkSvc,
|
||||
particleSvc: particleSvc,
|
||||
depotSvc: depotSvc,
|
||||
waitlistSvc: waitlistSvc,
|
||||
livekitClient: livekitClient,
|
||||
firestoreClient: firestoreClient,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -115,7 +121,8 @@ type RevokeInvitationRequest struct {
|
||||
// LiveKit DTOs
|
||||
|
||||
type GetLivekitTokenRequest struct {
|
||||
RoomId string `json:"room_id"`
|
||||
NetworkId string `json:"network_id"`
|
||||
StreamId string `json:"stream_id"`
|
||||
}
|
||||
|
||||
type GetLivekitTokenResponse struct {
|
||||
@@ -1076,14 +1083,21 @@ func (h *Handler) GetLivekitToken(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
|
||||
if req.RoomId == "" {
|
||||
http.Error(w, "room_id is required", http.StatusBadRequest)
|
||||
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
|
||||
}
|
||||
|
||||
token, err := h.livekitClient.GetJoinToken(req.RoomId, humanId, humanEmail)
|
||||
// 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, "roomId", req.RoomId)
|
||||
slog.Error("failed to generate livekit token", "error", err, "humanId", humanId, "roomName", roomName)
|
||||
http.Error(w, "failed to generate token", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
@@ -1092,6 +1106,72 @@ func (h *Handler) GetLivekitToken(w http.ResponseWriter, r *http.Request) {
|
||||
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 == "" {
|
||||
|
||||
@@ -1,29 +1,47 @@
|
||||
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 {
|
||||
// Name will show up in the participant data
|
||||
// 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
|
||||
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: utils.MustGetEnv("LIVEKIT_API_SECRET"),
|
||||
apiKey: utils.MustGetEnv("LIVEKIT_API_KEY"),
|
||||
hostUrl: utils.MustGetEnv("LIVEKIT_URL"),
|
||||
apiSecret: apiSecret,
|
||||
apiKey: apiKey,
|
||||
hostUrl: hostUrl,
|
||||
roomService: roomService,
|
||||
keyProvider: auth.NewSimpleKeyProvider(apiKey, apiSecret),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -44,3 +62,17 @@ func (c *clientImpl) GetJoinToken(room, humanId, name string) (string, error) {
|
||||
|
||||
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