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

package handler

import (
	"archive/zip"
	"bufio"
	"context"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"time"

	"git.frostfs.info/TrueCloudLab/frostfs-http-gw/internal/api"
	"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/response"
	"git.frostfs.info/TrueCloudLab/frostfs-http-gw/tree"
	"git.frostfs.info/TrueCloudLab/frostfs-http-gw/utils"
	"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/bearer"
	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"
)

var errObjectDeleted = errors.New("object deleted")

// DownloadByAddressOrBucketName handles download requests using simple cid/oid or bucketname/key format.
func (h *Handler) DownloadByAddressOrBucketName(c *fasthttp.RequestCtx) {
	cidParam := c.UserValue("cid").(string)
	oidParam := c.UserValue("oid").(string)
	downloadParam := c.QueryArgs().GetBool("download")

	ctx := utils.GetContextFromRequest(c)
	log := utils.GetReqLogOrDefault(ctx, h.log)

	bktInfo, err := h.getBucketInfo(ctx, cidParam, log)
	if err != nil {
		logAndSendBucketError(c, log, err)
		return
	}

	foundOid, err := h.checkS3ObjectExist(ctx, bktInfo, oidParam)
	if errors.Is(err, errObjectDeleted) {
		log.Error(logs.ObjectWasDeleted)
		response.Error(c, "object deleted", fasthttp.StatusNotFound)
		return
	}
	if err != nil && !errors.Is(err, tree.ErrNodeNotFound) {
		logAndSendBucketError(c, log, err)
		return
	}

	var objID oid.ID
	if foundOid != nil && shouldDownload(oidParam, downloadParam) {
		// Receive file via S3
		addr := newAddress(bktInfo.CID, foundOid.OID)
		h.receiveFile(ctx, h.newRequest(c, log), addr)
	} else if err = objID.DecodeString(oidParam); err == nil {
		// Receive file via native protocol
		addr := newAddress(bktInfo.CID, objID)
		h.receiveFile(ctx, h.newRequest(c, log), addr)
	} else {
		h.browseIndex(c)
	}
}

func shouldDownload(oidParam string, downloadParam bool) bool {
	return !isDir(oidParam) || downloadParam
}

func (h *Handler) checkS3ObjectExist(ctx context.Context, bktInfo *data.BucketInfo, oidParam string) (*api.NodeVersion, error) {
	unescapedKey, err := url.QueryUnescape(oidParam)
	if err != nil {
		return nil, err
	}
	foundOid, err := h.tree.GetLatestVersion(ctx, &bktInfo.CID, unescapedKey)
	if err != nil {
		return nil, err
	}
	if foundOid.DeleteMarker {
		return nil, errObjectDeleted
	}

	return foundOid, err
}

func (h *Handler) newRequest(ctx *fasthttp.RequestCtx, log *zap.Logger) request {
	return request{
		RequestCtx: ctx,
		log:        log,
	}
}

// DownloadByAttribute handles attribute-based download requests.
func (h *Handler) DownloadByAttribute(c *fasthttp.RequestCtx) {
	scid, _ := c.UserValue("cid").(string)
	key, _ := c.UserValue("attr_key").(string)
	val, _ := c.UserValue("attr_val").(string)

	ctx := utils.GetContextFromRequest(c)
	log := utils.GetReqLogOrDefault(ctx, h.log)

	key, err := url.QueryUnescape(key)
	if err != nil {
		log.Error(logs.FailedToUnescapeQuery, zap.String("cid", scid), zap.String("attr_key", key), zap.Error(err))
		response.Error(c, "could not unescape attr_key: "+err.Error(), fasthttp.StatusBadRequest)
		return
	}

	val, err = url.QueryUnescape(val)
	if err != nil {
		log.Error(logs.FailedToUnescapeQuery, zap.String("cid", scid), zap.String("attr_val", val), zap.Error(err))
		response.Error(c, "could not unescape attr_val: "+err.Error(), fasthttp.StatusBadRequest)
		return
	}

	log = log.With(zap.String("cid", scid), zap.String("attr_key", key), zap.String("attr_val", val))

	bktInfo, err := h.getBucketInfo(ctx, scid, log)
	if err != nil {
		logAndSendBucketError(c, log, err)
		return
	}

	res, err := h.search(ctx, bktInfo.CID, key, val, object.MatchStringEqual)
	if err != nil {
		log.Error(logs.CouldNotSearchForObjects, zap.Error(err))
		response.Error(c, "could not search for objects: "+err.Error(), fasthttp.StatusBadRequest)
		return
	}

	defer res.Close()

	buf := make([]oid.ID, 1)

	n, err := res.Read(buf)
	if n == 0 {
		if errors.Is(err, io.EOF) {
			log.Error(logs.ObjectNotFound, zap.Error(err))
			response.Error(c, "object not found", fasthttp.StatusNotFound)
			return
		}

		log.Error(logs.ReadObjectListFailed, zap.Error(err))
		response.Error(c, "read object list failed: "+err.Error(), fasthttp.StatusBadRequest)
		return
	}

	var addr oid.Address
	addr.SetContainer(bktInfo.CID)
	addr.SetObject(buf[0])

	h.receiveFile(ctx, h.newRequest(c, log), addr)
}

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)
}

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

	filePath := getZipFilePath(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(),
	})
}

// DownloadZipped handles zip by prefix requests.
func (h *Handler) DownloadZipped(c *fasthttp.RequestCtx) {
	scid, _ := c.UserValue("cid").(string)
	prefix, _ := c.UserValue("prefix").(string)

	ctx := utils.GetContextFromRequest(c)
	log := utils.GetReqLogOrDefault(ctx, h.log)

	prefix, err := url.QueryUnescape(prefix)
	if err != nil {
		log.Error(logs.FailedToUnescapeQuery, zap.String("cid", scid), zap.String("prefix", prefix), zap.Error(err))
		response.Error(c, "could not unescape prefix: "+err.Error(), fasthttp.StatusBadRequest)
		return
	}

	log = log.With(zap.String("cid", scid), zap.String("prefix", prefix))

	bktInfo, err := h.getBucketInfo(ctx, scid, log)
	if err != nil {
		logAndSendBucketError(c, log, err)
		return
	}

	resSearch, err := h.search(ctx, bktInfo.CID, object.AttributeFilePath, prefix, object.MatchCommonPrefix)
	if err != nil {
		log.Error(logs.CouldNotSearchForObjects, zap.Error(err))
		response.Error(c, "could not search for objects: "+err.Error(), fasthttp.StatusBadRequest)
		return
	}

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

	c.SetBodyStreamWriter(func(w *bufio.Writer) {
		defer resSearch.Close()

		zipWriter := zip.NewWriter(w)

		var bufZip []byte
		var addr oid.Address

		empty := true
		called := false
		btoken := bearerToken(ctx)
		addr.SetContainer(bktInfo.CID)

		errIter := resSearch.Iterate(func(id oid.ID) bool {
			called = true

			if empty {
				bufZip = make([]byte, 3<<20) // the same as for upload
			}
			empty = false

			addr.SetObject(id)
			if err = h.zipObject(ctx, zipWriter, addr, btoken, bufZip); err != nil {
				log.Error(logs.FailedToAddObjectToArchive, zap.String("oid", id.EncodeToString()), zap.Error(err))
			}

			return false
		})
		if errIter != nil {
			log.Error(logs.IteratingOverSelectedObjectsFailed, zap.Error(errIter))
		} else if !called {
			log.Error(logs.ObjectsNotFound)
		}

		if err = zipWriter.Close(); err != nil {
			log.Error(logs.CloseZipWriter, zap.Error(err))
		}
	})
}

func (h *Handler) zipObject(ctx context.Context, zipWriter *zip.Writer, addr oid.Address, btoken *bearer.Token, bufZip []byte) error {
	prm := PrmObjectGet{
		PrmAuth: PrmAuth{
			BearerToken: btoken,
		},
		Address: addr,
	}

	resGet, err := h.frostfs.GetObject(ctx, prm)
	if err != nil {
		return fmt.Errorf("get FrostFS object: %v", err)
	}

	objWriter, err := h.addObjectToZip(zipWriter, &resGet.Header)
	if err != nil {
		return fmt.Errorf("zip create header: %v", err)
	}

	if _, err = io.CopyBuffer(objWriter, resGet.Payload, bufZip); 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)
	}

	if err = zipWriter.Flush(); err != nil {
		return fmt.Errorf("flush zip writer: %v", err)
	}

	return nil
}

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

	return ""
}