Compare commits
No commits in common. "6ddef03790971e346fa0a7d33a462f39348bc6cc" and "76b6fb34fec9e384383000761038a1ae6498dc47" have entirely different histories.
6ddef03790
...
76b6fb34fe
|
@ -6,11 +6,9 @@ import (
|
|||
"errors"
|
||||
"git.lumeweb.com/LumeWeb/portal/db"
|
||||
"git.lumeweb.com/LumeWeb/portal/model"
|
||||
"git.lumeweb.com/LumeWeb/portal/shared"
|
||||
_validator "git.lumeweb.com/LumeWeb/portal/validator"
|
||||
"github.com/go-playground/validator/v10"
|
||||
"github.com/kataras/iris/v12"
|
||||
"go.uber.org/zap"
|
||||
"golang.org/x/crypto/bcrypt"
|
||||
"gorm.io/gorm"
|
||||
"reflect"
|
||||
|
@ -59,7 +57,6 @@ func hashPassword(password string) (string, error) {
|
|||
// Generate a new bcrypt hash from the provided password.
|
||||
hashedPassword, err := bcrypt.GenerateFromPassword([]byte(password), bcrypt.DefaultCost)
|
||||
if err != nil {
|
||||
shared.GetLogger().Error("failed to hash password", zap.Error(err))
|
||||
return "", err
|
||||
}
|
||||
|
||||
|
@ -71,7 +68,6 @@ func (a *AccountController) PostRegister() {
|
|||
var r RegisterRequest
|
||||
|
||||
if err := a.Ctx.ReadJSON(&r); err != nil {
|
||||
shared.GetLogger().Debug("failed to parse request", zap.Error(err))
|
||||
a.Ctx.StopWithError(iris.StatusBadRequest, err)
|
||||
return
|
||||
}
|
||||
|
@ -80,13 +76,11 @@ func (a *AccountController) PostRegister() {
|
|||
existingAccount := model.Account{}
|
||||
err := db.Get().Where("email = ?", r.Email).First(&existingAccount).Error
|
||||
if err == nil {
|
||||
shared.GetLogger().Debug("account with email already exists", zap.Error(err), zap.String("email", r.Email))
|
||||
// An account with the same email address already exists.
|
||||
// Return an error response to the client.
|
||||
a.Ctx.StopWithError(iris.StatusConflict, errors.New("an account with this email address already exists"))
|
||||
return
|
||||
} else if !errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
shared.GetLogger().Error("error querying accounts", zap.Error(err), zap.String("email", r.Email))
|
||||
// An unexpected error occurred while querying the database.
|
||||
// Return an error response to the client.
|
||||
a.Ctx.StopWithError(iris.StatusInternalServerError, err)
|
||||
|
@ -125,7 +119,6 @@ func (a *AccountController) PostRegister() {
|
|||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
shared.GetLogger().Error("failed to create account", zap.Error(err))
|
||||
a.Ctx.StopWithError(iris.StatusInternalServerError, err)
|
||||
return
|
||||
}
|
||||
|
|
|
@ -7,11 +7,9 @@ import (
|
|||
"fmt"
|
||||
"git.lumeweb.com/LumeWeb/portal/db"
|
||||
"git.lumeweb.com/LumeWeb/portal/model"
|
||||
"git.lumeweb.com/LumeWeb/portal/shared"
|
||||
"github.com/joomcode/errorx"
|
||||
"github.com/kataras/iris/v12"
|
||||
"github.com/kataras/jwt"
|
||||
"go.uber.org/zap"
|
||||
"golang.org/x/crypto/bcrypt"
|
||||
"time"
|
||||
)
|
||||
|
@ -75,7 +73,6 @@ func generateToken(maxAge time.Duration) (string, error) {
|
|||
token, err := jwt.Sign(jwt.HS256, sharedKey, claim, jwt.MaxAge(maxAge))
|
||||
|
||||
if err != nil {
|
||||
shared.GetLogger().Error("failed to sign jwt", zap.Error(err))
|
||||
return "", err
|
||||
}
|
||||
|
||||
|
@ -86,7 +83,6 @@ func generateAndSaveLoginToken(accountID uint, maxAge time.Duration) (string, er
|
|||
// Generate a JWT token for the authenticated user.
|
||||
token, err := generateToken(maxAge)
|
||||
if err != nil {
|
||||
shared.GetLogger().Error("failed to generate token", zap.Error(err))
|
||||
return "", fmt.Errorf("failed to generate token: %s", err)
|
||||
}
|
||||
|
||||
|
@ -103,9 +99,7 @@ func generateAndSaveLoginToken(accountID uint, maxAge time.Duration) (string, er
|
|||
}
|
||||
|
||||
if err := db.Get().Create(&session).Error; err != nil {
|
||||
msg := "failed to save token"
|
||||
shared.GetLogger().Error(msg, zap.Error(err))
|
||||
return "", errorx.Decorate(err, msg)
|
||||
return "", errorx.Decorate(err, "failed to save token")
|
||||
}
|
||||
|
||||
return token, nil
|
||||
|
@ -115,7 +109,6 @@ func generateAndSaveChallengeToken(accountID uint, maxAge time.Duration) (string
|
|||
// Generate a JWT token for the authenticated user.
|
||||
token, err := generateToken(maxAge)
|
||||
if err != nil {
|
||||
shared.GetLogger().Error("failed to generate token", zap.Error(err))
|
||||
return "", fmt.Errorf("failed to generate token: %s", err)
|
||||
}
|
||||
|
||||
|
@ -132,9 +125,7 @@ func generateAndSaveChallengeToken(accountID uint, maxAge time.Duration) (string
|
|||
}
|
||||
|
||||
if err := db.Get().Create(&keyChallenge).Error; err != nil {
|
||||
msg := "failed to save token"
|
||||
shared.GetLogger().Error(msg, zap.Error(err))
|
||||
return "", errorx.Decorate(err, msg)
|
||||
return "", errorx.Decorate(err, "failed to save token")
|
||||
}
|
||||
|
||||
return token, nil
|
||||
|
@ -146,7 +137,6 @@ func (a *AuthController) PostLogin() {
|
|||
|
||||
// Read the login request from the client.
|
||||
if err := a.Ctx.ReadJSON(&r); err != nil {
|
||||
shared.GetLogger().Debug("failed to parse request", zap.Error(err))
|
||||
a.Ctx.StopWithError(iris.StatusBadRequest, err)
|
||||
return
|
||||
}
|
||||
|
@ -154,24 +144,19 @@ func (a *AuthController) PostLogin() {
|
|||
// Retrieve the account for the given email.
|
||||
account := model.Account{}
|
||||
if err := db.Get().Where("email = ?", r.Email).First(&account).Error; err != nil {
|
||||
msg := "invalid email or password"
|
||||
shared.GetLogger().Debug(msg, zap.Error(err))
|
||||
a.Ctx.StopWithError(iris.StatusBadRequest, errors.New(msg))
|
||||
a.Ctx.StopWithError(iris.StatusBadRequest, errors.New("invalid email or password"))
|
||||
return
|
||||
}
|
||||
|
||||
// Verify the provided password against the hashed password stored in the database.
|
||||
if err := verifyPassword(*account.Password, r.Password); err != nil {
|
||||
msg := "invalid email or password"
|
||||
shared.GetLogger().Debug(msg, zap.Error(err))
|
||||
a.Ctx.StopWithError(iris.StatusBadRequest, errors.New(msg))
|
||||
a.Ctx.StopWithError(iris.StatusBadRequest, errors.New("invalid email or password"))
|
||||
return
|
||||
}
|
||||
|
||||
// Generate a JWT token for the authenticated user.
|
||||
token, err := generateAndSaveLoginToken(account.ID, 24*time.Hour)
|
||||
if err != nil {
|
||||
shared.GetLogger().Debug("failed to generate token", zap.Error(err))
|
||||
a.Ctx.StopWithError(iris.StatusInternalServerError, fmt.Errorf("failed to generate token: %s", err))
|
||||
return
|
||||
}
|
||||
|
@ -179,7 +164,7 @@ func (a *AuthController) PostLogin() {
|
|||
// Return the JWT token to the client.
|
||||
err = a.Ctx.JSON(&LoginResponse{Token: token})
|
||||
if err != nil {
|
||||
shared.GetLogger().Error("failed to generate response", zap.Error(err))
|
||||
panic(fmt.Errorf("Error with login attempt: %s \n", err))
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -189,7 +174,6 @@ func (a *AuthController) PostPubkeyChallenge() {
|
|||
|
||||
// Read the login request from the client.
|
||||
if err := a.Ctx.ReadJSON(&r); err != nil {
|
||||
shared.GetLogger().Debug("failed to parse request", zap.Error(err))
|
||||
a.Ctx.StopWithError(iris.StatusBadRequest, err)
|
||||
return
|
||||
}
|
||||
|
@ -221,7 +205,6 @@ func (a *AuthController) PostPubkeyLogin() {
|
|||
|
||||
// Read the key login request from the client.
|
||||
if err := a.Ctx.ReadJSON(&r); err != nil {
|
||||
shared.GetLogger().Debug("failed to parse request", zap.Error(err))
|
||||
a.Ctx.StopWithError(iris.StatusBadRequest, err)
|
||||
return
|
||||
}
|
||||
|
@ -229,33 +212,25 @@ func (a *AuthController) PostPubkeyLogin() {
|
|||
// Retrieve the key challenge for the given challenge.
|
||||
challenge := model.KeyChallenge{}
|
||||
if err := db.Get().Where("challenge = ?", r.Challenge).Preload("Key").First(&challenge).Error; err != nil {
|
||||
msg := "invalid key challenge"
|
||||
shared.GetLogger().Debug(msg, zap.Error(err), zap.String("challenge", r.Challenge))
|
||||
a.Ctx.StopWithError(iris.StatusBadRequest, errorx.RejectedOperation.New(msg))
|
||||
a.Ctx.StopWithError(iris.StatusBadRequest, errorx.RejectedOperation.New("invalid key challenge"))
|
||||
return
|
||||
}
|
||||
|
||||
verifiedToken, err := jwt.Verify(jwt.HS256, sharedKey, []byte(r.Challenge), blocklist)
|
||||
if err != nil {
|
||||
msg := "invalid key challenge"
|
||||
shared.GetLogger().Debug(msg, zap.Error(err), zap.String("challenge", r.Challenge))
|
||||
a.Ctx.StopWithError(iris.StatusBadRequest, errorx.RejectedOperation.New(msg))
|
||||
a.Ctx.StopWithError(iris.StatusBadRequest, errorx.RejectedOperation.New("invalid key challenge"))
|
||||
return
|
||||
}
|
||||
|
||||
rawPubKey, err := hex.DecodeString(r.Pubkey)
|
||||
if err != nil {
|
||||
msg := "invalid pubkey"
|
||||
shared.GetLogger().Debug(msg, zap.Error(err), zap.String("pubkey", r.Pubkey))
|
||||
a.Ctx.StopWithError(iris.StatusBadRequest, errorx.RejectedOperation.New(msg))
|
||||
a.Ctx.StopWithError(iris.StatusBadRequest, errorx.RejectedOperation.New("invalid pubkey"))
|
||||
return
|
||||
}
|
||||
|
||||
rawSignature, err := hex.DecodeString(r.Signature)
|
||||
if err != nil {
|
||||
msg := "invalid signature"
|
||||
shared.GetLogger().Debug(msg, zap.Error(err), zap.String("signature", r.Signature))
|
||||
a.Ctx.StopWithError(iris.StatusBadRequest, errorx.RejectedOperation.New(msg))
|
||||
a.Ctx.StopWithError(iris.StatusBadRequest, errorx.RejectedOperation.New("invalid signature"))
|
||||
return
|
||||
}
|
||||
|
||||
|
@ -263,9 +238,7 @@ func (a *AuthController) PostPubkeyLogin() {
|
|||
|
||||
// Verify the challenge signature.
|
||||
if !ed25519.Verify(publicKeyDecoded, []byte(r.Challenge), rawSignature) {
|
||||
msg := "invalid challenge"
|
||||
shared.GetLogger().Debug(msg, zap.Error(err), zap.String("challenge", r.Challenge))
|
||||
a.Ctx.StopWithError(iris.StatusBadRequest, errorx.RejectedOperation.New(msg))
|
||||
a.Ctx.StopWithError(iris.StatusBadRequest, errorx.RejectedOperation.New("invalid challenge"))
|
||||
}
|
||||
|
||||
// Generate a JWT token for the authenticated user.
|
||||
|
@ -277,23 +250,19 @@ func (a *AuthController) PostPubkeyLogin() {
|
|||
|
||||
err = blocklist.InvalidateToken(verifiedToken.Token, verifiedToken.StandardClaims)
|
||||
if err != nil {
|
||||
msg := "failed to invalidate token"
|
||||
shared.GetLogger().Error(msg, zap.Error(err), zap.String("token", hex.EncodeToString(verifiedToken.Token)))
|
||||
a.Ctx.StopWithError(iris.StatusInternalServerError, errorx.RejectedOperation.Wrap(err, msg))
|
||||
a.Ctx.StopWithError(iris.StatusInternalServerError, errorx.RejectedOperation.Wrap(err, "failed to invalidate token"))
|
||||
return
|
||||
}
|
||||
|
||||
if err := db.Get().Delete(&challenge).Error; err != nil {
|
||||
msg := "failed to delete key challenge"
|
||||
shared.GetLogger().Error(msg, zap.Error(err), zap.Any("key_challenge", challenge))
|
||||
a.Ctx.StopWithError(iris.StatusBadRequest, errorx.RejectedOperation.New(msg))
|
||||
a.Ctx.StopWithError(iris.StatusBadRequest, errorx.RejectedOperation.New("failed to delete key challenge"))
|
||||
return
|
||||
}
|
||||
|
||||
// Return the JWT token to the client.
|
||||
err = a.Ctx.JSON(&LoginResponse{Token: token})
|
||||
if err != nil {
|
||||
shared.GetLogger().Error("failed to create response", zap.Error(err))
|
||||
panic(fmt.Errorf("Error with login attempt: %s \n", err))
|
||||
}
|
||||
|
||||
}
|
||||
|
@ -304,7 +273,6 @@ func (a *AuthController) PostLogout() {
|
|||
|
||||
// Read the logout request from the client.
|
||||
if err := a.Ctx.ReadJSON(&r); err != nil {
|
||||
shared.GetLogger().Debug("failed to parse request", zap.Error(err))
|
||||
a.Ctx.StopWithError(iris.StatusBadRequest, err)
|
||||
return
|
||||
}
|
||||
|
@ -312,18 +280,13 @@ func (a *AuthController) PostLogout() {
|
|||
// Verify the provided token.
|
||||
claims, err := jwt.Verify(jwt.HS256, sharedKey, []byte(r.Token), blocklist)
|
||||
if err != nil {
|
||||
msg := "invalid token"
|
||||
shared.GetLogger().Debug(msg, zap.Error(err))
|
||||
a.Ctx.StopWithError(iris.StatusBadRequest, errors.New(msg))
|
||||
a.Ctx.StopWithError(iris.StatusBadRequest, errors.New("invalid token"))
|
||||
return
|
||||
}
|
||||
|
||||
err = blocklist.InvalidateToken(claims.Token, claims.StandardClaims)
|
||||
if err != nil {
|
||||
msg := "failed to invalidate token"
|
||||
shared.GetLogger().Error(msg, zap.Error(err), zap.String("token", hex.EncodeToString(claims.Token)))
|
||||
a.Ctx.StopWithError(iris.StatusBadRequest, errors.New(msg))
|
||||
return
|
||||
panic(fmt.Errorf("Error with logout: %s \n", err))
|
||||
}
|
||||
|
||||
// Return a success response to the client.
|
||||
|
|
|
@ -4,9 +4,7 @@ import (
|
|||
"errors"
|
||||
"git.lumeweb.com/LumeWeb/portal/cid"
|
||||
"git.lumeweb.com/LumeWeb/portal/service/files"
|
||||
"git.lumeweb.com/LumeWeb/portal/shared"
|
||||
"github.com/kataras/iris/v12"
|
||||
"go.uber.org/zap"
|
||||
"io"
|
||||
)
|
||||
|
||||
|
@ -22,29 +20,22 @@ func (f *FilesController) PostUpload() {
|
|||
|
||||
file, meta, err := f.Ctx.FormFile("file")
|
||||
if internalErrorCustom(ctx, err, errors.New("invalid file data")) {
|
||||
shared.GetLogger().Debug("invalid file data", zap.Error(err))
|
||||
return
|
||||
}
|
||||
|
||||
upload, err := files.Upload(file, meta.Size)
|
||||
|
||||
if internalError(ctx, err) {
|
||||
shared.GetLogger().Debug("failed uploading file", zap.Error(err))
|
||||
return
|
||||
}
|
||||
|
||||
cidString, err := cid.EncodeString(upload.Hash, uint64(meta.Size))
|
||||
|
||||
if internalError(ctx, err) {
|
||||
shared.GetLogger().Debug("failed creating cid", zap.Error(err))
|
||||
return
|
||||
}
|
||||
|
||||
err = ctx.JSON(&UploadResponse{Cid: cidString})
|
||||
|
||||
if err != nil {
|
||||
shared.GetLogger().Error("failed to create response", zap.Error(err))
|
||||
}
|
||||
_ = ctx.JSON(&UploadResponse{Cid: cidString})
|
||||
}
|
||||
|
||||
func (f *FilesController) GetDownloadBy(cidString string) {
|
||||
|
@ -52,15 +43,18 @@ func (f *FilesController) GetDownloadBy(cidString string) {
|
|||
|
||||
_, err := cid.Valid(cidString)
|
||||
if sendError(ctx, err, iris.StatusBadRequest) {
|
||||
shared.GetLogger().Debug("invalid cid", zap.Error(err))
|
||||
return
|
||||
}
|
||||
|
||||
cidObject, _ := cid.Decode(cidString)
|
||||
hashHex := cidObject.StringHash()
|
||||
|
||||
if internalError(ctx, err) {
|
||||
return
|
||||
}
|
||||
|
||||
download, err := files.Download(hashHex)
|
||||
if internalError(ctx, err) {
|
||||
shared.GetLogger().Debug("failed fetching file", zap.Error(err))
|
||||
return
|
||||
}
|
||||
|
||||
|
@ -69,9 +63,7 @@ func (f *FilesController) GetDownloadBy(cidString string) {
|
|||
_ = download.(io.Closer).Close()
|
||||
return err
|
||||
})
|
||||
if internalError(ctx, err) {
|
||||
shared.GetLogger().Debug("failed streaming file", zap.Error(err))
|
||||
}
|
||||
internalError(ctx, err)
|
||||
}
|
||||
|
||||
func sendErrorCustom(ctx iris.Context, err error, customError error, irisError int) bool {
|
||||
|
|
5
main.go
5
main.go
|
@ -8,14 +8,12 @@ import (
|
|||
_ "git.lumeweb.com/LumeWeb/portal/docs"
|
||||
"git.lumeweb.com/LumeWeb/portal/renterd"
|
||||
"git.lumeweb.com/LumeWeb/portal/service/files"
|
||||
"git.lumeweb.com/LumeWeb/portal/shared"
|
||||
"git.lumeweb.com/LumeWeb/portal/tus"
|
||||
"git.lumeweb.com/LumeWeb/portal/validator"
|
||||
"github.com/iris-contrib/swagger"
|
||||
"github.com/iris-contrib/swagger/swaggerFiles"
|
||||
"github.com/kataras/iris/v12"
|
||||
"github.com/kataras/iris/v12/mvc"
|
||||
"go.uber.org/zap"
|
||||
"log"
|
||||
"net/http"
|
||||
)
|
||||
|
@ -50,7 +48,6 @@ func main() {
|
|||
|
||||
renterd.Ready()
|
||||
|
||||
shared.Init()
|
||||
files.Init()
|
||||
|
||||
// Create a new Iris app instance
|
||||
|
@ -109,7 +106,7 @@ func main() {
|
|||
})
|
||||
|
||||
if err != nil {
|
||||
shared.GetLogger().Error("Failed starting webserver proof", zap.Error(err))
|
||||
log.Fatal(err)
|
||||
}
|
||||
|
||||
renterd.ShutdownComplete()
|
||||
|
|
|
@ -11,9 +11,7 @@ import (
|
|||
"git.lumeweb.com/LumeWeb/portal/renterd"
|
||||
"git.lumeweb.com/LumeWeb/portal/shared"
|
||||
"github.com/go-resty/resty/v2"
|
||||
"go.uber.org/zap"
|
||||
"io"
|
||||
"strings"
|
||||
)
|
||||
|
||||
var client *resty.Client
|
||||
|
@ -31,7 +29,6 @@ func Upload(r io.ReadSeeker, size int64) (model.Upload, error) {
|
|||
tree, hashBytes, err := bao.ComputeTree(r, size)
|
||||
|
||||
if err != nil {
|
||||
shared.GetLogger().Error("Failed to hash file", zap.Error(err))
|
||||
return upload, err
|
||||
}
|
||||
|
||||
|
@ -47,7 +44,6 @@ func Upload(r io.ReadSeeker, size int64) (model.Upload, error) {
|
|||
if (result.Error != nil && result.Error.Error() != "record not found") || result.RowsAffected > 0 {
|
||||
err := result.Row().Scan(&upload)
|
||||
if err != nil {
|
||||
shared.GetLogger().Error("Failed to query uploads table", zap.Error(err))
|
||||
return upload, err
|
||||
}
|
||||
}
|
||||
|
@ -55,39 +51,26 @@ func Upload(r io.ReadSeeker, size int64) (model.Upload, error) {
|
|||
objectExistsResult, err := client.R().Get(fmt.Sprintf("/worker/objects/%s", hashHex))
|
||||
|
||||
if err != nil {
|
||||
shared.GetLogger().Error("Failed query object", zap.Error(err))
|
||||
return upload, err
|
||||
}
|
||||
|
||||
statusCode := objectExistsResult.StatusCode()
|
||||
|
||||
if statusCode == 500 {
|
||||
bodyErr := objectExistsResult.String()
|
||||
if !strings.Contains(bodyErr, "no slabs found") {
|
||||
shared.GetLogger().Error("Failed fetching object", zap.String("error", objectExistsResult.String()))
|
||||
return upload, errors.New(fmt.Sprintf("error fetching file: %s", objectExistsResult.String()))
|
||||
}
|
||||
|
||||
statusCode = 404
|
||||
if objectExistsResult.StatusCode() != 404 {
|
||||
return upload, errors.New("file already exists in network, but missing in database")
|
||||
}
|
||||
|
||||
if statusCode != 404 {
|
||||
msg := "file already exists in network, but missing in database"
|
||||
shared.GetLogger().Error(msg)
|
||||
return upload, errors.New(msg)
|
||||
if err != nil {
|
||||
return upload, err
|
||||
}
|
||||
|
||||
ret, err := client.R().SetBody(r).Put(fmt.Sprintf("/worker/objects/%s", hashHex))
|
||||
if ret.StatusCode() != 200 {
|
||||
shared.GetLogger().Error("Failed uploading object", zap.String("error", ret.String()))
|
||||
err = errors.New(ret.String())
|
||||
err = errors.New(string(ret.Body()))
|
||||
return upload, err
|
||||
}
|
||||
|
||||
ret, err = client.R().SetBody(tree).Put(fmt.Sprintf("/worker/objects/%s.obao", hashHex))
|
||||
if ret.StatusCode() != 200 {
|
||||
shared.GetLogger().Error("Failed uploading proof", zap.String("error", ret.String()))
|
||||
err = errors.New(ret.String())
|
||||
err = errors.New(string(ret.Body()))
|
||||
return upload, err
|
||||
}
|
||||
|
||||
|
@ -96,7 +79,6 @@ func Upload(r io.ReadSeeker, size int64) (model.Upload, error) {
|
|||
}
|
||||
|
||||
if err = db.Get().Create(&upload).Error; err != nil {
|
||||
shared.GetLogger().Error("Failed adding upload to db", zap.Error(err))
|
||||
return upload, err
|
||||
}
|
||||
|
||||
|
@ -109,7 +91,6 @@ func Download(hash string) (io.Reader, error) {
|
|||
if uploadItem.Err() == nil {
|
||||
fetch, err := client.R().SetDoNotParseResponse(true).Get(fmt.Sprintf("/worker/objects/%s", hash))
|
||||
if err != nil {
|
||||
shared.GetLogger().Error("Failed downloading object", zap.Error(err))
|
||||
return nil, err
|
||||
}
|
||||
|
||||
|
@ -118,25 +99,21 @@ func Download(hash string) (io.Reader, error) {
|
|||
var tusData model.Tus
|
||||
err := tusItem.Scan(&tusData)
|
||||
if err != nil {
|
||||
shared.GetLogger().Error("Failed querying upload from db", zap.Error(err))
|
||||
return nil, err
|
||||
}
|
||||
|
||||
upload, err := shared.GetTusStore().GetUpload(context.Background(), tusData.Id)
|
||||
if err != nil {
|
||||
shared.GetLogger().Error("Failed querying tus upload", zap.Error(err))
|
||||
return nil, err
|
||||
}
|
||||
|
||||
reader, err := upload.GetReader(context.Background())
|
||||
if err != nil {
|
||||
shared.GetLogger().Error("Failed reading tus upload", zap.Error(err))
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return reader, nil
|
||||
} else {
|
||||
shared.GetLogger().Error("invalid file")
|
||||
return nil, errors.New("invalid file")
|
||||
}
|
||||
}
|
||||
|
|
|
@ -4,15 +4,11 @@ import (
|
|||
"github.com/golang-queue/queue"
|
||||
"github.com/tus/tusd/pkg/filestore"
|
||||
tusd "github.com/tus/tusd/pkg/handler"
|
||||
"go.uber.org/zap"
|
||||
_ "go.uber.org/zap"
|
||||
"log"
|
||||
)
|
||||
|
||||
var tusQueue *queue.Queue
|
||||
var tusStore *filestore.FileStore
|
||||
var tusComposer *tusd.StoreComposer
|
||||
var logger *zap.Logger
|
||||
|
||||
func SetTusQueue(q *queue.Queue) {
|
||||
tusQueue = q
|
||||
|
@ -37,17 +33,3 @@ func SetTusComposer(c *tusd.StoreComposer) {
|
|||
func GetTusComposer() *tusd.StoreComposer {
|
||||
return tusComposer
|
||||
}
|
||||
|
||||
func Init() {
|
||||
newLogger, err := zap.NewProduction()
|
||||
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
|
||||
logger = newLogger
|
||||
}
|
||||
|
||||
func GetLogger() *zap.Logger {
|
||||
return logger
|
||||
}
|
||||
|
|
49
tus/tus.go
49
tus/tus.go
|
@ -14,7 +14,6 @@ import (
|
|||
"github.com/tus/tusd/pkg/filestore"
|
||||
tusd "github.com/tus/tusd/pkg/handler"
|
||||
"github.com/tus/tusd/pkg/memorylocker"
|
||||
"go.uber.org/zap"
|
||||
"io"
|
||||
"log"
|
||||
)
|
||||
|
@ -35,7 +34,6 @@ func Init() *tusd.Handler {
|
|||
composer.UseConcater(store)
|
||||
composer.UseLocker(memorylocker.New())
|
||||
composer.UseTerminater(store)
|
||||
shared.SetTusComposer(composer)
|
||||
|
||||
handler, err := tusd.NewHandler(tusd.Config{
|
||||
BasePath: "/api/v1" + TUS_API_PATH,
|
||||
|
@ -44,9 +42,7 @@ func Init() *tusd.Handler {
|
|||
hash := hook.Upload.MetaData[HASH_META_HEADER]
|
||||
|
||||
if len(hash) == 0 {
|
||||
msg := "missing blake3-hash metadata"
|
||||
shared.GetLogger().Debug(msg)
|
||||
return errors.New(msg)
|
||||
return errors.New("missing blake3-hash metadata")
|
||||
}
|
||||
|
||||
var upload model.Upload
|
||||
|
@ -54,21 +50,18 @@ func Init() *tusd.Handler {
|
|||
if (result.Error != nil && result.Error.Error() != "record not found") || result.RowsAffected > 0 {
|
||||
hashBytes, err := hex.DecodeString(hash)
|
||||
if err != nil {
|
||||
shared.GetLogger().Debug("invalid hash", zap.Error(err))
|
||||
return err
|
||||
}
|
||||
|
||||
cidString, err := cid.Encode(hashBytes, uint64(hook.Upload.Size))
|
||||
|
||||
if err != nil {
|
||||
shared.GetLogger().Debug("failed to create cid", zap.Error(err))
|
||||
return err
|
||||
}
|
||||
|
||||
resp, err := json.Marshal(UploadResponse{Cid: cidString})
|
||||
|
||||
if err != nil {
|
||||
shared.GetLogger().Error("failed to create response", zap.Error(err))
|
||||
return err
|
||||
}
|
||||
|
||||
|
@ -84,19 +77,16 @@ func Init() *tusd.Handler {
|
|||
}
|
||||
|
||||
if err := db.Get().Create(tusEntry).Error; err != nil {
|
||||
shared.GetLogger().Error("failed to create tus entry", zap.Error(err))
|
||||
return err
|
||||
}
|
||||
|
||||
if err := shared.GetTusQueue().QueueTask(func(ctx context.Context) error {
|
||||
upload, err := store.GetUpload(nil, hook.Upload.ID)
|
||||
if err != nil {
|
||||
shared.GetLogger().Error("failed to query tus upload", zap.Error(err))
|
||||
return err
|
||||
}
|
||||
return tusWorker(&upload)
|
||||
}); err != nil {
|
||||
shared.GetLogger().Error("failed to queue tus upload", zap.Error(err))
|
||||
return err
|
||||
}
|
||||
|
||||
|
@ -125,7 +115,6 @@ func tusStartup() {
|
|||
if err := tusQueue.QueueTask(func(ctx context.Context) error {
|
||||
upload, err := store.GetUpload(nil, item.UploadID)
|
||||
if err != nil {
|
||||
shared.GetLogger().Error("failed to query tus upload", zap.Error(err))
|
||||
return err
|
||||
}
|
||||
return tusWorker(&upload)
|
||||
|
@ -136,23 +125,20 @@ func tusStartup() {
|
|||
}
|
||||
|
||||
func tusWorker(upload *tusd.Upload) error {
|
||||
info, err := (*upload).GetInfo(context.Background())
|
||||
info, err := (*upload).GetInfo(nil)
|
||||
if err != nil {
|
||||
shared.GetLogger().Error("failed to query tus upload metadata", zap.Error(err))
|
||||
log.Print(err)
|
||||
return err
|
||||
}
|
||||
file, err := (*upload).GetReader(context.Background())
|
||||
file, err := (*upload).GetReader(nil)
|
||||
if err != nil {
|
||||
shared.GetLogger().Error("failed reading upload", zap.Error(err))
|
||||
log.Print(err)
|
||||
return err
|
||||
}
|
||||
|
||||
_, err = files.Upload(file.(io.ReadSeeker), info.Size)
|
||||
if err != nil {
|
||||
err1 := terminateUpload(*upload)
|
||||
if err1 != nil {
|
||||
return err1
|
||||
}
|
||||
log.Print(err)
|
||||
return err
|
||||
}
|
||||
|
||||
|
@ -160,32 +146,17 @@ func tusWorker(upload *tusd.Upload) error {
|
|||
|
||||
var tusUpload model.Tus
|
||||
ret := db.Get().Where(&model.Tus{Hash: hash}).First(&tusUpload)
|
||||
|
||||
if ret.Error != nil && ret.Error.Error() != "record not found" {
|
||||
shared.GetLogger().Error("failed fetching tus entry", zap.Error(err))
|
||||
err1 := terminateUpload(*upload)
|
||||
if err1 != nil {
|
||||
return err1
|
||||
}
|
||||
log.Print(ret.Error)
|
||||
return err
|
||||
}
|
||||
|
||||
_ = db.Get().Delete(&tusUpload)
|
||||
ret = db.Get().Delete(&tusUpload)
|
||||
|
||||
err = terminateUpload(*upload)
|
||||
err = shared.GetTusComposer().Terminater.AsTerminatableUpload(*upload).Terminate(context.Background())
|
||||
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func terminateUpload(upload tusd.Upload) error {
|
||||
err := shared.GetTusComposer().Terminater.AsTerminatableUpload(upload).Terminate(context.Background())
|
||||
|
||||
if err != nil {
|
||||
shared.GetLogger().Error("failed deleting tus upload", zap.Error(err))
|
||||
log.Print(err)
|
||||
return err
|
||||
}
|
||||
|
||||
|
|
Loading…
Reference in New Issue