internal/api/objects.go - master - t3/s3-server

internal/api/objects.go
258 строк7 KB
Ivan Shibkikh
cline crit fixes
22 июл 2026, 15:35
22 июл 2026, 15:357e21419
package api

import (
	"errors"
	"fmt"
	"io"
	"net/http"

	"gitverse.ru/t3/s3-server/internal/storage"
)

// PutObject загружает новый объект или перезаписывает существующий.
func (h *Handler) PutObject(w http.ResponseWriter, r *http.Request) {
	bucket := r.PathValue("bucket")
	key := r.PathValue("key")

	if !validateBucket(bucket) {
		writeXMLError(w, r, http.StatusBadRequest, "InvalidBucketName", "Invalid bucket name")
		return
	}

	_, err := h.meta.GetBucket(r.Context(), bucket)
	if err != nil {
		if errors.Is(err, storage.ErrBucketNotFound) {
			writeXMLError(w, r, http.StatusNotFound, "NoSuchBucket", "The specified bucket does not exist")
			return
		}
		logInternalError(h, w, r, "GetBucket", err)
		return
	}

	// Remember the old object metadata for cleanup on overwrite.
	oldMeta, _ := h.meta.GetObjectMeta(r.Context(), bucket, key)

	r.Body = http.MaxBytesReader(w, r.Body, h.cfg.Server.MaxUploadSize)

	hash, size, diskIndexes, err := h.data.Write(r.Context(), r.Body)
	if err != nil {
		if err.Error() == "http: request body too large" {
			writeXMLError(w, r, http.StatusRequestEntityTooLarge, "EntityTooLarge", "Your proposed upload exceeds the maximum allowed object size.")
		} else {
			logInternalError(h, w, r, "DataWrite", err)
		}
		return
	}

	meta := storage.ObjectMeta{
		Bucket:      bucket,
		Key:         key,
		Hash:        hash,
		Size:        size,
		DiskIndexes: diskIndexes,
		ContentType: r.Header.Get("Content-Type"),
	}

	if err := h.meta.PutObjectMeta(r.Context(), meta); err != nil {
		count, _ := h.meta.CountObjectsByHash(r.Context(), hash)
		if count == 0 {
			_ = h.data.Delete(r.Context(), diskIndexes, hash)
		}
		logInternalError(h, w, r, "PutObjectMeta", err)
		return
	}

	// Clean up old blob if it was an overwrite and the old hash is now orphaned.
	if oldMeta != nil && oldMeta.Hash != hash {
		count, _ := h.meta.CountObjectsByHash(r.Context(), oldMeta.Hash)
		if count == 0 {
			if oldMeta.IsComposite {
				parts, _ := h.meta.GetObjectParts(r.Context(), oldMeta.Hash)
				for _, p := range parts {
					_ = h.data.Delete(r.Context(), p.DiskIndexes, p.Hash)
				}
				_ = h.meta.DeleteObjectParts(r.Context(), oldMeta.Hash)
			} else {
				_ = h.data.Delete(r.Context(), oldMeta.DiskIndexes, oldMeta.Hash)
			}
		}
	}

	w.Header().Set("ETag", fmt.Sprintf(`"%s"`, hash))
	w.WriteHeader(http.StatusOK)
}

// HeadObject возвращает метаданные объекта без тела.
func (h *Handler) HeadObject(w http.ResponseWriter, r *http.Request) {
	bucket := r.PathValue("bucket")
	key := r.PathValue("key")

	meta, err := h.meta.GetObjectMeta(r.Context(), bucket, key)
	if err != nil {
		if errors.Is(err, storage.ErrObjectNotFound) {
			w.WriteHeader(http.StatusNotFound)
			return
		}
		w.WriteHeader(http.StatusInternalServerError)
		return
	}

	if !checkConditionalRequests(w, r, meta) {
		return
	}

	w.Header().Set("Content-Type", meta.ContentType)
	w.Header().Set("Content-Length", fmt.Sprintf("%d", meta.Size))
	w.Header().Set("ETag", fmt.Sprintf(`"%s"`, meta.Hash))
	w.Header().Set("Last-Modified", meta.UpdatedAt.Format(http.TimeFormat))
	w.Header().Set("Accept-Ranges", "bytes")
	w.WriteHeader(http.StatusOK)
}

// DeleteObject удаляет объект (мягкое удаление).
func (h *Handler) DeleteObject(w http.ResponseWriter, r *http.Request) {
	bucket := r.PathValue("bucket")
	key := r.PathValue("key")

	hash, diskIndexes, isComposite, err := h.meta.DeleteObjectMeta(r.Context(), bucket, key)
	if err != nil {
		if errors.Is(err, storage.ErrObjectNotFound) {
			w.WriteHeader(http.StatusNoContent)
			return
		}
		logInternalError(h, w, r, "DeleteObjectMeta", err)
		return
	}

	count, err := h.meta.CountObjectsByHash(r.Context(), hash)
	if err == nil && count == 0 {
		if isComposite {
			parts, _ := h.meta.GetObjectParts(r.Context(), hash)
			for _, p := range parts {
				_ = h.data.Delete(r.Context(), p.DiskIndexes, p.Hash)
			}
			h.meta.DeleteObjectParts(r.Context(), hash)
		} else {
			_ = h.data.Delete(r.Context(), diskIndexes, hash)
		}
	}
	w.WriteHeader(http.StatusNoContent)
}

// GetObject скачивает объект (с поддержкой Range).
func (h *Handler) GetObject(w http.ResponseWriter, r *http.Request) {
	bucket := r.PathValue("bucket")
	key := r.PathValue("key")

	meta, err := h.meta.GetObjectMeta(r.Context(), bucket, key)
	if err != nil {
		if errors.Is(err, storage.ErrObjectNotFound) {
			writeXMLError(w, r, http.StatusNotFound, "NoSuchKey", err.Error())
			return
		}
		writeXMLError(w, r, http.StatusInternalServerError, "InternalError", err.Error())
		return
	}

	if !checkConditionalRequests(w, r, meta) {
		return
	}

	var reader io.ReadCloser
	if meta.IsComposite {
		parts, err := h.meta.GetObjectParts(r.Context(), meta.Hash)
		if err != nil {
			writeXMLError(w, r, http.StatusInternalServerError, "InternalError", err.Error())
			return
		}
		reader = &compositeReader{
			parts: parts,
			ctx:   r.Context(),
			data:  h.data,
		}
	} else {
		rc, err := h.data.Read(r.Context(), meta.DiskIndexes, meta.Hash)
		if err != nil {
			if errors.Is(err, storage.ErrObjectNotFound) {
				writeXMLError(w, r, http.StatusNotFound, "NoSuchKey", "File missing on disk")
				return
			}
			writeXMLError(w, r, http.StatusInternalServerError, "InternalError", err.Error())
			return
		}
		reader = rc
	}
	defer reader.Close()

	w.Header().Set("Content-Type", meta.ContentType)
	w.Header().Set("ETag", fmt.Sprintf(`"%s"`, meta.Hash))
	w.Header().Set("Last-Modified", meta.UpdatedAt.Format(http.TimeFormat))
	w.Header().Set("Accept-Ranges", "bytes")

	rangeHeader := r.Header.Get("Range")
	if rangeHeader != "" {
		start, end, err := parseRange(rangeHeader, meta.Size)
		if err != nil {
			w.Header().Set("Content-Range", fmt.Sprintf("bytes */%d", meta.Size))
			writeXMLError(w, r, http.StatusRequestedRangeNotSatisfiable, "InvalidRange", err.Error())
			return
		}

		// Skip to the requested start offset for any reader type.
		if start > 0 {
			if _, err := io.CopyN(io.Discard, reader, start); err != nil {
				writeXMLError(w, r, http.StatusInternalServerError, "InternalError", err.Error())
				return
			}
		}

		length := end - start + 1
		w.Header().Set("Content-Length", fmt.Sprintf("%d", length))
		w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", start, end, meta.Size))
		w.WriteHeader(http.StatusPartialContent)

		io.CopyN(w, reader, length)
		return
	}

	w.Header().Set("Content-Length", fmt.Sprintf("%d", meta.Size))
	w.WriteHeader(http.StatusOK)
	io.Copy(w, reader)
}

// handlePut диспетчеризирует PUT-запросы (PutObject или UploadPart).
func (h *Handler) handlePut(w http.ResponseWriter, r *http.Request) {
	uploadID := r.URL.Query().Get("uploadId")
	if uploadID != "" {
		h.UploadPart(w, r)
		return
	}
	h.PutObject(w, r)
}

// handleDelete диспетчеризирует DELETE-запросы (DeleteObject или AbortMultipart).
func (h *Handler) handleDelete(w http.ResponseWriter, r *http.Request) {
	uploadID := r.URL.Query().Get("uploadId")
	if uploadID != "" {
		h.AbortMultipart(w, r)
		return
	}
	h.DeleteObject(w, r)
}

// handlePost диспетчеризирует POST-запросы (InitiateMultipart или CompleteMultipart).
func (h *Handler) handlePost(w http.ResponseWriter, r *http.Request) {
	q := r.URL.Query()

	if q.Has("uploads") {
		h.InitiateMultipart(w, r)
		return
	}

	if q.Get("uploadId") != "" {
		h.CompleteMultipart(w, r)
		return
	}

	w.WriteHeader(http.StatusMethodNotAllowed)
}