diff --git a/constants/constants.go b/constants/constants.go index dfcb9ab..99736a3 100644 --- a/constants/constants.go +++ b/constants/constants.go @@ -87,7 +87,6 @@ const ( FILE_STATUS_MISSING FILE_STATUS_EXISTS FILE_STATUS_CHANGED - FILE_STATUS_NEW ) func (s FileStatus) String() string { @@ -100,8 +99,6 @@ func (s FileStatus) String() string { return "Exists" case FILE_STATUS_CHANGED: return "Changed" - case FILE_STATUS_NEW: - return "New" default: return fmt.Sprintf("%d", int(s)) } diff --git a/go.mod b/go.mod index a46e664..1cf601b 100644 --- a/go.mod +++ b/go.mod @@ -26,5 +26,4 @@ require ( golang.org/x/sync v0.7.0 // indirect golang.org/x/sys v0.20.0 // indirect golang.org/x/text v0.15.0 // indirect - gopkg.in/vansante/go-ffprobe.v2 v2.2.0 // indirect ) diff --git a/go.sum b/go.sum index 002d41e..576771c 100644 --- a/go.sum +++ b/go.sum @@ -68,8 +68,6 @@ golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU= golang.org/x/text v0.15.0 h1:h1V/4gjBv8v9cjcR6+AR5+/cIYK5N/WAgiv4xlsEtAk= golang.org/x/text v0.15.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= -gopkg.in/vansante/go-ffprobe.v2 v2.2.0 h1:iuOqTsbfYuqIz4tAU9NWh22CmBGxlGHdgj4iqP+NUmY= -gopkg.in/vansante/go-ffprobe.v2 v2.2.0/go.mod h1:qF0AlAjk7Nqzqf3y333Ly+KxN3cKF2JqA3JT5ZheUGE= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= nhooyr.io/websocket v1.8.11 h1:f/qXNc2/3DpoSZkHt1DQu6rj4zGC8JmkkLkWss0MgN0= nhooyr.io/websocket v1.8.11/go.mod h1:rN9OFWIUwuxg4fR5tELlYC04bXYowCP9GX47ivo2l+c= diff --git a/migrations/000017_alter_files_table_hash.down.sql b/migrations/000017_alter_files_table_hash.down.sql deleted file mode 100644 index 8f8179e..0000000 --- a/migrations/000017_alter_files_table_hash.down.sql +++ /dev/null @@ -1,3 +0,0 @@ -ALTER TABLE files -ALTER COLUMN md5 SET NOT NULL, -DROP IF EXISTS mod_time; diff --git a/migrations/000017_alter_files_table_hash.up.sql b/migrations/000017_alter_files_table_hash.up.sql deleted file mode 100644 index 35e5a23..0000000 --- a/migrations/000017_alter_files_table_hash.up.sql +++ /dev/null @@ -1,3 +0,0 @@ -ALTER TABLE files -ALTER COLUMN md5 DROP NOT NULL, -ADD mod_time TIMESTAMP NOT NULL DEFAULT current_timestamp; diff --git a/migrations/000018_alter_files_table_ffprobe_data.down.sql b/migrations/000018_alter_files_table_ffprobe_data.down.sql deleted file mode 100644 index b489a96..0000000 --- a/migrations/000018_alter_files_table_ffprobe_data.down.sql +++ /dev/null @@ -1,2 +0,0 @@ -ALTER TABLE files -DROP IF EXISTS ffprobe; diff --git a/migrations/000018_alter_files_table_ffprobe_data.up.sql b/migrations/000018_alter_files_table_ffprobe_data.up.sql deleted file mode 100644 index a0e64bf..0000000 --- a/migrations/000018_alter_files_table_ffprobe_data.up.sql +++ /dev/null @@ -1,2 +0,0 @@ -ALTER TABLE files -ADD ffprobe_data JSONB; diff --git a/server/hash.go b/server/hash.go deleted file mode 100644 index 6ff47a2..0000000 --- a/server/hash.go +++ /dev/null @@ -1,62 +0,0 @@ -package server - -import ( - "context" - "crypto/md5" - "fmt" - "io" - "log/slog" - "os" - "path/filepath" - "slices" - - "git.lastassault.de/speatzle/morffix/constants" -) - -func hashFile(ctx context.Context, id uint) error { - slog.InfoContext(ctx, "Hashing File", "id", id) - - var filePath string - var libraryPath string - var oldHash []byte - err := db.QueryRow(ctx, "SELECT f.path, l.path, md5 FROM files f INNER JOIN libraries l ON l.id = f.library_id WHERE f.id = $1", id).Scan(&filePath, &libraryPath, &oldHash) - if err != nil { - return fmt.Errorf("Get File: %w", err) - } - - fullPath := filepath.Join(libraryPath, filePath) - - file, err := os.Open(fullPath) - if err != nil { - return fmt.Errorf("Opening File: %w", err) - } - - hash := md5.New() - if _, err := io.Copy(hash, file); err != nil { - return fmt.Errorf("Reading File: %w", err) - } - newHash := hash.Sum(nil) - - if slices.Compare[[]byte](newHash, oldHash) != 0 { - // File has changed - // TODO Queue healthcheck / transcode if enabled - } - - tx, err := db.Begin(ctx) - if err != nil { - return fmt.Errorf("Begin Transaction: %w", err) - } - defer tx.Rollback(ctx) - - _, err = tx.Exec(ctx, "UPDATE files SET status = $2, md5 = $3 WHERE id = $1", id, constants.FILE_STATUS_EXISTS, newHash) - if err != nil { - return fmt.Errorf("Updating File in DB: %w", err) - } - - err = tx.Commit(ctx) - if err != nil { - return fmt.Errorf("Committing Changes: %w", err) - } - slog.InfoContext(ctx, "Hashing File Done", "id", id) - return nil -} diff --git a/server/library.go b/server/library.go index 85138c9..8e43f62 100644 --- a/server/library.go +++ b/server/library.go @@ -63,6 +63,10 @@ func handleLibrary(w http.ResponseWriter, r *http.Request) { Enable: enabled, } + if r.Method == "PUT" { + // TODO + } + rows, err := db.Query(r.Context(), "SELECT id, path, size, status, health, transcode, md5, updated_at FROM files where library_id = $1", id) if err != nil { slog.ErrorContext(r.Context(), "Query Files", "err", err) diff --git a/server/scan.go b/server/scan.go index 6b65947..4e9d49e 100644 --- a/server/scan.go +++ b/server/scan.go @@ -3,20 +3,18 @@ package server import ( "bytes" "context" + "crypto/md5" "errors" "fmt" + "io" "log/slog" "net/http" "os" "path/filepath" "slices" - "strconv" - "sync" - "time" "git.lastassault.de/speatzle/morffix/constants" "github.com/jackc/pgx/v5" - "gopkg.in/vansante/go-ffprobe.v2" ) var videoFileExtensions = []string{".mkv", ".mp4", ".webm", ".flv", ".avi"} @@ -27,16 +25,18 @@ func handleScan(w http.ResponseWriter, r *http.Request) { return } - id, err := strconv.Atoi(r.FormValue("id")) - if err != nil { - http.Error(w, "Parsing ID", http.StatusBadRequest) + id := r.PathValue("id") + if id == "" { + http.Error(w, "No ID Set", http.StatusBadRequest) return } + full := r.FormValue("full") == "on" + var name string var path string var enabled bool - err = db.QueryRow(r.Context(), "SELECT name, path, enable FROM libraries WHERE id = $1", id).Scan(&name, &path, &enabled) + err := db.QueryRow(r.Context(), "SELECT name, path, enable FROM libraries WHERE id = $1", id).Scan(&name, &path, &enabled) if err != nil { slog.ErrorContext(r.Context(), "Get Library", "err", err) http.Error(w, "Error Get Library: "+err.Error(), http.StatusInternalServerError) @@ -44,7 +44,7 @@ func handleScan(w http.ResponseWriter, r *http.Request) { } scanCtx := context.Background() - go scanLibrary(scanCtx, id) + go scan(scanCtx, id, full) message := "Scan Started" @@ -79,82 +79,18 @@ func scanStatus(w http.ResponseWriter, r *http.Request) { } } -func manageScan(stop chan bool) { - scanTicker := time.NewTicker(time.Minute) - hashTicker := time.NewTicker(time.Second) - scanRunning := false - hashRunning := false - ctx, cancel := context.WithCancel(context.Background()) - - for { - select { - case <-scanTicker.C: - if scanRunning { - continue - } - scanRunning = true - go func() { - defer func() { - scanRunning = false - }() - - rows, err := db.Query(ctx, "SELECT id FROM libraries WHERE enable = true") - if err != nil { - slog.ErrorContext(ctx, "Error getting Enabled Libraries for Scan", "err", err) - return - } - - libraries, err := pgx.CollectRows[int](rows, pgx.RowTo[int]) - if err != nil { - slog.ErrorContext(ctx, "Error Collecting Enabled Libraries for Scan", "err", err) - return - } - - for _, l := range libraries { - scanLibrary(ctx, l) - } - }() - - case <-hashTicker.C: - if hashRunning { - continue - } - - hashRunning = true - go func() { - defer func() { - hashRunning = false - }() - - var fileID uint - err := db.QueryRow(ctx, "SELECT id FROM files WHERE status = $1 OR status = $2 LIMIT 1", constants.FILE_STATUS_CHANGED, constants.FILE_STATUS_NEW).Scan(&fileID) - if err != nil { - if !errors.Is(err, pgx.ErrNoRows) { - slog.ErrorContext(ctx, "Error Getting Files for Hashing", "err", err) - } - return - } - - err = hashFile(ctx, fileID) - if err != nil { - slog.ErrorContext(ctx, "Error Hashing File", "err", err) - } - }() - case <-stop: - cancel() - return - } - } -} - -var scanLock sync.Mutex - -func scanLibrary(ctx context.Context, id int) { - slog.InfoContext(ctx, "Acquiring Scan Lock...") - scanLock.Lock() - defer scanLock.Unlock() +func scan(ctx context.Context, id string, full bool) { slog.InfoContext(ctx, "Starting Scan", "id", id) + // TODO Scan settings: + // - Auto Queue Healthcheck for Changed Files + // - Auto Queue Healthcheck for New Files + // - Auto Queue Transcode for New Files + // - Auto Queue Transcode for Changed Files (? Instead have library setting to queue transcode for changed files on healthcheck success) + // - Auto Queue Health/Transcode for Unkown Status ? (might result in requeue loop) + // - Schedule Scans Periodically + // - Add File Monitoring for Setting Changed status and triggering tasks + var name string var lpath string var enabled bool @@ -214,6 +150,20 @@ func scanLibrary(ctx context.Context, id int) { slog.InfoContext(ctx, "Skipping non video file", "path", fullPath) return nil } + slog.InfoContext(ctx, "Hashing File", "path", fullPath, "size", info.Size()) + + file, err := os.Open(fullPath) + if err != nil { + return fmt.Errorf("Opening File: %w", err) + } + + hash := md5.New() + if _, err := io.Copy(hash, file); err != nil { + return fmt.Errorf("Reading File: %w", err) + } + newMD5 := hash.Sum(nil) + + slog.InfoContext(ctx, "File MD5", "path", fullPath, "size", info.Size(), "md5", newMD5) fPath, err := filepath.Rel(lpath, fullPath) if err != nil { @@ -221,21 +171,14 @@ func scanLibrary(ctx context.Context, id int) { } var fileID int - var oldSize uint - var oldModTime time.Time - err = tx.QueryRow(ctx, "SELECT id, size, mod_time FROM files WHERE library_id = $1 AND path = $2", id, fPath).Scan(&fileID, &oldSize, &oldModTime) + var oldMD5 []byte + var health constants.FileHealth + err = tx.QueryRow(ctx, "SELECT id, md5, health FROM files WHERE library_id = $1 AND path = $2", id, fPath).Scan(&fileID, &oldMD5, &health) if errors.Is(err, pgx.ErrNoRows) { // File Does not Exist Yet - slog.InfoContext(ctx, "File is New, Running FFProbe...", "path", fullPath) - - ffprobeData, err := ffprobe.ProbeURL(ctx, fullPath) - if err != nil { - return fmt.Errorf("ffprobe New File: %w", err) - } - slog.InfoContext(ctx, "ffprobe Done", "path", fullPath) - - _, err = tx.Exec(ctx, "INSERT INTO files (library_id, path, size, status, health, ffprobe_data) VALUES ($1, $2, $3, $4, $5, $6)", id, fPath, info.Size(), constants.FILE_STATUS_NEW, constants.FILE_HEALTH_UNKNOWN, ffprobeData) + slog.InfoContext(ctx, "File is New", "path", fullPath) + _, err = tx.Exec(ctx, "INSERT INTO files (library_id, path, size, status, health, md5) VALUES ($1, $2, $3, $4, $5, $6)", id, fPath, info.Size(), constants.FILE_STATUS_EXISTS, constants.FILE_HEALTH_UNKNOWN, newMD5) if err != nil { return fmt.Errorf("Add New File to DB: %w", err) } @@ -244,31 +187,15 @@ func scanLibrary(ctx context.Context, id int) { return fmt.Errorf("Getting File: %w", err) } - // Remove Timezone and Round to nearest Second - newModTime := info.ModTime().UTC().Round(time.Second) - - // File Already Exists, Check if it has been changed - if newModTime.Equal(oldModTime) && uint(info.Size()) == oldSize { - slog.Debug("File stayed the same", "id", fileID) - _, err = tx.Exec(ctx, "UPDATE files SET status = $2 WHERE id = $1", fileID, constants.FILE_STATUS_EXISTS) - if err != nil { - return fmt.Errorf("Updating Non Changed File in DB: %w", err) - } - } else { - slog.InfoContext(ctx, "File Has Changed", "path", fullPath, "old_mod_time", oldModTime, "new_mod_time", newModTime, "old_size", oldSize, "new_size", info.Size()) - - ffprobeData, err := ffprobe.ProbeURL(ctx, fullPath) - if err != nil { - return fmt.Errorf("ffprobe Changed File: %w", err) - } - slog.InfoContext(ctx, "ffprobe Done", "path", fullPath) - - _, err = tx.Exec(ctx, "UPDATE files SET size = $2, status = $3, health = $4, mod_time = $5, ffprobe_data = $6 WHERE id = $1", fileID, info.Size(), constants.FILE_STATUS_CHANGED, constants.FILE_HEALTH_UNKNOWN, newModTime, ffprobeData) - if err != nil { - return fmt.Errorf("Updating Changed File in DB: %w", err) - } + if slices.Compare[[]byte](newMD5, oldMD5) != 0 { + // File has changed on disk so reset health + health = constants.FILE_HEALTH_UNKNOWN + } + // File Exists so update Size, status and hash + _, err = tx.Exec(ctx, "UPDATE files SET size = $2, status = $3, health = $4, md5 = $5 WHERE id = $1", fileID, info.Size(), constants.FILE_STATUS_EXISTS, health, newMD5) + if err != nil { + return fmt.Errorf("Updating File in DB: %w", err) } - return nil }) if err != nil { @@ -281,5 +208,7 @@ func scanLibrary(ctx context.Context, id int) { slog.ErrorContext(ctx, "Error Committing Changes", "err", err) return } + + // TODO, create health and transcode tasks if requested slog.InfoContext(ctx, "Scan Done", "id", id) } diff --git a/server/server.go b/server/server.go index d0c2e24..812f163 100644 --- a/server/server.go +++ b/server/server.go @@ -126,11 +126,8 @@ func Start(_conf config.Config, tmplFS embed.FS, staticFS embed.FS, migrationsFS serverClose <- true }() - stopWorkerManagement := make(chan bool, 1) - go manageWorkers(stopWorkerManagement) - - stopScanning := make(chan bool, 1) - go manageScan(stopScanning) + stopCleanup := make(chan bool, 1) + go manageWorkers(stopCleanup) sigs := make(chan os.Signal, 1) signal.Notify(sigs, os.Interrupt) @@ -142,10 +139,7 @@ func Start(_conf config.Config, tmplFS embed.FS, staticFS embed.FS, migrationsFS stopCtx, cancel := context.WithTimeout(context.Background(), time.Second*10) server.Shutdown(stopCtx) cancel() - slog.Info("Stopping Worker Management...") - stopWorkerManagement <- true - slog.Info("Stopping Scanning...") - stopScanning <- true + stopCleanup <- true slog.Info("Done") } }