package library import ( "context" "errors" "fmt" "log/slog" "math/rand/v2" "net/http" "os" "path" "path/filepath" "slices" "strings" "sync" "time" "unicode/utf8" "github.com/google/uuid" "pipu/api/internal/auth" "pipu/api/internal/library/meta" "pipu/api/internal/platform/apperr" "pipu/api/internal/storageagent" ) // Коды ошибок библиотеки. const ( CodeSourceNotFound = "library.source_not_found" CodeScanRunning = "library.scan_running" CodeSourceOffline = "library.source_offline" CodePatternInvalid = "validation.pattern_invalid" CodeNFSInvalid = "validation.nfs_invalid" CodeLocalPathInvalid = "validation.local_path_invalid" CodeNameInvalid = "validation.name_invalid" CodeIntervalInvalid = "validation.scan_interval_invalid" CodeNotFound = "catalog.not_found" ) // Config — пути к хранилищам в файловой системе api. type Config struct { LibraryRoot string // точки монтирования NFS: / HostRoot string // LIBRARY_HOST внутри контейнера, для local-источников } // Service управляет хранилищами и сканами. type Service struct { store *Store scanner *Scanner board *ProgressBoard agent Agent prober Prober cfg Config log *slog.Logger mu sync.Mutex // сериализует переподключение одного и того же источника // OnCatalogChanged — каталог изменился не сканом (удалено хранилище): переиндексация поиска. OnCatalogChanged func() } // NewService создаёт сервис. func NewService(store *Store, scanner *Scanner, board *ProgressBoard, agent Agent, prober Prober, cfg Config, log *slog.Logger) *Service { return &Service{store: store, scanner: scanner, board: board, agent: agent, prober: prober, cfg: cfg, log: log} } func forbidden() error { return apperr.Forbidden(auth.CodeForbidden, "not allowed") } func requirePerm(ctx context.Context, perm auth.Permission) error { p, ok := auth.PrincipalFrom(ctx) if !ok { return apperr.Unauthorized(auth.CodeUnauthorized, "not authenticated") } if !auth.Can(p.Role, perm) { return forbidden() } return nil } func notFound(err error) error { if errors.Is(err, ErrNotFound) { return apperr.NotFound(CodeSourceNotFound, "source not found") } return err } // Root — каталог источника в файловой системе api. func (s *Service) Root(src Source) (string, error) { switch src.Type { case SourceLocal: rel := "" if src.LocalPath != nil { rel = *src.LocalPath } return s.hostPath(rel) case SourceNFS: return filepath.Join(s.cfg.LibraryRoot, src.ID.String()), nil } return "", fmt.Errorf("library: unknown source type %q", src.Type) } // hostPath — путь внутри LIBRARY_HOST; выйти за его пределы нельзя. func (s *Service) hostPath(rel string) (string, error) { rel = strings.Trim(strings.ReplaceAll(rel, "\\", "/"), "/") clean := path.Clean("/" + rel) if clean != "/"+rel && rel != "" || strings.Contains(rel, "..") { return "", apperr.Validation("local_path", CodeLocalPathInvalid, "path must be relative without ..", nil) } return filepath.Join(s.cfg.HostRoot, filepath.FromSlash(strings.TrimPrefix(clean, "/"))), nil } // ListSources — хранилища с прогрессом сканов. func (s *Service) ListSources(ctx context.Context) ([]Source, error) { if err := requirePerm(ctx, auth.PermManageLibrary); err != nil { return nil, err } return s.store.ListSources(ctx) } // GetSource — хранилище по id. func (s *Service) GetSource(ctx context.Context, id uuid.UUID) (Source, error) { if err := requirePerm(ctx, auth.PermManageLibrary); err != nil { return Source{}, err } src, err := s.store.GetSource(ctx, id) return src, notFound(err) } // Progress — прогресс скана источника, если он был. func (s *Service) Progress(id uuid.UUID) (Progress, bool) { return s.board.Get(id) } // SourceInput — поля создания / изменения хранилища. nil — не задано. type SourceInput struct { Name *string Type SourceType NFSServer *string NFSExport *string NFSOptions *string LocalPath *string TagsFirst *bool ScanIntervalMinutes *int // 0 — выключить Enabled *bool } func (s *Service) apply(src *Source, in SourceInput) error { if in.Name != nil { name := strings.TrimSpace(*in.Name) if name == "" || utf8.RuneCountInString(name) > 100 { return apperr.Validation("name", CodeNameInvalid, "name must be 1-100 characters", map[string]any{"max": 100}) } src.Name = name } if in.TagsFirst != nil { src.TagsFirst = *in.TagsFirst } if in.ScanIntervalMinutes != nil { switch m := *in.ScanIntervalMinutes; { case m == 0: src.ScanInterval = nil case m < 5 || m > 60*24*30: return apperr.Validation("scan_interval_minutes", CodeIntervalInvalid, "interval must be 5 minutes .. 30 days", map[string]any{"min": 5}) default: d := time.Duration(m) * time.Minute src.ScanInterval = &d } } switch src.Type { case SourceLocal: if in.LocalPath != nil { rel := strings.Trim(strings.ReplaceAll(*in.LocalPath, "\\", "/"), "/") if _, err := s.hostPath(rel); err != nil { return err } src.LocalPath = &rel } if src.LocalPath == nil { empty := "" src.LocalPath = &empty } case SourceNFS: if in.NFSServer != nil { src.NFSServer = in.NFSServer } if in.NFSExport != nil { src.NFSExport = in.NFSExport } if in.NFSOptions != nil { src.NFSOptions = *in.NFSOptions } spec, err := nfsSpec(*src).Normalize(false) if err != nil { var ve *storageagent.ValidationError if errors.As(err, &ve) { return apperr.Validation(ve.Field, CodeNFSInvalid, ve.Reason, nil) } return err } src.NFSServer, src.NFSExport, src.NFSOptions = &spec.Server, &spec.Export, spec.Options default: return apperr.Validation("type", "validation.source_type_invalid", "unknown source type", nil) } return nil } func nfsSpec(src Source) storageagent.NFSSpec { var spec storageagent.NFSSpec if src.NFSServer != nil { spec.Server = *src.NFSServer } if src.NFSExport != nil { spec.Export = *src.NFSExport } spec.Options = src.NFSOptions return spec } // CreateSource сохраняет хранилище и сразу пытается подключить его. func (s *Service) CreateSource(ctx context.Context, in SourceInput) (Source, error) { if err := requirePerm(ctx, auth.PermManageLibrary); err != nil { return Source{}, err } src := Source{ID: newID(), Type: in.Type, TagsFirst: true, Status: "pending", NFSOptions: "ro,nfsvers=4.1,soft,timeo=150,retrans=3"} if in.Name == nil { in.Name = new(string) } if err := s.apply(&src, in); err != nil { return Source{}, err } src, err := s.store.SaveSource(ctx, src) if err != nil { return Source{}, err } return s.connect(ctx, src) } // UpdateSource меняет хранилище; смена адреса или папки переподключает его. func (s *Service) UpdateSource(ctx context.Context, id uuid.UUID, in SourceInput) (Source, error) { if err := requirePerm(ctx, auth.PermManageLibrary); err != nil { return Source{}, err } src, err := s.store.GetSource(ctx, id) if err != nil { return Source{}, notFound(err) } before := src in.Type = src.Type if err := s.apply(&src, in); err != nil { return Source{}, err } if src, err = s.store.SaveSource(ctx, src); err != nil { return Source{}, err } location := func(x Source) string { return fmt.Sprint(x.LocalPath != nil && *x.LocalPath != "", deref(x.LocalPath), deref(x.NFSServer), deref(x.NFSExport), x.NFSOptions) } switch { case in.Enabled != nil && !*in.Enabled: if src.Type == SourceNFS { _ = s.agent.Unmount(ctx, src.ID) } if err := s.store.SetSourceStatus(ctx, src.ID, "disabled", nil); err != nil { return Source{}, err } return s.store.GetSource(ctx, src.ID) case in.Enabled != nil && *in.Enabled, location(before) != location(src), src.Status == "error": return s.connect(ctx, src) } return src, nil } func deref(p *string) string { if p == nil { return "" } return *p } // connect подключает хранилище и записывает статус: local — проверка каталога, nfs — монтирование агентом. func (s *Service) connect(ctx context.Context, src Source) (Source, error) { s.mu.Lock() defer s.mu.Unlock() status, message := s.check(ctx, src, true) if err := s.store.SetSourceStatus(ctx, src.ID, status, message); err != nil { return Source{}, err } return s.store.GetSource(ctx, src.ID) } // check проверяет доступность; mount — смонтировать NFS (иначе только проверить точку). func (s *Service) check(ctx context.Context, src Source, mount bool) (string, *string) { msg := func(m string) *string { return &m } switch src.Type { case SourceLocal: root, err := s.Root(src) if err != nil { return "error", msg(err.Error()) } info, err := os.Stat(root) if err != nil || !info.IsDir() { return "error", msg("folder not found: " + deref(src.LocalPath)) } return "mounted", nil case SourceNFS: if !mount { root, _ := s.Root(src) if _, err := os.Stat(root); err != nil { return "error", msg(err.Error()) } return "mounted", nil } res, err := s.agent.Mount(ctx, src.ID, nfsSpec(src)) if err != nil { return "error", msg(err.Error()) } if res.Status != "mounted" { return "error", msg(res.Message) } return "mounted", nil } return "error", msg("unknown type") } // DeleteSource отключает хранилище и удаляет его файлы из каталога. func (s *Service) DeleteSource(ctx context.Context, id uuid.UUID) error { if err := requirePerm(ctx, auth.PermManageLibrary); err != nil { return err } src, err := s.store.GetSource(ctx, id) if err != nil { return notFound(err) } if p, ok := s.board.Get(id); ok && p.Running { return apperr.Conflict(CodeScanRunning, "scan is running") } if src.Type == SourceNFS { if err := s.agent.Unmount(ctx, id); err != nil { s.log.WarnContext(ctx, "unmount failed", "source_id", id, "err", err) } } if err := s.store.DeleteSource(ctx, id); err != nil { return notFound(err) } if err := s.store.CleanupOrphans(ctx); err != nil { return err } if s.OnCatalogChanged != nil { s.OnCatalogChanged() } return nil } // TestResult — итог проверки подключения. type TestResult struct { OK bool Message string AudioFiles int } // TestSource проверяет доступ к хранилищу и считает аудиофайлы (до 1000). func (s *Service) TestSource(ctx context.Context, id uuid.UUID) (TestResult, error) { if err := requirePerm(ctx, auth.PermManageLibrary); err != nil { return TestResult{}, err } src, err := s.store.GetSource(ctx, id) if err != nil { return TestResult{}, notFound(err) } if src.Type == SourceNFS { res, err := s.agent.Test(ctx, nfsSpec(src)) if err != nil { return TestResult{Message: err.Error()}, nil } return TestResult{OK: res.Status == "mounted", Message: res.Message, AudioFiles: res.AudioFiles}, nil } root, err := s.Root(src) if err != nil { return TestResult{Message: err.Error()}, nil } n := 0 err = filepath.WalkDir(root, func(p string, d os.DirEntry, err error) error { if err != nil { return err } if !d.IsDir() && IsAudio(d.Name()) { n++ if n >= 1000 { return filepath.SkipAll } } return nil }) if err != nil { return TestResult{Message: err.Error(), AudioFiles: n}, nil } return TestResult{OK: true, Message: fmt.Sprintf("found %d audio files", n), AudioFiles: n}, nil } // Patterns — шаблоны путей источника. func (s *Service) Patterns(ctx context.Context, id uuid.UUID) ([]PatternRow, error) { if err := requirePerm(ctx, auth.PermManageLibrary); err != nil { return nil, err } if _, err := s.store.GetSource(ctx, id); err != nil { return nil, notFound(err) } return s.store.Patterns(ctx, id) } func compile(patterns []PatternRow) ([]meta.Pattern, error) { var exprs []string idx := []int{} for i, p := range patterns { if p.IsEnabled { exprs = append(exprs, p.Regex) idx = append(idx, i) } } compiled, err := meta.CompilePatterns(exprs) var pe *meta.PatternError if errors.As(err, &pe) { at := idx[pe.Index] return nil, &apperr.Error{ Status: http.StatusUnprocessableEntity, Code: CodePatternInvalid, Detail: pe.Reason, Params: map[string]any{"index": at, "reason": pe.Reason}, Fields: []apperr.FieldError{{Field: fmt.Sprintf("patterns[%d].regex", at), Code: CodePatternInvalid}}, } } return compiled, err } // PutPatterns проверяет и сохраняет шаблоны. func (s *Service) PutPatterns(ctx context.Context, id uuid.UUID, patterns []PatternRow) ([]PatternRow, error) { if err := requirePerm(ctx, auth.PermManageLibrary); err != nil { return nil, err } if _, err := s.store.GetSource(ctx, id); err != nil { return nil, notFound(err) } if _, err := compile(patterns); err != nil { return nil, err } if err := s.store.ReplacePatterns(ctx, id, patterns); err != nil { return nil, err } return s.store.Patterns(ctx, id) } // PreviewItem — разбор одного файла. type PreviewItem struct { RelPath string PatternIndex *int Skipped bool PathFields map[string]string Tags map[string]string Result meta.Meta Error string } // Preview применяет шаблоны к выборке файлов без сохранения: из каталога, а до первого скана — из обхода. func (s *Service) Preview(ctx context.Context, id uuid.UUID, patterns []PatternRow, sample int) ([]PreviewItem, int, error) { if err := requirePerm(ctx, auth.PermManageLibrary); err != nil { return nil, 0, err } src, err := s.store.GetSource(ctx, id) if err != nil { return nil, 0, notFound(err) } compiled, err := compile(patterns) if err != nil { return nil, 0, err } sample = min(max(sample, 1), 200) root, err := s.Root(src) if err != nil { return nil, 0, err } paths, err := s.store.SampleFiles(ctx, id, sample) if err != nil { return nil, 0, err } if len(paths) == 0 { found, err := walk(root, func(string) bool { return false }) if err != nil { return nil, 0, apperr.New(http.StatusConflict, CodeSourceOffline, err.Error()) } rand.Shuffle(len(found), func(i, j int) { found[i], found[j] = found[j], found[i] }) for _, w := range found[:min(sample, len(found))] { paths = append(paths, w.rel) } } slices.Sort(paths) items := make([]PreviewItem, len(paths)) var wg sync.WaitGroup sem := make(chan struct{}, 8) for i, rel := range paths { wg.Add(1) go func() { defer wg.Done() sem <- struct{}{} defer func() { <-sem }() items[i] = s.previewOne(ctx, root, rel, compiled, src.TagsFirst) }() } wg.Wait() unmatched := 0 for _, it := range items { if it.PatternIndex == nil { unmatched++ } } return items, unmatched, nil } func (s *Service) previewOne(ctx context.Context, root, rel string, patterns []meta.Pattern, tagsFirst bool) PreviewItem { it := PreviewItem{RelPath: rel, PathFields: map[string]string{}, Tags: map[string]string{}} m := meta.MatchPath(patterns, rel) if m.Index >= 0 { idx := m.Index it.PatternIndex = &idx } it.Skipped = m.Skip for k, v := range m.Fields { it.PathFields[k] = v } p, err := s.prober.Probe(ctx, filepath.Join(root, filepath.FromSlash(rel))) if err != nil { it.Error = err.Error() } for k, v := range map[string]string{ "artist": p.Tags.Artist, "albumartist": p.Tags.AlbumArtist, "album": p.Tags.Album, "title": p.Tags.Title, "genre": p.Tags.Genre, "year": p.Tags.Year, "track": p.Tags.Track, "disc": p.Tags.Disc, } { if v != "" { it.Tags[k] = v } } it.Result = meta.Merge(p.Tags, meta.FromPath(m.Fields), tagsFirst, rel) if it.Result.Album == "" { it.Result.Album = UnknownAlbum } return it } // StartScan запускает скан источника в фоне. func (s *Service) StartScan(ctx context.Context, id uuid.UUID, full bool) (Progress, error) { if err := requirePerm(ctx, auth.PermManageLibrary); err != nil { return Progress{}, err } src, err := s.store.GetSource(ctx, id) if err != nil { return Progress{}, notFound(err) } if err := s.startScan(src, full); err != nil { return Progress{}, err } p, _ := s.board.Get(id) return p, nil } func (s *Service) startScan(src Source, full bool) error { if src.Status != "mounted" { return apperr.New(http.StatusConflict, CodeSourceOffline, "source is not connected") } root, err := s.Root(src) if err != nil { return err } if err := s.scanner.Start(src, root, full); errors.Is(err, ErrScanRunning) { return apperr.Conflict(CodeScanRunning, "scan is already running") } else if err != nil { return err } return nil } // Status — состояние библиотеки (admin, moderator). type Status struct { Counts Counts Sources []Source } // Status собирает счётчики и состояние источников. func (s *Service) Status(ctx context.Context) (Status, error) { if err := requirePerm(ctx, auth.PermViewLibraryStatus); err != nil { return Status{}, err } c, err := s.store.Counts(ctx) if err != nil { return Status{}, err } sources, err := s.store.ListSources(ctx) if err != nil { return Status{}, err } return Status{Counts: c, Sources: sources}, nil } // FileErrors — файлы с ошибками чтения. func (s *Service) FileErrors(ctx context.Context, after string, limit int) ([]FileError, error) { if err := requirePerm(ctx, auth.PermManageLibrary); err != nil { return nil, err } return s.store.FileErrors(ctx, after, limit) } // HostDirs — подпапки каталога LIBRARY_HOST для выбора local-хранилища. func (s *Service) HostDirs(ctx context.Context, rel string) (string, []string, error) { if err := requirePerm(ctx, auth.PermManageLibrary); err != nil { return "", nil, err } dir, err := s.hostPath(rel) if err != nil { return "", nil, err } entries, err := os.ReadDir(dir) if err != nil { return "", nil, apperr.Validation("path", CodeLocalPathInvalid, "folder not found", nil) } dirs := []string{} for _, e := range entries { if e.IsDir() && !strings.HasPrefix(e.Name(), ".") { dirs = append(dirs, e.Name()) } } slices.Sort(dirs) return strings.Trim(strings.ReplaceAll(rel, "\\", "/"), "/"), dirs, nil } // Restore при старте api переподключает NFS-хранилища и проверяет local (после перезапуска агента). func (s *Service) Restore(ctx context.Context) { sources, err := s.store.ListSources(ctx) if err != nil { s.log.Error("restore sources", "err", err) return } for _, src := range sources { if src.Status == "disabled" { continue } if _, err := s.connect(ctx, src); err != nil { s.log.Error("restore source", "source_id", src.ID, "err", err) } } } // RunSchedule раз в минуту проверяет доступность хранилищ и запускает сканы по расписанию. func (s *Service) RunSchedule(ctx context.Context) { t := time.NewTicker(time.Minute) defer t.Stop() for { select { case <-ctx.Done(): return case <-t.C: } sources, err := s.store.ListSources(ctx) if err != nil { s.log.Warn("schedule: list sources", "err", err) continue } for _, src := range sources { if src.Status == "disabled" { continue } s.healthCheck(ctx, src) if src.ScanInterval == nil || src.Status != "mounted" { continue } if src.LastScanAt == nil || time.Since(*src.LastScanAt) >= *src.ScanInterval { if err := s.startScan(src, false); err != nil && !errors.Is(err, apperr.Conflict(CodeScanRunning, "")) { s.log.Warn("scheduled scan", "source_id", src.ID, "err", err) } } } } } // healthCheck: недоступное хранилище — status=error (треки серые), вернулось — снова mounted. func (s *Service) healthCheck(ctx context.Context, src Source) { ctx, cancel := context.WithTimeout(ctx, 10*time.Second) defer cancel() done := make(chan struct{}) var status string var message *string go func() { status, message = s.check(ctx, src, false) close(done) }() select { case <-done: case <-ctx.Done(): m := "storage does not respond" status, message = "error", &m } if status == "error" && src.Type == SourceNFS && src.Status == "error" { // Попробовать перемонтировать, как при старте. if _, err := s.connect(ctx, src); err == nil { return } } if status != src.Status || deref(message) != deref(src.StatusMessage) { if err := s.store.SetSourceStatus(ctx, src.ID, status, message); err != nil { s.log.Warn("health check", "source_id", src.ID, "err", err) } } }