jukebox_maker/internal/copier/copier.go

package copier

import (
	"context"
	"errors"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"sync"

	"jukebox_maker/internal/config"
	"jukebox_maker/internal/db"
	"jukebox_maker/internal/disk"
	"jukebox_maker/internal/task"
)

type Options struct {
	DiskID         string
	MountPath      string
	MediaPath      string
	EnabledSources []string
	ReserveFreeGB  float64
	OverwriteMode  config.OverwriteMode
	FileSelectMode config.FileSelectMode
}

type Copier struct {
	tasks *task.Store

	mu     sync.Mutex
	cancel context.CancelFunc

	dbMu sync.RWMutex
	db   *db.DB
}

func New(tasks *task.Store) *Copier {
	return &Copier{tasks: tasks}
}

// SetDB replaces the active disk database (called when a disk connects or disconnects).
func (c *Copier) SetDB(d *db.DB) {
	c.dbMu.Lock()
	c.db = d
	c.dbMu.Unlock()
}

func (c *Copier) getDB() *db.DB {
	c.dbMu.RLock()
	defer c.dbMu.RUnlock()
	return c.db
}

func (c *Copier) Start(ctx context.Context, opts Options) (string, error) {
	c.mu.Lock()
	defer c.mu.Unlock()

	if _, active := c.tasks.ActiveTask(); active {
		return "", errors.New("copy already running")
	}

	database := c.getDB()
	if database == nil {
		return "", errors.New("no disk database available")
	}

	t := c.tasks.Create("copy")
	copyCtx, cancel := context.WithCancel(ctx)
	c.cancel = cancel

	go c.run(copyCtx, t.ID, opts, database)
	return t.ID, nil
}

func (c *Copier) Cancel() {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.cancel != nil {
		c.cancel()
	}
}

func (c *Copier) run(ctx context.Context, taskID string, opts Options, database *db.DB) {
	setStatus := func(s task.Status, msg string, prog int) {
		c.tasks.Update(taskID, func(t *task.Task) {
			t.Status = s
			t.Message = msg
			t.Progress = prog
		})
	}
	fail := func(err error) {
		c.tasks.Update(taskID, func(t *task.Task) {
			t.Status = task.StatusFailed
			t.Error = err.Error()
		})
	}

	setStatus(task.StatusRunning, "Подготовка…", 0)

	if opts.OverwriteMode == config.OverwriteDelete {
		setStatus(task.StatusRunning, "Удаление данных с диска…", 0)
		if err := deleteOurData(opts.MountPath); err != nil {
			fail(err)
			return
		}
	}

	var copiedPaths map[string]struct{}
	if opts.FileSelectMode == config.SelectNew {
		setStatus(task.StatusRunning, "Загрузка истории…", 0)
		var err error
		copiedPaths, err = database.CopiedPaths(opts.DiskID)
		if err != nil {
			fail(err)
			return
		}
	}

	setStatus(task.StatusRunning, "Сканирование источников…", 0)
	files, err := buildFileList(opts.MediaPath, opts.EnabledSources, copiedPaths)
	if err != nil {
		fail(err)
		return
	}
	if len(files) == 0 {
		setStatus(task.StatusSuccess, "Нет новых файлов для копирования.", 100)
		return
	}

	_, free, err := disk.DiskUsage(opts.MountPath)
	if err != nil {
		fail(err)
		return
	}
	reserveBytes := int64(opts.ReserveFreeGB * 1e9)
	available := free - reserveBytes
	if available <= 0 {
		setStatus(task.StatusSuccess, "Недостаточно свободного места на диске.", 100)
		return
	}

	total := len(files)
	copied := 0
	for i, f := range files {
		select {
		case <-ctx.Done():
			c.tasks.Update(taskID, func(t *task.Task) {
				t.Status = task.StatusCanceled
				t.Message = "Отменено"
			})
			return
		default:
		}

		if f.size > available {
			continue
		}

		msg := fmt.Sprintf("Копирование %s (%d/%d)", filepath.Base(f.srcAbs), i+1, total)
		prog := int(float64(i+1) / float64(total) * 100)
		setStatus(task.StatusRunning, msg, prog)

		dstAbs := filepath.Join(opts.MountPath, f.relPath)
		if err := copyFile(ctx, f.srcAbs, dstAbs); err != nil {
			if errors.Is(err, context.Canceled) {
				c.tasks.Update(taskID, func(t *task.Task) {
					t.Status = task.StatusCanceled
					t.Message = "Отменено"
				})
				return
			}
			continue
		}

		available -= f.size
		copied++
		_ = database.RecordCopy(db.CopyRecord{
			DiskID:     opts.DiskID,
			SourcePath: f.relPath,
			FileSize:   f.size,
		})
	}

	setStatus(task.StatusSuccess, fmt.Sprintf("Готово. Скопировано файлов: %d.", copied), 100)
}

type fileEntry struct {
	srcAbs  string
	relPath string
	size    int64
}

func buildFileList(mediaPath string, sources []string, skip map[string]struct{}) ([]fileEntry, error) {
	var result []fileEntry
	for _, src := range sources {
		dir := filepath.Join(mediaPath, src)
		err := filepath.WalkDir(dir, func(path string, d os.DirEntry, err error) error {
			if err != nil || d.IsDir() {
				return nil
			}
			rel, _ := filepath.Rel(mediaPath, path)
			if _, skipped := skip[rel]; skipped {
				return nil
			}
			info, err := d.Info()
			if err != nil {
				return nil
			}
			result = append(result, fileEntry{srcAbs: path, relPath: rel, size: info.Size()})
			return nil
		})
		if err != nil {
			return nil, err
		}
	}
	return result, nil
}

func deleteOurData(mountPath string) error {
	entries, err := os.ReadDir(mountPath)
	if err != nil {
		return err
	}
	for _, e := range entries {
		if e.Name() == ".jukebox" {
			continue
		}
		if err := os.RemoveAll(filepath.Join(mountPath, e.Name())); err != nil {
			return err
		}
	}
	return nil
}

func copyFile(ctx context.Context, src, dst string) error {
	if err := os.MkdirAll(filepath.Dir(dst), 0o755); err != nil {
		return err
	}
	in, err := os.Open(src)
	if err != nil {
		return err
	}
	defer in.Close()

	tmp := dst + ".juketmp"
	out, err := os.Create(tmp)
	if err != nil {
		return err
	}

	buf := make([]byte, 512*1024)
	for {
		select {
		case <-ctx.Done():
			out.Close()
			os.Remove(tmp)
			return ctx.Err()
		default:
		}
		n, readErr := in.Read(buf)
		if n > 0 {
			if _, werr := out.Write(buf[:n]); werr != nil {
				out.Close()
				os.Remove(tmp)
				return werr
			}
		}
		if errors.Is(readErr, io.EOF) {
			break
		}
		if readErr != nil {
			out.Close()
			os.Remove(tmp)
			return readErr
		}
	}
	out.Close()
	return os.Rename(tmp, dst)
}