live-sfu-demo/internal/server/gateway.go

package server

import (
	"encoding/json"
	"fmt"
	"net/http"
	"time"

	"google.golang.org/protobuf/encoding/protojson"
	"google.golang.org/protobuf/proto"
	"sync-live/gen"
	"sync-live/internal/room"
)

func (s *Server) writeProto(w http.ResponseWriter, m proto.Message, err error) {
	if err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	w.Header().Set("Content-Type", "application/json")
	data, merr := protojson.Marshal(m)
	if merr != nil {
		http.Error(w, merr.Error(), http.StatusInternalServerError)
		return
	}
	w.Write(data)
}

func (s *Server) handleConfig(w http.ResponseWriter, r *http.Request) {
	resp, err := s.svc.GetConfig(r.Context(), &gen.GetConfigRequest{})
	s.writeProto(w, resp, err)
}

func (s *Server) handleRooms(w http.ResponseWriter, r *http.Request) {
	resp, err := s.svc.ListRooms(r.Context(), &gen.ListRoomsRequest{})
	s.writeProto(w, resp, err)
}

func (s *Server) handlePublish(w http.ResponseWriter, r *http.Request) {
	var req gen.PublishRequest
	if err := protojson.Unmarshal(readBody(r), &req); err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	resp, err := s.svc.Publish(r.Context(), &req)
	s.writeProto(w, resp, err)
}

func (s *Server) handleSubscribe(w http.ResponseWriter, r *http.Request) {
	var req gen.SubscribeRequest
	if err := protojson.Unmarshal(readBody(r), &req); err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	resp, err := s.svc.Subscribe(r.Context(), &req)
	s.writeProto(w, resp, err)
}

func (s *Server) handleStop(w http.ResponseWriter, r *http.Request) {
	var req gen.StopStreamRequest
	if err := protojson.Unmarshal(readBody(r), &req); err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	resp, err := s.svc.StopStream(r.Context(), &req)
	s.writeProto(w, resp, err)
}

// handleRoomEvents 以 SSE 推送房间分发拓扑（分流状态），浏览器据此自动拉流。
func (s *Server) handleRoomEvents(w http.ResponseWriter, r *http.Request) {
	room := r.PathValue("room")
	flusher, ok := w.(http.Flusher)
	if !ok {
		http.Error(w, "streaming unsupported", http.StatusInternalServerError)
		return
	}
	w.Header().Set("Content-Type", "text/event-stream")
	w.Header().Set("Cache-Control", "no-cache")
	w.Header().Set("Connection", "keep-alive")

	ch, unsub := s.hub.subscribe(room)
	defer unsub()

	sendRoom := func(targets []*gen.StreamTarget, dists []*gen.DistributionTarget) {
		data, _ := protojson.Marshal(&gen.RoomEvent{Room: room, Targets: targets, Distributions: dists})
		fmt.Fprintf(w, "event: room\ndata: %s\n\n", data)
		flusher.Flush()
	}
	sendTyped := func(eventType string, raw json.RawMessage) {
		fmt.Fprintf(w, "event: %s\ndata: %s\n\n", eventType, raw)
		flusher.Flush()
	}
	sendRoom(s.hub.targets(room), s.hub.distributions(room))

	ticker := time.NewTicker(20 * time.Second)
	defer ticker.Stop()
	for {
		select {
		case <-r.Context().Done():
			return
		case msg := <-ch:
			var probe struct{ Type string `json:"type"` }
			if err := json.Unmarshal(msg, &probe); err == nil && probe.Type != "" {
				sendTyped(probe.Type, json.RawMessage(msg))
			} else {
				targets, dists := decodeRoomEvent(msg)
				if targets == nil {
					targets = s.hub.targets(room)
				}
				if dists == nil {
					dists = s.hub.distributions(room)
				}
				sendRoom(targets, dists)
			}
		case <-ticker.C:
			sendRoom(s.hub.targets(room), s.hub.distributions(room))
		}
	}
}


// handleRoomsEvents 以 SSE 推送全局房间列表变更（创建/删除/LIVE 状态变化），供 /rooms 实时刷新。
func (s *Server) handleRoomsEvents(w http.ResponseWriter, r *http.Request) {
	flusher, ok := w.(http.Flusher)
	if !ok {
		http.Error(w, "streaming unsupported", http.StatusInternalServerError)
		return
	}
	w.Header().Set("Content-Type", "text/event-stream")
	w.Header().Set("Cache-Control", "no-cache")
	w.Header().Set("Connection", "keep-alive")

	ch, unsub := s.hub.subscribeRooms()
	defer unsub()

	// initial sync hint
	fmt.Fprintf(w, "event: rooms\ndata: {\"type\":\"rooms\",\"event\":\"sync\",\"ts\":%d}\n\n", time.Now().Unix())
	flusher.Flush()

	ticker := time.NewTicker(25 * time.Second)
	defer ticker.Stop()
	for {
		select {
		case <-r.Context().Done():
			return
		case msg := <-ch:
			fmt.Fprintf(w, "event: rooms\ndata: %s\n\n", msg)
			flusher.Flush()
		case <-ticker.C:
			fmt.Fprintf(w, ": ping\n\n")
			flusher.Flush()
		}
	}
}

func (s *Server) handleSRSStreams(w http.ResponseWriter, r *http.Request) {
	resp, err := http.Get(s.cfg.SRSBaseURL + "/api/v1/streams/")
	if err != nil {
		http.Error(w, "srs unreachable: "+err.Error(), http.StatusBadGateway)
		return
	}
	defer resp.Body.Close()
	var parsed struct {
		Code    int `json:"code"`
		Streams []struct {
			App  string `json:"app"`
			Name string `json:"name"`
		} `json:"streams"`
	}
	if err := json.NewDecoder(resp.Body).Decode(&parsed); err != nil {
		http.Error(w, "srs decode: "+err.Error(), http.StatusBadGateway)
		return
	}
	names := make([]string, 0, len(parsed.Streams))
	for _, st := range parsed.Streams {
		names = append(names, st.Name)
	}
	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(map[string]interface{}{"code": parsed.Code, "streams": names})
}

func readBody(r *http.Request) []byte {
	buf := make([]byte, 0, 4096)
	tmp := make([]byte, 4096)
	for {
		n, err := r.Body.Read(tmp)
		if n > 0 {
			buf = append(buf, tmp[:n]...)
		}
		if err != nil {
			break
		}
		if len(buf) > 1<<20 {
			break
		}
	}
	return buf
}

func decodeTargets(msg []byte) []*gen.StreamTarget {
	targets, _ := decodeRoomEvent(msg)
	return targets
}

func decodeRoomEvent(msg []byte) ([]*gen.StreamTarget, []*gen.DistributionTarget) {
	var m struct {
		Targets       map[string]*gen.StreamTarget       `json:"targets"`
		Distributions map[string]*gen.DistributionTarget `json:"distributions"`
	}
	if err := json.Unmarshal(msg, &m); err != nil {
		return nil, nil
	}
	targets := make([]*gen.StreamTarget, 0, len(m.Targets))
	for _, t := range m.Targets {
		targets = append(targets, t)
	}
	dists := make([]*gen.DistributionTarget, 0, len(m.Distributions))
	for _, d := range m.Distributions {
		dists = append(dists, d)
	}
	return targets, dists
}

func (s *Server) handleRoomsQuery(w http.ResponseWriter, r *http.Request) {
	q := r.URL.Query()
	page := 1
	pageSize := 20
	if v := q.Get("page"); v != "" {
		fmt.Sscanf(v, "%d", &page)
	}
	if v := q.Get("page_size"); v != "" {
		fmt.Sscanf(v, "%d", &pageSize)
	}
	if v := q.Get("pageSize"); v != "" {
		fmt.Sscanf(v, "%d", &pageSize)
	}
	search := q.Get("search")
	var searchPtr *string
	if search != "" {
		searchPtr = &search
	}
	statusStr := q.Get("status")
	var statusPtr *room.RoomStatus
	if statusStr != "" {
		if st, err := room.ParseRoomStatus(statusStr); err == nil {
			statusPtr = &st
		}
	}
	creator := q.Get("creator")
	var creatorPtr *string
	if creator != "" {
		creatorPtr = &creator
	}
	var isPublicPtr *bool
	if v := q.Get("is_public"); v != "" {
		b := v == "true" || v == "1"
		isPublicPtr = &b
	}
	sortBy := room.RoomListSortBy(q.Get("sort_by"))
	if sortBy == "" {
		sortBy = room.SortByCreatedAt
	}
	sortDir := room.SortDirection(q.Get("sort_direction"))
	if sortDir == "" {
		sortDir = room.SortDesc
	}
	query := room.RoomListQuery{
		Pagination: room.PageParams{Page: page, PageSize: pageSize},
		Search:     searchPtr,
		Status:     statusPtr,
		Creator:    creatorPtr,
		IsPublic:   isPublicPtr,
		SortBy:     sortBy,
		SortDirection: sortDir,
	}
	// prefer DB-backed query if available
	if s.roomStore != nil && s.roomStore.DB() != nil {
		rooms, total, err := s.roomStore.QueryRooms(r.Context(), query)
		if err != nil {
			http.Error(w, err.Error(), http.StatusInternalServerError)
			return
		}
		w.Header().Set("Content-Type", "application/json")
		json.NewEncoder(w).Encode(map[string]interface{}{
			"rooms": rooms,
			"total": total,
			"page": page,
			"page_size": pageSize,
		})
		return
	}
	// fallback to memory hub
	all := s.hub.list()
	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(map[string]interface{}{
		"rooms": all,
		"total": len(all),
		"page": 1,
		"page_size": len(all),
	})
}