feat(voice): вебхуки LiveKit и сторож голосовых состояний
AGENT.md 7.14: - POST /api/v1/livekit/webhook: подпись проверяется JWT-токеном, подписанным API-секретом, с проверкой sha256 тела (иначе 401), при выключенном голосе — 503; обработка идемпотентна по event.id (повторная доставка → duplicate); - события: participant_joined восстанавливает состояние после рестарта приложения, participant_left и participant_connection_aborted убирают участника, room_finished очищает комнату целиком, track_published и track_unpublished обновляют флаги камеры и демонстрации экрана; каждое изменение рассылает VOICE_STATE_UPDATE участникам сервера; - сторож `StartVoiceWatchdog` раз в 15 минут удаляет состояния без обновлений дольше шести часов (сбои без вебхука) и уведомляет клиентов; - тесты: подпись (чужой секрет, другое тело, отсутствие заголовка), полный цикл событий SFU и идемпотентность повторной доставки.
This commit is contained in:
@@ -0,0 +1,241 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/go-chi/chi/v5"
|
||||
|
||||
"glchat/internal/store"
|
||||
"glchat/internal/voice"
|
||||
)
|
||||
|
||||
// livekitWebhookEvent — событие вебхука LiveKit (AGENT.md 7.14).
|
||||
type livekitWebhookEvent struct {
|
||||
Event string `json:"event"`
|
||||
ID string `json:"id"`
|
||||
Room struct {
|
||||
Name string `json:"name"`
|
||||
SID string `json:"sid"`
|
||||
} `json:"room"`
|
||||
Participant struct {
|
||||
Identity string `json:"identity"`
|
||||
SID string `json:"sid"`
|
||||
} `json:"participant"`
|
||||
Track struct {
|
||||
Type string `json:"type"`
|
||||
Source string `json:"source"`
|
||||
} `json:"track"`
|
||||
}
|
||||
|
||||
// webhookTTL — сколько помним идентификаторы обработанных событий.
|
||||
const webhookTTL = time.Hour
|
||||
|
||||
// registerVoiceWebhook вешает обработчик вебхуков LiveKit (AGENT.md 7.14).
|
||||
func (s *Server) registerVoiceWebhook(router chi.Router) {
|
||||
router.Post("/livekit/webhook", s.handleLiveKitWebhook)
|
||||
}
|
||||
|
||||
// handleLiveKitWebhook принимает события SFU: подпись проверяется, обработка
|
||||
// идемпотентна по идентификатору события, состояния синхронизируются с БД.
|
||||
func (s *Server) handleLiveKitWebhook(w http.ResponseWriter, r *http.Request) {
|
||||
if !s.voice.Enabled() {
|
||||
httpxWriteJSONError(w, http.StatusServiceUnavailable, "voice.disabled", "голос не настроен")
|
||||
return
|
||||
}
|
||||
body, err := io.ReadAll(io.LimitReader(r.Body, 1<<20))
|
||||
if err != nil {
|
||||
httpxWriteJSONError(w, http.StatusBadRequest, "request.bad", "cannot read body")
|
||||
return
|
||||
}
|
||||
if err := s.voice.VerifyWebhook(r.Header.Get("Authorization"), body); err != nil {
|
||||
s.logger.WarnContext(r.Context(), "livekit webhook rejected", slog.Any("error", err))
|
||||
httpxWriteJSONError(w, http.StatusUnauthorized, "voice.webhook_unauthorized", "invalid webhook signature")
|
||||
return
|
||||
}
|
||||
|
||||
var event livekitWebhookEvent
|
||||
if err := json.Unmarshal(body, &event); err != nil {
|
||||
httpxWriteJSONError(w, http.StatusBadRequest, "request.bad", "malformed event")
|
||||
return
|
||||
}
|
||||
if event.ID != "" && !s.markWebhookProcessed(event.ID) {
|
||||
// Повторная доставка: отвечаем успехом, ничего не меняя.
|
||||
httpxWriteJSON(w, http.StatusOK, map[string]any{"ok": true, "duplicate": true})
|
||||
return
|
||||
}
|
||||
s.applyLiveKitEvent(r.Context(), event)
|
||||
httpxWriteJSON(w, http.StatusOK, map[string]any{"ok": true})
|
||||
}
|
||||
|
||||
// markWebhookProcessed возвращает false, если событие уже обрабатывалось.
|
||||
func (s *Server) markWebhookProcessed(id string) bool {
|
||||
now := time.Now()
|
||||
s.webhookMu.Lock()
|
||||
defer s.webhookMu.Unlock()
|
||||
if len(s.webhookSeen) > 4096 {
|
||||
cutoff := now.Add(-webhookTTL)
|
||||
for key, at := range s.webhookSeen {
|
||||
if at.Before(cutoff) {
|
||||
delete(s.webhookSeen, key)
|
||||
}
|
||||
}
|
||||
}
|
||||
if _, ok := s.webhookSeen[id]; ok {
|
||||
return false
|
||||
}
|
||||
s.webhookSeen[id] = now
|
||||
return true
|
||||
}
|
||||
|
||||
// applyLiveKitEvent обновляет голосовые состояния по событию SFU.
|
||||
func (s *Server) applyLiveKitEvent(ctx context.Context, event livekitWebhookEvent) {
|
||||
guildID, channelID, ok := voice.ParseRoomName(event.Room.Name)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
switch event.Event {
|
||||
case "participant_joined":
|
||||
userID := parseVoiceIdentity(event.Participant.Identity)
|
||||
if userID == 0 {
|
||||
return
|
||||
}
|
||||
// Восстанавливаем состояние, если приложение перезапускалось: комната
|
||||
// уже существует, значит участник действительно в голосовой комнате.
|
||||
current, err := s.store.GetVoiceState(ctx, guildID, userID)
|
||||
if err == nil && current.ChannelID == channelID {
|
||||
return
|
||||
}
|
||||
params := store.VoiceStateParams{UserID: userID, GuildID: guildID, ChannelID: channelID}
|
||||
if err == nil {
|
||||
params.SelfMute = current.SelfMute
|
||||
params.SelfDeaf = current.SelfDeaf
|
||||
params.ServerMute = current.ServerMute
|
||||
params.ServerDeaf = current.ServerDeaf
|
||||
}
|
||||
state, err := s.store.UpsertVoiceState(ctx, params)
|
||||
if err != nil {
|
||||
s.logger.WarnContext(ctx, "livekit webhook: upsert state failed", slog.Any("error", err))
|
||||
return
|
||||
}
|
||||
s.dispatchVoiceState(ctx, state)
|
||||
|
||||
case "participant_left", "participant_connection_aborted":
|
||||
userID := parseVoiceIdentity(event.Participant.Identity)
|
||||
if userID == 0 {
|
||||
return
|
||||
}
|
||||
if err := s.store.DeleteVoiceState(ctx, guildID, userID); err != nil {
|
||||
return
|
||||
}
|
||||
s.dispatchVoiceLeave(ctx, guildID, channelID, userID)
|
||||
|
||||
case "room_finished":
|
||||
states, err := s.store.ListVoiceStates(ctx, guildID)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
for _, state := range states {
|
||||
if state.ChannelID != channelID {
|
||||
continue
|
||||
}
|
||||
if err := s.store.DeleteVoiceState(ctx, guildID, state.UserID); err != nil {
|
||||
continue
|
||||
}
|
||||
s.dispatchVoiceLeave(ctx, guildID, channelID, state.UserID)
|
||||
}
|
||||
|
||||
case "track_published", "track_unpublished":
|
||||
userID := parseVoiceIdentity(event.Participant.Identity)
|
||||
if userID == 0 {
|
||||
return
|
||||
}
|
||||
state, err := s.store.GetVoiceState(ctx, guildID, userID)
|
||||
if err != nil || state.ChannelID != channelID {
|
||||
return
|
||||
}
|
||||
published := event.Event == "track_published"
|
||||
params := store.VoiceStateParams{
|
||||
UserID: userID, GuildID: guildID, ChannelID: channelID, SessionID: state.SessionID,
|
||||
SelfMute: state.SelfMute, SelfDeaf: state.SelfDeaf,
|
||||
ServerMute: state.ServerMute, ServerDeaf: state.ServerDeaf,
|
||||
Camera: state.Camera, Screen: state.Screen,
|
||||
}
|
||||
switch strings.ToUpper(event.Track.Source) {
|
||||
case "CAMERA":
|
||||
params.Camera = published
|
||||
case "SCREEN_SHARE", "SCREEN_SHARE_AUDIO":
|
||||
params.Screen = published
|
||||
default:
|
||||
return
|
||||
}
|
||||
updated, err := s.store.UpsertVoiceState(ctx, params)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
s.dispatchVoiceState(ctx, updated)
|
||||
}
|
||||
}
|
||||
|
||||
// parseVoiceIdentity превращает identity участника LiveKit в идентификатор.
|
||||
func parseVoiceIdentity(identity string) uint64 {
|
||||
id, err := parseID("identity", identity)
|
||||
if err != nil {
|
||||
return 0
|
||||
}
|
||||
return id
|
||||
}
|
||||
|
||||
// startVoiceWatchdog периодически убирает «призрачные» состояния: записи без
|
||||
// обновлений дольше шести часов (сервер мог упасть и не получить вебхук).
|
||||
func (s *Server) startVoiceWatchdog(ctx context.Context, interval time.Duration) {
|
||||
if interval <= 0 {
|
||||
interval = 15 * time.Minute
|
||||
}
|
||||
go func() {
|
||||
ticker := time.NewTicker(interval)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
s.cleanupStaleVoiceStates(ctx)
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// cleanupStaleVoiceStates удаляет давно не обновлявшиеся голосовые состояния.
|
||||
func (s *Server) cleanupStaleVoiceStates(ctx context.Context) {
|
||||
cutoff := time.Now().UTC().Add(-6 * time.Hour)
|
||||
guilds, err := s.store.ListAllGuilds(ctx)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
removed := 0
|
||||
for _, guild := range guilds {
|
||||
states, err := s.store.ListVoiceStates(ctx, guild.ID)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
for _, state := range states {
|
||||
if state.UpdatedAt.After(cutoff) {
|
||||
continue
|
||||
}
|
||||
if err := s.store.DeleteVoiceState(ctx, guild.ID, state.UserID); err != nil {
|
||||
continue
|
||||
}
|
||||
removed++
|
||||
s.dispatchVoiceLeave(ctx, guild.ID, state.ChannelID, state.UserID)
|
||||
}
|
||||
}
|
||||
if removed > 0 {
|
||||
s.logger.InfoContext(ctx, "stale voice states cleaned", slog.Int("removed", removed))
|
||||
}
|
||||
}
|
||||
@@ -1053,3 +1053,105 @@ func TestVoiceStateFlow(t *testing.T) {
|
||||
t.Fatalf("после выхода = %+v", empty.States)
|
||||
}
|
||||
}
|
||||
|
||||
// TestLiveKitWebhook проверяет подпись, идемпотентность и синхронизацию состояний.
|
||||
func TestLiveKitWebhook(t *testing.T) {
|
||||
f := newMessagingFixture(t)
|
||||
f.srv.cfg.LiveKitAPIKey = "testkey"
|
||||
f.srv.cfg.LiveKitAPISecret = "testsecret"
|
||||
f.srv.cfg.LiveKitURL = "wss://gl.test/rtc"
|
||||
f.srv.voice = voice.NewIssuer("testkey", "testsecret", time.Hour)
|
||||
|
||||
voiceRec := doJSON(t, f.srv, http.MethodPost, "/api/v1/guilds/"+f.guildID+"/channels",
|
||||
`{"name":"Голос","type":"voice"}`, f.ownerCookie)
|
||||
voiceChannel := decodeResponse[struct {
|
||||
Channel struct {
|
||||
ID string `json:"id"`
|
||||
} `json:"channel"`
|
||||
}](t, voiceRec)
|
||||
channelID := guildIDOf(t, voiceChannel.Channel.ID)
|
||||
guildID := guildIDOf(t, f.guildID)
|
||||
room := voice.RoomName(guildID, channelID)
|
||||
|
||||
// Событие без подписи отклоняется.
|
||||
noAuth := doJSON(t, f.srv, http.MethodPost, "/api/v1/livekit/webhook",
|
||||
`{"event":"participant_joined"}`, f.ownerCookie)
|
||||
if noAuth.Code != http.StatusUnauthorized {
|
||||
t.Fatalf("webhook без подписи = %d, want 401", noAuth.Code)
|
||||
}
|
||||
|
||||
post := func(body string) *httptest.ResponseRecorder {
|
||||
t.Helper()
|
||||
signature, err := f.srv.voice.IssueWebhookForTest([]byte(body))
|
||||
if err != nil {
|
||||
t.Fatalf("подпись вебхука: %v", err)
|
||||
}
|
||||
req := httptest.NewRequestWithContext(t.Context(), http.MethodPost, "/api/v1/livekit/webhook",
|
||||
strings.NewReader(body))
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
req.Header.Set("Authorization", signature)
|
||||
rec := httptest.NewRecorder()
|
||||
f.srv.Handler().ServeHTTP(rec, req)
|
||||
return rec
|
||||
}
|
||||
|
||||
// Участник «появился» на стороне SFU — состояние восстанавливается.
|
||||
joined := post(`{"event":"participant_joined","id":"evt-1","room":{"name":"` + room + `"},` +
|
||||
`"participant":{"identity":"` + f.memberID + `"}}`)
|
||||
if joined.Code != http.StatusOK {
|
||||
t.Fatalf("participant_joined = %d, body = %s", joined.Code, joined.Body.String())
|
||||
}
|
||||
states := doJSON(t, f.srv, http.MethodGet, "/api/v1/guilds/"+f.guildID+"/voice-states", "", f.ownerCookie)
|
||||
list := decodeResponse[struct {
|
||||
States []struct {
|
||||
UserID string `json:"user_id"`
|
||||
ChannelID string `json:"channel_id"`
|
||||
} `json:"voice_states"`
|
||||
}](t, states)
|
||||
if len(list.States) != 1 || list.States[0].UserID != f.memberID {
|
||||
t.Fatalf("состояния после вебхука = %+v", list.States)
|
||||
}
|
||||
|
||||
// Повторная доставка того же события ничего не меняет.
|
||||
duplicate := post(`{"event":"participant_joined","id":"evt-1","room":{"name":"` + room + `"},` +
|
||||
`"participant":{"identity":"` + f.memberID + `"}}`)
|
||||
var duplicateBody struct {
|
||||
Duplicate bool `json:"duplicate"`
|
||||
}
|
||||
if err := json.Unmarshal(duplicate.Body.Bytes(), &duplicateBody); err != nil {
|
||||
t.Fatalf("decode duplicate: %v", err)
|
||||
}
|
||||
if !duplicateBody.Duplicate {
|
||||
t.Fatalf("повторное событие должно помечаться duplicate: %s", duplicate.Body.String())
|
||||
}
|
||||
|
||||
// Публикация камеры отмечает флаг.
|
||||
camera := post(`{"event":"track_published","id":"evt-2","room":{"name":"` + room + `"},` +
|
||||
`"participant":{"identity":"` + f.memberID + `"},"track":{"source":"CAMERA"}}`)
|
||||
if camera.Code != http.StatusOK {
|
||||
t.Fatalf("track_published = %d, body = %s", camera.Code, camera.Body.String())
|
||||
}
|
||||
cameraState, err := f.srv.store.GetVoiceState(t.Context(), guildID, guildIDOf(t, f.memberID))
|
||||
if err != nil {
|
||||
t.Fatalf("GetVoiceState: %v", err)
|
||||
}
|
||||
if !cameraState.Camera {
|
||||
t.Fatal("флаг камеры не выставлен по вебхуку")
|
||||
}
|
||||
|
||||
// Уход участника убирает состояние.
|
||||
left := post(`{"event":"participant_left","id":"evt-3","room":{"name":"` + room + `"},` +
|
||||
`"participant":{"identity":"` + f.memberID + `"}}`)
|
||||
if left.Code != http.StatusOK {
|
||||
t.Fatalf("participant_left = %d", left.Code)
|
||||
}
|
||||
after := doJSON(t, f.srv, http.MethodGet, "/api/v1/guilds/"+f.guildID+"/voice-states", "", f.ownerCookie)
|
||||
empty := decodeResponse[struct {
|
||||
States []struct {
|
||||
UserID string `json:"user_id"`
|
||||
} `json:"voice_states"`
|
||||
}](t, after)
|
||||
if len(empty.States) != 0 {
|
||||
t.Fatalf("после participant_left = %+v", empty.States)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -65,10 +65,13 @@ type Server struct {
|
||||
// presence — время последнего обновления last_seen по пользователю.
|
||||
presenceMu sync.Mutex
|
||||
presence map[uint64]time.Time
|
||||
logger *slog.Logger
|
||||
http *http.Server
|
||||
static *staticHandler
|
||||
api huma.API
|
||||
// webhookSeen — идентификаторы обработанных вебхуков LiveKit (идемпотентность).
|
||||
webhookMu sync.Mutex
|
||||
webhookSeen map[string]time.Time
|
||||
logger *slog.Logger
|
||||
http *http.Server
|
||||
static *staticHandler
|
||||
api huma.API
|
||||
}
|
||||
|
||||
func New(cfg config.Config, db *database.DB, logger *slog.Logger, deps Deps) *Server {
|
||||
@@ -92,6 +95,7 @@ func New(cfg config.Config, db *database.DB, logger *slog.Logger, deps Deps) *Se
|
||||
inviteLimiter: httpx.NewRateLimiterWindow(10, 24*time.Hour, 10),
|
||||
slowmode: map[string]time.Time{},
|
||||
presence: map[uint64]time.Time{},
|
||||
webhookSeen: map[string]time.Time{},
|
||||
voice: voice.NewIssuer(cfg.LiveKitAPIKey, cfg.LiveKitAPISecret, time.Hour),
|
||||
}
|
||||
switch {
|
||||
@@ -119,6 +123,7 @@ func New(cfg config.Config, db *database.DB, logger *slog.Logger, deps Deps) *Se
|
||||
s.registerSocialRoutes(s.api)
|
||||
s.registerEmojiRoutes(s.api, apiRouter)
|
||||
s.registerVoiceRoutes(s.api)
|
||||
s.registerVoiceWebhook(apiRouter)
|
||||
}
|
||||
apiRouter.Get("/openapi.json", s.handleOpenAPI)
|
||||
})
|
||||
@@ -187,6 +192,11 @@ func (s *Server) ListenAndServe() error { return s.http.ListenAndServe() }
|
||||
|
||||
func (s *Server) Shutdown(ctx context.Context) error { return s.http.Shutdown(ctx) }
|
||||
|
||||
// StartVoiceWatchdog запускает периодическую очистку голосовых состояний.
|
||||
func (s *Server) StartVoiceWatchdog(ctx context.Context, interval time.Duration) {
|
||||
s.startVoiceWatchdog(ctx, interval)
|
||||
}
|
||||
|
||||
func (s *Server) handleHealthz(w http.ResponseWriter, _ *http.Request) {
|
||||
httpx.WriteJSON(w, http.StatusOK, map[string]any{
|
||||
"status": "ok",
|
||||
|
||||
Reference in New Issue
Block a user