diff --git a/api/graphql/notification/Notification.go b/api/graphql/notification/Notification.go index 7c7d3de1..b845c978 100644 --- a/api/graphql/notification/Notification.go +++ b/api/graphql/notification/Notification.go @@ -2,9 +2,10 @@ package notification import ( "errors" - "github.com/viktorstrate/photoview/api/graphql/models" "log" "sync" + + "github.com/viktorstrate/photoview/api/graphql/models" ) type NotificationChannel = chan<- *models.Notification @@ -68,6 +69,10 @@ func DeregisterListener(listenerID int) error { func BroadcastNotification(notification *models.Notification) { + if notification == nil { + return + } + log.Printf("Broadcasting notification: %s\n", notification.Header) notificationLock.Lock() diff --git a/api/graphql/resolvers/scanner.go b/api/graphql/resolvers/scanner.go index 788994a1..e57105e1 100644 --- a/api/graphql/resolvers/scanner.go +++ b/api/graphql/resolvers/scanner.go @@ -9,35 +9,21 @@ import ( ) 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 - // } + err := scanner.AddAllToQueue() + if err != nil { + return nil, err + } - // startMessage := "Scanner started" + startMessage := "Scanner started" - // return &models.ScannerResult{ - // Finished: false, - // Success: true, - // Message: &startMessage, - // }, nil - panic("not implemented") + return &models.ScannerResult{ + Finished: false, + Success: true, + Message: &startMessage, + }, nil } 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 - // } - row := r.Database.QueryRow("SELECT * FROM user WHERE user_id = ?", userID) user, err := models.NewUserFromRow(row) if err != nil { diff --git a/api/routes/photos.go b/api/routes/photos.go index 1e6bd02a..55ce82f5 100644 --- a/api/routes/photos.go +++ b/api/routes/photos.go @@ -151,7 +151,7 @@ func RegisterPhotoRoutes(db *sql.DB, router *mux.Router) { return } - err = scanner.ProcessPhoto(tx, photo) + _, err = scanner.ProcessPhoto(tx, photo) if err != nil { log.Printf("ERROR: processing image not found in cache: %s\n", err) w.WriteHeader(http.StatusInternalServerError) diff --git a/api/scanner/process_photo.go b/api/scanner/process_photo.go index a4fc6f94..284ccb33 100644 --- a/api/scanner/process_photo.go +++ b/api/scanner/process_photo.go @@ -43,41 +43,43 @@ func makePhotoURLChecker(tx *sql.Tx, photoID int) (func(purpose models.PhotoPurp }, nil } -func ProcessPhoto(tx *sql.Tx, photo *models.Photo) error { +func ProcessPhoto(tx *sql.Tx, photo *models.Photo) (bool, error) { log.Printf("Processing photo: %s\n", photo.Path) + didProcess := false + imageData := EncodeImageData{ photo: photo, } photoUrlFromDB, err := makePhotoURLChecker(tx, photo.PhotoID) if err != nil { - return err + return false, err } // original photo url origURL, err := photoUrlFromDB(models.PhotoOriginal) if err != nil { - return err + return false, err } // Thumbnail thumbURL, err := photoUrlFromDB(models.PhotoThumbnail) if err != nil { - return errors.Wrap(err, "error processing thumbnail") + return false, errors.Wrap(err, "error processing thumbnail") } // Highres highResURL, err := photoUrlFromDB(models.PhotoHighRes) if err != nil { - return errors.Wrap(err, "error processing highres") + return false, errors.Wrap(err, "error processing highres") } // Make sure photo cache directory exists photoCachePath, err := makePhotoCacheDir(photo) if err != nil { - return errors.Wrap(err, "cache directory error") + return false, errors.Wrap(err, "cache directory error") } // Generate high res jpeg @@ -86,9 +88,11 @@ func ProcessPhoto(tx *sql.Tx, photo *models.Photo) error { if highResURL == nil { + didProcess = true + contentType, err := imageData.ContentType() if err != nil { - return err + return false, err } if !contentType.isWebCompatible() { @@ -101,19 +105,19 @@ func ProcessPhoto(tx *sql.Tx, photo *models.Photo) error { err = imageData.EncodeHighRes(tx, baseImagePath) if err != nil { - return errors.Wrap(err, "creating high-res cached image") + return false, errors.Wrap(err, "creating high-res cached image") } photoDimensions, err = GetPhotoDimensions(baseImagePath) if err != nil { - return err + return false, err } _, err = tx.Exec("INSERT INTO photo_url (photo_id, photo_name, width, height, purpose, content_type) VALUES (?, ?, ?, ?, ?, ?)", photo.PhotoID, highres_name, photoDimensions.Width, photoDimensions.Height, models.PhotoHighRes, "image/jpeg") if err != nil { log.Printf("Could not insert highres photo url: %d, %s\n", photo.PhotoID, path.Base(photo.Path)) - return err + return false, err } } } else { @@ -122,31 +126,36 @@ func ProcessPhoto(tx *sql.Tx, photo *models.Photo) error { if _, err := os.Stat(baseImagePath); os.IsNotExist(err) { fmt.Printf("High-res photo found in database but not in cache, re-encoding photo to cache: %s\n", highResURL.PhotoName) + didProcess = true err = imageData.EncodeHighRes(tx, baseImagePath) if err != nil { - return errors.Wrap(err, "creating high-res cached image") + return false, errors.Wrap(err, "creating high-res cached image") } } } // Save original photo to database if origURL == nil { + didProcess = true + // Make sure photo dimensions is set if photoDimensions == nil { photoDimensions, err = GetPhotoDimensions(baseImagePath) if err != nil { - return err + return false, err } } if err = saveOriginalPhotoToDB(tx, photo, imageData, photoDimensions); err != nil { - return errors.Wrap(err, "saving original photo to database") + return false, errors.Wrap(err, "saving original photo to database") } } // Save thumbnail to cache if thumbURL == nil { + didProcess = true + thumbnail_name := fmt.Sprintf("thumbnail_%s_%s", path.Base(photo.Path), utils.GenerateToken()) thumbnail_name = strings.ReplaceAll(thumbnail_name, ".", "_") thumbnail_name = strings.ReplaceAll(thumbnail_name, " ", "_") @@ -161,28 +170,29 @@ func ProcessPhoto(tx *sql.Tx, photo *models.Photo) error { thumbSize, err := EncodeThumbnail(baseImagePath, thumbOutputPath) if err != nil { - return errors.Wrap(err, "could not create thumbnail cached image") + return false, errors.Wrap(err, "could not create thumbnail cached image") } _, err = tx.Exec("INSERT INTO photo_url (photo_id, photo_name, width, height, purpose, content_type) VALUES (?, ?, ?, ?, ?, ?)", photo.PhotoID, thumbnail_name, thumbSize.Width, thumbSize.Height, models.PhotoThumbnail, "image/jpeg") if err != nil { - return err + return false, err } } else { // Verify that thumbnail photo still exists in cache thumbPath := path.Join(*photoCachePath, thumbURL.PhotoName) if _, err := os.Stat(thumbPath); os.IsNotExist(err) { + didProcess = true fmt.Printf("Thumbnail photo found in database but not in cache, re-encoding photo to cache: %s\n", thumbURL.PhotoName) _, err := EncodeThumbnail(baseImagePath, thumbPath) if err != nil { - return errors.Wrap(err, "could not create thumbnail cached image") + return false, errors.Wrap(err, "could not create thumbnail cached image") } } } - return nil + return didProcess, nil } func makePhotoCacheDir(photo *models.Photo) (*string, error) { diff --git a/api/scanner/queue.go b/api/scanner/queue.go index 1ca5cce2..ccd9643d 100644 --- a/api/scanner/queue.go +++ b/api/scanner/queue.go @@ -2,10 +2,13 @@ package scanner import ( "database/sql" + "fmt" "log" "sync" + "github.com/pkg/errors" "github.com/viktorstrate/photoview/api/graphql/models" + "github.com/viktorstrate/photoview/api/graphql/notification" ) type ScannerJob struct { @@ -78,9 +81,28 @@ func (queue *ScannerQueue) startBackgroundWorker() { }() } - log.Printf("Waiting jobs: %d\n", len(queue.up_next)) + in_progress_length := len(global_scanner_queue.in_progress) + up_next_length := len(global_scanner_queue.up_next) queue.mutex.Unlock() + + if in_progress_length+up_next_length == 0 { + notification.BroadcastNotification(&models.Notification{ + Key: "global-scanner-progress", + Type: models.NotificationTypeMessage, + Header: fmt.Sprintf("Scanner complete"), + Content: fmt.Sprintf("All jobs have been scanned"), + Positive: true, + }) + } else { + notification.BroadcastNotification(&models.Notification{ + Key: "global-scanner-progress", + Type: models.NotificationTypeMessage, + Header: fmt.Sprintf("Scanning photos"), + Content: fmt.Sprintf("%d jobs in progress\n%d jobs waiting", in_progress_length, up_next_length), + }) + } + } } @@ -94,11 +116,29 @@ func (queue *ScannerQueue) notify() bool { } } -func AddUserToQueue(user *models.User) { +func AddAllToQueue() error { + rows, err := global_scanner_queue.db.Query("SELECT * FROM user") + if err != nil { + return errors.Wrap(err, "get all users from database") + } + + users, err := models.NewUsersFromRows(rows) + if err != nil { + return errors.Wrap(err, "parse all users from db") + } + + for _, user := range users { + AddUserToQueue(user) + } + + return nil +} + +func AddUserToQueue(user *models.User) error { album_cache := MakeAlbumCache() albums, album_errors := findAlbumsForUser(global_scanner_queue.db, user, album_cache) for _, err := range album_errors { - log.Printf("User scanner error: %s", err) + return errors.Wrapf(err, "find albums for user (user_id: %s)", user.UserID) } global_scanner_queue.mutex.Lock() @@ -109,6 +149,8 @@ func AddUserToQueue(user *models.User) { }) } global_scanner_queue.mutex.Unlock() + + return nil } // Queue should be locked prior to calling this function diff --git a/api/scanner/scanner_album.go b/api/scanner/scanner_album.go index 07f4af63..d5917cb9 100644 --- a/api/scanner/scanner_album.go +++ b/api/scanner/scanner_album.go @@ -2,46 +2,81 @@ package scanner import ( "database/sql" + "fmt" "io/ioutil" "path" + "time" "github.com/viktorstrate/photoview/api/graphql/models" + "github.com/viktorstrate/photoview/api/graphql/notification" + "github.com/viktorstrate/photoview/api/utils" ) func scanAlbum(album *models.Album, cache *AlbumScannerCache, db *sql.DB) { + + album_notify_key := utils.GenerateToken() + notifyThrottle := utils.NewThrottle(500 * time.Millisecond) + notifyThrottle.Trigger(nil) + // 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 newPhoto { + notifyThrottle.Trigger(func() { + notification.BroadcastNotification(&models.Notification{ + Key: album_notify_key, + Type: models.NotificationTypeMessage, + Header: fmt.Sprintf("Found new photos in album '%s'", album.Title), + Content: fmt.Sprintf("Found photo %s", photo.Path), + }) + }) + } }) if err != nil { ScannerError("Failed to find photos for album (%s): %s", album.Path, err) } - for _, photo := range albumPhotos { + album_has_changes := false + + for count, photo := range albumPhotos { tx, err := db.Begin() if err != nil { ScannerError("Failed to begin database transaction: %s", err) } - err = ProcessPhoto(tx, photo) + processing_was_needed, err := ProcessPhoto(tx, photo) if err != nil { tx.Rollback() ScannerError("Failed to process photo (%s): %s", photo.Path, err) } + if processing_was_needed { + album_has_changes = true + progress := float64(count) / float64(len(albumPhotos)) * 100.0 + notification.BroadcastNotification(&models.Notification{ + Key: album_notify_key, + Type: models.NotificationTypeProgress, + Header: fmt.Sprintf("Processing photo for album '%s'", album.Title), + Content: fmt.Sprintf("Processed photo at %s", photo.Path), + Progress: &progress, + }) + } + err = tx.Commit() if err != nil { ScannerError("Failed to commit database transaction: %s", err) } + } - // TODO: Broadcast progress + if album_has_changes { + timeoutDelay := 2000 + notification.BroadcastNotification(&models.Notification{ + Key: album_notify_key, + Type: models.NotificationTypeMessage, + Positive: true, + Header: fmt.Sprintf("Done processing photos for album '%s'", album.Title), + Content: fmt.Sprintf("All photos have been processed"), + Timeout: &timeoutDelay, + }) } } diff --git a/api/utils/Throttle.go b/api/utils/Throttle.go index a0850d98..10d20856 100644 --- a/api/utils/Throttle.go +++ b/api/utils/Throttle.go @@ -15,6 +15,9 @@ func NewThrottle(interval time.Duration) Throttle { } func (t *Throttle) Trigger(action func()) { + if action == nil { + return + } if time.Now().After(t.lastAction.Add(t.interval)) { t.lastAction = time.Now() action()