speatzle/morffix
1
0
Fork 0
Code Issues Pull requests Projects Releases 2 Packages Wiki Activity Actions

scan_rework #1

Merged
speatzle merged 3 commits from scan_rework into main 2024-07-06 23:53:28 +00:00
Conversation 0 Commits 3 Files changed 11 +206 -55
11 changed files with 206 additions and 55 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
constants/constants.go
View file

@ -87,6 +87,7 @@ const (
FILE_STATUS_MISSING
FILE_STATUS_EXISTS
FILE_STATUS_CHANGED
FILE_STATUS_NEW
)
func (s FileStatus) String() string {
@ -99,6 +100,8 @@ 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))
}

1
go.mod
View file

@ -26,4 +26,5 @@ 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
)

2
go.sum
View file

@ -68,6 +68,8 @@ 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=

3
migrations/000017_alter_files_table_hash.down.sql Normal file
View file

@ -0,0 +1,3 @@
ALTER TABLE files
ALTER COLUMN md5 SET NOT NULL,
DROP IF EXISTS mod_time;

3
migrations/000017_alter_files_table_hash.up.sql Normal file
View file

@ -0,0 +1,3 @@
ALTER TABLE files
ALTER COLUMN md5 DROP NOT NULL,
ADD mod_time TIMESTAMP NOT NULL DEFAULT current_timestamp;

2
migrations/000018_alter_files_table_ffprobe_data.down.sql Normal file
View file

@ -0,0 +1,2 @@
ALTER TABLE files
DROP IF EXISTS ffprobe;

2
migrations/000018_alter_files_table_ffprobe_data.up.sql Normal file
View file

@ -0,0 +1,2 @@
ALTER TABLE files
ADD ffprobe_data JSONB;

62
server/hash.go Normal file
View file

@ -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
}

4
server/library.go
View file

@ -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)

165
server/scan.go
View file

@ -3,18 +3,20 @@ 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"}
@ -25,18 +27,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
}
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 scan(scanCtx, id, full)
go scanLibrary(scanCtx, id)
message := "Scan Started"
@ -79,17 +79,81 @@ func scanStatus(w http.ResponseWriter, r *http.Request) {
}
}
func scan(ctx context.Context, id string, full bool) {
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
@ -150,20 +214,6 @@ func scan(ctx context.Context, id string, full bool) {
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 {
@ -171,14 +221,21 @@ func scan(ctx context.Context, id string, full bool) {
}
var fileID int
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)
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
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)
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)
if err != nil {
return fmt.Errorf("Add New File to DB: %w", err)
}
@ -187,15 +244,31 @@ func scan(ctx context.Context, id string, full bool) {
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
}
// 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)
// 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 File in DB: %w", err)
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)
}
}
return nil
})
if err != nil {
@ -208,7 +281,5 @@ func scan(ctx context.Context, id string, full bool) {
slog.ErrorContext(ctx, "Error Committing Changes", "err", err)
return
}
// TODO, create health and transcode tasks if requested
slog.InfoContext(ctx, "Scan Done", "id", id)
}

12
server/server.go
View file

@ -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")
}
}