package upload import ( "context" "crypto/sha256" "errors" "fmt" "git.bvbej.com/bvbej/base-golang/pkg/color" "git.bvbej.com/bvbej/base-golang/pkg/token" "github.com/rs/cors" "github.com/tus/tusd/pkg/filestore" tus "github.com/tus/tusd/pkg/handler" "go.uber.org/zap" "io" "net/http" "os" "strings" "sync" "time" ) var _ Server = (*server)(nil) type Server interface { GetUploadToken(string, string, time.Duration) string GetFileInfo(id string) (*tus.FileInfo, error) Start(func(string, string, tus.FileInfo)) error Stop() error } type server struct { headerTokenKey string uploading sync.Map config Config token token.Token store filestore.FileStore logger *zap.Logger httpServer *http.Server completedEvent func(user, param string, info tus.FileInfo) } type Config struct { ListenAddr string Path string Dir string Secret string DisableDownload bool Debug bool } func New(conf Config, logger *zap.Logger) Server { return &server{ config: conf, uploading: sync.Map{}, headerTokenKey: "Authorization", logger: logger, token: token.New(conf.Secret), } } func (s *server) GetUploadToken(user, param string, ttl time.Duration) string { sign, _ := s.token.JwtSign(user, param, ttl) return sign } func (s *server) GetFileInfo(id string) (*tus.FileInfo, error) { upload, err := s.store.GetUpload(context.Background(), id) if err != nil { return nil, err } info, err := upload.GetInfo(context.Background()) if err != nil { return nil, err } return &info, nil } func (s *server) Start(completedEvent func(user, param string, info tus.FileInfo)) error { s.completedEvent = completedEvent composer := tus.NewStoreComposer() if err := os.MkdirAll(s.config.Dir, os.ModePerm); err != nil { return err } s.store = filestore.New(s.config.Dir) s.store.UseIn(composer) handler, err := tus.NewHandler(tus.Config{ StoreComposer: composer, BasePath: s.config.Path, Logger: zap.NewStdLog(s.logger), NotifyCompleteUploads: true, DisableTermination: true, DisableDownload: s.config.DisableDownload, RespectForwardedHeaders: strings.Contains(s.config.ListenAddr, "127.0.0.1"), PreUploadCreateCallback: func(hook tus.HookEvent) error { authStr := hook.HTTPRequest.Header.Get(s.headerTokenKey) jwtParse, err := s.token.JwtParse(authStr) if err == nil { _, ok := s.uploading.Load(jwtParse.ID) if !ok { s.uploading.Store(jwtParse.ID, time.Now()) return nil } } return errors.New("unauthorized") }, PreFinishResponseCallback: func(hook tus.HookEvent) error { authStr := hook.HTTPRequest.Header.Get(s.headerTokenKey) jwtParse, err := s.token.JwtParse(authStr) if err != nil { return errors.New("token expired") } _, ok := s.uploading.Load(jwtParse.ID) if ok { s.uploading.Delete(jwtParse.ID) } upload, err := s.store.GetUpload(context.Background(), hook.Upload.ID) if err != nil { return err } reader, err := upload.GetReader(context.Background()) if err != nil { return err } all, err := io.ReadAll(reader) if err != nil { return err } _ = reader.(io.Closer).Close() hash := sha256.New() hash.Write(all) sha256Byte := hash.Sum(nil) sha256String := fmt.Sprintf("%x", sha256Byte) if !s.config.Debug && sha256String != strings.ToLower(jwtParse.ID) { info, _ := upload.GetInfo(context.Background()) path, exist := info.Storage["Path"] if exist { _ = os.Remove(path) _ = os.Remove(path + ".info") } return errors.New("file check error") } return nil }, }) if err != nil { return err } go func() { for { event := <-handler.CompleteUploads authStr := event.HTTPRequest.Header.Get(s.headerTokenKey) jwtParse, _ := s.token.JwtParse(authStr) if s.completedEvent != nil { go func() { s.completedEvent(jwtParse.ID, jwtParse.Subject, event.Upload) }() } } }() //监听服务 addr := s.config.ListenAddr mux := http.NewServeMux() mux.Handle(s.config.Path, http.StripPrefix(s.config.Path, handler)) s.httpServer = &http.Server{ Addr: addr, Handler: cors.AllowAll().Handler(mux), } go func() { if err = s.httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed { s.logger.Sugar().Fatal("upload server startup err", zap.Error(err)) } }() fmt.Println(color.Green(fmt.Sprintf("* [register tusd listen %s]", addr))) return nil } func (s *server) Stop() error { return s.httpServer.Close() }