chore: moved business logic out of repository layer

music-player/pkg/repository/scan.go

@@ -1,122 +0,0 @@
-package repository
-
-import (
-	"github.com/felamaslen/go-music-player/pkg/config"
-	"github.com/felamaslen/go-music-player/pkg/database"
-	"github.com/felamaslen/go-music-player/pkg/logger"
-	"github.com/felamaslen/go-music-player/pkg/read"
-	"github.com/lib/pq"
-)
-
-const BATCH_SIZE = 100
-const LOG_EVERY = 100;
-
-func InsertMusicIntoDatabase(songs chan *read.Song) {
-  var l = logger.CreateLogger(config.GetConfig().LogLevel)
-
-  db := database.GetConnection()
-
-  var batch [BATCH_SIZE]*read.Song
-  var batchSize = 0
-  var numAdded = 0
-
-  var processBatch = func() {
-    if batchSize == 0 {
-      return
-    }
-
-    l.Debug("[INSERT] Processing batch\n")
-
-    var trackNumbers pq.Int64Array
-    var titles pq.StringArray
-    var artists pq.StringArray
-    var albums pq.StringArray
-    var durations pq.Int64Array
-
-    var modifiedDates pq.Int64Array
-
-    var basePaths pq.StringArray
-    var relativePaths pq.StringArray
-
-    for i := 0; i < batchSize; i++ {
-      trackNumbers = append(trackNumbers, int64(batch[i].TrackNumber))
-      titles = append(titles, batch[i].Title)
-      artists = append(artists, batch[i].Artist)
-      albums = append(albums, batch[i].Album)
-      durations = append(durations, int64(batch[i].Duration))
-
-      modifiedDates = append(modifiedDates, batch[i].ModifiedDate)
-
-      basePaths = append(basePaths, batch[i].BasePath)
-      relativePaths = append(relativePaths, batch[i].RelativePath)
-    }
-
-    db.MustExec(
-      `
-      insert into songs (
-        track_number
-        ,title
-        ,artist
-        ,album
-        ,duration
-        ,modified_date
-        ,base_path
-        ,relative_path
-      )
-      select * from unnest(
-        $1::integer[]
-        ,$2::varchar[]
-        ,$3::varchar[]
-        ,$4::varchar[]
-        ,$5::integer[]
-        ,$6::bigint[]
-        ,$7::varchar[]
-        ,$8::varchar[]
-      )
-      on conflict (base_path, relative_path) do update
-      set
-        track_number = excluded.track_number
-        ,title = excluded.title
-        ,artist = excluded.artist
-        ,album = excluded.album
-        ,duration = excluded.duration
-        ,modified_date = excluded.modified_date
-      `,
-      trackNumbers,
-      titles,
-      artists,
-      albums,
-      durations,
-      modifiedDates,
-      basePaths,
-      relativePaths,
-    )
-
-    l.Debug("[INSERT] Processed batch\n")
-
-    batchSize = 0
-  }
-
-  for {
-    select {
-    case song, more := <- songs:
-      if !more {
-        processBatch()
-        l.Verbose("[INSERT] Finished inserting %d songs\n", numAdded)
-        return
-      }
-
-      batch[batchSize] = song
-      batchSize++
-
-      numAdded++
-      if numAdded % LOG_EVERY == 0 {
-        l.Verbose("[INSERT] Inserted %d\n", numAdded)
-      }
-
-      if batchSize >= BATCH_SIZE {
-        processBatch()
-      }
-    }
-  }
-}

music-player/pkg/repository/scan_test.go

@@ -1,152 +0,0 @@
-package repository_test
-
-import (
-	. "github.com/onsi/ginkgo"
-	. "github.com/onsi/gomega"
-
-	"github.com/felamaslen/go-music-player/pkg/database"
-	"github.com/felamaslen/go-music-player/pkg/read"
-	"github.com/felamaslen/go-music-player/pkg/repository"
-	setup "github.com/felamaslen/go-music-player/pkg/testing"
-)
-
-var _ = Describe("scanning repository", func() {
-  db := database.GetConnection()
-
-  BeforeEach(func() {
-    setup.PrepareDatabaseForTesting()
-  })
-
-  Describe("when the channel sends two files", func() {
-    var songs chan *read.Song
-
-    var testInsertSongs = func() {
-      songs = make(chan *read.Song)
-
-      go func() {
-	defer close(songs)
-	songs <- &read.Song{
-	  TrackNumber: 7,
-	  Title: "Hey Jude",
-	  Artist: "The Beatles",
-	  Album: "",
-	  Duration: 431,
-	  BasePath: "/path/to",
-	  RelativePath: "file.ogg",
-	  ModifiedDate: 8876,
-	}
-
-	songs <- &read.Song{
-	  TrackNumber: 11,
-	  Title: "Starman",
-	  Artist: "David Bowie",
-	  Album: "The Rise and Fall of Ziggy Stardust and the Spiders from Mars",
-	  Duration: 256,
-	  BasePath: "/different/path",
-	  RelativePath: "otherFile.ogg",
-	  ModifiedDate: 11883,
-	}
-      }()
-
-      repository.InsertMusicIntoDatabase(songs)
-    }
-
-    Context("when the songs do not already exist in the database", func() {
-      BeforeEach(testInsertSongs)
-
-      It("should insert the correct number of songs", func() {
-	var count int
-	db.Get(&count, "select count(*) from songs")
-	Expect(count).To(Equal(2))
-      })
-
-      It("should insert both songs", func() {
-	var songs []read.Song
-
-	db.Select(&songs, `
-	select track_number, title, artist, album, duration, base_path, relative_path, modified_date
-	from songs
-	order by title
-	`)
-
-	Expect(songs[0]).To(Equal(read.Song{
-	  TrackNumber: 7,
-	  Title: "Hey Jude",
-	  Artist: "The Beatles",
-	  Album: "",
-	  Duration: 431,
-	  BasePath: "/path/to",
-	  RelativePath: "file.ogg",
-	  ModifiedDate: 8876,
-	}))
-
-	Expect(songs[1]).To(Equal(read.Song{
-	  TrackNumber: 11,
-	  Title: "Starman",
-	  Artist: "David Bowie",
-	  Album: "The Rise and Fall of Ziggy Stardust and the Spiders from Mars",
-	  Duration: 256,
-	  BasePath: "/different/path",
-	  RelativePath: "otherFile.ogg",
-	  ModifiedDate: 11883,
-	}))
-      })
-    })
-
-    Context("when there is already a file in the database with the same name", func() {
-      BeforeEach(func() {
-	db.MustExec(
-	  `
-	  insert into songs (title, artist, album, base_path, relative_path, modified_date)
-	  values ($1, $2, $3, $4, $5, $6)
-	  `,
-	  "my title",
-	  "my artist",
-	  "my album",
-	  "/path/to",
-	  "file.ogg",
-	  7782,
-	)
-
-	testInsertSongs()
-      })
-
-      It("should not add an additional row for the same file", func() {
-	var count int
-	db.Get(&count, `
-	select count(*) from songs
-	where base_path = '/path/to' and relative_path = 'file.ogg'
-	`)
-
-	Expect(count).To(Equal(1))
-      })
-
-      It("should upsert the existing item", func() {
-	var songs []read.Song
-	db.Select(&songs, `
-	select
-	  track_number
-	  ,title
-	  ,artist
-	  ,album
-	  ,duration
-	  ,base_path
-	  ,relative_path
-	  ,modified_date
-	from songs
-	where base_path = '/path/to' and relative_path = 'file.ogg'
-	`)
-
-	Expect(songs).To(HaveLen(1))
-	var song = songs[0]
-
-	Expect(song.TrackNumber).To(Equal(7))
-	Expect(song.Title).To(Equal("Hey Jude"))
-	Expect(song.Artist).To(Equal("The Beatles"))
-	Expect(song.Album).To(Equal(""))
-	Expect(song.Duration).To(Equal(431))
-	Expect(song.ModifiedDate).To(Equal(int64(8876)))
-      })
-    })
-  })
-})

music-player/pkg/repository/songs.go

@@ -5,8 +5,11 @@ import (
 
 	"github.com/felamaslen/go-music-player/pkg/read"
 	"github.com/jmoiron/sqlx"
+	"github.com/lib/pq"
 )
 
+const BATCH_SIZE = 100
+
 func SelectSong(db *sqlx.DB, id int) (song *read.Song, err error) {
   var songs []*read.Song
 
@@ -32,3 +35,72 @@ func SelectSong(db *sqlx.DB, id int) (song *read.Song, err error) {
 
   return
 }
+
+func BatchUpsertSongs(db *sqlx.DB, batch *[BATCH_SIZE]*read.Song, batchSize int) error {
+  var trackNumbers pq.Int64Array
+  var titles pq.StringArray
+  var artists pq.StringArray
+  var albums pq.StringArray
+  var durations pq.Int64Array
+
+  var modifiedDates pq.Int64Array
+
+  var basePaths pq.StringArray
+  var relativePaths pq.StringArray
+
+  for i := 0; i < batchSize; i++ {
+    trackNumbers = append(trackNumbers, int64((*batch)[i].TrackNumber))
+    titles = append(titles, (*batch)[i].Title)
+    artists = append(artists, (*batch)[i].Artist)
+    albums = append(albums, (*batch)[i].Album)
+    durations = append(durations, int64((*batch)[i].Duration))
+
+    modifiedDates = append(modifiedDates, (*batch)[i].ModifiedDate)
+
+    basePaths = append(basePaths, (*batch)[i].BasePath)
+    relativePaths = append(relativePaths, (*batch)[i].RelativePath)
+  }
+
+  _, err := db.Exec(
+    `
+    insert into songs (
+      track_number
+      ,title
+      ,artist
+      ,album
+      ,duration
+      ,modified_date
+      ,base_path
+      ,relative_path
+    )
+    select * from unnest(
+      $1::integer[]
+      ,$2::varchar[]
+      ,$3::varchar[]
+      ,$4::varchar[]
+      ,$5::integer[]
+      ,$6::bigint[]
+      ,$7::varchar[]
+      ,$8::varchar[]
+    )
+    on conflict (base_path, relative_path) do update
+    set
+      track_number = excluded.track_number
+      ,title = excluded.title
+      ,artist = excluded.artist
+      ,album = excluded.album
+      ,duration = excluded.duration
+      ,modified_date = excluded.modified_date
+    `,
+    trackNumbers,
+    titles,
+    artists,
+    albums,
+    durations,
+    modifiedDates,
+    basePaths,
+    relativePaths,
+  )
+
+  return err
+}

music-player/pkg/repository/songs_test.go

@@ -17,7 +17,7 @@ var _ = Describe("songs repository", func() {
     setup.PrepareDatabaseForTesting()
   })
 
-  Context("when reading", func() {
+  Describe("SelectSong", func() {
     var id int64
 
     BeforeEach(func() {
@@ -66,4 +66,76 @@ var _ = Describe("songs repository", func() {
       })
     })
   })
+
+  Describe("BatchUpsertSongs", func() {
+    songs := [100]*read.Song{
+      {
+	TrackNumber: 1,
+	Title: "Title A",
+	Artist: "Artist A",
+	Album: "Album A",
+	Duration: 123,
+	BasePath: "/base/path/1",
+	RelativePath: "song1.ogg",
+	ModifiedDate: 8886663103,
+      },
+      {
+	TrackNumber: 2,
+	Title: "Title B",
+	Artist: "Artist B",
+	Album: "Album B",
+	Duration: 456,
+	BasePath: "/base/path/2",
+	RelativePath: "song2.ogg",
+	ModifiedDate: 2711291992,
+      },
+    }
+
+    Context("when the songs do not already exist", func() {
+      BeforeEach(func() {
+	repository.BatchUpsertSongs(db, &songs, 2)
+      })
+
+      It("should insert the batch into the database", func() {
+	var result []*read.Song
+	db.Select(&result, `
+	select track_number, title, artist, album, duration, base_path, relative_path, modified_date
+	from songs
+	`)
+
+	Expect(result).To(HaveLen(2))
+	Expect(songs[0]).To(BeElementOf(result))
+	Expect(songs[1]).To(BeElementOf(result))
+      })
+    })
+
+    Context("when the songs already exist", func() {
+      var result []*read.Song
+      var modifiedBatch [100]*read.Song
+
+      modifiedBatch[0] = songs[0]
+      modifiedBatch[1] = songs[1]
+
+      modifiedBatch[0].Title = "Title A modified"
+
+      BeforeEach(func() {
+	repository.BatchUpsertSongs(db, &songs, 2)
+
+	repository.BatchUpsertSongs(db, &modifiedBatch, 2)
+
+	db.Select(&result, `
+	select track_number, title, artist, album, duration, base_path, relative_path, modified_date
+	from songs
+	`)
+      })
+
+      It("should not create any additional rows", func() {
+	Expect(result).To(HaveLen(2))
+      })
+
+      It("should update the rows with any changes", func() {
+	Expect(modifiedBatch[0]).To(BeElementOf(result))
+      })
+    })
+  })
 })

music-player/pkg/services/scanner.go

@@ -2,11 +2,63 @@ package services
 
 import (
 	"github.com/felamaslen/go-music-player/pkg/config"
+	"github.com/felamaslen/go-music-player/pkg/database"
 	"github.com/felamaslen/go-music-player/pkg/logger"
 	"github.com/felamaslen/go-music-player/pkg/read"
 	"github.com/felamaslen/go-music-player/pkg/repository"
 )
 
+const LOG_EVERY = 100;
+
+const BATCH_SIZE = 100
+
+func UpsertSongsFromChannel(songs chan *read.Song) {
+  var l = logger.CreateLogger(config.GetConfig().LogLevel)
+
+  db := database.GetConnection()
+
+  var batch [BATCH_SIZE]*read.Song
+  var batchSize = 0
+  var numAdded = 0
+
+  var processBatch = func() {
+    if batchSize == 0 {
+      return
+    }
+
+    l.Debug("[INSERT] Processing batch\n")
+    if err := repository.BatchUpsertSongs(db, &batch, batchSize); err != nil {
+      panic(err)
+    }
+    l.Debug("[INSERT] Processed batch\n")
+
+    batchSize = 0
+  }
+
+  for {
+    select {
+    case song, more := <- songs:
+      if !more {
+        processBatch()
+        l.Verbose("[INSERT] Finished inserting %d songs\n", numAdded)
+        return
+      }
+
+      batch[batchSize] = song
+      batchSize++
+
+      numAdded++
+      if numAdded % LOG_EVERY == 0 {
+        l.Verbose("[INSERT] Inserted %d\n", numAdded)
+      }
+
+      if batchSize >= BATCH_SIZE {
+        processBatch()
+      }
+    }
+  }
+}
+
 func ScanAndInsert(musicDirectory string) {
   var l = logger.CreateLogger(config.GetConfig().LogLevel)
 
@@ -17,8 +69,7 @@ func ScanAndInsert(musicDirectory string) {
   songs := read.ReadMultipleFiles(musicDirectory, files)
 
   l.Info("Inserting data...\n")
-  repository.InsertMusicIntoDatabase(songs)
+  UpsertSongsFromChannel(songs)
 
   l.Info("Finished scan and insert\n")
 }
-

music-player/pkg/services/scanner_test.go

@@ -10,43 +10,177 @@ import (
 	setup "github.com/felamaslen/go-music-player/pkg/testing"
 )
 
-var _ = Describe("music scanner (integration test)", func() {
+var _ = Describe("Music scanner service", func() {
+  db := database.GetConnection()
 
   BeforeEach(func() {
     setup.PrepareDatabaseForTesting()
   })
 
-  It("should recursively scan files from a directory and add them to the database", func() {
-    services.ScanAndInsert(read.TestDirectory)
-
-    db := database.GetConnection()
-
-    var songs []read.Song
-    err := db.Select(&songs, `
-      select title, artist, album, duration, base_path, relative_path
-      from songs
-    `)
-
-    Expect(err).To(BeNil())
-
-    Expect(songs).To(HaveLen(2))
-
-    Expect(read.Song{
-      Title: read.TestSong.Title,
-      Artist: read.TestSong.Artist,
-      Album: read.TestSong.Album,
-      Duration: read.TestSong.Duration,
-      BasePath: read.TestSong.BasePath,
-      RelativePath: read.TestSong.RelativePath,
-    }).To(BeElementOf(songs))
-
-    Expect(read.Song{
-      Title: read.TestSongNested.Title,
-      Artist: read.TestSongNested.Artist,
-      Album: read.TestSongNested.Album,
-      Duration: read.TestSongNested.Duration,
-      BasePath: read.TestSongNested.BasePath,
-      RelativePath: read.TestSongNested.RelativePath,
-    }).To(BeElementOf(songs))
+  Describe("UpsertSongsFromChannel", func() {
+    var songs chan *read.Song
+
+    var testScanSongs = func() {
+      songs = make(chan *read.Song)
+
+      go func() {
+	defer close(songs)
+	songs <- &read.Song{
+	  TrackNumber: 7,
+	  Title: "Hey Jude",
+	  Artist: "The Beatles",
+	  Album: "",
+	  Duration: 431,
+	  BasePath: "/path/to",
+	  RelativePath: "file.ogg",
+	  ModifiedDate: 8876,
+	}
+
+	songs <- &read.Song{
+	  TrackNumber: 11,
+	  Title: "Starman",
+	  Artist: "David Bowie",
+	  Album: "The Rise and Fall of Ziggy Stardust and the Spiders from Mars",
+	  Duration: 256,
+	  BasePath: "/different/path",
+	  RelativePath: "otherFile.ogg",
+	  ModifiedDate: 11883,
+	}
+      }()
+
+      services.UpsertSongsFromChannel(songs)
+    }
+
+    Context("when the songs do not already exist in the database", func() {
+      BeforeEach(testScanSongs)
+
+      It("should insert the correct number of songs", func() {
+	var count int
+	db.Get(&count, "select count(*) from songs")
+	Expect(count).To(Equal(2))
+      })
+
+      It("should insert both songs", func() {
+	var songs []read.Song
+
+	db.Select(&songs, `
+	select track_number, title, artist, album, duration, base_path, relative_path, modified_date
+	from songs
+	order by title
+	`)
+
+	Expect(songs[0]).To(Equal(read.Song{
+	  TrackNumber: 7,
+	  Title: "Hey Jude",
+	  Artist: "The Beatles",
+	  Album: "",
+	  Duration: 431,
+	  BasePath: "/path/to",
+	  RelativePath: "file.ogg",
+	  ModifiedDate: 8876,
+	}))
+
+	Expect(songs[1]).To(Equal(read.Song{
+	  TrackNumber: 11,
+	  Title: "Starman",
+	  Artist: "David Bowie",
+	  Album: "The Rise and Fall of Ziggy Stardust and the Spiders from Mars",
+	  Duration: 256,
+	  BasePath: "/different/path",
+	  RelativePath: "otherFile.ogg",
+	  ModifiedDate: 11883,
+	}))
+      })
+    })
+
+    Context("when there is already a file in the database with the same name", func() {
+      BeforeEach(func() {
+	db.MustExec(
+	  `
+	  insert into songs (title, artist, album, base_path, relative_path, modified_date)
+	  values ($1, $2, $3, $4, $5, $6)
+	  `,
+	  "my title",
+	  "my artist",
+	  "my album",
+	  "/path/to",
+	  "file.ogg",
+	  7782,
+	)
+
+	testScanSongs()
+      })
+
+      It("should not add an additional row for the same file", func() {
+	var count int
+	db.Get(&count, `
+	select count(*) from songs
+	where base_path = '/path/to' and relative_path = 'file.ogg'
+	`)
+
+	Expect(count).To(Equal(1))
+      })
+
+      It("should upsert the existing item", func() {
+	var songs []read.Song
+	db.Select(&songs, `
+	select
+	  track_number
+	  ,title
+	  ,artist
+	  ,album
+	  ,duration
+	  ,base_path
+	  ,relative_path
+	  ,modified_date
+	from songs
+	where base_path = '/path/to' and relative_path = 'file.ogg'
+	`)
+
+	Expect(songs).To(HaveLen(1))
+	var song = songs[0]
+
+	Expect(song.TrackNumber).To(Equal(7))
+	Expect(song.Title).To(Equal("Hey Jude"))
+	Expect(song.Artist).To(Equal("The Beatles"))
+	Expect(song.Album).To(Equal(""))
+	Expect(song.Duration).To(Equal(431))
+	Expect(song.ModifiedDate).To(Equal(int64(8876)))
+      })
+    })
+  })
+
+  Describe("ScanAndInsert", func() {
+    It("should recursively scan files from a directory and add them to the database", func() {
+      services.ScanAndInsert(read.TestDirectory)
+
+      var songs []read.Song
+      err := db.Select(&songs, `
+	select title, artist, album, duration, base_path, relative_path
+	from songs
+      `)
+
+      Expect(err).To(BeNil())
+
+      Expect(songs).To(HaveLen(2))
+
+      Expect(read.Song{
+	Title: read.TestSong.Title,
+	Artist: read.TestSong.Artist,
+	Album: read.TestSong.Album,
+	Duration: read.TestSong.Duration,
+	BasePath: read.TestSong.BasePath,
+	RelativePath: read.TestSong.RelativePath,
+      }).To(BeElementOf(songs))
+
+      Expect(read.Song{
+	Title: read.TestSongNested.Title,
+	Artist: read.TestSongNested.Artist,
+	Album: read.TestSongNested.Album,
+	Duration: read.TestSongNested.Duration,
+	BasePath: read.TestSongNested.BasePath,
+	RelativePath: read.TestSongNested.RelativePath,
+      }).To(BeElementOf(songs))
+    })
   })
 })