MyGO/mygo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(file)!: stream uploads through staged storage

Browse Source 
- fix: replace multipart form parsing with streaming multipart reads and
  apply request body limits when max_upload_size is configured.
- refactor: route uploads through staging paths before promotion to
  long-term data paths, keeping incomplete uploads out of durable
  storage records.
- test: cover oversized uploads, unlimited uploads, staged cleanup, and
  local storage promotion boundaries.
- docs: document the staged upload model and multipart parent_id query
  parameter.
This commit is contained in:
Sunday
2026-07-05 17:18:19 +08:00
parent bfeb4b26e4
commit b988b4b15e
10 changed files with 457 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/service/file.go
+76 -10
View File
@@ -41,6 +41,8 @@ type FileList struct {
Total int64 `json:"total"`
}
var errUploadTooLarge = errors.New("upload exceeds maximum size")
// FileService handles file management business logic.
type FileService struct {
fileRepo repository.FileRepository
@@ -67,6 +69,12 @@ func NewFileService(
}
}
// MaxUploadSize returns the configured maximum upload size in bytes.
// A value of 0 means uploads are not size-limited by MyGO.
func (s *FileService) MaxUploadSize() int64 {
return s.maxUploadSize
}
// Upload stores a file's content and creates its metadata record.
// The MIME type is always detected server-side from the file content.
func (s *FileService) Upload(ctx context.Context, userID string, parentID *string, fileName string, reader io.Reader) (*FileInfo, error) {
@@ -87,14 +95,16 @@ func (s *FileService) Upload(ctx context.Context, userID string, parentID *strin
return nil, model.NewInternalError("check name conflict", err)
}
// Limit the reader if a max upload size is configured.
if s.maxUploadSize > 0 {
reader = io.LimitReader(reader, s.maxUploadSize+1)
reader = &uploadLimitReader{reader: reader, remaining: s.maxUploadSize}
}
// Detect MIME type from file content.
head := make([]byte, 512)
n, readErr := io.ReadFull(reader, head)
if isUploadTooLargeError(readErr) {
return nil, uploadTooLargeError(s.maxUploadSize)
}
if readErr != nil && readErr != io.ErrUnexpectedEOF && readErr != io.EOF {
return nil, model.NewInternalError("read for mime detection", readErr)
}
@@ -103,21 +113,30 @@ func (s *FileService) Upload(ctx context.Context, userID string, parentID *strin
reader = io.MultiReader(bytes.NewReader(head), reader)
fileID := uuid.NewString()
storagePath := fmt.Sprintf("%s/%s", userID, fileID)
stagedPath := fmt.Sprintf("staging/%s/%s", userID, uuid.NewString())
storagePath := fmt.Sprintf("data/%s/%s", userID, fileID)
// Compute SHA-256 hash while writing to storage.
// Compute SHA-256 hash while writing to staging storage.
hasher := sha256.New()
teeReader := io.TeeReader(reader, hasher)
written, err := s.storage.Save(ctx, storagePath, teeReader)
written, err := s.storage.SaveStaged(ctx, stagedPath, teeReader)
if err != nil {
return nil, model.NewInternalError("save file", err)
_ = s.storage.DeleteStaged(ctx, stagedPath)
if isUploadTooLargeError(err) {
return nil, uploadTooLargeError(s.maxUploadSize)
}
return nil, model.NewInternalError("save staged file", err)
}
if s.maxUploadSize > 0 && written > s.maxUploadSize {
// Clean up the oversized file.
_ = s.storage.Delete(ctx, storagePath)
return nil, &model.AppError{Status: http.StatusBadRequest, Message: fmt.Sprintf("file size exceeds the maximum allowed size of %d bytes", s.maxUploadSize)}
_ = s.storage.DeleteStaged(ctx, stagedPath)
return nil, uploadTooLargeError(s.maxUploadSize)
}
if err := s.storage.PromoteStaged(ctx, stagedPath, storagePath); err != nil {
_ = s.storage.DeleteStaged(ctx, stagedPath)
return nil, model.NewInternalError("promote staged file", err)
}
file := &model.File{
@@ -134,7 +153,7 @@ func (s *FileService) Upload(ctx context.Context, userID string, parentID *strin
}
if err := s.fileRepo.Create(ctx, file); err != nil {
// Compensate: remove the stored file.
// Compensate: remove the promoted file.
_ = s.storage.Delete(ctx, storagePath)
if errors.Is(err, model.ErrDuplicate) {
return nil, &model.AppError{Status: http.StatusConflict, Message: "a file with this name already exists in this directory"}
@@ -145,6 +164,53 @@ func (s *FileService) Upload(ctx context.Context, userID string, parentID *strin
return modelToFileInfo(file), nil
}
type uploadLimitReader struct {
reader io.Reader
remaining int64
}
func (r *uploadLimitReader) Read(p []byte) (int, error) {
if len(p) == 0 {
return 0, nil
}
if r.remaining == 0 {
var one [1]byte
n, err := r.reader.Read(one[:])
if n > 0 {
p[0] = one[0]
return 1, errUploadTooLarge
}
return 0, err
}
if int64(len(p)) > r.remaining {
p = p[:r.remaining]
}
n, err := r.reader.Read(p)
r.remaining -= int64(n)
return n, err
}
func isUploadTooLargeError(err error) bool {
if err == nil {
return false
}
if errors.Is(err, errUploadTooLarge) {
return true
}
var maxBytesErr *http.MaxBytesError
return errors.As(err, &maxBytesErr)
}
func uploadTooLargeError(maxUploadSize int64) *model.AppError {
if maxUploadSize > 0 {
return &model.AppError{Status: http.StatusRequestEntityTooLarge, Message: fmt.Sprintf("file size exceeds the maximum allowed size of %d bytes", maxUploadSize)}
}
return &model.AppError{Status: http.StatusRequestEntityTooLarge, Message: "request body is too large"}
}
// Download returns a reader for the file's content and its metadata.
// The caller must close the returned ReadCloser.
func (s *FileService) Download(ctx context.Context, userID, fileID string) (io.ReadCloser, *FileInfo, error) {
internal/service/file_test.go
+52 -3
View File
@@ -43,7 +43,7 @@ func newMemStorage() *memStorage {
return &memStorage{files: make(map[string][]byte)}
}
func (s *memStorage) Save(_ context.Context, path string, reader io.Reader) (int64, error) {
func (s *memStorage) SaveStaged(_ context.Context, path string, reader io.Reader) (int64, error) {
data, err := io.ReadAll(reader)
if err != nil {
return 0, err
@@ -54,6 +54,19 @@ func (s *memStorage) Save(_ context.Context, path string, reader io.Reader) (int
return int64(len(data)), nil
}
func (s *memStorage) PromoteStaged(_ context.Context, stagedPath, finalPath string) error {
s.mu.Lock()
defer s.mu.Unlock()
data, ok := s.files[stagedPath]
if !ok {
return io.ErrUnexpectedEOF
}
s.files[finalPath] = data
delete(s.files, stagedPath)
return nil
}
func (s *memStorage) Open(_ context.Context, path string) (io.ReadCloser, error) {
s.mu.RLock()
data, ok := s.files[path]
@@ -64,6 +77,10 @@ func (s *memStorage) Open(_ context.Context, path string) (io.ReadCloser, error)
return io.NopCloser(bytes.NewReader(data)), nil
}
func (s *memStorage) DeleteStaged(ctx context.Context, path string) error {
return s.Delete(ctx, path)
}
func (s *memStorage) Delete(_ context.Context, path string) error {
s.mu.Lock()
delete(s.files, path)
@@ -74,6 +91,10 @@ func (s *memStorage) Delete(_ context.Context, path string) error {
var _ storage.StorageBackend = (*memStorage)(nil)
func setupFileServiceWithStorage(t *testing.T) (*FileService, *memStorage) {
return setupFileServiceWithStorageAndMaxUploadSize(t, 0)
}
func setupFileServiceWithStorageAndMaxUploadSize(t *testing.T, maxUploadSize int64) (*FileService, *memStorage) {
t.Helper()
db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{})
@@ -87,7 +108,7 @@ func setupFileServiceWithStorage(t *testing.T) (*FileService, *memStorage) {
fileRepo := repository.NewFileRepository(db)
store := newMemStorage()
return NewFileService(fileRepo, store, 0, nil), store // 0 = unlimited, nil logger defaults to slog.Default()
return NewFileService(fileRepo, store, maxUploadSize, nil), store // nil logger defaults to slog.Default()
}
func setupFileService(t *testing.T) *FileService {
@@ -125,6 +146,34 @@ func TestFileService_Upload(t *testing.T) {
}
}
func TestFileService_UploadRejectsOversizedFileAndCleansStaging(t *testing.T) {
svc, store := setupFileServiceWithStorageAndMaxUploadSize(t, 5)
ctx := context.Background()
_, err := svc.Upload(ctx, "user1", nil, "too-large.txt", strings.NewReader("123456"))
assertAppError(t, err, http.StatusRequestEntityTooLarge)
store.mu.RLock()
defer store.mu.RUnlock()
if len(store.files) != 0 {
t.Fatalf("storage files = %#v, want empty after oversized upload", store.files)
}
}
func TestFileService_UploadUnlimitedWhenMaxUploadSizeIsZero(t *testing.T) {
svc := setupFileService(t)
ctx := context.Background()
content := strings.Repeat("a", 128)
info, err := svc.Upload(ctx, "user1", nil, "unlimited.txt", strings.NewReader(content))
if err != nil {
t.Fatalf("Upload = %v", err)
}
if info.Size != int64(len(content)) {
t.Errorf("Size = %d, want %d", info.Size, len(content))
}
}
func TestFileService_UploadIntoDirectory(t *testing.T) {
svc := setupFileService(t)
ctx := context.Background()
@@ -498,7 +547,7 @@ func TestFileService_SoftDeleteKeepsStorage(t *testing.T) {
t.Fatalf("Delete = %v", err)
}
storagePath := "user1/" + info.ID
storagePath := "data/user1/" + info.ID
reader, err := store.Open(ctx, storagePath)
if err != nil {
t.Fatalf("storage should retain content after soft-delete, got: %v", err)