Feat/support db transaction

cmd/daemon.go

@@ -51,12 +51,11 @@ var daemonCmd = &cobra.Command{
 			fx_opt.Override(new(block.Adapter), factory.BuildBlockAdapter),
 			//database
 			fx_opt.Override(new(*bun.DB), models.SetupDatabase),
-			fx_opt.Override(fx_opt.NextInvoke(), migrations.MigrateDatabase),
-			fx_opt.Override(new(models.IUserRepo), models.NewUserRepo),
-			fx_opt.Override(new(models.IRepositoryRepo), models.NewRepositoryRepo),
-			fx_opt.Override(new(models.IRefRepo), models.NewRefRepo),
-			fx_opt.Override(new(models.IObjectRepo), models.NewObjectRepo),
+			fx_opt.Override(new(models.IRepo), func(db *bun.DB) models.IRepo {
+				return models.NewRepo(db)
+			}),
 
+			fx_opt.Override(fx_opt.NextInvoke(), migrations.MigrateDatabase),
 			//api
 			fx_opt.Override(fx_opt.NextInvoke(), apiImpl.SetupAPI),
 		)

config/default.go

@@ -1,5 +1,7 @@
 package config
 
+import "github.com/jiaozifs/jiaozifs/utils"
+
 var DefaultLocalBSPath = "~/.jiaozifs/blockstore"
 
 var defaultCfg = Config{
@@ -12,7 +14,7 @@ var defaultCfg = Config{
 	},
 	Blockstore: BlockStoreConfig{
 		Type:                   "local",
-		DefaultNamespacePrefix: nil,
+		DefaultNamespacePrefix: utils.String("data"),
 		Local: (*struct {
 			Path                    string   `mapstructure:"path"`
 			ImportEnabled           bool     `mapstructure:"import_enabled"`

controller/object_ctl.go

@@ -9,13 +9,10 @@ import (
 	"net/http"
 	"time"
 
-	"github.com/jiaozifs/jiaozifs/config"
-
-	"github.com/jiaozifs/jiaozifs/versionmgr"
-
+	"github.com/go-openapi/swag"
 	"github.com/jiaozifs/jiaozifs/models/filemode"
 
-	"github.com/go-openapi/swag"
+	"github.com/jiaozifs/jiaozifs/versionmgr"
 
 	"github.com/jiaozifs/jiaozifs/block"
 
@@ -33,13 +30,7 @@ type ObjectController struct {
 
 	BlockAdapter block.Adapter
 
-	StashRepo  models.StashRepo
-	UserRepo   models.IUserRepo
-	Repository models.RepositoryRepo
-	Object     models.ObjectRepo
-	Ref        models.RefRepo
-
-	Cfg config.BlockStoreConfig
+	Repo models.IRepo
 }
 
 func (oct ObjectController) DeleteObject(ctx context.Context, w *api.JiaozifsResponse, r *http.Request, user string, repository string, params api.DeleteObjectParams) { //nolint
@@ -53,13 +44,13 @@ func (oct ObjectController) GetObject(ctx context.Context, w *api.JiaozifsRespon
 }
 
 func (oct ObjectController) HeadObject(ctx context.Context, w *api.JiaozifsResponse, r *http.Request, userName string, repository string, params api.HeadObjectParams) { //nolint
-	user, err := oct.UserRepo.Get(ctx, &models.GetUserParam{Name: utils.String(userName)})
+	user, err := oct.Repo.UserRepo().Get(ctx, &models.GetUserParam{Name: utils.String(userName)})
 	if err != nil {
 		w.Error(err)
 		return
 	}
 
-	repo, err := oct.Repository.Get(ctx, &models.GetRepoParams{
+	repo, err := oct.Repo.RepositoryRepo().Get(ctx, &models.GetRepoParams{
 		CreateID: user.ID,
 		Name:     utils.String(repository),
 	})
@@ -68,7 +59,7 @@ func (oct ObjectController) HeadObject(ctx context.Context, w *api.JiaozifsRespo
 		return
 	}
 
-	ref, err := oct.Ref.Get(ctx, &models.GetRefParams{
+	ref, err := oct.Repo.RefRepo().Get(ctx, &models.GetRefParams{
 		RepositoryID: repo.ID,
 		Name:         utils.String(params.Branch),
 	})
@@ -77,46 +68,44 @@ func (oct ObjectController) HeadObject(ctx context.Context, w *api.JiaozifsRespo
 		return
 	}
 
-	commit, err := oct.Object.Commit(ctx, ref.CommitHash)
+	commit, err := oct.Repo.ObjectRepo().Commit(ctx, ref.CommitHash)
 	if err != nil {
 		w.Error(err)
 		return
 	}
 
-	treeOp := versionmgr.NewTreeOp(oct.Object)
-
-	treeNode, err := oct.Object.TreeNode(ctx, commit.TreeHash)
+	objRepo := oct.Repo.ObjectRepo()
+	treeOp := versionmgr.NewTreeOp(oct.Repo.ObjectRepo())
+	treeNode, err := objRepo.TreeNode(ctx, commit.TreeHash)
 	if err != nil {
 		w.Error(err)
 		return
 	}
-
 	existNodes, missingPath, err := treeOp.MatchPath(ctx, treeNode, params.Path)
 	if err != nil {
 		w.Error(err)
 		return
 	}
-
 	if len(missingPath) == 0 {
 		w.Error(versionmgr.ErrPathNotFound)
 		return
 	}
 
-	blobWithName := existNodes[len(existNodes)-1]
+	objectWithName := existNodes[len(existNodes)-1]
 
-	blob, err := oct.Object.Blob(ctx, blobWithName.Node.Hash)
+	blob, err := objRepo.Blob(ctx, objectWithName.Node.Hash)
 	if err != nil {
 		w.Error(err)
 		return
 	}
 
 	//lookup files
-	etag := httputil.ETag(blobWithName.Node.Hash.Hex())
+	etag := httputil.ETag(objectWithName.Node.Hash.Hex())
 	w.Header().Set("ETag", etag)
-	lastModified := httputil.HeaderTimestamp(blobWithName.Node.CreatedAt)
+	lastModified := httputil.HeaderTimestamp(objectWithName.Node.CreatedAt)
 	w.Header().Set("Last-Modified", lastModified)
 	w.Header().Set("Accept-Ranges", "bytes")
-	w.Header().Set("Content-Type", httputil.ExtensionsByType(blobWithName.Name))
+	w.Header().Set("Content-Type", httputil.ExtensionsByType(objectWithName.Name))
 	// for security, make sure the browser and any proxies en route don't cache the response
 	w.Header().Set("Cache-Control", "no-store, must-revalidate")
 	w.Header().Set("Expires", "0")
@@ -144,16 +133,9 @@ func (oct ObjectController) UploadObject(ctx context.Context, w *api.JiaozifsRes
 		w.Error(err)
 		return
 	}
-	treeOp := versionmgr.TreeOp{Object: oct.Object}
-	var blob *models.Blob
-	if mediaType != "multipart/form-data" {
-		// handle non-multipart, direct content upload
-		blob, err = treeOp.WriteBlob(ctx, oct.BlockAdapter, r.Body, r.ContentLength, block.PutOpts{})
-		if err != nil {
-			w.Error(err)
-			return
-		}
-	} else {
+
+	reader := r.Body
+	if mediaType == "multipart/form-data" {
 		// handle multipart upload
 		boundary, ok := p["boundary"]
 		if !ok {
@@ -162,9 +144,9 @@ func (oct ObjectController) UploadObject(ctx context.Context, w *api.JiaozifsRes
 		}
 
 		contentUploaded := false
-		reader := multipart.NewReader(r.Body, boundary)
+		partReader := multipart.NewReader(r.Body, boundary)
 		for !contentUploaded {
-			part, err := reader.NextPart()
+			part, err := partReader.NextPart()
 			if err == io.EOF {
 				break
 			}
@@ -175,73 +157,74 @@ func (oct ObjectController) UploadObject(ctx context.Context, w *api.JiaozifsRes
 			contentType = part.Header.Get("Content-Type")
 			partName := part.FormName()
 			if partName == "content" {
-				// upload the first "content" and exit the loop
-				blob, err = treeOp.WriteBlob(ctx, oct.BlockAdapter, part, -1, block.PutOpts{})
-				if err != nil {
-					_ = part.Close()
-					w.Error(err)
-					return
-				}
+				reader = part
 				contentUploaded = true
+			} else { //close not target part
+				_ = part.Close()
 			}
-			_ = part.Close()
+
 		}
 		if !contentUploaded {
 			w.Error(fmt.Errorf("multipart upload missing key 'content': %w", http.ErrMissingFile))
 			return
 		}
 	}
+	defer reader.Close() //nolint
 
-	user, err := oct.UserRepo.Get(ctx, &models.GetUserParam{Name: utils.String(userName)})
-	if err != nil {
-		w.Error(err)
-		return
-	}
+	var response api.ObjectStats
+	err = oct.Repo.Transaction(ctx, func(dRepo models.IRepo) error {
+		treeOp := versionmgr.TreeOp{Object: dRepo.ObjectRepo()}
+		blob, err := treeOp.WriteBlob(ctx, oct.BlockAdapter, reader, r.ContentLength, block.PutOpts{})
+		if err != nil {
+			return err
+		}
 
-	repo, err := oct.Repository.Get(ctx, &models.GetRepoParams{
-		CreateID: user.ID,
-		Name:     utils.String(repository),
-	})
-	if err != nil {
-		w.Error(err)
-		return
-	}
+		user, err := dRepo.UserRepo().Get(ctx, &models.GetUserParam{Name: utils.String(userName)})
+		if err != nil {
+			return err
+		}
 
-	stash, err := oct.StashRepo.Get(ctx, &models.GetStashParam{
-		RepositoryID: repo.ID,
-		CreateID:     user.ID,
-	})
-	if err != nil {
-		w.Error(err)
-		return
-	}
+		repo, err := dRepo.RepositoryRepo().Get(ctx, &models.GetRepoParams{
+			CreateID: user.ID,
+			Name:     utils.String(repository),
+		})
+		if err != nil {
+			return err
+		}
 
-	workingTreeID, err := oct.Object.TreeNode(ctx, stash.CurrentTree)
-	if err != nil {
-		w.Error(err)
-		return
-	}
+		stash, err := dRepo.WipRepo().Get(ctx, &models.GetWipParam{
+			RepositoryID: repo.ID,
+			CreateID:     user.ID,
+		})
+		if err != nil {
+			return err
+		}
 
-	newRoot, err := treeOp.AddLeaf(ctx, workingTreeID, params.Path, blob)
-	if err != nil {
-		w.Error(err)
-		return
-	}
+		workingTreeID, err := dRepo.ObjectRepo().TreeNode(ctx, stash.CurrentTree)
+		if err != nil {
+			return err
+		}
+
+		newRoot, err := treeOp.AddLeaf(ctx, workingTreeID, params.Path, blob)
+		if err != nil {
+			return err
+		}
+		response = api.ObjectStats{
+			Checksum:    blob.Hash.Hex(),
+			Mtime:       time.Now().Unix(),
+			Path:        params.Path,
+			PathMode:    utils.Uint32(uint32(filemode.Regular)),
+			SizeBytes:   swag.Int64(blob.Size),
+			ContentType: &contentType,
+			Metadata:    &api.ObjectUserMetadata{},
+		}
+		return dRepo.WipRepo().UpdateCurrentHash(ctx, stash.ID, newRoot.Hash)
+	})
 
-	err = oct.StashRepo.UpdateCurrentHash(ctx, stash.ID, newRoot.Hash)
 	if err != nil {
 		w.Error(err)
 		return
 	}
 
-	response := api.ObjectStats{
-		Checksum:    blob.Hash.Hex(),
-		Mtime:       time.Now().Unix(),
-		Path:        params.Path,
-		PathMode:    utils.Uint32(uint32(filemode.Regular)),
-		SizeBytes:   swag.Int64(blob.Size),
-		ContentType: &contentType,
-		Metadata:    &api.ObjectUserMetadata{},
-	}
 	w.JSON(response)
 }

controller/user_ctl.go

@@ -20,11 +20,11 @@ const (
 type UserController struct {
 	fx.In
 
-	Repo   models.IUserRepo
+	Repo   models.IRepo
 	Config *config.Config
 }
 
-func (A UserController) Login(ctx context.Context, w *api.JiaozifsResponse, r *http.Request) {
+func (userCtl UserController) Login(ctx context.Context, w *api.JiaozifsResponse, r *http.Request) {
 	// Decode requestBody
 	var login auth.Login
 	decoder := json.NewDecoder(r.Body)
@@ -34,15 +34,15 @@ func (A UserController) Login(ctx context.Context, w *api.JiaozifsResponse, r *h
 	}
 
 	// perform login
-	authToken, err := login.Login(ctx, A.Repo, A.Config)
+	authToken, err := login.Login(ctx, userCtl.Repo.UserRepo(), userCtl.Config)
 	if err != nil {
 		w.Error(err)
 		return
 	}
 	w.JSON(authToken)
 }
 
-func (A UserController) Register(ctx context.Context, w *api.JiaozifsResponse, r *http.Request) {
+func (userCtl UserController) Register(ctx context.Context, w *api.JiaozifsResponse, r *http.Request) {
 	// Decode requestBody
 	var register auth.Register
 	decoder := json.NewDecoder(r.Body)
@@ -51,21 +51,21 @@ func (A UserController) Register(ctx context.Context, w *api.JiaozifsResponse, r
 		return
 	}
 	// perform register
-	err := register.Register(ctx, A.Repo)
+	err := register.Register(ctx, userCtl.Repo.UserRepo())
 	if err != nil {
 		w.Error(err)
 		return
 	}
 	w.OK()
 }
 
-func (A UserController) GetUserInfo(ctx context.Context, w *api.JiaozifsResponse, r *http.Request) {
+func (userCtl UserController) GetUserInfo(ctx context.Context, w *api.JiaozifsResponse, r *http.Request) {
 	// Get token from Header
 	tokenString := r.Header.Get(AuthHeader)
 	userInfo := &auth.UserInfo{Token: tokenString}
 
 	// perform GetUserInfo
-	usrInfo, err := userInfo.UserProfile(ctx, A.Repo, A.Config)
+	usrInfo, err := userInfo.UserProfile(ctx, userCtl.Repo.UserRepo(), userCtl.Config)
 	if err != nil {
 		w.Error(err)
 		return

models/object.go

@@ -295,11 +295,11 @@ type IObjectRepo interface {
 var _ IObjectRepo = (*ObjectRepo)(nil)
 
 type ObjectRepo struct {
-	db *bun.DB
+	db bun.IDB
 }
 
-func NewObjectRepo(db *bun.DB) IObjectRepo {
-	return &ObjectRepo{db}
+func NewObjectRepo(db bun.IDB) IObjectRepo {
+	return &ObjectRepo{db: db}
 }
 
 func (o ObjectRepo) Insert(ctx context.Context, obj *Object) (*Object, error) {

models/ref.go

@@ -41,11 +41,11 @@ type IRefRepo interface {
 var _ IRefRepo = (*RefRepo)(nil)
 
 type RefRepo struct {
-	db *bun.DB
+	db bun.IDB
 }
 
-func NewRefRepo(db *bun.DB) IRefRepo {
-	return &RefRepo{db}
+func NewRefRepo(db bun.IDB) IRefRepo {
+	return &RefRepo{db: db}
 }
 
 func (r RefRepo) Insert(ctx context.Context, ref *Ref) (*Ref, error) {