Reintregrate notifications with scanner

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

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 {

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)

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

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

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,
+		})
 	}
 }
 

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