package services import ( "context" "errors" "fmt" "io" "io/fs" "log" "os" "path/filepath" "runtime" "strings" "sync" "github.com/kurin/blazer/b2" "golang.org/x/sync/semaphore" ) const writers = 10 const maxConcurrentWeight = 4 const largeFileSize = 500 * 1024 * 1024 // 500 MB type BackBalze struct { bucketName string dir string filePath string maxWorkers int bbID string bbKey string } func NewBackBlaze(bbID, bbKey string) *BackBalze { log.Println("runtime.NumCPU()", runtime.NumCPU()) return &BackBalze{ bbID: bbID, bbKey: bbKey, maxWorkers: runtime.NumCPU(), } } func (b *BackBalze) WithBucket(bucketName string) *BackBalze { b.bucketName = bucketName return b } func (b *BackBalze) WithDir(dir string) *BackBalze { b.dir = dir return b } func (b *BackBalze) WithFile(filePath string) *BackBalze { b.filePath = filePath return b } func (b *BackBalze) Sync() error { if b.bucketName == "" && (b.filePath == "" || b.dir == "") { return fmt.Errorf("bucket name is %v | filePath is %v | dir is %v", b.bucketName, b.filePath, b.dir) } if b.filePath != "" && b.dir != "" { return errors.New("you must select just 1 option, dir or file") } ctx := context.Background() b2Client, err := b2.NewClient(ctx, b.bbID, b.bbKey) if err != nil { return fmt.Errorf("b2.NewClient %w", err) } log.Println("b2Client ok") bc, err := b2Client.Bucket(ctx, b.bucketName) if err != nil { return fmt.Errorf("b2Client.Bucket %w", err) } if bc == nil { return fmt.Errorf("bucket doesn't exist %s", b.bucketName) } log.Println("bucket found:", bc.Name()) if b.filePath != "" { log.Println("file:", b.filePath) if err := copyFile(ctx, bc, b.filePath); err != nil { return fmt.Errorf("copyFile %w", err) } return nil } if b.dir != "" { oldFiles, err := bucketFiles(ctx, bc) if err != nil { return fmt.Errorf("bucketFiles %w", err) } log.Println(strings.Repeat("*", 40)) log.Println("oldFiles to clean:\n\t\t" + strings.Join(oldFiles, "\n\t\t")) log.Println(strings.Repeat("*", 40)) fileChan := make(chan string) var wg sync.WaitGroup for i := 0; i < b.maxWorkers; i++ { wg.Add(1) go func() { defer wg.Done() for src := range fileChan { log.Println("start copying file", src) if err := copyFile(ctx, bc, src); err != nil { log.Printf("error copying file %s: %v\n", src, err) } } }() } // Walk the directory and send files to the channel for uploading err = filepath.WalkDir(b.dir, func(path string, d fs.DirEntry, err error) error { if err != nil { return err } if !d.IsDir() { fileChan <- path } return nil }) if err != nil { return fmt.Errorf("error walking the directory: %v", err) } // Close the channel (no more files to send) close(fileChan) wg.Wait() // Cleanup old files after backup is completed if err := cleanBucket(ctx, bc, oldFiles); err != nil { return fmt.Errorf("cleanBucket %w", err) } } log.Println("copied successfully") return nil } func (b *BackBalze) OldSync() error { if b.bucketName == "" && (b.filePath == "" || b.dir == "") { return fmt.Errorf("bucket name is %v | filePath is %v | dir is %v", b.bucketName, b.filePath, b.dir) } if b.filePath != "" && b.dir != "" { return errors.New("you must select just 1 option, dir or file") } ctx := context.Background() b2Client, err := b2.NewClient(ctx, b.bbID, b.bbKey) if err != nil { return fmt.Errorf("b2.NewClient %w", err) } log.Println("b2Client ok") bc, err := b2Client.Bucket(ctx, b.bucketName) if err != nil { return fmt.Errorf("b2Client.Bucket %w", err) } if bc == nil { return fmt.Errorf("bucket doesn't exist %s", b.bucketName) } log.Println("bucket found:", bc.Name()) if b.filePath != "" { log.Println("file:", b.filePath) if err := copyFile(ctx, bc, b.filePath); err != nil { return fmt.Errorf("copyFile %w", err) } return nil } if b.dir != "" { oldFiles, err := bucketFiles(ctx, bc) if err != nil { return fmt.Errorf("bucketFiles %w", err) } log.Println(strings.Repeat("*", 40)) log.Println("oldFiles to clean:\n\t\t" + strings.Join(oldFiles, "\n\t\t")) log.Println(strings.Repeat("*", 40)) fileChan := make(chan string) uploadSem := semaphore.NewWeighted(maxConcurrentWeight) var wg sync.WaitGroup for i := 0; i < b.maxWorkers; i++ { wg.Add(1) go func() { defer wg.Done() for src := range fileChan { info, err := os.Stat(src) if err != nil { log.Printf("error getting file info %s: %v\n", src, err) continue } weight := int64(1) if info.Size() > largeFileSize { weight = 2 } if err := uploadSem.Acquire(ctx, weight); err == nil { log.Println("start copying file", src) if err := copyFile(ctx, bc, src); err != nil { log.Printf("error copying file %s: %v\n", src, err) } uploadSem.Release(weight) } else { log.Printf("error acquiring semaphore: %v\n", err) } } }() } // Walk the directory and send files to the channel for uploading err = filepath.WalkDir(b.dir, func(path string, d fs.DirEntry, err error) error { if err != nil { return err } if !d.IsDir() { fileChan <- path } return nil }) if err != nil { return fmt.Errorf("error walking the directory: %v", err) } // Close the channel (no more files to send) close(fileChan) wg.Wait() // Cleanup old files after backup is completed if err := cleanBucket(ctx, bc, oldFiles); err != nil { return fmt.Errorf("cleanBucket %w", err) } } log.Println("copied successfully") return nil } func copyFile(ctx context.Context, bucket *b2.Bucket, src string) error { f, err := os.Open(src) if err != nil { return err } defer f.Close() fi, err := f.Stat() if err != nil { return err } w := bucket.Object(fi.Name()).NewWriter(ctx) w.ConcurrentUploads = writers if _, err := io.Copy(w, f); err != nil { w.Close() return err } return w.Close() } func cleanBucket(ctx context.Context, bucket *b2.Bucket, files []string) error { var errorBuilder strings.Builder for _, v := range files { obj := bucket.Object(v) if obj == nil { log.Println("object is nil", v) continue } if err := obj.Delete(ctx); err != nil { errorBuilder.WriteString(fmt.Errorf("error deleting %s : %w", v, err).Error()) errorBuilder.WriteString("; ") } } if errorBuilder.Len() > 0 { return errors.New(errorBuilder.String()) } return nil } func bucketFiles(ctx context.Context, bucket *b2.Bucket) ([]string, error) { bucketIter := bucket.List(ctx) if bucketIter == nil { return nil, fmt.Errorf("bucket list cannot be nil") } var files []string for { if !bucketIter.Next() { if bucketIter.Err() != nil { return nil, fmt.Errorf("bucketIter err %w", bucketIter.Err()) } break } if bucketIter.Object() == nil { log.Println("bucketIter Object is nil") continue } files = append(files, bucketIter.Object().Name()) } return files, nil }