Start to integrate new scanner system with api

api/graphql/resolvers/scanner.go

@@ -3,7 +3,9 @@ package resolvers
 import (
 	"context"
 
+	"github.com/pkg/errors"
 	"github.com/viktorstrate/photoview/api/graphql/models"
+	"github.com/viktorstrate/photoview/api/scanner"
 )
 
 func (r *mutationResolver) ScanAll(ctx context.Context) (*models.ScannerResult, error) {
@@ -36,12 +38,18 @@ func (r *mutationResolver) ScanUser(ctx context.Context, userID int) (*models.Sc
 	// 	}, nil
 	// }
 
-	// startMessage := "Scanner started"
+	row := r.Database.QueryRow("SELECT * FROM user WHERE user_id = ?", userID)
+	user, err := models.NewUserFromRow(row)
+	if err != nil {
+		return nil, errors.Wrap(err, "get user from database")
+	}
 
-	// return &models.ScannerResult{
-	// 	Finished: false,
-	// 	Success:  true,
-	// 	Message:  &startMessage,
-	// }, nil
-	panic("not implemented")
+	scanner.AddUserToQueue(user)
+
+	startMessage := "Scanner started"
+	return &models.ScannerResult{
+		Finished: false,
+		Success:  true,
+		Message:  &startMessage,
+	}, nil
 }

api/scanner/cache.go

@@ -1,10 +1,14 @@
 package scanner
 
-import "path"
+import (
+	"path"
+	"sync"
+)
 
 type AlbumScannerCache struct {
 	path_contains_photos map[string]bool
 	photo_types          map[string]ImageType
+	mutex                sync.Mutex
 }
 
 func MakeAlbumCache() *AlbumScannerCache {
@@ -16,6 +20,8 @@ func MakeAlbumCache() *AlbumScannerCache {
 
 // Insert single album directory in cache
 func (c *AlbumScannerCache) InsertAlbumPath(path string, contains_photo bool) {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
 	c.path_contains_photos[path] = contains_photo
 }
 
@@ -24,6 +30,9 @@ func (c *AlbumScannerCache) InsertAlbumPaths(end_path string, root string, conta
 	curr_path := path.Clean(end_path)
 	root_path := path.Clean(root)
 
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+
 	for curr_path != root_path || curr_path == "." {
 
 		c.InsertAlbumPath(curr_path, contains_photo)
@@ -33,6 +42,9 @@ func (c *AlbumScannerCache) InsertAlbumPaths(end_path string, root string, conta
 }
 
 func (c *AlbumScannerCache) AlbumContainsPhotos(path string) *bool {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+
 	contains_photo, found := c.path_contains_photos[path]
 	if found {
 		// log.Printf("Album cache hit: %s\n", path)
@@ -43,10 +55,16 @@ func (c *AlbumScannerCache) AlbumContainsPhotos(path string) *bool {
 }
 
 func (c *AlbumScannerCache) InsertPhotoType(path string, content_type ImageType) {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+
 	(c.photo_types)[path] = content_type
 }
 
 func (c *AlbumScannerCache) GetPhotoType(path string) *ImageType {
+	c.mutex.Lock()
+	defer c.mutex.Unlock()
+
 	result, found := c.photo_types[path]
 	if found {
 		// log.Printf("Image cache hit: %s\n", path)

api/scanner/queue.go

@@ -46,21 +46,25 @@ func InitializeScannerQueue(db *sql.DB) {
 
 func (queue *ScannerQueue) startBackgroundWorker() {
 	for {
+		log.Println("Queue waiting")
 		<-queue.idle_chan
+		log.Println("Queue waiting for lock")
 		queue.mutex.Lock()
-		defer queue.mutex.Unlock()
+		log.Println("Queue running")
 
 		for len(queue.in_progress) < queue.settings.max_concurrent_tasks && len(queue.up_next) > 0 {
+			log.Println("Queue starting job")
 			nextJob := queue.up_next[0]
 			queue.up_next = queue.up_next[1:]
 			queue.in_progress = append(queue.in_progress, nextJob)
 
 			go func() {
+				log.Println("Starting job")
 				nextJob.Run(queue.db)
-				queue.mutex.Lock()
-				defer queue.mutex.Unlock()
+				log.Println("Job finished")
 
 				// Delete finished job from queue
+				queue.mutex.Lock()
 				for i, x := range queue.in_progress {
 					if x == nextJob {
 						queue.in_progress[i] = queue.in_progress[len(queue.in_progress)-1]
@@ -68,15 +72,20 @@ func (queue *ScannerQueue) startBackgroundWorker() {
 						break
 					}
 				}
+				queue.mutex.Unlock()
 
-				queue.Notify()
+				queue.notify()
 			}()
 		}
+
+		log.Printf("Waiting jobs: %d\n", len(queue.up_next))
+
+		queue.mutex.Unlock()
 	}
 }
 
 // Notifies the queue that the jobs has changed
-func (queue *ScannerQueue) Notify() bool {
+func (queue *ScannerQueue) notify() bool {
 	select {
 	case queue.idle_chan <- true:
 		return true
@@ -85,21 +94,21 @@ func (queue *ScannerQueue) Notify() bool {
 	}
 }
 
-func (queue *ScannerQueue) ScanUser(user *models.User) {
+func AddUserToQueue(user *models.User) {
 	album_cache := MakeAlbumCache()
-	albums, album_errors := findAlbumsForUser(queue.db, user, album_cache)
+	albums, album_errors := findAlbumsForUser(global_scanner_queue.db, user, album_cache)
 	for _, err := range album_errors {
 		log.Printf("User scanner error: %s", err)
 	}
 
-	queue.mutex.Lock()
+	global_scanner_queue.mutex.Lock()
 	for _, album := range albums {
-		queue.addJob(&ScannerJob{
+		global_scanner_queue.addJob(&ScannerJob{
 			album: album,
 			cache: album_cache,
 		})
 	}
-	queue.mutex.Unlock()
+	global_scanner_queue.mutex.Unlock()
 }
 
 // Queue should be locked prior to calling this function
@@ -108,7 +117,7 @@ func (queue *ScannerQueue) addJob(job *ScannerJob) error {
 		return err
 	}
 	queue.up_next = append(queue.up_next, *job)
-	queue.Notify()
+	queue.notify()
 
 	return nil
 }

api/scanner/scanner_album.go

@@ -24,17 +24,23 @@ func scanAlbum(album *models.Album, cache *AlbumScannerCache, db *sql.DB) {
 		ScannerError("Failed to find photos for album (%s): %s", album.Path, err)
 	}
 
+	for _, photo := range albumPhotos {
 		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 {
+			tx.Rollback()
 			ScannerError("Failed to process photo (%s): %s", photo.Path, err)
 		}
 
+		err = tx.Commit()
+		if err != nil {
+			ScannerError("Failed to commit database transaction: %s", err)
+		}
+
 		// TODO: Broadcast progress
 	}
 }
@@ -60,7 +66,7 @@ func findPhotosForAlbum(album *models.Album, cache *AlbumScannerCache, db *sql.D
 
 			photo, isNewPhoto, err := ScanPhoto(tx, photoPath, album.AlbumID)
 			if err != nil {
-				ScannerError("Scanning image %s: %s", photoPath, err)
+				ScannerError("Scanning image (%s): %s", photoPath, err)
 				tx.Rollback()
 				continue
 			}

api/scanner/scanner_user.go

@@ -164,7 +164,7 @@ func deleteOldUserAlbums(db *sql.DB, scannedAlbums []*models.Album, user *models
 
 	albumPaths := make([]interface{}, len(scannedAlbums))
 	for i, album := range scannedAlbums {
-		albumPaths[i] = album.AlbumID
+		albumPaths[i] = album.Path
 	}
 
 	// Delete old albums

api/server.go

@@ -13,6 +13,7 @@ import (
 	"github.com/viktorstrate/photoview/api/database"
 	"github.com/viktorstrate/photoview/api/graphql/auth"
 	"github.com/viktorstrate/photoview/api/routes"
+	"github.com/viktorstrate/photoview/api/scanner"
 	"github.com/viktorstrate/photoview/api/server"
 	"github.com/viktorstrate/photoview/api/utils"
 
@@ -40,6 +41,8 @@ func main() {
 		log.Panicf("Could not migrate database: %s\n", err)
 	}
 
+	scanner.InitializeScannerQueue(db)
+
 	rootRouter := mux.NewRouter()
 
 	rootRouter.Use(auth.Middleware(db))