OwnCord/Server/ws/livekit_webhook.go

package ws

import (
	"context"
	"fmt"
	"log/slog"
	"net/http"
	"strconv"
	"strings"

	"github.com/livekit/protocol/auth"
	"github.com/livekit/protocol/livekit"
	"github.com/livekit/protocol/webhook"
)

// webhookMaxBodyBytes bounds the webhook request body to prevent unbounded
// reads from an unauthenticated caller.
const webhookMaxBodyBytes = 64 * 1024

// NewLiveKitWebhookHandler returns an HTTP handler that processes LiveKit
// webhook events. It synchronises LiveKit room state back into OwnCord's
// voice_states DB — primarily for crash recovery when a participant
// disconnects from LiveKit without sending a WS voice_leave.
//
// Speaker detection is handled client-side via LiveKit's
// RoomEvent.ActiveSpeakersChanged (lower latency than webhooks).
func (h *Hub) NewLiveKitWebhookHandler(apiKey, apiSecret string) http.HandlerFunc {
	// The SDK receiver verifies the token signature AND that the token's sha256
	// claim matches the request body hash, binding verification to the body so
	// a captured token cannot be replayed against a forged payload.
	provider := auth.NewSimpleKeyProvider(apiKey, apiSecret)

	return func(w http.ResponseWriter, r *http.Request) {
		// Check Authorization header BEFORE reading the body to avoid
		// allocating memory for unauthenticated requests.
		if r.Header.Get("Authorization") == "" {
			slog.Warn("livekit webhook: missing Authorization header")
			http.Error(w, "unauthorized", http.StatusUnauthorized)
			return
		}

		// Bound the body before the SDK reads it (ReceiveWebhookEvent uses an
		// unbounded io.ReadAll internally).
		r.Body = http.MaxBytesReader(w, r.Body, webhookMaxBodyBytes)

		// ReceiveWebhookEvent verifies the JWT signature, the token's body-hash
		// claim, and the exp/nbf claims, then parses the payload. This replaces
		// the previous manual ParseAPIToken/Verify sequence, which was not bound
		// to the request body (forgery/replay).
		event, err := webhook.ReceiveWebhookEvent(r, provider)
		if err != nil {
			slog.Warn("livekit webhook: verification failed", "error", err)
			http.Error(w, "unauthorized", http.StatusUnauthorized)
			return
		}

		slog.Info("livekit webhook received",
			"event", event.Event,
			"room", event.GetRoom().GetName(),
			"participant", event.GetParticipant().GetIdentity(),
		)

		switch event.Event {
		case "participant_joined":
			h.handleWebhookParticipantJoined(r.Context(), event)
		case "participant_left":
			h.handleWebhookParticipantLeft(r.Context(), event)
		default:
			slog.Debug("livekit webhook: unhandled event", "event", event.Event)
		}

		w.WriteHeader(http.StatusOK)
	}
}

// parseParticipantIdentity extracts a user ID and optional join token from a
// LiveKit participant identity formatted as "user-{id}" or
// "user-{id}:{joinToken}".
func parseParticipantIdentity(identity string) (int64, string, error) {
	if !strings.HasPrefix(identity, "user-") {
		return 0, "", fmt.Errorf("invalid identity format: %s", identity)
	}
	body := identity[5:]
	idPart, joinToken, _ := strings.Cut(body, ":")
	userID, err := strconv.ParseInt(idPart, 10, 64)
	if err != nil {
		return 0, "", err
	}
	return userID, joinToken, nil
}

// parseRoomChannelID extracts a channel ID from a LiveKit room name
// formatted as "channel-{id}".
func parseRoomChannelID(roomName string) (int64, error) {
	if !strings.HasPrefix(roomName, "channel-") {
		return 0, fmt.Errorf("invalid room name format: %s", roomName)
	}
	return strconv.ParseInt(roomName[8:], 10, 64)
}

func (h *Hub) handleWebhookParticipantJoined(ctx context.Context, event *livekit.WebhookEvent) {
	// Detach from the triggering HTTP request before doing any cleanup work,
	// mirroring every sibling teardown path (readPump's defer and
	// unregisterFailedHandshake use context.WithoutCancel in serve.go /
	// serve_pumps.go, rollbackVoiceJoin uses it in voice_join.go, the hub
	// sweeps use context.Background in hub_sweep.go). Without this, a webhook
	// sender (LiveKit) that hangs up mid-request cancels r.Context(), and the
	// rogue-participant GetVoiceState/RemoveParticipant calls below would
	// either wrongly skip (treating a cancelled read as a transient error, ok)
	// or fail outright instead of completing the eviction.
	ctx = context.WithoutCancel(ctx)

	p := event.GetParticipant()
	room := event.GetRoom()
	if p == nil || room == nil {
		return
	}

	userID, joinToken, err := parseParticipantIdentity(p.Identity)
	if err != nil {
		slog.Warn("livekit webhook: participant_joined bad identity",
			"identity", p.Identity, "error", err)
		return
	}

	channelID, err := parseRoomChannelID(room.Name)
	if err != nil {
		slog.Warn("livekit webhook: participant_joined bad room",
			"room", room.Name, "error", err)
		return
	}

	slog.Info("livekit webhook: participant joined",
		"user_id", userID,
		"channel_id", channelID,
		"room", room.Name)

	// Validate that the participant has a matching voice_states row (BUG-127).
	// A replayed token from a previous session will not have a matching row,
	// so we remove the rogue participant from LiveKit.
	if h.db != nil {
		h.webhookJoinedEnforceVoiceState(ctx, userID, channelID, joinToken)
	}
}

// webhookJoinedEnforceVoiceState is the voice_states reconciliation stage of
// handleWebhookParticipantJoined: it matches the joining participant against
// their DB row and removes them from the SFU when the row is missing, points
// at another channel, or carries a different join token. Callers guarantee
// h.db != nil.
func (h *Hub) webhookJoinedEnforceVoiceState(ctx context.Context, userID, channelID int64, joinToken string) {
	state, stateErr := h.db.GetVoiceState(ctx, userID)
	if stateErr != nil {
		// A transient read failure (I/O error, lock contention, a
		// maintenance window) is not proof of a rogue participant —
		// treating it as one would eject a legitimate participant from
		// the SFU on a single bad read. Mirrors sweepStaleVoiceStates'
		// hasChannelPermChecked guard: skip and let the participant be;
		// a later webhook retry or sweep tick resolves it.
		slog.Error("livekit webhook: GetVoiceState failed, skipping rogue-participant check",
			"error", stateErr, "user_id", userID, "channel_id", channelID)
		return
	}
	if state == nil || state.ChannelID != channelID {
		slog.Warn("livekit webhook: rogue participant_joined — no matching voice state, removing",
			"user_id", userID, "channel_id", channelID)
		if h.livekit != nil {
			if rmErr := h.livekit.RemoveParticipant(ctx, channelID, userID, joinToken); rmErr != nil {
				slog.Error("livekit webhook: failed to remove rogue participant",
					"error", rmErr, "user_id", userID, "channel_id", channelID)
			}
		}
		return
	}
	// Verify join token matches to prevent token replay from old sessions.
	if joinToken != "" && state.JoinedAt != joinToken {
		slog.Warn("livekit webhook: stale join token on participant_joined, removing",
			"user_id", userID, "channel_id", channelID,
			"expected_token", state.JoinedAt, "got_token", joinToken)
		if h.livekit != nil {
			if rmErr := h.livekit.RemoveParticipant(ctx, channelID, userID, joinToken); rmErr != nil {
				slog.Error("livekit webhook: failed to remove stale participant",
					"error", rmErr, "user_id", userID, "channel_id", channelID)
			}
		}
		return
	}
}

func (h *Hub) handleWebhookParticipantLeft(ctx context.Context, event *livekit.WebhookEvent) {
	// Detach from the triggering HTTP request before doing any cleanup work
	// (OC-0018), mirroring every sibling teardown path (readPump's defer and
	// unregisterFailedHandshake use context.WithoutCancel in serve.go /
	// serve_pumps.go, rollbackVoiceJoin uses it in voice_join.go, the hub
	// sweeps use context.Background in hub_sweep.go). Without this, a webhook
	// sender (LiveKit) that hangs up mid-request cancels r.Context(), which
	// makes channelReadAudience's GetChannel call fail and fail closed to an
	// empty audience (hub_broadcast.go) — silently dropping the voice_leave
	// for anyone who has READ_MESSAGES on the channel but is not currently in
	// the room. Unlike the DB row, no sweep ever re-emits that missed
	// broadcast. The same cancellation would also make both
	// LeaveVoiceChannelIfMatch branches below fail on their synchronous first
	// attempt.
	ctx = context.WithoutCancel(ctx)

	p := event.GetParticipant()
	room := event.GetRoom()
	if p == nil || room == nil {
		return
	}

	userID, joinToken, err := parseParticipantIdentity(p.Identity)
	if err != nil {
		slog.Warn("livekit webhook: participant_left bad identity",
			"identity", p.Identity, "error", err)
		return
	}

	channelID, err := parseRoomChannelID(room.Name)
	if err != nil {
		slog.Warn("livekit webhook: participant_left bad room",
			"room", room.Name, "error", err)
		return
	}

	slog.Info("livekit webhook: participant left",
		"user_id", userID,
		"channel_id", channelID)

	// Clean up voice state if the user disconnected from LiveKit
	// without sending a WS voice_leave (e.g. crash, network loss, F5 reload).
	h.mu.RLock()
	c, exists := h.clients[userID]
	h.mu.RUnlock()

	if exists {
		h.webhookLeftCleanupClient(ctx, c, userID, channelID, joinToken)
	} else if h.db != nil {
		// Client already disconnected from WS — use channel-conditional delete
		// to avoid wiping a newer row if the user reconnected and rejoined.
		deleted, dbErr := h.db.LeaveVoiceChannelIfMatch(ctx, userID, channelID, joinToken)
		if dbErr != nil {
			slog.Error("livekit webhook: LeaveVoiceChannelIfMatch failed (client gone)",
				"error", dbErr, "user_id", userID, "channel_id", channelID)
		} else if deleted {
			h.broadcastVoiceEvent(ctx, channelID, buildVoiceLeave(channelID, userID))
		}
	}
}

// webhookLeftCleanupClient is the still-connected-client stage of
// handleWebhookParticipantLeft: it compare-and-clears the client's voice
// fields for this exact join instance, then either finishes the leave or, when
// the client has already moved on, clears the stale DB row.
func (h *Hub) webhookLeftCleanupClient(ctx context.Context, c *Client, userID, channelID int64, joinToken string) {
	// Atomic compare-and-clear under c.voiceMu, replacing the previous
	// read-then-read-then-clear: two independent unlocked getVoiceState
	// snapshots followed by an unconditional clearVoiceState is not a
	// guard at all — no lock spans the second read and the clear, so a
	// voice_join committed on the readPump goroutine in between (a
	// channel switch, or a same-channel rejoin with a fresh token) is
	// wiped out from under the new session, dropping its VoiceTopic
	// subscription along with it. client.go's clearVoiceStateIfMatch
	// only compares the channel, not the token, so it would still be
	// fooled by a same-channel rejoin — this compares both, inlined here
	// via direct field access (same package as client.go) under the
	// client's own voiceMu.
	c.voiceMu.Lock()
	matched := c.voiceChID == channelID && c.voiceJoinToken != "" && c.voiceJoinToken == joinToken
	if matched {
		c.voiceChID = 0
		c.voiceJoinToken = ""
		c.e2eePubKey = ""
		c.e2eeSignature = ""
	}
	c.voiceMu.Unlock()

	if matched {
		h.webhookLeftFinishLeave(ctx, c, userID, channelID, joinToken)
	} else if h.db != nil {
		// Client has voiceChID=0 or moved to a different channel (e.g.
		// after F5 reload), or this webhook is for an older join instance.
		deleted, dbErr := h.db.LeaveVoiceChannelIfMatch(ctx, userID, channelID, joinToken)
		if dbErr != nil {
			slog.Error("livekit webhook: LeaveVoiceChannelIfMatch failed (stale DB row)",
				"error", dbErr, "user_id", userID, "channel_id", channelID)
		} else if deleted {
			h.broadcastVoiceEvent(ctx, channelID, buildVoiceLeave(channelID, userID))
			slog.Info("livekit webhook: cleaned stale DB voice row after reconnect",
				"user_id", userID, "channel_id", channelID)
		}
	}
}

// webhookLeftFinishLeave is the tear-down stage of webhookLeftCleanupClient,
// reached once the client's voice fields matched this join instance and were
// cleared: drop the voice subscription, clear the DB row, move the E2EE key
// holder on, and broadcast the leave.
func (h *Hub) webhookLeftFinishLeave(ctx context.Context, c *Client, userID, channelID int64, joinToken string) {
	h.pubsub.Unsubscribe(c, VoiceTopic(channelID))

	if h.db != nil {
		if err := leaveVoiceChannelWithRetry(ctx, h, userID, channelID, joinToken); err != nil {
			slog.Error("livekit webhook: LeaveVoiceChannel exhausted retries",
				"error", err, "user_id", userID, "channel_id", channelID)
		}
	}

	// This participant is out of voice, so the E2EE key holder may
	// need to move. Without this the map keeps naming the departed
	// user and the real lowest-uid participant's rekey offers are
	// rejected with NOT_KEY_HOLDER. Safe here: no locks are held.
	h.updateKeyHolder(channelID)

	// The leaver's own client state was just cleared above, so
	// broadcastVoiceEvent's still-in-the-room union can no longer see
	// them — without broadcastVoiceEventWithLeaver's extra term, a
	// participant without READ_MESSAGES on this channel (voice
	// membership needs only CONNECT_VOICE) never learns the server
	// already tore down their call. Mirrors finishVoiceLeave and
	// CleanupVoiceForChannel, which add the leaver for the same reason.
	h.broadcastVoiceEventWithLeaver(ctx, channelID, buildVoiceLeave(channelID, userID), userID)
	slog.Info("livekit webhook: cleaned up stale voice state",
		"user_id", userID,
		"channel_id", channelID)
}

// MountWebhookRoute is a helper for the router to mount the webhook endpoint.
func MountWebhookRoute(h *Hub, apiKey, apiSecret string) http.HandlerFunc {
	return h.NewLiveKitWebhookHandler(apiKey, apiSecret)
}