openreplay/backend/pkg/spot/api/handlers.go

605 lines
18 KiB
Go

package api
import (
"bytes"
"context"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"strconv"
"strings"
"time"
"github.com/gorilla/mux"
metrics "openreplay/backend/pkg/metrics/spot"
"openreplay/backend/pkg/objectstorage"
"openreplay/backend/pkg/spot/auth"
"openreplay/backend/pkg/spot/service"
)
func (e *Router) createSpot(w http.ResponseWriter, r *http.Request) {
startTime := time.Now()
bodySize := 0
bodyBytes, err := e.readBody(w, r, e.cfg.JsonSizeLimit)
if err != nil {
e.ResponseWithError(r.Context(), w, http.StatusRequestEntityTooLarge, err, startTime, r.URL.Path, bodySize)
return
}
bodySize = len(bodyBytes)
req := &CreateSpotRequest{}
if err := json.Unmarshal(bodyBytes, req); err != nil {
e.ResponseWithError(r.Context(), w, http.StatusBadRequest, err, startTime, r.URL.Path, bodySize)
return
}
// Creat a spot
currUser := r.Context().Value("userData").(*auth.User)
newSpot, err := e.services.Spots.Add(currUser, req.Name, req.Comment, req.Duration, req.Crop)
if err != nil {
e.ResponseWithError(r.Context(), w, http.StatusInternalServerError, err, startTime, r.URL.Path, bodySize)
return
}
// Parse and upload preview image
previewImage, err := getSpotPreview(req.Preview)
if err != nil {
e.ResponseWithError(r.Context(), w, http.StatusBadRequest, err, startTime, r.URL.Path, bodySize)
return
}
previewName := fmt.Sprintf("%d/preview.jpeg", newSpot.ID)
if err = e.services.ObjStorage.Upload(bytes.NewReader(previewImage), previewName, "image/jpeg", objectstorage.NoCompression); err != nil {
e.log.Error(r.Context(), "can't upload preview image: %s", err)
e.ResponseWithError(r.Context(), w, http.StatusInternalServerError, errors.New("can't upload preview image"), startTime, r.URL.Path, bodySize)
return
}
mobURL, err := e.getUploadMobURL(newSpot.ID)
if err != nil {
e.ResponseWithError(r.Context(), w, http.StatusInternalServerError, err, startTime, r.URL.Path, bodySize)
return
}
videoURL, err := e.getUploadVideoURL(newSpot.ID)
if err != nil {
e.ResponseWithError(r.Context(), w, http.StatusInternalServerError, err, startTime, r.URL.Path, bodySize)
return
}
resp := &CreateSpotResponse{
ID: strconv.Itoa(int(newSpot.ID)),
MobURL: mobURL,
VideoURL: videoURL,
}
e.ResponseWithJSON(r.Context(), w, resp, startTime, r.URL.Path, bodySize)
}
func getSpotPreview(preview string) ([]byte, error) {
parts := strings.Split(preview, ",")
if len(parts) != 2 {
return nil, fmt.Errorf("invalid preview format")
}
base64Str := parts[1]
data, err := base64.StdEncoding.DecodeString(base64Str)
if err != nil {
return nil, fmt.Errorf("can't decode base64 preview: %s", err)
}
return data, nil
}
func (e *Router) getUploadMobURL(spotID uint64) (string, error) {
mobKey := fmt.Sprintf("%d/events.mob", spotID)
mobURL, err := e.services.ObjStorage.GetPreSignedUploadUrl(mobKey)
if err != nil {
return "", fmt.Errorf("can't get mob URL: %s", err)
}
return mobURL, nil
}
func (e *Router) getUploadVideoURL(spotID uint64) (string, error) {
mobKey := fmt.Sprintf("%d/video.webm", spotID)
mobURL, err := e.services.ObjStorage.GetPreSignedUploadUrl(mobKey)
if err != nil {
return "", fmt.Errorf("can't get video URL: %s", err)
}
return mobURL, nil
}
func getSpotID(r *http.Request) (uint64, error) {
vars := mux.Vars(r)
idStr := vars["id"]
if idStr == "" {
return 0, fmt.Errorf("empty spot id")
}
id, err := strconv.ParseUint(idStr, 10, 64)
if err != nil {
return 0, fmt.Errorf("invalid spot id")
}
if id <= 0 {
return 0, fmt.Errorf("invalid spot id")
}
return id, nil
}
func getSpotsRequest(r *http.Request) (*GetSpotsRequest, error) {
params := r.URL.Query()
page := params.Get("page")
limit := params.Get("limit")
pageNum, _ := strconv.ParseUint(page, 10, 64)
limitNum, _ := strconv.ParseUint(limit, 10, 64)
req := &GetSpotsRequest{
Query: params.Get("query"),
FilterBy: params.Get("filterBy"),
Order: params.Get("order"),
Page: pageNum,
Limit: limitNum,
}
return req, nil
}
func (e *Router) getPreviewURL(spotID uint64) (string, error) {
previewKey := fmt.Sprintf("%d/preview.jpeg", spotID)
previewURL, err := e.services.ObjStorage.GetPreSignedDownloadUrl(previewKey)
if err != nil {
return "", fmt.Errorf("can't get preview URL: %s", err)
}
return previewURL, nil
}
func (e *Router) getMobURL(spotID uint64) (string, error) {
mobKey := fmt.Sprintf("%d/events.mob", spotID)
mobURL, err := e.services.ObjStorage.GetPreSignedDownloadUrl(mobKey)
if err != nil {
return "", fmt.Errorf("can't get mob URL: %s", err)
}
return mobURL, nil
}
func (e *Router) getVideoURL(spotID uint64) (string, error) {
mobKey := fmt.Sprintf("%d/video.webm", spotID) // TODO: later return url to m3u8 file
mobURL, err := e.services.ObjStorage.GetPreSignedDownloadUrl(mobKey)
if err != nil {
return "", fmt.Errorf("can't get video URL: %s", err)
}
return mobURL, nil
}
func (e *Router) getSpot(w http.ResponseWriter, r *http.Request) {
startTime := time.Now()
bodySize := 0
id, err := getSpotID(r)
if err != nil {
e.ResponseWithError(r.Context(), w, http.StatusBadRequest, err, startTime, r.URL.Path, bodySize)
return
}
user := r.Context().Value("userData").(*auth.User)
res, err := e.services.Spots.GetByID(user, id)
if err != nil {
e.ResponseWithError(r.Context(), w, http.StatusInternalServerError, err, startTime, r.URL.Path, bodySize)
return
}
if res == nil {
e.ResponseWithError(r.Context(), w, http.StatusNotFound, fmt.Errorf("spot not found"), startTime, r.URL.Path, bodySize)
return
}
previewUrl, err := e.getPreviewURL(id)
if err != nil {
e.log.Error(r.Context(), "can't get preview URL: %s", err)
}
mobURL, err := e.getMobURL(id)
if err != nil {
e.ResponseWithError(r.Context(), w, http.StatusInternalServerError, err, startTime, r.URL.Path, bodySize)
return
}
videoURL, err := e.getVideoURL(id)
if err != nil {
e.ResponseWithError(r.Context(), w, http.StatusInternalServerError, err, startTime, r.URL.Path, bodySize)
return
}
spotInfo := &Info{
Name: res.Name,
UserEmail: res.UserEmail,
Duration: res.Duration,
Comments: res.Comments,
CreatedAt: res.CreatedAt,
PreviewURL: previewUrl,
MobURL: mobURL,
VideoURL: videoURL,
}
playlist, err := e.services.Transcoder.GetSpotStreamPlaylist(id)
if err != nil {
e.log.Warn(r.Context(), "can't get stream playlist: %s", err)
} else {
spotInfo.StreamFile = base64.StdEncoding.EncodeToString(playlist)
}
e.ResponseWithJSON(r.Context(), w, &GetSpotResponse{Spot: spotInfo}, startTime, r.URL.Path, bodySize)
}
func (e *Router) updateSpot(w http.ResponseWriter, r *http.Request) {
startTime := time.Now()
bodySize := 0
id, err := getSpotID(r)
if err != nil {
e.ResponseWithError(r.Context(), w, http.StatusBadRequest, err, startTime, r.URL.Path, bodySize)
return
}
bodyBytes, err := e.readBody(w, r, e.cfg.JsonSizeLimit)
if err != nil {
e.ResponseWithError(r.Context(), w, http.StatusRequestEntityTooLarge, err, startTime, r.URL.Path, bodySize)
return
}
bodySize = len(bodyBytes)
req := &UpdateSpotRequest{}
if err := json.Unmarshal(bodyBytes, req); err != nil {
e.ResponseWithError(r.Context(), w, http.StatusBadRequest, err, startTime, r.URL.Path, bodySize)
return
}
user := r.Context().Value("userData").(*auth.User)
_, err = e.services.Spots.UpdateName(user, id, req.Name)
if err != nil {
e.ResponseWithError(r.Context(), w, http.StatusInternalServerError, err, startTime, r.URL.Path, bodySize)
return
}
e.ResponseOK(r.Context(), w, startTime, r.URL.Path, bodySize)
}
func (e *Router) getSpots(w http.ResponseWriter, r *http.Request) {
startTime := time.Now()
bodySize := 0
req, err := getSpotsRequest(r)
if err != nil {
e.ResponseWithError(r.Context(), w, http.StatusBadRequest, err, startTime, r.URL.Path, bodySize)
return
}
user := r.Context().Value("userData").(*auth.User)
opts := &service.GetOpts{
NameFilter: req.Query, Order: req.Order, Page: req.Page, Limit: req.Limit}
switch req.FilterBy {
case "own":
opts.UserID = user.ID
default:
opts.TenantID = user.TenantID
}
spots, total, tenantHasSpots, err := e.services.Spots.Get(user, opts)
if err != nil {
e.ResponseWithError(r.Context(), w, http.StatusInternalServerError, err, startTime, r.URL.Path, bodySize)
return
}
res := make([]ShortInfo, 0, len(spots))
for _, spot := range spots {
previewUrl, err := e.getPreviewURL(spot.ID)
if err != nil {
e.log.Error(r.Context(), "can't get preview URL: %s", err)
}
res = append(res, ShortInfo{
ID: strconv.Itoa(int(spot.ID)),
Name: spot.Name,
UserEmail: spot.UserEmail,
Duration: spot.Duration,
CreatedAt: spot.CreatedAt,
PreviewURL: previewUrl,
})
}
e.ResponseWithJSON(r.Context(), w, &GetSpotsResponse{Spots: res, Total: total, TenantHasSpots: tenantHasSpots}, startTime, r.URL.Path, bodySize)
}
func (e *Router) deleteSpots(w http.ResponseWriter, r *http.Request) {
startTime := time.Now()
bodySize := 0
bodyBytes, err := e.readBody(w, r, e.cfg.JsonSizeLimit)
if err != nil {
e.ResponseWithError(r.Context(), w, http.StatusRequestEntityTooLarge, err, startTime, r.URL.Path, bodySize)
return
}
bodySize = len(bodyBytes)
req := &DeleteSpotRequest{}
if err := json.Unmarshal(bodyBytes, req); err != nil {
e.ResponseWithError(r.Context(), w, http.StatusBadRequest, err, startTime, r.URL.Path, bodySize)
return
}
spotsToDelete := make([]uint64, 0, len(req.SpotIDs))
for _, idStr := range req.SpotIDs {
id, err := strconv.ParseUint(idStr, 10, 64)
if err != nil {
e.ResponseWithError(r.Context(), w, http.StatusBadRequest, fmt.Errorf("invalid spot id: %s", idStr), startTime, r.URL.Path, bodySize)
return
}
spotsToDelete = append(spotsToDelete, id)
}
user := r.Context().Value("userData").(*auth.User)
if err := e.services.Spots.Delete(user, spotsToDelete); err != nil {
e.ResponseWithError(r.Context(), w, http.StatusInternalServerError, err, startTime, r.URL.Path, bodySize)
return
}
e.ResponseOK(r.Context(), w, startTime, r.URL.Path, bodySize)
}
func (e *Router) addComment(w http.ResponseWriter, r *http.Request) {
startTime := time.Now()
bodySize := 0
id, err := getSpotID(r)
if err != nil {
e.ResponseWithError(r.Context(), w, http.StatusBadRequest, err, startTime, r.URL.Path, bodySize)
return
}
bodyBytes, err := e.readBody(w, r, e.cfg.JsonSizeLimit)
if err != nil {
e.ResponseWithError(r.Context(), w, http.StatusRequestEntityTooLarge, err, startTime, r.URL.Path, bodySize)
return
}
bodySize = len(bodyBytes)
req := &AddCommentRequest{}
if err := json.Unmarshal(bodyBytes, req); err != nil {
e.ResponseWithError(r.Context(), w, http.StatusBadRequest, err, startTime, r.URL.Path, bodySize)
return
}
user := r.Context().Value("userData").(*auth.User)
updatedSpot, err := e.services.Spots.AddComment(user, id, &service.Comment{UserName: req.UserName, Text: req.Comment})
if err != nil {
e.ResponseWithError(r.Context(), w, http.StatusInternalServerError, err, startTime, r.URL.Path, bodySize)
return
}
mobURL, err := e.getMobURL(id)
if err != nil {
e.ResponseWithError(r.Context(), w, http.StatusInternalServerError, err, startTime, r.URL.Path, bodySize)
return
}
videoURL, err := e.getVideoURL(id)
if err != nil {
e.ResponseWithError(r.Context(), w, http.StatusInternalServerError, err, startTime, r.URL.Path, bodySize)
return
}
spotInfo := &Info{
Name: updatedSpot.Name,
Duration: updatedSpot.Duration,
Comments: updatedSpot.Comments,
CreatedAt: updatedSpot.CreatedAt,
MobURL: mobURL,
VideoURL: videoURL,
}
e.ResponseWithJSON(r.Context(), w, &GetSpotResponse{Spot: spotInfo}, startTime, r.URL.Path, bodySize)
}
func (e *Router) uploadedSpot(w http.ResponseWriter, r *http.Request) {
startTime := time.Now()
bodySize := 0
id, err := getSpotID(r)
if err != nil {
e.ResponseWithError(r.Context(), w, http.StatusBadRequest, err, startTime, r.URL.Path, bodySize)
return
}
user := r.Context().Value("userData").(*auth.User)
spot, err := e.services.Spots.GetByID(user, id) // check if spot exists
if err != nil {
e.ResponseWithError(r.Context(), w, http.StatusInternalServerError, err, startTime, r.URL.Path, bodySize)
return
}
e.log.Info(r.Context(), "uploaded spot %+v, from user: %+v", spot, user)
if err := e.services.Transcoder.Process(spot); err != nil {
e.log.Error(r.Context(), "can't add transcoding task: %s", err)
}
e.ResponseOK(r.Context(), w, startTime, r.URL.Path, bodySize)
}
func (e *Router) getSpotVideo(w http.ResponseWriter, r *http.Request) {
startTime := time.Now()
bodySize := 0
id, err := getSpotID(r)
if err != nil {
e.ResponseWithError(r.Context(), w, http.StatusBadRequest, err, startTime, r.URL.Path, bodySize)
return
}
key := fmt.Sprintf("%d/video.webm", id)
videoURL, err := e.services.ObjStorage.GetPreSignedDownloadUrl(key)
if err != nil {
e.ResponseWithError(r.Context(), w, http.StatusInternalServerError, err, startTime, r.URL.Path, bodySize)
return
}
resp := map[string]interface{}{
"url": videoURL,
}
e.ResponseWithJSON(r.Context(), w, resp, startTime, r.URL.Path, bodySize)
}
func (e *Router) getSpotStream(w http.ResponseWriter, r *http.Request) {
startTime := time.Now()
bodySize := 0
id, err := getSpotID(r)
if err != nil {
e.ResponseWithError(r.Context(), w, http.StatusBadRequest, err, startTime, r.URL.Path, bodySize)
return
}
// Example data to serve as the file content
streamPlaylist, err := e.services.Transcoder.GetSpotStreamPlaylist(id)
if err != nil {
e.ResponseWithError(r.Context(), w, http.StatusInternalServerError, err, startTime, r.URL.Path, bodySize)
return
}
// Create a buffer with the file content
buffer := bytes.NewBuffer(streamPlaylist)
// Set the headers for the response
w.Header().Set("Content-Disposition", "attachment; filename=index.m3u8")
w.Header().Set("Content-Type", "application/vnd.apple.mpegurl") //"application/octet-stream")
w.Header().Set("Content-Length", string(len(streamPlaylist)))
// Write the content of the buffer to the response writer
if _, err := buffer.WriteTo(w); err != nil {
e.ResponseWithError(r.Context(), w, http.StatusInternalServerError, err, startTime, r.URL.Path, bodySize)
return
}
}
func (e *Router) getPublicKey(w http.ResponseWriter, r *http.Request) {
startTime := time.Now()
bodySize := 0
id, err := getSpotID(r)
if err != nil {
e.ResponseWithError(r.Context(), w, http.StatusBadRequest, err, startTime, r.URL.Path, bodySize)
return
}
user := r.Context().Value("userData").(*auth.User)
key, err := e.services.Keys.Get(id, user)
if err != nil {
if strings.Contains(err.Error(), "not found") {
e.ResponseWithError(r.Context(), w, http.StatusNotFound, err, startTime, r.URL.Path, bodySize)
} else {
e.ResponseWithError(r.Context(), w, http.StatusInternalServerError, err, startTime, r.URL.Path, bodySize)
}
return
}
resp := map[string]interface{}{
"key": key,
}
e.ResponseWithJSON(r.Context(), w, resp, startTime, r.URL.Path, bodySize)
}
func (e *Router) updatePublicKey(w http.ResponseWriter, r *http.Request) {
startTime := time.Now()
bodySize := 0
id, err := getSpotID(r)
if err != nil {
e.ResponseWithError(r.Context(), w, http.StatusBadRequest, err, startTime, r.URL.Path, bodySize)
return
}
bodyBytes, err := e.readBody(w, r, e.cfg.JsonSizeLimit)
if err != nil {
e.ResponseWithError(r.Context(), w, http.StatusRequestEntityTooLarge, err, startTime, r.URL.Path, bodySize)
return
}
bodySize = len(bodyBytes)
req := &UpdateSpotPublicKeyRequest{}
if err := json.Unmarshal(bodyBytes, req); err != nil {
e.ResponseWithError(r.Context(), w, http.StatusBadRequest, err, startTime, r.URL.Path, bodySize)
return
}
user := r.Context().Value("userData").(*auth.User)
key, err := e.services.Keys.Set(id, req.Expiration, user)
if err != nil {
e.ResponseWithError(r.Context(), w, http.StatusInternalServerError, err, startTime, r.URL.Path, bodySize)
return
}
resp := map[string]interface{}{
"key": key,
}
e.ResponseWithJSON(r.Context(), w, resp, startTime, r.URL.Path, bodySize)
}
func (e *Router) spotStatus(w http.ResponseWriter, r *http.Request) {
startTime := time.Now()
bodySize := 0
id, err := getSpotID(r)
if err != nil {
e.ResponseWithError(r.Context(), w, http.StatusBadRequest, err, startTime, r.URL.Path, bodySize)
return
}
user := r.Context().Value("userData").(*auth.User)
status, err := e.services.Spots.GetStatus(user, id)
if err != nil {
e.ResponseWithError(r.Context(), w, http.StatusInternalServerError, err, startTime, r.URL.Path, bodySize)
return
}
resp := map[string]interface{}{
"status": status,
}
e.ResponseWithJSON(r.Context(), w, resp, startTime, r.URL.Path, bodySize)
}
func recordMetrics(requestStart time.Time, url string, code, bodySize int) {
if bodySize > 0 {
metrics.RecordRequestSize(float64(bodySize), url, code)
}
metrics.IncreaseTotalRequests()
metrics.RecordRequestDuration(float64(time.Now().Sub(requestStart).Milliseconds()), url, code)
}
func (e *Router) readBody(w http.ResponseWriter, r *http.Request, limit int64) ([]byte, error) {
body := http.MaxBytesReader(w, r.Body, limit)
bodyBytes, err := io.ReadAll(body)
// Close body
if closeErr := body.Close(); closeErr != nil {
e.log.Warn(r.Context(), "error while closing request body: %s", closeErr)
}
if err != nil {
return nil, err
}
return bodyBytes, nil
}
func (e *Router) ResponseOK(ctx context.Context, w http.ResponseWriter, requestStart time.Time, url string, bodySize int) {
w.WriteHeader(http.StatusOK)
e.log.Info(ctx, "response ok")
recordMetrics(requestStart, url, http.StatusOK, bodySize)
}
func (e *Router) ResponseWithJSON(ctx context.Context, w http.ResponseWriter, res interface{}, requestStart time.Time, url string, bodySize int) {
e.log.Info(ctx, "response ok")
body, err := json.Marshal(res)
if err != nil {
e.log.Error(ctx, "can't marshal response: %s", err)
}
w.Header().Set("Content-Type", "application/json")
w.Write(body)
recordMetrics(requestStart, url, http.StatusOK, bodySize)
}
type response struct {
Error string `json:"error"`
}
func (e *Router) ResponseWithError(ctx context.Context, w http.ResponseWriter, code int, err error, requestStart time.Time, url string, bodySize int) {
e.log.Error(ctx, "response error, code: %d, error: %s", code, err)
body, err := json.Marshal(&response{err.Error()})
if err != nil {
e.log.Error(ctx, "can't marshal response: %s", err)
}
w.WriteHeader(code)
w.Write(body)
recordMetrics(requestStart, url, code, bodySize)
}