backblaze-backup/internal/services/backblaze.go

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 {
	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)
		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
}