From a22d00214651f3c1f202db24d74e5d8b0a8acd2e Mon Sep 17 00:00:00 2001 From: viktorstrate Date: Mon, 22 Jun 2020 23:52:41 +0200 Subject: [PATCH] Major rework of scanner --- api/graphql/resolvers/scanner.go | 60 +++--- api/scanner/cache.go | 56 +++-- api/scanner/photo_type.go | 6 +- api/scanner/queue.go | 127 +++++------ api/scanner/queue_test.go | 168 +++++++-------- api/scanner/scanner_album.go | 35 ++- api/scanner/scanner_user.go | 353 +++++++++---------------------- 7 files changed, 338 insertions(+), 467 deletions(-) diff --git a/api/graphql/resolvers/scanner.go b/api/graphql/resolvers/scanner.go index dab821fe..f14c48dd 100644 --- a/api/graphql/resolvers/scanner.go +++ b/api/graphql/resolvers/scanner.go @@ -2,46 +2,46 @@ package resolvers import ( "context" - "fmt" "github.com/viktorstrate/photoview/api/graphql/models" - "github.com/viktorstrate/photoview/api/scanner" ) func (r *mutationResolver) ScanAll(ctx context.Context) (*models.ScannerResult, error) { - if err := scanner.ScanAll(r.Database); err != nil { - errorMessage := fmt.Sprintf("Error starting scanner: %s", err.Error()) - return &models.ScannerResult{ - Finished: false, - Success: false, - Message: &errorMessage, - }, nil - } + // if err := scanner.ScanAll(r.Database); err != nil { + // errorMessage := fmt.Sprintf("Error starting scanner: %s", err.Error()) + // return &models.ScannerResult{ + // Finished: false, + // Success: false, + // Message: &errorMessage, + // }, nil + // } - startMessage := "Scanner started" + // startMessage := "Scanner started" - return &models.ScannerResult{ - Finished: false, - Success: true, - Message: &startMessage, - }, nil + // return &models.ScannerResult{ + // Finished: false, + // Success: true, + // Message: &startMessage, + // }, nil + panic("not implemented") } func (r *mutationResolver) ScanUser(ctx context.Context, userID int) (*models.ScannerResult, error) { - if err := scanner.ScanUser(r.Database, userID); err != nil { - errorMessage := fmt.Sprintf("Error scanning user: %s", err.Error()) - return &models.ScannerResult{ - Finished: false, - Success: false, - Message: &errorMessage, - }, nil - } + // if err := scanner.ScanUser(r.Database, userID); err != nil { + // errorMessage := fmt.Sprintf("Error scanning user: %s", err.Error()) + // return &models.ScannerResult{ + // Finished: false, + // Success: false, + // Message: &errorMessage, + // }, nil + // } - startMessage := "Scanner started" + // startMessage := "Scanner started" - return &models.ScannerResult{ - Finished: false, - Success: true, - Message: &startMessage, - }, nil + // return &models.ScannerResult{ + // Finished: false, + // Success: true, + // Message: &startMessage, + // }, nil + panic("not implemented") } diff --git a/api/scanner/cache.go b/api/scanner/cache.go index 40e7f7f3..ce2be92d 100644 --- a/api/scanner/cache.go +++ b/api/scanner/cache.go @@ -2,54 +2,38 @@ package scanner import "path" -type ScannerCache struct { - cache map[string]interface{} - photo_paths_scanned []interface{} - album_paths_scanned []interface{} +type AlbumScannerCache struct { + path_contains_photos map[string]bool + photo_types map[string]ImageType } -func MakeScannerCache() ScannerCache { - return ScannerCache{ - cache: make(map[string]interface{}), - photo_paths_scanned: make([]interface{}, 0), - album_paths_scanned: make([]interface{}, 0), +func MakeAlbumCache() *AlbumScannerCache { + return &AlbumScannerCache{ + path_contains_photos: make(map[string]bool), + photo_types: make(map[string]ImageType), } } -func (c *ScannerCache) insert_photo_type(path string, content_type ImageType) { - (c.cache)["photo_type//"+path] = content_type -} - -func (c *ScannerCache) get_photo_type(path string) *string { - result, found := (c.cache)["photo_type//"+path].(string) - if found { - // log.Printf("Image cache hit: %s\n", path) - return &result - } - - return nil -} - // Insert single album directory in cache -func (c *ScannerCache) insert_album_path(path string, contains_photo bool) { - (c.cache)["album_path//"+path] = contains_photo +func (c *AlbumScannerCache) InsertAlbumPath(path string, contains_photo bool) { + c.path_contains_photos[path] = contains_photo } // Insert album path and all parent directories up to the given root directory in cache -func (c *ScannerCache) insert_album_paths(end_path string, root string, contains_photo bool) { +func (c *AlbumScannerCache) InsertAlbumPaths(end_path string, root string, contains_photo bool) { curr_path := path.Clean(end_path) root_path := path.Clean(root) for curr_path != root_path || curr_path == "." { - c.insert_album_path(curr_path, contains_photo) + c.InsertAlbumPath(curr_path, contains_photo) curr_path = path.Dir(curr_path) } } -func (c *ScannerCache) album_contains_photo(path string) *bool { - contains_photo, found := (c.cache)["album_path//"+path].(bool) +func (c *AlbumScannerCache) AlbumContainsPhotos(path string) *bool { + contains_photo, found := c.path_contains_photos[path] if found { // log.Printf("Album cache hit: %s\n", path) return &contains_photo @@ -57,3 +41,17 @@ func (c *ScannerCache) album_contains_photo(path string) *bool { return nil } + +func (c *AlbumScannerCache) InsertPhotoType(path string, content_type ImageType) { + (c.photo_types)[path] = content_type +} + +func (c *AlbumScannerCache) GetPhotoType(path string) *ImageType { + result, found := c.photo_types[path] + if found { + // log.Printf("Image cache hit: %s\n", path) + return &result + } + + return nil +} diff --git a/api/scanner/photo_type.go b/api/scanner/photo_type.go index dc35da85..43b595e2 100644 --- a/api/scanner/photo_type.go +++ b/api/scanner/photo_type.go @@ -234,8 +234,8 @@ func getImageType(path string) (*ImageType, error) { return nil, nil } -func isPathImage(path string, cache *ScannerCache) bool { - if cache.get_photo_type(path) != nil { +func isPathImage(path string, cache *AlbumScannerCache) bool { + if cache.GetPhotoType(path) != nil { return true } @@ -252,7 +252,7 @@ func isPathImage(path string, cache *ScannerCache) bool { return false } - cache.insert_photo_type(path, *imageType) + cache.InsertPhotoType(path, *imageType) return true } diff --git a/api/scanner/queue.go b/api/scanner/queue.go index 25036e6d..f73d1a60 100644 --- a/api/scanner/queue.go +++ b/api/scanner/queue.go @@ -2,43 +2,23 @@ package scanner import ( "database/sql" - "errors" + "log" "sync" "github.com/viktorstrate/photoview/api/graphql/models" ) -type ScannerJobScope int - -const ( - JOB_SCAN_USER ScannerJobScope = iota - JOB_SCAN_ALBUM -) - type ScannerJob struct { - scope ScannerJobScope - // Either models.User, models.Album or nil depending on the value of scope - model interface{} + album *models.Album + cache *AlbumScannerCache } -func (job *ScannerJob) modelAsUser() (*models.User, error) { - user, ok := job.model.(models.User) - if !ok { - return nil, errors.New("scanner job not of type User") - } - return &user, nil +func (job *ScannerJob) Run(db *sql.DB) { + scanAlbum(job.album, job.cache, db) } -func (job *ScannerJob) modelAsAlbum() (*models.Album, error) { - album, ok := job.model.(models.Album) - if !ok { - return nil, errors.New("scanner job not of type Album") - } - return &album, nil -} - -func (job *ScannerJob) Run() { - // TODO: Not implemented +type ScannerQueueSettings struct { + max_concurrent_tasks int } type ScannerQueue struct { @@ -47,6 +27,7 @@ type ScannerQueue struct { in_progress []ScannerJob up_next []ScannerJob db *sql.DB + settings ScannerQueueSettings } var global_scanner_queue ScannerQueue @@ -57,7 +38,10 @@ func InitializeScannerQueue(db *sql.DB) { in_progress: make([]ScannerJob, 0), up_next: make([]ScannerJob, 0), db: db, + settings: ScannerQueueSettings{max_concurrent_tasks: 3}, } + + go global_scanner_queue.startBackgroundWorker() } func (queue *ScannerQueue) startBackgroundWorker() { @@ -65,22 +49,33 @@ func (queue *ScannerQueue) startBackgroundWorker() { <-queue.idle_chan queue.mutex.Lock() defer queue.mutex.Unlock() + + for len(queue.in_progress) < queue.settings.max_concurrent_tasks && len(queue.up_next) > 0 { + nextJob := queue.up_next[0] + queue.up_next = queue.up_next[1:] + queue.in_progress = append(queue.in_progress, nextJob) + + go func() { + nextJob.Run(queue.db) + queue.mutex.Lock() + defer queue.mutex.Unlock() + + // Delete finished job from queue + for i, x := range queue.in_progress { + if x == nextJob { + queue.in_progress[i] = queue.in_progress[len(queue.in_progress)-1] + queue.in_progress = queue.in_progress[0 : len(queue.in_progress)-1] + break + } + } + + queue.Notify() + }() + } } } -func (queue *ScannerQueue) AddJob(job *ScannerJob) error { - queue.mutex.Lock() - defer queue.mutex.Unlock() - - if exists, err := queue.jobOnQueue(job); exists || err != nil { - return err - } - queue.up_next = append(queue.up_next, *job) - queue.Notify() - - return nil -} - +// Notifies the queue that the jobs has changed func (queue *ScannerQueue) Notify() bool { select { case queue.idle_chan <- true: @@ -90,35 +85,43 @@ func (queue *ScannerQueue) Notify() bool { } } +func (queue *ScannerQueue) ScanUser(user *models.User) { + album_cache := MakeAlbumCache() + albums, album_errors := findAlbumsForUser(queue.db, user, album_cache) + for _, err := range album_errors { + log.Printf("User scanner error: %s", err) + } + + queue.mutex.Lock() + for _, album := range albums { + queue.addJob(&ScannerJob{ + album: album, + cache: album_cache, + }) + } + queue.mutex.Unlock() +} + +// Queue should be locked prior to calling this function +func (queue *ScannerQueue) addJob(job *ScannerJob) error { + if exists, err := queue.jobOnQueue(job); exists || err != nil { + return err + } + queue.up_next = append(queue.up_next, *job) + queue.Notify() + + return nil +} + +// Queue should be locked prior to calling this function func (queue *ScannerQueue) jobOnQueue(job *ScannerJob) (bool, error) { scannerJobs := append(queue.in_progress, queue.up_next...) for _, scannerJob := range scannerJobs { - - if scannerJob == *job { + if scannerJob.album.AlbumID == job.album.AlbumID { return true, nil } - - if scannerJob.scope == JOB_SCAN_USER { - user, err := scannerJob.modelAsUser() - if err != nil { - return true, err - } - - if job.scope == JOB_SCAN_ALBUM { - album, err := job.modelAsAlbum() - if err != nil { - return true, err - } - - if album.OwnerID == user.UserID { - return true, nil - } - - } - } - } return false, nil diff --git a/api/scanner/queue_test.go b/api/scanner/queue_test.go index 2c8501b6..555d7a37 100644 --- a/api/scanner/queue_test.go +++ b/api/scanner/queue_test.go @@ -1,107 +1,101 @@ package scanner -import ( - "testing" +// func TestScannerQueue_AddJob(t *testing.T) { - "github.com/viktorstrate/photoview/api/graphql/models" -) +// scannerJobs := []ScannerJob{ +// {scope: JOB_SCAN_ALBUM, model: models.Album{AlbumID: 100, OwnerID: 123}}, +// {scope: JOB_SCAN_USER, model: models.User{UserID: 20}}, +// } -func TestScannerQueue_AddJob(t *testing.T) { +// mockScannerQueue := ScannerQueue{ +// idle_chan: make(chan bool, 1), +// in_progress: make([]ScannerJob, 0), +// up_next: scannerJobs, +// db: nil, +// } - scannerJobs := []ScannerJob{ - {scope: JOB_SCAN_ALBUM, model: models.Album{AlbumID: 100, OwnerID: 123}}, - {scope: JOB_SCAN_USER, model: models.User{UserID: 20}}, - } +// t.Run("add new job to scanner queue", func(t *testing.T) { +// newJob := ScannerJob{ +// scope: JOB_SCAN_USER, +// model: models.User{UserID: 253}, +// } - mockScannerQueue := ScannerQueue{ - idle_chan: make(chan bool, 1), - in_progress: make([]ScannerJob, 0), - up_next: scannerJobs, - db: nil, - } +// startingJobs := len(mockScannerQueue.up_next) - t.Run("add new job to scanner queue", func(t *testing.T) { - newJob := ScannerJob{ - scope: JOB_SCAN_USER, - model: models.User{UserID: 253}, - } +// err := mockScannerQueue.AddJob(&newJob) +// if err != nil { +// t.Errorf(".AddJob() returned an unexpected error: %s", err) +// } - startingJobs := len(mockScannerQueue.up_next) +// if len(mockScannerQueue.up_next) != startingJobs+1 { +// t.Errorf("Expected scanner queue length to be %d but got %d", startingJobs+1, len(mockScannerQueue.up_next)) +// } else if mockScannerQueue.up_next[len(mockScannerQueue.up_next)-1] != newJob { +// t.Errorf("Expected scanner queue to contain the job that was added: %+v", newJob) +// } - err := mockScannerQueue.AddJob(&newJob) - if err != nil { - t.Errorf(".AddJob() returned an unexpected error: %s", err) - } +// }) - if len(mockScannerQueue.up_next) != startingJobs+1 { - t.Errorf("Expected scanner queue length to be %d but got %d", startingJobs+1, len(mockScannerQueue.up_next)) - } else if mockScannerQueue.up_next[len(mockScannerQueue.up_next)-1] != newJob { - t.Errorf("Expected scanner queue to contain the job that was added: %+v", newJob) - } +// t.Run("add existing job to scanner queue", func(t *testing.T) { +// startingJobs := len(mockScannerQueue.up_next) - }) +// err := mockScannerQueue.AddJob(&ScannerJob{ +// scope: JOB_SCAN_USER, +// model: models.User{UserID: 20}, +// }) +// if err != nil { +// t.Errorf(".AddJob() returned an unexpected error: %s", err) +// } - t.Run("add existing job to scanner queue", func(t *testing.T) { - startingJobs := len(mockScannerQueue.up_next) +// if len(mockScannerQueue.up_next) != startingJobs { +// t.Errorf("Expected scanner queue length not to change: start length %d, new length %d", startingJobs, len(mockScannerQueue.up_next)) +// } - err := mockScannerQueue.AddJob(&ScannerJob{ - scope: JOB_SCAN_USER, - model: models.User{UserID: 20}, - }) - if err != nil { - t.Errorf(".AddJob() returned an unexpected error: %s", err) - } +// }) - if len(mockScannerQueue.up_next) != startingJobs { - t.Errorf("Expected scanner queue length not to change: start length %d, new length %d", startingJobs, len(mockScannerQueue.up_next)) - } +// } - }) +// func TestScannerQueue_JobOnQueue(t *testing.T) { -} +// scannerJobs := []ScannerJob{ +// {scope: JOB_SCAN_ALBUM, model: models.Album{AlbumID: 100, OwnerID: 123}}, +// {scope: JOB_SCAN_USER, model: models.User{UserID: 20}}, +// } -func TestScannerQueue_JobOnQueue(t *testing.T) { +// mockScannerQueue := ScannerQueue{ +// idle_chan: make(chan bool, 1), +// in_progress: make([]ScannerJob, 0), +// up_next: scannerJobs, +// db: nil, +// } - scannerJobs := []ScannerJob{ - {scope: JOB_SCAN_ALBUM, model: models.Album{AlbumID: 100, OwnerID: 123}}, - {scope: JOB_SCAN_USER, model: models.User{UserID: 20}}, - } +// onQueueTests := []struct { +// string +// bool +// ScannerJob +// }{ +// {"user that is already on the queue", true, ScannerJob{ +// scope: JOB_SCAN_USER, +// model: models.User{UserID: 20}, +// }}, +// {"album which owner is already on the queue", true, ScannerJob{ +// scope: JOB_SCAN_ALBUM, +// model: models.Album{AlbumID: 40, OwnerID: 20}, +// }}, +// {"album that is not on the queue", false, ScannerJob{ +// scope: JOB_SCAN_ALBUM, +// model: models.Album{AlbumID: 321, OwnerID: 11}, +// }}, +// } - mockScannerQueue := ScannerQueue{ - idle_chan: make(chan bool, 1), - in_progress: make([]ScannerJob, 0), - up_next: scannerJobs, - db: nil, - } +// for _, test := range onQueueTests { +// t.Run(test.string, func(t *testing.T) { +// onQueue, err := mockScannerQueue.jobOnQueue(&test.ScannerJob) +// if err != nil { +// t.Error("Expected jobOnQueue not to return an error") +// } else if onQueue != test.bool { +// t.Fail() +// } +// }) +// } - onQueueTests := []struct { - string - bool - ScannerJob - }{ - {"user that is already on the queue", true, ScannerJob{ - scope: JOB_SCAN_USER, - model: models.User{UserID: 20}, - }}, - {"album which owner is already on the queue", true, ScannerJob{ - scope: JOB_SCAN_ALBUM, - model: models.Album{AlbumID: 40, OwnerID: 20}, - }}, - {"album that is not on the queue", false, ScannerJob{ - scope: JOB_SCAN_ALBUM, - model: models.Album{AlbumID: 321, OwnerID: 11}, - }}, - } - - for _, test := range onQueueTests { - t.Run(test.string, func(t *testing.T) { - onQueue, err := mockScannerQueue.jobOnQueue(&test.ScannerJob) - if err != nil { - t.Error("Expected jobOnQueue not to return an error") - } else if onQueue != test.bool { - t.Fail() - } - }) - } - -} +// } diff --git a/api/scanner/scanner_album.go b/api/scanner/scanner_album.go index b0547b03..7084c2bb 100644 --- a/api/scanner/scanner_album.go +++ b/api/scanner/scanner_album.go @@ -8,7 +8,38 @@ import ( "github.com/viktorstrate/photoview/api/graphql/models" ) -func findPhotosForAlbum(album *models.Album, cache *ScannerCache, db *sql.DB, onScanPhoto func(photo *models.Photo, newPhoto bool)) ([]*models.Photo, error) { +func scanAlbum(album *models.Album, cache *AlbumScannerCache, db *sql.DB) { + // Scan for photos + albumPhotos, err := findPhotosForAlbum(album, cache, db, func(photo *models.Photo, newPhoto bool) { + // notifyThrottle.Trigger(func() { + // notification.BroadcastNotification(&models.Notification{ + // Key: processKey, + // Type: models.NotificationTypeMessage, + // Header: fmt.Sprintf("Scanning photo for user '%s'", user.Username), + // Content: fmt.Sprintf("Scanning image at %s", photo.Path), + // }) + // }) + }) + if err != nil { + ScannerError("Failed to find photos for album (%s): %s", album.Path, err) + } + + tx, err := db.Begin() + if err != nil { + ScannerError("Failed to begin database transaction: %s", err) + } + + for _, photo := range albumPhotos { + err = ProcessPhoto(tx, photo) + if err != nil { + ScannerError("Failed to process photo (%s): %s", photo.Path, err) + } + + // TODO: Broadcast progress + } +} + +func findPhotosForAlbum(album *models.Album, cache *AlbumScannerCache, db *sql.DB, onScanPhoto func(photo *models.Photo, newPhoto bool)) ([]*models.Photo, error) { newPhotos := make([]*models.Photo, 0) @@ -27,8 +58,6 @@ func findPhotosForAlbum(album *models.Album, cache *ScannerCache, db *sql.DB, on continue } - cache.photo_paths_scanned = append(cache.photo_paths_scanned, photoPath) - photo, isNewPhoto, err := ScanPhoto(tx, photoPath, album.AlbumID) if err != nil { ScannerError("Scanning image %s: %s", photoPath, err) diff --git a/api/scanner/scanner_user.go b/api/scanner/scanner_user.go index 7f64c03a..36817ad5 100644 --- a/api/scanner/scanner_user.go +++ b/api/scanner/scanner_user.go @@ -10,77 +10,24 @@ import ( "path" "strconv" "strings" - "time" + "github.com/pkg/errors" "github.com/viktorstrate/photoview/api/graphql/models" "github.com/viktorstrate/photoview/api/graphql/notification" "github.com/viktorstrate/photoview/api/utils" ) -func ScanAll(database *sql.DB) error { - rows, err := database.Query("SELECT * FROM user") - if err != nil { - log.Printf("Could not fetch all users from database: %s\n", err.Error()) - return err - } - - users, err := models.NewUsersFromRows(rows) - if err != nil { - log.Printf("Could not convert users: %s\n", err) - return err - } - - for _, user := range users { - go scan(database, user) - } - - return nil -} - -func ScanUser(database *sql.DB, userId int) error { - - row := database.QueryRow("SELECT * FROM user WHERE user_id = ?", userId) - user, err := models.NewUserFromRow(row) - if err != nil { - log.Printf("Could not find user to scan: %s\n", err.Error()) - return err - } - - log.Printf("Starting scan for user '%s'\n", user.Username) - go scan(database, user) - - return nil -} - -func scan(database *sql.DB, user *models.User) { +func findAlbumsForUser(db *sql.DB, user *models.User, album_cache *AlbumScannerCache) ([]*models.Album, []error) { // Check if user directory exists on the file system if _, err := os.Stat(user.RootPath); err != nil { if os.IsNotExist(err) { - ScannerError("Photo directory for user '%s' does not exist '%s'\n", user.Username, user.RootPath) + return nil, []error{errors.Errorf("Photo directory for user '%s' does not exist '%s'\n", user.Username, user.RootPath)} } else { - ScannerError("Could not read photo directory for user '%s': %s\n", user.Username, user.RootPath) + return nil, []error{errors.Errorf("Could not read photo directory for user '%s': %s\n", user.Username, user.RootPath)} } - - return } - notifyKey := utils.GenerateToken() - processKey := utils.GenerateToken() - notifyThrottle := utils.NewThrottle(500 * time.Millisecond) - - timeout := 3000 - notification.BroadcastNotification(&models.Notification{ - Key: notifyKey, - Type: models.NotificationTypeMessage, - Header: "User scan started", - Content: fmt.Sprintf("Scanning has started for user '%s'", user.Username), - Timeout: &timeout, - }) - - // Start scanning - cache := MakeScannerCache() - type scanInfo struct { path string parentId *int @@ -92,7 +39,9 @@ func scan(database *sql.DB, user *models.User) { parentId: nil, }) - newPhotos := make([]*models.Photo, 0) + userAlbums := make([]*models.Album, 0) + albumErrors := make([]error, 0) + // newPhotos := make([]*models.Photo, 0) for scanQueue.Front() != nil { albumInfo := scanQueue.Front().Value.(scanInfo) @@ -101,18 +50,16 @@ func scan(database *sql.DB, user *models.User) { albumPath := albumInfo.path albumParentId := albumInfo.parentId - cache.album_paths_scanned = append(cache.album_paths_scanned, albumPath) - // Read path dirContent, err := ioutil.ReadDir(albumPath) if err != nil { - ScannerError("Could not read directory: %s\n", err.Error()) + albumErrors = append(albumErrors, errors.Wrapf(err, "read directory (%s)", albumPath)) continue } - tx, err := database.Begin() + tx, err := db.Begin() if err != nil { - ScannerError("Could not begin database transaction: %s\n", err) + albumErrors = append(albumErrors, errors.Wrap(err, "begin database transaction")) continue } @@ -122,7 +69,7 @@ func scan(database *sql.DB, user *models.User) { albumTitle := path.Base(albumPath) _, err = tx.Exec("INSERT IGNORE INTO album (title, parent_album, owner_id, path) VALUES (?, ?, ?, ?)", albumTitle, albumParentId, user.UserID, albumPath) if err != nil { - ScannerError("Could not insert album into database: %s\n", err) + albumErrors = append(albumErrors, errors.Wrap(err, "insert album into database")) tx.Rollback() continue } @@ -130,33 +77,18 @@ func scan(database *sql.DB, user *models.User) { row := tx.QueryRow("SELECT * FROM album WHERE path = ?", albumPath) album, err := models.NewAlbumFromRow(row) if err != nil { - ScannerError("Could not get album: %s\n", err) + albumErrors = append(albumErrors, errors.Wrapf(err, "get album from database (%s)", albumPath)) tx.Rollback() - return + continue } + userAlbums = append(userAlbums, album) // Commit album transaction if err := tx.Commit(); err != nil { - log.Printf("ERROR: Could not commit database transaction: %s\n", err) - return + albumErrors = append(albumErrors, errors.Wrap(err, "commit database transaction")) + continue } - // Scan for photos - newFoundPhotos, err := findPhotosForAlbum(album, &cache, database, func(photo *models.Photo, newPhoto bool) { - notifyThrottle.Trigger(func() { - notification.BroadcastNotification(&models.Notification{ - Key: processKey, - Type: models.NotificationTypeMessage, - Header: fmt.Sprintf("Scanning photo for user '%s'", user.Username), - Content: fmt.Sprintf("Scanning image at %s", photo.Path), - }) - }) - }) - if err != nil { - ScannerError("Failed to scan album for new photos (album_id %d)", album.AlbumID) - } - newPhotos = append(newPhotos, newFoundPhotos...) - // Scan for sub-albums for _, item := range dirContent { subalbumPath := path.Join(albumPath, item.Name()) @@ -166,7 +98,7 @@ func scan(database *sql.DB, user *models.User) { continue } - if item.IsDir() && directoryContainsPhotos(subalbumPath, &cache) { + if item.IsDir() && directoryContainsPhotos(subalbumPath, album_cache) { scanQueue.PushBack(scanInfo{ path: subalbumPath, parentId: &album.AlbumID, @@ -175,32 +107,15 @@ func scan(database *sql.DB, user *models.User) { } } - completeMessage := "No new photos were found" - if len(newPhotos) > 0 { - completeMessage = fmt.Sprintf("%d new photos were found", len(newPhotos)) - } + deleteErrors := deleteOldUserAlbums(db, userAlbums, user) + albumErrors = append(albumErrors, deleteErrors...) - notification.BroadcastNotification(&models.Notification{ - Key: notifyKey, - Type: models.NotificationTypeMessage, - Header: fmt.Sprintf("Scan completed for user '%s'", user.Username), - Content: completeMessage, - Positive: true, - }) - - cleanupCache(database, &cache, user) - - err := processUnprocessedPhotos(database, user, notifyKey) - if err != nil { - log.Printf("ERROR: processing photos: %s\n", err) - } - - log.Printf("Done scanning user '%s'\n", user.Username) + return userAlbums, albumErrors } -func directoryContainsPhotos(rootPath string, cache *ScannerCache) bool { +func directoryContainsPhotos(rootPath string, cache *AlbumScannerCache) bool { - if contains_image := cache.album_contains_photo(rootPath); contains_image != nil { + if contains_image := cache.AlbumContainsPhotos(rootPath); contains_image != nil { return *contains_image } @@ -228,7 +143,7 @@ func directoryContainsPhotos(rootPath string, cache *ScannerCache) bool { scanQueue.PushBack(filePath) } else { if isPathImage(filePath, cache) { - cache.insert_album_paths(dirPath, rootPath, true) + cache.InsertAlbumPaths(dirPath, rootPath, true) return true } } @@ -237,189 +152,121 @@ func directoryContainsPhotos(rootPath string, cache *ScannerCache) bool { } for _, scanned_path := range scanned_directories { - cache.insert_album_path(scanned_path, false) + cache.InsertAlbumPath(scanned_path, false) } return false } -func processUnprocessedPhotos(database *sql.DB, user *models.User, notifyKey string) error { - - processKey := utils.GenerateToken() - notifyThrottle := utils.NewThrottle(500 * time.Millisecond) - - rows, err := database.Query(` - SELECT photo.* FROM photo JOIN album ON photo.album_id = album.album_id - WHERE album.owner_id = ? - AND photo.photo_id NOT IN ( - SELECT photo_id FROM photo_url WHERE photo_url.photo_id = photo.photo_id - ) - `, user.UserID) - if err != nil { - ScannerError("Could not get photos to process from db") - return err +func deleteOldUserAlbums(db *sql.DB, scannedAlbums []*models.Album, user *models.User) []error { + if len(scannedAlbums) == 0 { + return nil } - photosToProcess, err := models.NewPhotosFromRows(rows) - if err != nil { - if err == sql.ErrNoRows { - // No photos to process - return nil - } - - ScannerError("Could not parse photos to process from db %s", err) - return err - } - - // Proccess all photos - for count, photo := range photosToProcess { - - tx, err := database.Begin() - if err != nil { - ScannerError("Could not start database transaction: %s", err) - continue - } - - notifyThrottle.Trigger(func() { - var progress float64 = float64(count) / float64(len(photosToProcess)) * 100.0 - - notification.BroadcastNotification(&models.Notification{ - Key: processKey, - Type: models.NotificationTypeProgress, - Header: fmt.Sprintf("Processing photos (%d of %d) for user '%s'", count, len(photosToProcess), user.Username), - Content: fmt.Sprintf("Processing photo at %s", photo.Path), - Progress: &progress, - }) - }) - - err = ProcessPhoto(tx, photo) - if err != nil { - tx.Rollback() - ScannerError("Could not process photo (%s): %s", photo.Path, err) - continue - } - - err = tx.Commit() - if err != nil { - ScannerError("Could not commit db transaction: %s", err) - continue - } - } - - if len(photosToProcess) > 0 { - notification.BroadcastNotification(&models.Notification{ - Key: notifyKey, - Type: models.NotificationTypeMessage, - Header: fmt.Sprintf("Processing photos for user '%s' has completed", user.Username), - Content: fmt.Sprintf("%d photos have been processed", len(photosToProcess)), - Positive: true, - }) - - notification.BroadcastNotification(&models.Notification{ - Key: processKey, - Type: models.NotificationTypeClose, - }) - } - - return nil -} - -func cleanupCache(database *sql.DB, cache *ScannerCache, user *models.User) { - if len(cache.album_paths_scanned) == 0 { - return + albumPaths := make([]interface{}, len(scannedAlbums)) + for i, album := range scannedAlbums { + albumPaths[i] = album.AlbumID } // Delete old albums album_args := make([]interface{}, 0) album_args = append(album_args, user.UserID) - album_args = append(album_args, cache.album_paths_scanned...) + album_args = append(album_args, albumPaths...) - albums_questions := strings.Repeat("?,", len(cache.album_paths_scanned))[:len(cache.album_paths_scanned)*2-1] - rows, err := database.Query("SELECT album_id FROM album WHERE album.owner_id = ? AND path NOT IN ("+albums_questions+")", album_args...) + albums_questions := strings.Repeat("?,", len(albumPaths))[:len(albumPaths)*2-1] + rows, err := db.Query("SELECT album_id FROM album WHERE album.owner_id = ? AND path NOT IN ("+albums_questions+")", album_args...) if err != nil { - ScannerError("Could not get albums from database: %s\n", err) - return + return []error{errors.Wrap(err, "get albums to be deleted from database")} } defer rows.Close() + deleteErrors := make([]error, 0) + deleted_album_ids := make([]interface{}, 0) for rows.Next() { var album_id int if err := rows.Scan(&album_id); err != nil { - ScannerError("Could not parse album to be removed (album_id %d): %s\n", album_id, err) + deleteErrors = append(deleteErrors, errors.Wrapf(err, "parse album to be removed (album_id %d)", album_id)) + continue } deleted_album_ids = append(deleted_album_ids, album_id) cache_path := path.Join("./photo_cache", strconv.Itoa(album_id)) err := os.RemoveAll(cache_path) if err != nil { - ScannerError("Could not delete unused cache folder: %s\n%s\n", cache_path, err) + deleteErrors = append(deleteErrors, errors.Wrapf(err, "delete unused cache folder (%s)", cache_path)) } } if len(deleted_album_ids) > 0 { albums_questions = strings.Repeat("?,", len(deleted_album_ids))[:len(deleted_album_ids)*2-1] - if _, err := database.Exec("DELETE FROM album WHERE album_id IN ("+albums_questions+")", deleted_album_ids...); err != nil { + if _, err := db.Exec("DELETE FROM album WHERE album_id IN ("+albums_questions+")", deleted_album_ids...); err != nil { ScannerError("Could not delete old albums from database:\n%s\n", err) + deleteErrors = append(deleteErrors, errors.Wrap(err, "delete old albums from database")) } } - // Delete old photos - photo_args := make([]interface{}, 0) - photo_args = append(photo_args, user.UserID) - photo_args = append(photo_args, cache.photo_paths_scanned...) - - photo_questions := strings.Repeat("?,", len(cache.photo_paths_scanned))[:len(cache.photo_paths_scanned)*2-1] - - rows, err = database.Query(` - SELECT photo.photo_id as photo_id, album.album_id as album_id FROM photo JOIN album ON photo.album_id = album.album_id - WHERE album.owner_id = ? AND photo.path NOT IN (`+photo_questions+`) - `, photo_args...) - if err != nil { - ScannerError("Could not get deleted photos from database: %s\n", err) - return - } - defer rows.Close() - - deleted_photo_ids := make([]interface{}, 0) - - for rows.Next() { - var photo_id int - var album_id int - - if err := rows.Scan(&photo_id, &album_id); err != nil { - ScannerError("Could not parse photo to be removed (album_id %d, photo_id %d): %s\n", album_id, photo_id, err) - } - - deleted_photo_ids = append(deleted_photo_ids, photo_id) - cache_path := path.Join("./photo_cache", strconv.Itoa(album_id), strconv.Itoa(photo_id)) - err := os.RemoveAll(cache_path) - if err != nil { - ScannerError("Could not delete unused cache photo folder: %s\n%s\n", cache_path, err) - } - } - - if len(deleted_photo_ids) > 0 { - photo_questions = strings.Repeat("?,", len(deleted_photo_ids))[:len(deleted_photo_ids)*2-1] - - if _, err := database.Exec("DELETE FROM photo WHERE photo_id IN ("+photo_questions+")", deleted_photo_ids...); err != nil { - ScannerError("Could not delete old photos from database:\n%s\n", err) - } - } - - if len(deleted_album_ids) > 0 || len(deleted_photo_ids) > 0 { - timeout := 3000 - notification.BroadcastNotification(&models.Notification{ - Key: utils.GenerateToken(), - Type: models.NotificationTypeMessage, - Header: "Deleted old photos", - Content: fmt.Sprintf("Deleted %d albums and %d photos, that was not found on disk", len(deleted_album_ids), len(deleted_photo_ids)), - Timeout: &timeout, - }) - } - + return deleteErrors } +// func cleanupCache(database *sql.DB, cache *ScannerCache, user *models.User) { + +// // Delete old photos +// photo_args := make([]interface{}, 0) +// photo_args = append(photo_args, user.UserID) +// photo_args = append(photo_args, cache.photo_paths_scanned...) + +// photo_questions := strings.Repeat("?,", len(cache.photo_paths_scanned))[:len(cache.photo_paths_scanned)*2-1] + +// rows, err = database.Query(` +// SELECT photo.photo_id as photo_id, album.album_id as album_id FROM photo JOIN album ON photo.album_id = album.album_id +// WHERE album.owner_id = ? AND photo.path NOT IN (`+photo_questions+`) +// `, photo_args...) +// if err != nil { +// ScannerError("Could not get deleted photos from database: %s\n", err) +// return +// } +// defer rows.Close() + +// deleted_photo_ids := make([]interface{}, 0) + +// for rows.Next() { +// var photo_id int +// var album_id int + +// if err := rows.Scan(&photo_id, &album_id); err != nil { +// ScannerError("Could not parse photo to be removed (album_id %d, photo_id %d): %s\n", album_id, photo_id, err) +// } + +// deleted_photo_ids = append(deleted_photo_ids, photo_id) +// cache_path := path.Join("./photo_cache", strconv.Itoa(album_id), strconv.Itoa(photo_id)) +// err := os.RemoveAll(cache_path) +// if err != nil { +// ScannerError("Could not delete unused cache photo folder: %s\n%s\n", cache_path, err) +// } +// } + +// if len(deleted_photo_ids) > 0 { +// photo_questions = strings.Repeat("?,", len(deleted_photo_ids))[:len(deleted_photo_ids)*2-1] + +// if _, err := database.Exec("DELETE FROM photo WHERE photo_id IN ("+photo_questions+")", deleted_photo_ids...); err != nil { +// ScannerError("Could not delete old photos from database:\n%s\n", err) +// } +// } + +// if len(deleted_album_ids) > 0 || len(deleted_photo_ids) > 0 { +// timeout := 3000 +// notification.BroadcastNotification(&models.Notification{ +// Key: utils.GenerateToken(), +// Type: models.NotificationTypeMessage, +// Header: "Deleted old photos", +// Content: fmt.Sprintf("Deleted %d albums and %d photos, that was not found on disk", len(deleted_album_ids), len(deleted_photo_ids)), +// Timeout: &timeout, +// }) +// } + +// } + func ScannerError(format string, args ...interface{}) { message := fmt.Sprintf(format, args...)