feat(upload): add programmatic Ingest service and file-upload skill

Introduce upload/ingest as the single domain entry for storing files,
writing w_uploads records, and maintaining incremental stats. Refactor
HTTP UploadFile and delete handlers to delegate to ingest, fix stats
decrement ordering on Remove, and document usage in the file-upload skill.
This commit is contained in:
ryan
2026-06-18 14:56:19 +08:00
parent 9c6c697d95
commit 1cad0c5c55
13 changed files with 995 additions and 192 deletions
+4 -116
View File
@@ -4,20 +4,13 @@
package handler
import (
"bytes"
"context"
"errors"
"sort"
"strings"
"github.com/Rain-kl/Wavelet/internal/apps/upload/shared"
uploadstats "github.com/Rain-kl/Wavelet/internal/apps/upload/stats"
uploadstorage "github.com/Rain-kl/Wavelet/internal/apps/upload/storage"
"github.com/Rain-kl/Wavelet/internal/db/idgen"
"github.com/Rain-kl/Wavelet/internal/apps/upload/ingest"
"github.com/Rain-kl/Wavelet/internal/model"
"github.com/Rain-kl/Wavelet/internal/repository"
"github.com/Rain-kl/Wavelet/internal/storage"
"github.com/Rain-kl/Wavelet/pkg/logger"
"gorm.io/gorm"
)
@@ -31,30 +24,11 @@ func listMyUploadFiles(ctx context.Context, userID uint64, filter repository.Upl
}
func softDeleteUpload(ctx context.Context, uploadID uint64) (model.Upload, error) {
upload, err := repository.GetActiveUploadByID(ctx, uploadID)
if err != nil {
return model.Upload{}, err
}
if err := repository.SoftDeleteUpload(ctx, &upload); err != nil {
return model.Upload{}, err
}
uploadstats.RecordUploadStatsRemove(ctx, &upload)
return upload, nil
return ingest.Remove(ctx, uploadID)
}
func softDeleteOwnedUpload(ctx context.Context, userID, uploadID uint64) (model.Upload, error) {
upload, err := repository.GetActiveUploadByID(ctx, uploadID)
if err != nil {
return model.Upload{}, err
}
if upload.UserID != userID {
return model.Upload{}, errUploadForbidden
}
if err := repository.SoftDeleteUpload(ctx, &upload); err != nil {
return model.Upload{}, err
}
uploadstats.RecordUploadStatsRemove(ctx, &upload)
return upload, nil
return ingest.RemoveOwned(ctx, userID, uploadID)
}
func listDistinctUploadTypes(ctx context.Context) ([]string, error) {
@@ -77,7 +51,7 @@ func updateOwnedUpload(ctx context.Context, userID, uploadID uint64, input updat
return model.Upload{}, err
}
if upload.UserID != userID {
return model.Upload{}, errUploadForbidden
return model.Upload{}, ingest.ErrForbidden
}
updates := make(map[string]any)
@@ -103,96 +77,10 @@ func listUploadsForBatchDownload(ctx context.Context, ids []uint64) ([]model.Upl
return repository.ListUploadsByIDs(ctx, ids)
}
type instantUploadInput struct {
UserID uint64
FileHash string
Size int64
MimeType string
Extension string
OrigName string
UploadType string
AccessMode int
}
func createInstantUpload(ctx context.Context, existing model.Upload, input instantUploadInput) (model.Upload, error) {
newUpload := model.Upload{
ID: idgen.NextUint64ID(),
UserID: input.UserID,
FileName: input.OrigName,
FilePath: existing.FilePath,
FileSize: input.Size,
MimeType: input.MimeType,
Extension: input.Extension,
Hash: input.FileHash,
Type: input.UploadType,
Status: model.UploadStatusUsed,
AccessMode: input.AccessMode,
Metadata: existing.Metadata,
}
if err := repository.CreateUpload(ctx, &newUpload); err != nil {
return model.Upload{}, err
}
uploadstats.RecordUploadStatsAdd(ctx, &newUpload)
logger.InfoF(ctx, "文件触发秒传成功! ID: %d, Path: %s", newUpload.ID, existing.FilePath)
return newUpload, nil
}
func findReusableUpload(ctx context.Context, hash string, size int64) (model.Upload, error) {
return repository.FindReusableUploadByHash(ctx, hash, size)
}
func saveNewUploadRecord(ctx context.Context, upload *model.Upload, filePath string) error {
if err := repository.CreateUpload(ctx, upload); err != nil {
_, backend, backendErr := storage.Active(ctx)
if backendErr == nil {
if deleteErr := backend.Delete(ctx, filePath); deleteErr != nil {
logger.WarnF(ctx, "清理未写入数据库的上传对象失败: %v", deleteErr)
}
}
return err
}
uploadstats.RecordUploadStatsAdd(ctx, upload)
return nil
}
func loadUploadStats(ctx context.Context) ([]model.UploadStat, error) {
return repository.ListUploadStats(ctx)
}
var errUploadForbidden = errors.New("upload forbidden")
func storeUploadObject(ctx context.Context, subPath string, size int64, mimeType string, buf *bytes.Buffer, meta *model.UploadMetadata) (string, error) {
if uploadstorage.ReadOnly(ctx) {
return "", errors.New(shared.ErrStorageReadOnly)
}
driver, backend, err := storage.Active(ctx)
if err != nil {
logger.ErrorF(ctx, "初始化活动存储失败: %v", err)
return "", errors.New(shared.ErrSaveFileFailed)
}
result, err := backend.Put(ctx, subPath, bytes.NewReader(buf.Bytes()), size, mimeType)
if err != nil {
logger.ErrorF(ctx, "写入 %s 存储失败: %v", driver, err)
return "", errors.New(shared.ErrSaveFileFailed)
}
meta.Bucket = result.Bucket
return result.Key, nil
}
func validateUploadAllowedExtension(ctx context.Context, ext string) string {
sc, err := repository.GetSystemConfigByKey(ctx, model.ConfigKeyUploadAllowedExtensions)
if err != nil || sc.Value == "" {
return ""
}
allowedExts := strings.Split(strings.ToLower(sc.Value), ",")
for _, allowedExt := range allowedExts {
if strings.TrimSpace(allowedExt) == ext {
return ""
}
}
return shared.ErrUnsupportedFormat
}
func isRecordNotFound(err error) bool {
return errors.Is(err, gorm.ErrRecordNotFound)
}