diff --git a/migrations/000015_alter_files_table_hash.down.sql b/migrations/000015_alter_files_table_hash.down.sql deleted file mode 100644 index ee61bdc..0000000 --- a/migrations/000015_alter_files_table_hash.down.sql +++ /dev/null @@ -1,2 +0,0 @@ -ALTER TABLE files -ALTER COLUMN hash bigint SET NOT NULL; \ No newline at end of file diff --git a/migrations/000015_alter_files_table_hash.up.sql b/migrations/000015_alter_files_table_hash.up.sql deleted file mode 100644 index c282260..0000000 --- a/migrations/000015_alter_files_table_hash.up.sql +++ /dev/null @@ -1,2 +0,0 @@ -ALTER TABLE files -ALTER COLUMN hash bigint DROP NOT NULL; \ No newline at end of file diff --git a/migrations/000017_alter_files_table_hash.down.sql b/migrations/000017_alter_files_table_hash.down.sql new file mode 100644 index 0000000..8f8179e --- /dev/null +++ b/migrations/000017_alter_files_table_hash.down.sql @@ -0,0 +1,3 @@ +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 new file mode 100644 index 0000000..35e5a23 --- /dev/null +++ b/migrations/000017_alter_files_table_hash.up.sql @@ -0,0 +1,3 @@ +ALTER TABLE files +ALTER COLUMN md5 DROP NOT NULL, +ADD mod_time TIMESTAMP NOT NULL DEFAULT current_timestamp; diff --git a/server/hash.go b/server/hash.go new file mode 100644 index 0000000..6ff47a2 --- /dev/null +++ b/server/hash.go @@ -0,0 +1,62 @@ +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 8e43f62..85138c9 100644 --- a/server/library.go +++ b/server/library.go @@ -63,10 +63,6 @@ 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 5d3bb1a..af3f322 100644 --- a/server/scan.go +++ b/server/scan.go @@ -10,6 +10,9 @@ import ( "os" "path/filepath" "slices" + "strconv" + "sync" + "time" "git.lastassault.de/speatzle/morffix/constants" "github.com/jackc/pgx/v5" @@ -23,16 +26,16 @@ func handleScan(w http.ResponseWriter, r *http.Request) { return } - id := r.PathValue("id") - if id == "" { - http.Error(w, "No ID Set", http.StatusBadRequest) + id, err := strconv.Atoi(r.FormValue("id")) + if err != nil { + http.Error(w, "Parsing ID", http.StatusBadRequest) return } 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) @@ -40,7 +43,7 @@ func handleScan(w http.ResponseWriter, r *http.Request) { } scanCtx := context.Background() - go scan(scanCtx, id) + go scanLibrary(scanCtx, id) message := "Scan Started" @@ -75,17 +78,81 @@ func scanStatus(w http.ResponseWriter, r *http.Request) { } } -func scan(ctx context.Context, id string) { - slog.InfoContext(ctx, "Starting Scan", "id", id) +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()) - // 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 + 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() + slog.InfoContext(ctx, "Starting Scan", "id", id) var name string var lpath string @@ -146,31 +213,16 @@ func scan(ctx context.Context, id string) { 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 { return fmt.Errorf("Getting Relative Path: %w", err) } var fileID int - var size uint - var health constants.FileHealth - err = tx.QueryRow(ctx, "SELECT id, size, health FROM files WHERE library_id = $1 AND path = $2", id, fPath).Scan(&fileID, &size, &health) + 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) if errors.Is(err, pgx.ErrNoRows) { // File Does not Exist Yet @@ -184,17 +236,24 @@ func scan(ctx context.Context, id string) { return fmt.Errorf("Getting File: %w", err) } - /* - if slices.Compare[[]byte](newMD5, oldMD5) != 0 { - // File has changed on disk so reset health - health = constants.FILE_HEALTH_UNKNOWN + // 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()) + _, err = tx.Exec(ctx, "UPDATE files SET size = $2, status = $3, health = $4, mod_time = $5 WHERE id = $1", fileID, info.Size(), constants.FILE_STATUS_CHANGED, constants.FILE_HEALTH_UNKNOWN, newModTime) + if err != nil { + return fmt.Errorf("Updating Changed File in DB: %w", err) } - */ - // File Exists so update Size, status and hash - _, err = tx.Exec(ctx, "UPDATE files SET size = $2, status = $3, health = $4 WHERE id = $1", fileID, info.Size(), constants.FILE_STATUS_EXISTS, health) - if err != nil { - return fmt.Errorf("Updating File in DB: %w", err) } + return nil }) if err != nil { @@ -207,7 +266,5 @@ func scan(ctx context.Context, id string) { 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 812f163..d0c2e24 100644 --- a/server/server.go +++ b/server/server.go @@ -126,8 +126,11 @@ func Start(_conf config.Config, tmplFS embed.FS, staticFS embed.FS, migrationsFS serverClose <- true }() - stopCleanup := make(chan bool, 1) - go manageWorkers(stopCleanup) + stopWorkerManagement := make(chan bool, 1) + go manageWorkers(stopWorkerManagement) + + stopScanning := make(chan bool, 1) + go manageScan(stopScanning) sigs := make(chan os.Signal, 1) signal.Notify(sigs, os.Interrupt) @@ -139,7 +142,10 @@ 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() - stopCleanup <- true + slog.Info("Stopping Worker Management...") + stopWorkerManagement <- true + slog.Info("Stopping Scanning...") + stopScanning <- true slog.Info("Done") } }