fix(library): expose scan activity and failed analysis file details
ci / api / api (push) Successful in 55s
ci / contract / contract (сгенерированный код совпадает с openapi.yaml) (push) Successful in 1m6s
ci / db / db (push) Successful in 52s
ci / mobile / mobile (Metro-бандл Android) (push) Successful in 1m38s
release / android (APK в релиз) (push) Successful in 3m50s
ci / ts / ts (packages + web) (push) Successful in 3m20s
release / release (версия и заметки) (push) Successful in 5s
release / desktop (Linux и Windows в релиз) (push) Successful in 44m45s
release / prune (старые файлы и прогоны) (push) Successful in 7s

This commit is contained in:
mrleo1nid committed 2026-10-22 15:25:48 +03:00
1 parent 17b6dd2041
commit 0dc0384c78
20 files changed
+1114 -326

No files matched your search

@@ -0,0 +1,60 @@
package library
import (
"context"
"encoding/base64"
"testing"
"github.com/google/uuid"
"github.com/stretchr/testify/require"
"pipu/api/internal/api/gen"
"pipu/api/internal/auth"
)
type analysisErrorStore struct {
SourceStore
rows []AnalysisError
after uuid.UUID
limit int
}
func (s *analysisErrorStore) AnalysisErrors(_ context.Context, after uuid.UUID, limit int) ([]AnalysisError, error) {
s.after, s.limit = after, limit
return s.rows, nil
}
func TestAnalysisErrorsPermissionsAndPagination(t *testing.T) {
first, second := uuid.New(), uuid.New()
store := &analysisErrorStore{rows: []AnalysisError{{TrackID: first, Title: "One"}, {TrackID: second, Title: "Two"}}}
h := &Handler{svc: &Service{SourceService: &SourceService{store: store}}}
limit := 1
req := gen.ListAnalysisErrorsRequestObject{Params: gen.ListAnalysisErrorsParams{Limit: &limit}}
for _, role := range []auth.Role{auth.RoleUser, auth.RoleModerator} {
_, err := h.ListAnalysisErrors(auth.WithPrincipal(context.Background(), auth.Principal{Role: role}), req)
require.Error(t, err)
require.Zero(t, store.limit, "unauthorized requests must not reach the store")
}
_, err := h.ListAnalysisErrors(context.Background(), req)
require.Error(t, err)
ctx := auth.WithPrincipal(context.Background(), auth.Principal{Role: auth.RoleAdmin})
resp, err := h.ListAnalysisErrors(ctx, req)
require.NoError(t, err)
page := resp.(gen.ListAnalysisErrors200JSONResponse)
require.Len(t, page.Items, 1)
require.Equal(t, first, page.Items[0].TrackId)
require.Nil(t, page.Items[0].RelPath, "a missing file must not hide the failed job")
require.Equal(t, 2, store.limit)
require.NotNil(t, page.NextCursor)
store.rows = store.rows[1:]
req.Params.Cursor = page.NextCursor
resp, err = h.ListAnalysisErrors(ctx, req)
require.NoError(t, err)
require.Equal(t, first, store.after)
require.Nil(t, resp.(gen.ListAnalysisErrors200JSONResponse).NextCursor)
for _, cursor := range []string{"!invalid", base64.RawURLEncoding.EncodeToString([]byte("not-a-uuid"))} {
req.Params.Cursor = &cursor
_, err = h.ListAnalysisErrors(ctx, req)
require.Error(t, err)
}
}
+36
View File
@@ -11,6 +11,8 @@ import (
"os"
"time"
"github.com/google/uuid"
"pipu/api/internal/api/gen"
"pipu/api/internal/library/meta"
"pipu/api/internal/platform/apperr"
@@ -80,7 +82,9 @@ func (h *Handler) sourceDTO(s Source) gen.Source {
}
func progressDTO(p Progress) gen.ScanProgress {
phase := gen.ScanProgressPhase(p.Phase)
out := gen.ScanProgress{
Phase: &phase, WalkedDirs: &p.WalkedDirs, CurrentDir: &p.CurrentDir,
Running: p.Running, Full: p.Full, StartedAt: p.StartedAt, FinishedAt: p.FinishedAt, Found: p.Found,
ToProcess: p.ToProcess, Processed: p.Processed, Added: p.Added, Updated: p.Updated, Moved: p.Moved,
Missing: p.Missing, Errors: p.Errors,
@@ -365,6 +369,38 @@ func (h *Handler) ListLibraryErrors(ctx context.Context, req gen.ListLibraryErro
return out, nil
}
// ListAnalysisErrors implements the paginated analysis diagnostics endpoint.
func (h *Handler) ListAnalysisErrors(ctx context.Context, req gen.ListAnalysisErrorsRequestObject) (gen.ListAnalysisErrorsResponseObject, error) {
limit := 50
if req.Params.Limit != nil {
limit = min(max(*req.Params.Limit, 1), 200)
}
after := uuid.Nil
if req.Params.Cursor != nil {
raw, err := base64.RawURLEncoding.DecodeString(*req.Params.Cursor)
if err == nil {
after, err = uuid.Parse(string(raw))
}
if err != nil {
return nil, apperr.Validation("cursor", "validation.cursor_invalid", "malformed cursor", nil)
}
}
rows, err := h.svc.AnalysisErrors(ctx, after, limit+1)
if err != nil {
return nil, err
}
out := gen.ListAnalysisErrors200JSONResponse{Items: []gen.AnalysisError{}}
if len(rows) > limit {
rows = rows[:limit]
next := base64.RawURLEncoding.EncodeToString([]byte(rows[limit-1].TrackID.String()))
out.NextCursor = &next
}
for _, r := range rows {
out.Items = append(out.Items, gen.AnalysisError{TrackId: r.TrackID, Title: r.Title, SourceId: r.SourceID, SourceName: r.SourceName, RelPath: r.RelPath, Error: r.Error})
}
return out, nil
}
// ListHostDirs реализует gen.StrictServerInterface.
func (h *Handler) ListHostDirs(ctx context.Context, req gen.ListHostDirsRequestObject) (gen.ListHostDirsResponseObject, error) {
rel := ""
+27 -8
View File
@@ -28,6 +28,9 @@ const MissingRetention = 30 * 24 * time.Hour
// Progress — состояние скана одного источника (для админки).
type Progress struct {
Phase string
WalkedDirs int
CurrentDir string
SourceID uuid.UUID
Running bool
Full bool
@@ -82,7 +85,7 @@ func (b *ProgressBoard) start(id uuid.UUID, full bool, now time.Time) bool {
if p, ok := b.m[id]; ok && p.Running {
return false
}
b.m[id] = &Progress{SourceID: id, Running: true, Full: full, StartedAt: now}
b.m[id] = &Progress{SourceID: id, Running: true, Full: full, StartedAt: now, Phase: "walking"}
return true
}
@@ -202,14 +205,18 @@ func (sc *Scanner) run(ctx context.Context, src Source, root string, full bool)
if err != nil {
return err
}
found, err := walk(root, func(rel string) bool { return meta.MatchPath(patterns, rel).Skip })
found, err := walk(root, func(rel string) bool { return meta.MatchPath(patterns, rel).Skip }, func(found, dirs int, currentDir string) {
sc.board.update(src.ID, func(p *Progress) {
p.Found, p.WalkedDirs, p.CurrentDir = found, dirs, currentDir
})
})
if err != nil {
return err
}
if len(found) == 0 && hasPresent(existing) {
return ErrSourceEmpty
}
sc.board.update(src.ID, func(p *Progress) { p.Found = len(found) })
sc.board.update(src.ID, func(p *Progress) { p.Found, p.Phase = len(found), "preparing" })
var (
jobs []job
@@ -256,11 +263,12 @@ func (sc *Scanner) run(ctx context.Context, src Source, root string, full bool)
for _, w := range rest {
jobs = append(jobs, job{walked: w})
}
sc.board.update(src.ID, func(p *Progress) { p.ToProcess, p.Moved = len(jobs), len(moves) })
sc.board.update(src.ID, func(p *Progress) { p.ToProcess, p.Moved, p.Phase = len(jobs), len(moves), "processing" })
if err := sc.process(ctx, src, root, patterns, jobs); err != nil {
return err
}
sc.board.update(src.ID, func(p *Progress) { p.Phase = "finalizing" })
var goneIDs []uuid.UUID
affected := map[uuid.UUID]bool{}
@@ -462,11 +470,12 @@ func (sc *Scanner) apply(ctx context.Context, src Source, patterns []meta.Patter
// walk обходит дерево источника и возвращает аудиофайлы. Скрытые файлы и каталоги пропускаются;
// skip — пути, которые шаблоны велят игнорировать. Ошибки отдельных каталогов не прерывают обход.
func walk(root string, skip func(rel string) bool) ([]walked, error) {
func walk(root string, skip func(rel string) bool, report func(found, dirs int, currentDir string)) ([]walked, error) {
if _, err := os.Stat(root); err != nil {
return nil, fmt.Errorf("library: source root: %w", err)
}
var out []walked
dirs := 0
err := filepath.WalkDir(root, func(p string, d fs.DirEntry, err error) error {
if err != nil {
if p == root {
@@ -481,14 +490,21 @@ func walk(root string, skip func(rel string) bool) ([]walked, error) {
}
return nil
}
if d.IsDir() || !IsAudio(name) {
return nil
}
rel, err := filepath.Rel(root, p)
if err != nil {
return nil
}
rel = filepath.ToSlash(rel)
if d.IsDir() {
dirs++
if report != nil {
report(len(out), dirs, rel)
}
return nil
}
if !IsAudio(name) {
return nil
}
if skip(rel) {
return nil
}
@@ -498,6 +514,9 @@ func walk(root string, skip func(rel string) bool) ([]walked, error) {
}
// PostgreSQL хранит время с точностью до микросекунд.
out = append(out, walked{rel: rel, size: info.Size(), mtime: info.ModTime().UTC().Truncate(time.Microsecond)})
if report != nil {
report(len(out), dirs, filepath.ToSlash(filepath.Dir(rel)))
}
return nil
})
if err != nil {
+34 -2
View File
@@ -55,7 +55,7 @@ func TestWalkAndHash(t *testing.T) {
write("_incoming/new.mp3", 10)
write("big.mp3", 3<<20)
found, err := walk(root, func(rel string) bool { return rel == "_incoming/new.mp3" })
found, err := walk(root, func(rel string) bool { return rel == "_incoming/new.mp3" }, nil)
require.NoError(t, err)
assert.ElementsMatch(t, []string{"Artist/Album/01 - Song.FLAC", "big.mp3"}, rels(found))
for _, w := range found {
@@ -69,10 +69,42 @@ func TestWalkAndHash(t *testing.T) {
require.NoError(t, err)
assert.Equal(t, h1, h2, "hash survives a rename")
_, err = walk(filepath.Join(root, "nope"), func(string) bool { return false })
_, err = walk(filepath.Join(root, "nope"), func(string) bool { return false }, nil)
assert.Error(t, err)
}
func TestWalkReportsIncrementalProgress(t *testing.T) {
root := t.TempDir()
for _, dir := range []string{"a-empty", "b-album", ".hidden"} {
require.NoError(t, os.MkdirAll(filepath.Join(root, dir), 0o755))
}
for _, rel := range []string{"b-album/1.mp3", "b-album/2.flac", "b-album/skip.mp3", ".hidden/3.mp3"} {
require.NoError(t, os.WriteFile(filepath.Join(root, filepath.FromSlash(rel)), []byte("audio"), 0o644))
}
var counts, folders []int
var paths []string
found, err := walk(root, func(rel string) bool { return rel == "b-album/skip.mp3" }, func(n, dirs int, path string) {
counts, folders, paths = append(counts, n), append(folders, dirs), append(paths, path)
})
require.NoError(t, err)
require.Len(t, found, 2)
assert.Equal(t, []int{0, 0, 0, 1, 2}, counts, "reports activity before any file is found and after each accepted audio file")
assert.Equal(t, []int{1, 2, 3, 3, 3}, folders)
assert.Equal(t, []string{".", "a-empty", "b-album", "b-album", "b-album"}, paths)
}
func TestProgressDTOIncludesWalkActivity(t *testing.T) {
p := Progress{Phase: "walking", Running: true, Found: 7, WalkedDirs: 4, CurrentDir: "Artist/Album"}
dto := progressDTO(p)
require.NotNil(t, dto.Phase)
assert.Equal(t, "walking", string(*dto.Phase))
assert.Equal(t, p.Found, dto.Found)
require.NotNil(t, dto.WalkedDirs)
assert.Equal(t, p.WalkedDirs, *dto.WalkedDirs)
require.NotNil(t, dto.CurrentDir)
assert.Equal(t, p.CurrentDir, *dto.CurrentDir)
}
func TestParseFFprobe(t *testing.T) {
raw := []byte(`{
"streams": [
+9
View File
@@ -33,6 +33,7 @@ type SourceStore interface {
AnalysisCounts(context.Context) (AnalysisCounts, error)
WaveMetrics(context.Context) (WaveMetrics, error)
FileErrors(context.Context, string, int) ([]FileError, error)
AnalysisErrors(context.Context, uuid.UUID, int) ([]AnalysisError, error)
}
// SourceService owns source configuration, connectivity and diagnostics.
@@ -440,6 +441,14 @@ func (s *SourceService) FileErrors(ctx context.Context, after string, limit int)
return s.store.FileErrors(ctx, after, limit)
}
// AnalysisErrors lists failed analysis jobs for library administrators.
func (s *SourceService) AnalysisErrors(ctx context.Context, after uuid.UUID, limit int) ([]AnalysisError, error) {
if err := requirePerm(ctx, auth.PermManageLibrary); err != nil {
return nil, err
}
return s.store.AnalysisErrors(ctx, after, limit)
}
// HostDirs — подпапки каталога LIBRARY_HOST для выбора local-хранилища.
func (s *SourceService) HostDirs(ctx context.Context, rel string) (string, []string, error) {
if err := requirePerm(ctx, auth.PermManageLibrary); err != nil {
+34 -3
View File
@@ -511,7 +511,7 @@ func (st *Store) Counts(ctx context.Context) (Counts, error) {
err := st.pool.QueryRow(ctx, `
SELECT (SELECT count(*) FROM artists), (SELECT count(*) FROM albums), (SELECT count(*) FROM tracks),
(SELECT count(*) FROM media_files), (SELECT count(*) FROM media_files WHERE missing_since IS NOT NULL),
(SELECT count(*) FROM media_files WHERE scan_error IS NOT NULL)`).
(SELECT count(*) FROM media_files WHERE scan_error IS NOT NULL AND missing_since IS NULL)`).
Scan(&c.Artists, &c.Albums, &c.Tracks, &c.Files, &c.Missing, &c.Errors)
if err != nil {
return Counts{}, fmt.Errorf("library: counts: %w", err)
@@ -531,7 +531,7 @@ type FileError struct {
func (st *Store) FileErrors(ctx context.Context, after string, limit int) ([]FileError, error) {
rows, err := st.pool.Query(ctx, `
SELECT id, source_id, rel_path, scan_error FROM media_files
WHERE scan_error IS NOT NULL AND rel_path > $1 ORDER BY rel_path LIMIT $2`, after, limit)
WHERE scan_error IS NOT NULL AND missing_since IS NULL AND rel_path > $1 ORDER BY rel_path LIMIT $2`, after, limit)
if err != nil {
return nil, fmt.Errorf("library: file errors: %w", err)
}
@@ -624,7 +624,38 @@ func (st *Store) AnalysisCounts(ctx context.Context) (AnalysisCounts, error) {
return c, nil
}
// RetryFailedAnalysis возвращает неудавшиеся задачи анализа в очередь с нуля попыток.
// AnalysisError is a failed analysis job with a known file path, when available.
type AnalysisError struct {
TrackID uuid.UUID
Title string
SourceID *uuid.UUID
SourceName *string
RelPath *string
Error string
}
// AnalysisErrors includes failed jobs even when their file is no longer available.
func (st *Store) AnalysisErrors(ctx context.Context, after uuid.UUID, limit int) ([]AnalysisError, error) {
rows, err := st.pool.Query(ctx, `
SELECT t.id, t.title, f.source_id, s.name, f.rel_path, COALESCE(a.error, '')
FROM analysis_jobs a JOIN tracks t ON t.id = a.track_id
LEFT JOIN LATERAL (
SELECT mf.source_id, mf.rel_path FROM media_files mf WHERE mf.track_id = t.id
ORDER BY (mf.id = t.primary_file_id) DESC NULLS LAST, (mf.missing_since IS NULL) DESC, mf.id
LIMIT 1
) f ON true
LEFT JOIN library_sources s ON s.id = f.source_id
WHERE a.status = 'failed' AND t.id > $1 ORDER BY t.id LIMIT $2`, after, limit)
if err != nil {
return nil, fmt.Errorf("library: analysis errors: %w", err)
}
return pgx.CollectRows(rows, func(r pgx.CollectableRow) (AnalysisError, error) {
var e AnalysisError
return e, r.Scan(&e.TrackID, &e.Title, &e.SourceID, &e.SourceName, &e.RelPath, &e.Error)
})
}
// RetryFailedAnalysis returns failed jobs to the queue with attempts reset.
func (st *Store) RetryFailedAnalysis(ctx context.Context) (int, error) {
tag, err := st.pool.Exec(ctx, `
UPDATE analysis_jobs SET status = 'queued', attempts = 0, error = NULL, updated_at = now()
+1 -1
View File
@@ -116,7 +116,7 @@ func (s *TemplateService) Preview(ctx context.Context, id uuid.UUID, patterns []
return nil, 0, err
}
if len(paths) == 0 {
found, err := walk(root, func(string) bool { return false })
found, err := walk(root, func(string) bool { return false }, nil)
if err != nil {
return nil, 0, apperr.New(http.StatusConflict, CodeSourceOffline, err.Error())
}