book-comic-library/backend/internal/scanner/scanner.go

package scanner

import (
	"context"
	"errors"
	"fmt"
	"io"
	"io/fs"
	"log"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"booklib/internal/bookfile"
	"booklib/internal/config"
	"booklib/internal/media"
	"booklib/internal/redispkg"
	"booklib/internal/store"
)

// Sweeper 是 scanner 每轮顺手调用的清理钩子;upload.U 满足它(B16)。
type Sweeper interface {
	Sweep(ctx context.Context) error
}

type Scanner struct {
	st       *store.Store
	cfg      *config.Config
	rdb      *redispkg.R
	sweepers []Sweeper
	// B9-②: per-library single-flight — concurrent scan triggers for the same
	// library are merged into one execution, even without redis.
	flights sync.Map // map[int64]*sync.WaitGroup
}

func New(st *store.Store, cfg *config.Config, rdb *redispkg.R, sweepers ...Sweeper) *Scanner {
	return &Scanner{st: st, cfg: cfg, rdb: rdb, sweepers: sweepers}
}

func (s *Scanner) Run(ctx context.Context) {
	t := time.NewTicker(s.cfg.ScanInterval)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-t.C:
			for _, sw := range s.sweepers { // B16: 上传会话清扫随扫描周期跑
				if err := sw.Sweep(ctx); err != nil {
					log.Printf("scan: sweep: %v", err)
				}
			}
			libs, err := s.st.ListLibraries(ctx)
			if err != nil {
				log.Printf("scan: list libraries: %v", err)
				continue
			}
			for _, l := range libs {
				s.scanOnce(ctx, l)
			}
		}
	}
}

func (s *Scanner) ScanLibraryByID(ctx context.Context, id int64) {
	lib, err := s.st.GetLibrary(ctx, id)
	if err != nil {
		log.Printf("scan: library %d: %v", id, err)
		return
	}
	s.scanOnce(ctx, lib)
}

// scanOnce ensures only one scan per library runs concurrently in this process.
// Concurrent callers block until the in-flight scan completes (B9-②).
func (s *Scanner) scanOnce(ctx context.Context, lib store.Library) {
	wg := &sync.WaitGroup{}
	wg.Add(1)
	if existing, loaded := s.flights.LoadOrStore(lib.ID, wg); loaded {
		existing.(*sync.WaitGroup).Wait()
		return
	}
	defer func() {
		s.flights.Delete(lib.ID)
		wg.Done()
	}()
	s.ScanLibrary(ctx, lib)
}

func (s *Scanner) ScanLibrary(ctx context.Context, lib store.Library) {
	// B9-①: ScanLock auto-renews every TTL/2 during long scans.
	unlock, ok := s.rdb.ScanLock(ctx, fmt.Sprintf("scan:%d", lib.ID), 5*time.Minute)
	if !ok {
		return // 别的副本在扫
	}
	defer unlock()

	root, err := filepath.EvalSymlinks(filepath.Clean(lib.RootPath))
	if err != nil || !bookfile.Contains(s.cfg.BooksDir, root) {
		log.Printf("scan: library %d root %q rejected", lib.ID, lib.RootPath)
		return
	}
	disk, err := walk(root)
	if err != nil {
		log.Printf("scan: walk %s: %v", root, err)
		return
	}
	dbMeta, err := s.st.ListBookMeta(ctx, lib.ID)
	if err != nil {
		log.Printf("scan: list books: %v", err)
		return
	}
	for rel, ds := range disk {
		old, exists := dbMeta[rel]
		delete(dbMeta, rel)
		switch {
		case !exists:
			s.ingest(ctx, lib.ID, 0, root, rel, ds, true)
		case old.Size != ds.size || old.ModTS != ds.modTS:
			s.ingest(ctx, lib.ID, old.ID, root, rel, ds, false)
		}
	}
	for rel := range dbMeta { // 只剩被删的文件
		if err := s.st.DeleteBookByPath(ctx, lib.ID, rel); err != nil {
			log.Printf("scan: delete %s: %v", rel, err)
		}
	}
	s.sweepCache(ctx)
}

type diskStat struct{ size, modTS int64 }

func walk(root string) (map[string]diskStat, error) {
	out := map[string]diskStat{}
	err := filepath.WalkDir(root, func(p string, d fs.DirEntry, err error) error {
		if err != nil {
			log.Printf("scan: walk %s: %v", p, err)
			return nil // 单点失败不中断
		}
		if d.IsDir() {
			return nil
		}
		if bookfile.FormatFromExt(d.Name()) == "" {
			return nil
		}
		info, err := d.Info()
		if err != nil {
			return nil
		}
		rel, err := filepath.Rel(root, p)
		if err != nil {
			return err
		}
		out[filepath.ToSlash(rel)] = diskStat{info.Size(), info.ModTime().Unix()}
		return nil
	})
	return out, err
}

func titleOf(rel string) string {
	base := filepath.Base(rel)
	return strings.TrimSpace(strings.ReplaceAll(strings.TrimSuffix(base, filepath.Ext(base)), "_", " "))
}

// ingest 是 add/update 的合一实现(Task 24):isNew 决定走 Insert 还是 UpdateBookFile,
// 之后的错误处理与封面生成完全共享。bookID 仅在 isNew=false 时有意义。
func (s *Scanner) ingest(ctx context.Context, libID, bookID int64, root, rel string, ds diskStat, isNew bool) {
	format := bookfile.FormatFromExt(filepath.Base(rel))
	pageCount := 0
	var idxErr error
	if format == "cbz" {
		idx, err := s.zipIndex(root, rel)
		pageCount = len(idx)
		idxErr = err
		if idxErr == nil && pageCount == 0 {
			idxErr = errors.New("no images in archive")
		}
	}
	if isNew {
		id, err := s.st.InsertBook(ctx, libID, rel, titleOf(rel), format, ds.size, ds.modTS, pageCount)
		if err != nil {
			log.Printf("scan: insert %s: %v", rel, err)
			return
		}
		bookID = id
	} else if err := s.st.UpdateBookFile(ctx, bookID, ds.size, ds.modTS, pageCount); err != nil {
		log.Printf("scan: update %s: %v", rel, err)
		return
	}
	if idxErr != nil {
		// B10: log SetBookState errors instead of discarding.
		if e := s.st.SetBookState(ctx, bookID, "error", idxErr.Error()); e != nil {
			log.Printf("scan: SetBookState %s: %v", rel, e)
		}
		return
	}
	s.cover(ctx, bookID, root, rel, format, ds)
}

func (s *Scanner) zipIndex(root, rel string) ([]string, error) {
	f, size, err := bookfile.OpenReaderAt(root, rel)
	if err != nil {
		return nil, err
	}
	defer f.Close()
	return bookfile.PageIndex(f, size)
}

// cover writes the cover image to the cache dir via media.WriteAtomic.
// B11: write failures are logged; orphan .tmp files are cleaned only on failure.
func (s *Scanner) cover(ctx context.Context, id int64, root, rel, format string, ds diskStat) {
	var fn func(io.ReaderAt, int64) ([]byte, string, error)
	switch format {
	case "cbz":
		fn = bookfile.CBZCover
	case "epub":
		fn = bookfile.EPUBCover
	default:
		return // pdf/txt/md 用占位 SVG,不落盘
	}
	f, size, err := bookfile.OpenReaderAt(root, rel)
	if err != nil {
		log.Printf("scan: cover %s: %v", rel, err)
		return
	}
	defer f.Close()
	img, ext, err := fn(f, size)
	if err != nil {
		log.Printf("scan: cover %s: %v", rel, err)
		return
	}
	dir := bookfile.CoverDir(s.cfg.CacheDir, bookfile.DirKey(id, bookfile.Hash(ds.size, ds.modTS)))
	if e := media.WriteAtomic(dir, "cover"+ext, img); e != nil {
		log.Printf("scan: cover write %s: %v", rel, e)
	}
}

func (s *Scanner) sweepCache(ctx context.Context) {
	hashes, err := s.st.BookHashes(ctx)
	if err != nil {
		return
	}
	live := map[string]bool{}
	for id, v := range hashes {
		live[bookfile.DirKey(id, bookfile.Hash(v[0], v[1]))] = true
	}
	if n, err := bookfile.SweepStale(s.cfg.CacheDir, live); err != nil {
		log.Printf("scan: sweep: %v", err)
	} else if n > 0 {
		log.Printf("scan: swept %d stale cache dirs", n)
	}
}