edge/pkg/module/blob/http.go

package blob

import (
	"encoding/json"
	"io"
	"io/fs"
	"mime/multipart"
	"net/http"
	"os"
	"time"

	"forge.cadoles.com/arcad/edge/pkg/bus"
	edgehttp "forge.cadoles.com/arcad/edge/pkg/http"
	"forge.cadoles.com/arcad/edge/pkg/storage"
	"github.com/go-chi/chi/v5"
	"github.com/pkg/errors"
	"gitlab.com/wpetit/goweb/logger"
)

type uploadResponse struct {
	Bucket string         `json:"bucket"`
	BlobID storage.BlobID `json:"blobId"`
}

func Mount(uploadMaxFileSize int64) func(r chi.Router) {
	return func(r chi.Router) {
		r.Post("/api/v1/upload", getAppUploadHandler(uploadMaxFileSize))
		r.Get("/api/v1/download/{bucket}/{blobID}", handleAppDownload)
	}
}

func getAppUploadHandler(uploadMaxFileSize int64) func(w http.ResponseWriter, r *http.Request) {
	return func(w http.ResponseWriter, r *http.Request) {
		ctx := r.Context()

		r.Body = http.MaxBytesReader(w, r.Body, uploadMaxFileSize)

		if err := r.ParseMultipartForm(uploadMaxFileSize); err != nil {
			logger.Error(ctx, "could not parse multipart form", logger.CapturedE(errors.WithStack(err)))
			edgehttp.JSONError(w, http.StatusBadRequest, edgehttp.ErrCodeBadRequest)

			return
		}

		_, fileHeader, err := r.FormFile("file")
		if err != nil {
			logger.Error(ctx, "could not read form file", logger.CapturedE(errors.WithStack(err)))
			edgehttp.JSONError(w, http.StatusBadRequest, edgehttp.ErrCodeBadRequest)

			return
		}

		var metadata map[string]any

		rawMetadata := r.Form.Get("metadata")
		if rawMetadata != "" {
			if err := json.Unmarshal([]byte(rawMetadata), &metadata); err != nil {
				logger.Error(ctx, "could not parse metadata", logger.CapturedE(errors.WithStack(err)))
				edgehttp.JSONError(w, http.StatusBadRequest, edgehttp.ErrCodeBadRequest)

				return
			}
		}

		requestEnv := NewUploadRequestEnvelope(ctx, fileHeader, metadata)

		bus, ok := edgehttp.ContextBus(ctx)
		if !ok {
			logger.Error(ctx, "could find bus on context")
			edgehttp.JSONError(w, http.StatusInternalServerError, edgehttp.ErrCodeInternalError)

			return
		}

		reply, err := bus.Request(ctx, requestEnv)
		if err != nil {
			logger.Error(ctx, "could not retrieve file", logger.CapturedE(errors.WithStack(err)))
			edgehttp.JSONError(w, http.StatusInternalServerError, edgehttp.ErrCodeInternalError)

			return
		}

		logger.Debug(ctx, "upload reply", logger.F("reply", reply))

		replyMessage, ok := reply.Message().(*UploadResponse)
		if !ok {
			logger.Error(
				ctx, "unexpected upload response message",
				logger.F("message", reply.Message()),
			)
			edgehttp.JSONError(w, http.StatusInternalServerError, edgehttp.ErrCodeInternalError)

			return
		}

		if !replyMessage.Allow {
			edgehttp.JSONError(w, http.StatusForbidden, edgehttp.ErrCodeForbidden)

			return
		}

		encoder := json.NewEncoder(w)
		res := &uploadResponse{
			Bucket: replyMessage.Bucket,
			BlobID: replyMessage.BlobID,
		}

		if err := encoder.Encode(res); err != nil {
			panic(errors.Wrap(err, "could not encode upload response"))
		}
	}
}

func handleAppDownload(w http.ResponseWriter, r *http.Request) {
	bucket := chi.URLParam(r, "bucket")
	blobID := chi.URLParam(r, "blobID")

	ctx := logger.With(r.Context(), logger.F("blobID", blobID), logger.F("bucket", bucket))

	requestMsg := NewDownloadRequestEnvelope(ctx, bucket, storage.BlobID(blobID))

	bs, ok := edgehttp.ContextBus(ctx)
	if !ok {
		logger.Error(ctx, "could find bus on context")
		edgehttp.JSONError(w, http.StatusInternalServerError, edgehttp.ErrCodeInternalError)

		return
	}

	reply, err := bs.Request(ctx, requestMsg)
	if err != nil {
		logger.Error(ctx, "could not retrieve file", logger.CapturedE(errors.WithStack(err)))
		http.Error(w, http.StatusText(http.StatusInternalServerError), http.StatusInternalServerError)

		return
	}

	replyMessage, ok := reply.Message().(*DownloadResponse)
	if !ok {
		logger.Error(
			ctx, "unexpected download response message",
			logger.CapturedE(errors.WithStack(bus.ErrUnexpectedMessage)),
			logger.F("message", reply),
		)
		edgehttp.JSONError(w, http.StatusInternalServerError, edgehttp.ErrCodeInternalError)

		return
	}

	if !replyMessage.Allow {
		edgehttp.JSONError(w, http.StatusForbidden, edgehttp.ErrCodeForbidden)

		return
	}

	if replyMessage.Blob == nil {
		edgehttp.JSONError(w, http.StatusNotFound, edgehttp.ErrCodeNotFound)

		return
	}

	defer func() {
		if err := replyMessage.Blob.Close(); err != nil {
			logger.Error(ctx, "could not close blob", logger.CapturedE(errors.WithStack(err)))
		}
	}()

	// TODO Fix usage of ServeContent
	// http.ServeContent(w, r, string(replyMessage.BlobInfo.ID()), replyMessage.BlobInfo.ModTime(), replyMessage.Blob)

	w.Header().Add("Content-Type", replyMessage.BlobInfo.ContentType())

	if _, err := io.Copy(w, replyMessage.Blob); err != nil {
		logger.Error(ctx, "could not write blob", logger.CapturedE(errors.WithStack(err)))
	}
}

type uploadedFile struct {
	multipart.File
	header  *multipart.FileHeader
	modTime time.Time
}

// Stat implements fs.File
func (f *uploadedFile) Stat() (fs.FileInfo, error) {
	return &uploadedFileInfo{
		header:  f.header,
		modTime: f.modTime,
	}, nil
}

type uploadedFileInfo struct {
	header  *multipart.FileHeader
	modTime time.Time
}

// IsDir implements fs.FileInfo
func (i *uploadedFileInfo) IsDir() bool {
	return false
}

// ModTime implements fs.FileInfo
func (i *uploadedFileInfo) ModTime() time.Time {
	return i.modTime
}

// Mode implements fs.FileInfo
func (i *uploadedFileInfo) Mode() fs.FileMode {
	return os.ModePerm
}

// Name implements fs.FileInfo
func (i *uploadedFileInfo) Name() string {
	return i.header.Filename
}

// Size implements fs.FileInfo
func (i *uploadedFileInfo) Size() int64 {
	return i.header.Size
}

// Sys implements fs.FileInfo
func (i *uploadedFileInfo) Sys() any {
	return nil
}

var (
	_ fs.File     = &uploadedFile{}
	_ fs.FileInfo = &uploadedFileInfo{}
)