pipu-music/services/api/internal/uploads/tus.go

package uploads

import (
	"context"
	"errors"
	"io"
	"log/slog"
	"net/http"
	"os"
	"path/filepath"
	"strconv"
	"sync"

	"github.com/go-chi/chi/v5"
	"github.com/google/uuid"

	"pipu/api/internal/auth"
	"pipu/api/internal/platform/httpx"
)

// Ядро протокола tus 1.0.0 (https://tus.io/protocols/resumable-upload): HEAD — сколько принято, PATCH — следующий
// кусок с Upload-Offset. Создание (extension creation) не нужно: файл заводит POST /uploads вместе с резервом квоты,
// а клиент получает готовый tus_url. Принимаемое смещение — фактический размер файла в staging.

const tusVersion = "1.0.0"

// MaxChunkBytes — предел одного PATCH (клиент шлёт по 8 МБ; nginx пропускает до 16 МБ).
const MaxChunkBytes = 64 << 20

type tusFiles interface {
	File(ctx context.Context, id uuid.UUID) (File, error)
	SetReceived(ctx context.Context, id uuid.UUID, offset int64) error
}

// TUS — приём файлов очереди.
type TUS struct {
	files      tusFiles
	dir        string                                  // каталог staging
	onComplete func(ctx context.Context, f File) error // файл принят целиком
	log        *slog.Logger
	locks      sync.Map // id → *sync.Mutex: один PATCH на файл одновременно
}

// NewTUS создаёт приёмник.
func NewTUS(files tusFiles, dir string, onComplete func(ctx context.Context, f File) error, log *slog.Logger) *TUS {
	return &TUS{files: files, dir: dir, onComplete: onComplete, log: log}
}

// Mount регистрирует маршруты tus на роутере (`/api/v1/uploads/files/{id}`).
func (t *TUS) Mount(r chi.Router) {
	r.Options("/api/v1/uploads/files/{id}", t.options)
	r.Head("/api/v1/uploads/files/{id}", t.head)
	r.Patch("/api/v1/uploads/files/{id}", t.patch)
}

func (t *TUS) options(w http.ResponseWriter, _ *http.Request) {
	w.Header().Set("Tus-Resumable", tusVersion)
	w.Header().Set("Tus-Version", tusVersion)
	w.Header().Set("Tus-Max-Size", strconv.FormatInt(MaxChunkBytes, 10))
	w.WriteHeader(http.StatusNoContent)
}

// own — файл автора запроса в статусе, где tus имеет смысл.
func (t *TUS) own(w http.ResponseWriter, r *http.Request) (File, bool) {
	w.Header().Set("Tus-Resumable", tusVersion)
	w.Header().Set("Cache-Control", "no-store")
	p, ok := auth.PrincipalFrom(r.Context())
	if !ok {
		httpx.WriteProblem(w, http.StatusUnauthorized, auth.CodeUnauthorized, "not authenticated")
		return File{}, false
	}
	id, err := uuid.Parse(chi.URLParam(r, "id"))
	if err != nil {
		httpx.WriteProblem(w, http.StatusNotFound, CodeNotFound, "upload not found")
		return File{}, false
	}
	f, err := t.files.File(r.Context(), id)
	if err != nil || f.AuthorID != p.UserID {
		httpx.WriteProblem(w, http.StatusNotFound, CodeNotFound, "upload not found")
		return File{}, false
	}
	return f, true
}

func (t *TUS) path(f File) string { return filepath.Join(t.dir, filepath.FromSlash(f.StagingPath)) }

// onDisk — сколько байт уже лежит в staging.
func (t *TUS) onDisk(f File) int64 {
	info, err := os.Stat(t.path(f))
	if err != nil {
		return 0
	}
	return info.Size()
}

func (t *TUS) head(w http.ResponseWriter, r *http.Request) {
	f, ok := t.own(w, r)
	if !ok {
		return
	}
	offset := f.Size // принят целиком и уже обрабатывается
	switch {
	case f.Status == StatusUploading:
		offset = t.onDisk(f)
	case !f.Status.Occupies():
		w.WriteHeader(http.StatusGone)
		return
	}
	w.Header().Set("Upload-Offset", strconv.FormatInt(offset, 10))
	w.Header().Set("Upload-Length", strconv.FormatInt(f.Size, 10))
	w.WriteHeader(http.StatusOK)
}

func (t *TUS) patch(w http.ResponseWriter, r *http.Request) {
	f, ok := t.own(w, r)
	if !ok {
		return
	}
	if r.Header.Get("Tus-Resumable") != tusVersion {
		w.Header().Set("Tus-Version", tusVersion)
		w.WriteHeader(http.StatusPreconditionFailed)
		return
	}
	if r.Header.Get("Content-Type") != "application/offset+octet-stream" {
		httpx.WriteProblem(w, http.StatusUnsupportedMediaType, "common.bad_request", "content type must be application/offset+octet-stream")
		return
	}
	if f.Status != StatusUploading {
		httpx.WriteProblem(w, http.StatusConflict, CodeWrongStatus, "upload is not accepting data")
		return
	}
	mu, _ := t.locks.LoadOrStore(f.ID, &sync.Mutex{})
	lock := mu.(*sync.Mutex)
	if !lock.TryLock() {
		httpx.WriteProblem(w, http.StatusLocked, CodeOffset, "another request is writing this upload")
		return
	}
	defer lock.Unlock()

	offset, err := strconv.ParseInt(r.Header.Get("Upload-Offset"), 10, 64)
	have := t.onDisk(f)
	if err != nil || offset != have {
		w.Header().Set("Upload-Offset", strconv.FormatInt(have, 10))
		httpx.WriteProblem(w, http.StatusConflict, CodeOffset, "offset does not match")
		return
	}
	remaining := f.Size - have
	if r.ContentLength > remaining || r.ContentLength > MaxChunkBytes {
		httpx.WriteProblem(w, http.StatusRequestEntityTooLarge, CodeSizeMismatch, "chunk exceeds the declared size")
		return
	}

	n, werr := t.append(f, r.Body, min(remaining, MaxChunkBytes))
	now := have + n
	if err := t.files.SetReceived(r.Context(), f.ID, now); err != nil {
		t.log.ErrorContext(r.Context(), "tus: save offset", "upload_id", f.ID, "err", err)
	}
	if werr != nil {
		if errors.Is(werr, errTooMuch) {
			httpx.WriteProblem(w, http.StatusRequestEntityTooLarge, CodeSizeMismatch, "data exceeds the declared size")
			return
		}
		// Обрыв посреди куска: принятое сохранено, клиент спросит HEAD и продолжит.
		t.log.WarnContext(r.Context(), "tus: write interrupted", "upload_id", f.ID, "offset", now, "err", werr)
		httpx.WriteProblem(w, http.StatusInternalServerError, "common.internal", "write interrupted")
		return
	}
	if now == f.Size {
		if err := t.onComplete(r.Context(), f); err != nil {
			t.log.ErrorContext(r.Context(), "tus: complete", "upload_id", f.ID, "err", err)
			httpx.WriteProblem(w, http.StatusInternalServerError, "common.internal", "cannot finish upload")
			return
		}
	}
	w.Header().Set("Upload-Offset", strconv.FormatInt(now, 10))
	w.WriteHeader(http.StatusNoContent)
}

var errTooMuch = errors.New("uploads: more data than declared")

// append дописывает тело в конец файла staging, не больше limit байт; лишние данные — ошибка.
func (t *TUS) append(f File, body io.Reader, limit int64) (int64, error) {
	p := t.path(f)
	if err := os.MkdirAll(filepath.Dir(p), 0o750); err != nil {
		return 0, err
	}
	out, err := os.OpenFile(p, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o640)
	if err != nil {
		return 0, err
	}
	n, err := io.Copy(out, io.LimitReader(body, limit))
	if cerr := out.Close(); err == nil {
		err = cerr
	}
	if err != nil {
		return n, err
	}
	var extra [1]byte
	if m, _ := body.Read(extra[:]); m > 0 {
		// Лишнее на диск не попало — его отсёк LimitReader.
		return n, errTooMuch
	}
	return n, nil
}