frostfs-http-gw/internal/handler/download.go

package handler

import (
	"archive/tar"
	"archive/zip"
	"bufio"
	"compress/gzip"
	"context"
	"errors"
	"fmt"
	"io"
	"net/url"
	"strings"
	"time"

	"git.frostfs.info/TrueCloudLab/frostfs-http-gw/internal/data"
	"git.frostfs.info/TrueCloudLab/frostfs-http-gw/internal/logs"
	"git.frostfs.info/TrueCloudLab/frostfs-http-gw/tree"
	"git.frostfs.info/TrueCloudLab/frostfs-http-gw/utils"
	"git.frostfs.info/TrueCloudLab/frostfs-observability/tracing"
	cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
	oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
	"github.com/valyala/fasthttp"
	"go.uber.org/zap"
)

// DownloadByAddressOrBucketName handles download requests using simple cid/oid or bucketname/key format.
func (h *Handler) DownloadByAddressOrBucketName(req *fasthttp.RequestCtx) {
	ctx, span := tracing.StartSpanFromContext(utils.GetContextFromRequest(req), "handler.DownloadByAddressOrBucketName")
	defer span.End()

	cidParam := req.UserValue("cid").(string)
	oidParam := req.UserValue("oid").(string)

	ctx = utils.SetReqLog(ctx, h.reqLogger(ctx).With(
		zap.String("cid", cidParam),
		zap.String("oid", oidParam),
	))

	path, err := url.QueryUnescape(oidParam)
	if err != nil {
		h.logAndSendError(ctx, req, logs.FailedToUnescapePath, err)
		return
	}

	bktInfo, err := h.getBucketInfo(ctx, cidParam)
	if err != nil {
		h.logAndSendError(ctx, req, logs.FailedToGetBucketInfo, err)
		return
	}

	checkS3Err := h.tree.CheckSettingsNodeExists(ctx, bktInfo)
	if checkS3Err != nil && !errors.Is(checkS3Err, tree.ErrNodeNotFound) {
		h.logAndSendError(ctx, req, logs.FailedToCheckIfSettingsNodeExist, checkS3Err)
		return
	}

	prm := MiddlewareParam{
		Context: ctx,
		Request: req,
		BktInfo: bktInfo,
		Path:    path,
	}

	indexPageEnabled := h.config.IndexPageEnabled()

	if checkS3Err == nil {
		run(prm, h.errorMiddleware(logs.ObjectNotFound, ErrObjectNotFound),
			Middleware{Func: h.byS3PathMiddleware(h.receiveFile, noopFormer), Enabled: true},
			Middleware{Func: h.byS3PathMiddleware(h.receiveFile, indexFormer), Enabled: indexPageEnabled},
			Middleware{Func: h.browseIndexMiddleware(h.getDirObjectsS3), Enabled: indexPageEnabled},
		)
	} else {
		slashFallbackEnabled := h.config.EnableFilepathSlashFallback()
		fileNameFallbackEnabled := h.config.EnableFilepathFallback()

		run(prm, h.errorMiddleware(logs.ObjectNotFound, ErrObjectNotFound),
			Middleware{Func: h.byAddressMiddleware(h.receiveFile), Enabled: true},
			Middleware{Func: h.byAttributeSearchMiddleware(h.receiveFile, object.AttributeFilePath, noopFormer), Enabled: true},
			Middleware{Func: h.byAttributeSearchMiddleware(h.receiveFile, object.AttributeFilePath, reverseLeadingSlash), Enabled: slashFallbackEnabled},
			Middleware{Func: h.byAttributeSearchMiddleware(h.receiveFile, object.AttributeFileName, noopFormer), Enabled: fileNameFallbackEnabled},
			Middleware{Func: h.byAttributeSearchMiddleware(h.receiveFile, object.AttributeFileName, reverseLeadingSlash), Enabled: fileNameFallbackEnabled && slashFallbackEnabled},
			Middleware{Func: h.byAttributeSearchMiddleware(h.receiveFile, object.AttributeFilePath, indexFormer), Enabled: indexPageEnabled},
			Middleware{Func: h.byAttributeSearchMiddleware(h.receiveFile, object.AttributeFileName, indexFormer), Enabled: fileNameFallbackEnabled && indexPageEnabled},
			Middleware{Func: h.browseIndexMiddleware(h.getDirObjectsNative), Enabled: indexPageEnabled},
		)
	}
}

type MiddlewareFunc func(param MiddlewareParam) bool

type MiddlewareParam struct {
	Context context.Context
	Request *fasthttp.RequestCtx
	BktInfo *data.BucketInfo
	Path    string
}

type Middleware struct {
	Func    MiddlewareFunc
	Enabled bool
}

func run(prm MiddlewareParam, defaultMiddleware MiddlewareFunc, middlewares ...Middleware) {
	for _, m := range middlewares {
		if m.Enabled && !m.Func(prm) {
			return
		}
	}

	defaultMiddleware(prm)
}

func indexFormer(path string) string {
	indexPath := path
	if indexPath != "" && !strings.HasSuffix(indexPath, "/") {
		indexPath += "/"
	}

	return indexPath + "index.html"
}

func reverseLeadingSlash(path string) string {
	if path == "" || path == "/" {
		return path
	}

	if path[0] == '/' {
		return path[1:]
	}

	return "/" + path
}

func noopFormer(path string) string {
	return path
}

func (h *Handler) byS3PathMiddleware(handler func(context.Context, *fasthttp.RequestCtx, oid.Address), pathFormer func(string) string) MiddlewareFunc {
	return func(prm MiddlewareParam) bool {
		ctx, span := tracing.StartSpanFromContext(prm.Context, "handler.byS3Path")
		defer span.End()

		path := pathFormer(prm.Path)

		foundOID, err := h.tree.GetLatestVersion(ctx, &prm.BktInfo.CID, path)
		if err == nil {
			if foundOID.IsDeleteMarker {
				h.logAndSendError(ctx, prm.Request, logs.IndexWasDeleted, ErrObjectNotFound)
				return false
			}

			addr := newAddress(prm.BktInfo.CID, foundOID.OID)
			handler(ctx, prm.Request, addr)
			return false
		}

		if !errors.Is(err, tree.ErrNodeNotFound) {
			h.logAndSendError(ctx, prm.Request, logs.FailedToGetLatestVersionOfIndexObject, err, zap.String("path", path))
			return false
		}

		return true
	}
}

func (h *Handler) byAttributeSearchMiddleware(handler func(context.Context, *fasthttp.RequestCtx, oid.Address), attr string, pathFormer func(string) string) MiddlewareFunc {
	return func(prm MiddlewareParam) bool {
		ctx, span := tracing.StartSpanFromContext(prm.Context, "handler.byAttributeSearch")
		defer span.End()

		path := pathFormer(prm.Path)

		res, err := h.search(ctx, prm.BktInfo.CID, attr, path, object.MatchStringEqual)
		if err != nil {
			h.logAndSendError(ctx, prm.Request, logs.FailedToFindObjectByAttribute, err)
			return false
		}
		defer res.Close()

		buf := make([]oid.ID, 1)
		n, err := res.Read(buf)
		if err == nil && n > 0 {
			addr := newAddress(prm.BktInfo.CID, buf[0])
			handler(ctx, prm.Request, addr)
			return false
		}

		if !errors.Is(err, io.EOF) {
			h.logAndSendError(ctx, prm.Request, logs.FailedToFindObjectByAttribute, err)
			return false
		}

		return true
	}
}

func (h *Handler) byAddressMiddleware(handler func(context.Context, *fasthttp.RequestCtx, oid.Address)) MiddlewareFunc {
	return func(prm MiddlewareParam) bool {
		ctx, span := tracing.StartSpanFromContext(prm.Context, "handler.byAddress")
		defer span.End()

		var objID oid.ID
		if objID.DecodeString(prm.Path) == nil {
			handler(ctx, prm.Request, newAddress(prm.BktInfo.CID, objID))
			return false
		}

		return true
	}
}

// DownloadByAttribute handles attribute-based download requests.
func (h *Handler) DownloadByAttribute(req *fasthttp.RequestCtx) {
	ctx, span := tracing.StartSpanFromContext(utils.GetContextFromRequest(req), "handler.DownloadByAttribute")
	defer span.End()

	h.byAttribute(ctx, req, h.receiveFile)
}

func (h *Handler) search(ctx context.Context, cnrID cid.ID, key, val string, op object.SearchMatchType) (ResObjectSearch, error) {
	filters := object.NewSearchFilters()
	filters.AddRootFilter()
	filters.AddFilter(key, val, op)

	prm := PrmObjectSearch{
		PrmAuth: PrmAuth{
			BearerToken: bearerToken(ctx),
		},
		Container: cnrID,
		Filters:   filters,
	}

	return h.frostfs.SearchObjects(ctx, prm)
}

// DownloadZip handles zip by prefix requests.
func (h *Handler) DownloadZip(req *fasthttp.RequestCtx) {
	ctx, span := tracing.StartSpanFromContext(utils.GetContextFromRequest(req), "handler.DownloadZip")
	defer span.End()

	scid, _ := req.UserValue("cid").(string)
	prefix, _ := req.UserValue("prefix").(string)

	ctx = utils.SetReqLog(ctx, h.reqLogger(ctx).With(zap.String("cid", scid), zap.String("prefix", prefix)))

	bktInfo, err := h.getBucketInfo(ctx, scid)
	if err != nil {
		h.logAndSendError(ctx, req, logs.FailedToGetBucketInfo, err)
		return
	}

	resSearch, err := h.searchObjectsByPrefix(ctx, bktInfo.CID, prefix)
	if err != nil {
		return
	}

	req.Response.Header.Set(fasthttp.HeaderContentType, "application/zip")
	req.Response.Header.Set(fasthttp.HeaderContentDisposition, "attachment; filename=\"archive.zip\"")

	req.SetBodyStreamWriter(h.getZipResponseWriter(ctx, resSearch, bktInfo))
}

func (h *Handler) getZipResponseWriter(ctx context.Context, resSearch ResObjectSearch, bktInfo *data.BucketInfo) func(w *bufio.Writer) {
	return func(w *bufio.Writer) {
		defer resSearch.Close()

		buf := make([]byte, 3<<20)
		zipWriter := zip.NewWriter(w)
		var objectsWritten int

		errIter := resSearch.Iterate(h.putObjectToArchive(ctx, bktInfo.CID, buf,
			func(obj *object.Object) (io.Writer, error) {
				objectsWritten++
				return h.createZipFile(zipWriter, obj)
			}),
		)
		if errIter != nil {
			h.reqLogger(ctx).Error(logs.IteratingOverSelectedObjectsFailed, zap.Error(errIter), logs.TagField(logs.TagDatapath))
			return
		} else if objectsWritten == 0 {
			h.reqLogger(ctx).Warn(logs.ObjectsNotFound, logs.TagField(logs.TagDatapath))
		}
		if err := zipWriter.Close(); err != nil {
			h.reqLogger(ctx).Error(logs.CloseZipWriter, zap.Error(err), logs.TagField(logs.TagDatapath))
		}
	}
}

func (h *Handler) createZipFile(zw *zip.Writer, obj *object.Object) (io.Writer, error) {
	method := zip.Store
	if h.config.ArchiveCompression() {
		method = zip.Deflate
	}

	filePath := getFilePath(obj)
	if len(filePath) == 0 || filePath[len(filePath)-1] == '/' {
		return nil, fmt.Errorf("invalid filepath '%s'", filePath)
	}

	return zw.CreateHeader(&zip.FileHeader{
		Name:     filePath,
		Method:   method,
		Modified: time.Now(),
	})
}

// DownloadTar forms tar.gz from objects by prefix.
func (h *Handler) DownloadTar(req *fasthttp.RequestCtx) {
	ctx, span := tracing.StartSpanFromContext(utils.GetContextFromRequest(req), "handler.DownloadTar")
	defer span.End()

	scid, _ := req.UserValue("cid").(string)
	prefix, _ := req.UserValue("prefix").(string)

	ctx = utils.SetReqLog(ctx, h.reqLogger(ctx).With(zap.String("cid", scid), zap.String("prefix", prefix)))

	bktInfo, err := h.getBucketInfo(ctx, scid)
	if err != nil {
		h.logAndSendError(ctx, req, logs.FailedToGetBucketInfo, err)
		return
	}

	resSearch, err := h.searchObjectsByPrefix(ctx, bktInfo.CID, prefix)
	if err != nil {
		return
	}

	req.Response.Header.Set(fasthttp.HeaderContentType, "application/gzip")
	req.Response.Header.Set(fasthttp.HeaderContentDisposition, "attachment; filename=\"archive.tar.gz\"")

	req.SetBodyStreamWriter(h.getTarResponseWriter(ctx, resSearch, bktInfo))
}

func (h *Handler) getTarResponseWriter(ctx context.Context, resSearch ResObjectSearch, bktInfo *data.BucketInfo) func(w *bufio.Writer) {
	return func(w *bufio.Writer) {
		defer resSearch.Close()

		compressionLevel := gzip.NoCompression
		if h.config.ArchiveCompression() {
			compressionLevel = gzip.DefaultCompression
		}

		// ignore error because it's not nil only if compressionLevel argument is invalid
		gzipWriter, _ := gzip.NewWriterLevel(w, compressionLevel)
		tarWriter := tar.NewWriter(gzipWriter)

		defer func() {
			if err := tarWriter.Close(); err != nil {
				h.reqLogger(ctx).Error(logs.CloseTarWriter, zap.Error(err), logs.TagField(logs.TagDatapath))
			}
			if err := gzipWriter.Close(); err != nil {
				h.reqLogger(ctx).Error(logs.CloseGzipWriter, zap.Error(err), logs.TagField(logs.TagDatapath))
			}
		}()

		var objectsWritten int
		buf := make([]byte, 3<<20) // the same as for upload

		errIter := resSearch.Iterate(h.putObjectToArchive(ctx, bktInfo.CID, buf,
			func(obj *object.Object) (io.Writer, error) {
				objectsWritten++
				return h.createTarFile(tarWriter, obj)
			}),
		)
		if errIter != nil {
			h.reqLogger(ctx).Error(logs.IteratingOverSelectedObjectsFailed, zap.Error(errIter), logs.TagField(logs.TagDatapath))
		} else if objectsWritten == 0 {
			h.reqLogger(ctx).Warn(logs.ObjectsNotFound, logs.TagField(logs.TagDatapath))
		}
	}
}

func (h *Handler) createTarFile(tw *tar.Writer, obj *object.Object) (io.Writer, error) {
	filePath := getFilePath(obj)
	if len(filePath) == 0 || filePath[len(filePath)-1] == '/' {
		return nil, fmt.Errorf("invalid filepath '%s'", filePath)
	}

	return tw, tw.WriteHeader(&tar.Header{
		Name: filePath,
		Mode: 0655,
		Size: int64(obj.PayloadSize()),
	})
}

func (h *Handler) putObjectToArchive(ctx context.Context, cnrID cid.ID, buf []byte, createArchiveHeader func(obj *object.Object) (io.Writer, error)) func(id oid.ID) bool {
	return func(id oid.ID) bool {
		logger := h.reqLogger(ctx).With(zap.String("oid", id.EncodeToString()))

		prm := PrmObjectGet{
			PrmAuth: PrmAuth{
				BearerToken: bearerToken(ctx),
			},
			Address: newAddress(cnrID, id),
		}

		resGet, err := h.frostfs.GetObject(ctx, prm)
		if err != nil {
			logger.Error(logs.FailedToGetObject, zap.Error(err), logs.TagField(logs.TagExternalStorage))
			return false
		}

		fileWriter, err := createArchiveHeader(&resGet.Header)
		if err != nil {
			logger.Error(logs.FailedToAddObjectToArchive, zap.Error(err), logs.TagField(logs.TagDatapath))
			return false
		}

		if err = writeToArchive(resGet, fileWriter, buf); err != nil {
			logger.Error(logs.FailedToAddObjectToArchive, zap.Error(err), logs.TagField(logs.TagDatapath))
			return false
		}

		return false
	}
}

func (h *Handler) searchObjectsByPrefix(ctx context.Context, cnrID cid.ID, prefix string) (ResObjectSearch, error) {
	prefix, err := url.QueryUnescape(prefix)
	if err != nil {
		return nil, fmt.Errorf("unescape prefix: %w", err)
	}

	resSearch, err := h.search(ctx, cnrID, object.AttributeFilePath, prefix, object.MatchCommonPrefix)
	if err != nil {
		return nil, fmt.Errorf("search objects by prefix: %w", err)
	}

	return resSearch, nil
}

func writeToArchive(resGet *Object, objWriter io.Writer, buf []byte) error {
	var err error
	if _, err = io.CopyBuffer(objWriter, resGet.Payload, buf); err != nil {
		return fmt.Errorf("copy object payload to zip file: %v", err)
	}

	if err = resGet.Payload.Close(); err != nil {
		return fmt.Errorf("object body close error: %w", err)
	}

	return nil
}

func getFilePath(obj *object.Object) string {
	for _, attr := range obj.Attributes() {
		if attr.Key() == object.AttributeFilePath {
			return attr.Value()
		}
	}

	return ""
}