listenbrainz: added a listens synchronizer. fixes #142

conf/lms.conf

@@ -35,10 +35,14 @@ deploy-path = "/";
 http-server-thread-count = 0;
 
 # ListenBrainz root API
-listenbrainz-api-url = "https://api.listenbrainz.org/1/";
+listenbrainz-api-base-url = "https://api.listenbrainz.org";
+# How many listens to retrieve when syncing (0 disables sync)
+listenbrainz-max-sync-listen-count = 1000;
+# How often to resync listens (0 disables sync)
+listenbrainz-sync-listens-period-hours = 1;
 
 # Acousticbrainz root API
-acousticbrainz-api-url = "https://acousticbrainz.org/api/v1/";
+acousticbrainz-api-base-url = "https://acousticbrainz.org/api";
 
 # Authentication
 # Available backends: "internal", "PAM", "http-headers"

src/libs/database/impl/Track.cpp

@@ -30,6 +30,7 @@
 #include "utils/Logger.hpp"
 
 #include "SqlQuery.hpp"
+#include "StringViewTraits.hpp"
 
 namespace Database {
 
@@ -151,12 +152,12 @@ Track::getById(Session& session, IdType id)
 }
 
 Track::pointer
-Track::getByMBID(Session& session, const UUID& mbid)
+Track::getByRecordingMBID(Session& session, const UUID& mbid)
 {
 	session.checkSharedLocked();
 
 	return session.getDboSession().find<Track>()
-		.where("mbid = ?").bind(std::string {mbid.getAsString()});
+		.where("recording_mbid = ?").bind(std::string {mbid.getAsString()});
 }
 
 Track::pointer
@@ -354,6 +355,18 @@ Track::getByFilter(Session& session,
 	return res;
 }
 
+std::vector<Track::pointer>
+Track::getByNameAndReleaseName(Session& session, std::string_view trackName, std::string_view releaseName)
+{
+	session.checkSharedLocked();
+	Wt::Dbo::collection<pointer> collection = session.getDboSession().query<Track::pointer>("SELECT t from track t")
+		.join("release r ON t.release_id = r.id")
+		.where("t.name = ?").bind(trackName)
+		.where("r.name = ?").bind(releaseName);
+
+	return std::vector<pointer>(collection.begin(), collection.end());
+}
+
 std::vector<Track::pointer>
 Track::getSimilarTracks(Session& session,
 				const std::unordered_set<IdType>& tracks,

src/libs/database/impl/TrackList.cpp

@@ -148,6 +148,18 @@ TrackList::getEntries(std::optional<std::size_t> offset, std::optional<std::size
 	return std::vector<Wt::Dbo::ptr<TrackListEntry>>(entries.begin(), entries.end());
 }
 
+Wt::Dbo::ptr<TrackListEntry>
+TrackList::getEntryByTrackAndDateTime(Wt::Dbo::ptr<Track> track, const Wt::WDateTime& dateTime) const
+{
+	assert(session());
+	assert(IdIsValid(self()->id()));
+
+	return session()->find<TrackListEntry>()
+			.where("tracklist_id = ?").bind(self().id())
+			.where("track_id = ?").bind(track.id())
+			.where("date_time = ?").bind(Wt::WDateTime::fromTime_t(dateTime.toTime_t()));
+}
+
 static
 Wt::Dbo::Query<Artist::pointer>
 createArtistsQuery(Wt::Dbo::Session& session, const std::string& queryStr, IdType tracklistId, const std::set<IdType>& clusterIds, std::optional<TrackArtistLinkType> linkType)

src/libs/database/impl/User.cpp

@@ -84,6 +84,15 @@ User::getAll(Session& session)
 	return std::vector<pointer>(res.begin(), res.end());
 }
 
+std::vector<IdType>
+User::getAllIds(Session& session)
+{
+	session.checkSharedLocked();
+
+	Wt::Dbo::collection<IdType> res = session.getDboSession().query<IdType>("SELECT id FROM user");
+	return std::vector<IdType>(res.begin(), res.end());
+}
+
 User::pointer
 User::getDemo(Session& session)
 {

src/libs/database/include/database/Track.hpp

@@ -23,6 +23,7 @@
 #include <filesystem>
 #include <optional>
 #include <string>
+#include <string_view>
 #include <unordered_set>
 #include <vector>
 
@@ -61,7 +62,7 @@ class Track : public Wt::Dbo::Dbo<Track>
 		static std::size_t getCount(Session& session);
 		static pointer getByPath(Session& session, const std::filesystem::path& p);
 		static pointer getById(Session& session, IdType id);
-		static pointer getByMBID(Session& session, const UUID& MBID);
+		static pointer getByRecordingMBID(Session& session, const UUID& MBID);
 		static std::vector<pointer>	getSimilarTracks(Session& session,
 							const std::unordered_set<IdType>& trackIds,
 							std::optional<std::size_t> offset = {},
@@ -73,6 +74,7 @@ class Track : public Wt::Dbo::Dbo<Track>
 							const std::vector<std::string>& keywords,    // if non empty, name must match all of these keywords
 							std::optional<Range> range,
 							bool& moreExpected);
+		static std::vector<pointer> getByNameAndReleaseName(Session& session, std::string_view trackName, std::string_view releaseName);
 
 		static std::vector<pointer>	getAll(Session& session, std::optional<std::size_t> limit = std::nullopt);
 		static std::vector<pointer>	getAllRandom(Session& session, const std::set<IdType>& clusters, std::optional<std::size_t> limit = std::nullopt);
@@ -188,28 +190,28 @@ class Track : public Wt::Dbo::Dbo<Track>
 		static const std::size_t _maxCopyrightLength = 128;
 		static const std::size_t _maxCopyrightURLLength = 128;
 
-		int					_scanVersion {};
+		int						_scanVersion {};
-		int					_trackNumber {};
+		int						_trackNumber {};
-		int					_discNumber {};
+		int						_discNumber {};
-		std::string			_discSubtitle;
+		std::string				_discSubtitle;
-		int					_totalTrack {};
+		int						_totalTrack {};
-		int					_totalDisc {};
+		int						_totalDisc {};
 		std::string				_name;
 		std::string				_artistName;
 		std::string				_releaseName;
-		std::chrono::duration<int, std::milli>	_duration;
+		std::chrono::duration<int, std::milli>	_duration {};
-		int					_year {};
+		int						_year {};
-		int					_originalYear {};
+		int						_originalYear {};
 		std::string				_filePath;
-		Wt::WDateTime				_fileLastWrite;
+		Wt::WDateTime			_fileLastWrite;
-		Wt::WDateTime				_fileAdded;
+		Wt::WDateTime			_fileAdded;
 		bool					_hasCover {};
 		std::string				_trackMBID;
 		std::string				_recordingMBID;
 		std::string				_copyright;
 		std::string				_copyrightURL;
-		std::optional<float>			_trackReplayGain;
+		std::optional<float>	_trackReplayGain;
-		std::optional<float>			_releaseReplayGain;
+		std::optional<float>	_releaseReplayGain;
 
 		Wt::Dbo::ptr<Release>				_release;
 		Wt::Dbo::collection<Wt::Dbo::ptr<TrackArtistLink>> _trackArtistLinks;

src/libs/database/include/database/TrackList.hpp

@@ -84,6 +84,9 @@ class TrackList : public Wt::Dbo::Dbo<TrackList>
 		std::size_t getCount() const;
 		Wt::Dbo::ptr<TrackListEntry> getEntry(std::size_t pos) const;
 		std::vector<Wt::Dbo::ptr<TrackListEntry>> getEntries(std::optional<std::size_t> offset = {}, std::optional<std::size_t> size = {}) const;
+		Wt::Dbo::ptr<TrackListEntry> getEntryByTrackAndDateTime(Wt::Dbo::ptr<Track> track, const Wt::WDateTime& dateTime) const;
+
+		// Get track bya
 
 		std::vector<Wt::Dbo::ptr<Artist>> getArtistsReverse(const std::set<IdType>& clusterIds, std::optional<TrackArtistLinkType> linkType, std::optional<Range> range, bool& moreResults) const;
 		std::vector<Wt::Dbo::ptr<Release>> getReleasesReverse(const std::set<IdType>& clusterIds, std::optional<Range> range, bool& moreResults) const;

src/libs/database/include/database/User.hpp

@@ -139,6 +139,7 @@ class User : public Wt::Dbo::Dbo<User>
 		static pointer			getById(Session& session, IdType id);
 		static pointer			getByLoginName(Session& session, std::string_view loginName);
 		static std::vector<pointer>	getAll(Session& session);
+		static std::vector<IdType>	getAllIds(Session& session);
 		static pointer			getDemo(Session& session);
 		static std::size_t		getCount(Session& session);
 

src/libs/scanner/impl/AcousticBrainzUtils.cpp

@@ -38,9 +38,9 @@ static
 std::string
 getJsonData(const UUID& mbid)
 {
-	static constexpr std::string_view defaultAPIURL {"https://acousticbrainz.org/api/v1/"};
+	static constexpr std::string_view defaultAPIURL {"https://acousticbrainz.org/api"};
 
-	const std::string url {std::string {Service<IConfig>::get()->getString("acousticbrainz-api-url", defaultAPIURL)} + std::string {mbid.getAsString()} + "/low-level"};
+	const std::string url {std::string {Service<IConfig>::get()->getString("acousticbrainz-api-base-url", defaultAPIURL)} + std::string {mbid.getAsString()} + "/low-level"};
 
 	boost::asio::io_service ioService;
 

src/libs/scrobbling/CMakeLists.txt

@@ -2,6 +2,9 @@
 add_library(lmsscrobbling SHARED
 	impl/internal/InternalScrobbler.cpp
 	impl/listenbrainz/ListenBrainzScrobbler.cpp
+	impl/listenbrainz/ListensSynchronizer.cpp
+	impl/listenbrainz/SendQueue.cpp
+	impl/listenbrainz/Utils.cpp
 	impl/Scrobbling.cpp
 	)
 
@@ -14,9 +17,12 @@ target_include_directories(lmsscrobbling PRIVATE
 	impl
 	)
 
+target_link_libraries(lmsscrobbling PRIVATE
+	lmsutils
+	)
+
 target_link_libraries(lmsscrobbling PUBLIC
 	lmsdatabase
-	lmsutils
 	)
 
 install(TARGETS lmsscrobbling DESTINATION lib)

src/libs/scrobbling/impl/IScrobbler.hpp

@@ -46,7 +46,7 @@ namespace Scrobbling
 			virtual void listenStarted(const Listen& listen) = 0;
 			virtual void listenFinished(const Listen& listen, std::optional<std::chrono::seconds> duration) = 0;
 
-			virtual void addListen(const Listen& listen, const Wt::WDateTime& timePoint) = 0;
+			virtual void addTimedListen(const TimedListen& listen) = 0;
 
 			virtual Wt::Dbo::ptr<Database::TrackList> getListensTrackList(Database::Session& session, Wt::Dbo::ptr<Database::User> user)  = 0;
 	};

src/libs/scrobbling/impl/Scrobbling.cpp

@@ -30,37 +30,37 @@
 namespace Scrobbling
 {
 	std::unique_ptr<IScrobbling>
-	createScrobbling(Database::Db& db)
+	createScrobbling(boost::asio::io_context& ioContext, Database::Db& db)
 	{
-		return std::make_unique<Scrobbling>(db);
+		return std::make_unique<Scrobbling>(ioContext, db);
 	}
 
-	Scrobbling::Scrobbling(Database::Db& db)
+	Scrobbling::Scrobbling(boost::asio::io_context& ioContext, Database::Db& db)
 		: _db {db}
 	{
 		_scrobblers.emplace(Database::Scrobbler::Internal, std::make_unique<InternalScrobbler>(_db));
-		_scrobblers.emplace(Database::Scrobbler::ListenBrainz, std::make_unique<ListenBrainzScrobbler>(_db));
+		_scrobblers.emplace(Database::Scrobbler::ListenBrainz, std::make_unique<ListenBrainz::Scrobbler>(ioContext, _db));
 	}
 
 	void
 	Scrobbling::listenStarted(const Listen& listen)
 	{
-		if (auto scrobbler {getUserScrobbler(listen.userId)})
+		if (std::optional<Database::Scrobbler> scrobbler {getUserScrobbler(listen.userId)})
 			_scrobblers[*scrobbler]->listenStarted(listen);
 	}
 
 	void
 	Scrobbling::listenFinished(const Listen& listen, std::optional<std::chrono::seconds> duration)
 	{
-		if (auto scrobbler {getUserScrobbler(listen.userId)})
+		if (std::optional<Database::Scrobbler> scrobbler {getUserScrobbler(listen.userId)})
 			_scrobblers[*scrobbler]->listenFinished(listen, duration);
 	}
 
 	void
-	Scrobbling::addListen(const Listen& listen, Wt::WDateTime timePoint)
+	Scrobbling::addTimedListen(const TimedListen& listen)
 	{
-		if (auto scrobbler {getUserScrobbler(listen.userId)})
+		if (std::optional<Database::Scrobbler> scrobbler {getUserScrobbler(listen.userId)})
-			_scrobblers[*scrobbler]->addListen(listen, timePoint);
+			_scrobblers[*scrobbler]->addTimedListen(listen);
 	}
 
 	std::optional<Database::Scrobbler>

src/libs/scrobbling/impl/Scrobbling.hpp

@@ -31,12 +31,12 @@ namespace Scrobbling
 	class Scrobbling : public IScrobbling
 	{
 		public:
-			Scrobbling(Database::Db& db);
+			Scrobbling(boost::asio::io_context& ioContext, Database::Db& db);
 
 		private:
 			void listenStarted(const Listen& listen) override;
 			void listenFinished(const Listen& listen, std::optional<std::chrono::seconds> duration) override;
-			void addListen(const Listen& listen, Wt::WDateTime timePoint) override;
+			void addTimedListen(const TimedListen& listen) override;
 
 			std::vector<Wt::Dbo::ptr<Database::Artist>> getRecentArtists(Database::Session& session,
 																	Wt::Dbo::ptr<Database::User> user,

src/libs/scrobbling/impl/internal/InternalScrobbler.cpp

@@ -47,11 +47,11 @@ namespace Scrobbling
 		if (duration && *duration < std::chrono::seconds {5})
 			return;
 
-		addListen(listen, Wt::WDateTime::currentDateTime());
+		addTimedListen({listen, Wt::WDateTime::currentDateTime()});
 	}
 
 	void
-	InternalScrobbler::addListen(const Listen& listen, const Wt::WDateTime& timePoint)
+	InternalScrobbler::addTimedListen(const TimedListen& listen)
 	{
 		Database::Session& session {_db.getTLSSession()};
 
@@ -69,7 +69,7 @@ namespace Scrobbling
 		if (!track)
 			return;
 
-		Database::TrackListEntry::create(session, track, getListensTrackList(session, user), timePoint);
+		Database::TrackListEntry::create(session, track, getListensTrackList(session, user), listen.listenedAt);
 	}
 
 	Wt::Dbo::ptr<Database::TrackList>

src/libs/scrobbling/impl/internal/InternalScrobbler.hpp

@@ -32,7 +32,7 @@ namespace Scrobbling
 			void listenStarted(const Listen& listen) override;
 			void listenFinished(const Listen& listen, std::optional<std::chrono::seconds> duration) override;
 
-			void addListen(const Listen& listen, const Wt::WDateTime& timePoint) override;
+			void addTimedListen(const TimedListen& listen) override;
 
 			Wt::Dbo::ptr<Database::TrackList> getListensTrackList(Database::Session& session, Wt::Dbo::ptr<Database::User> user) override;
 

src/libs/scrobbling/impl/listenbrainz/ListenBrainzScrobbler.cpp

@@ -34,41 +34,12 @@
 #include "utils/IConfig.hpp"
 #include "utils/Logger.hpp"
 #include "utils/Service.hpp"
+#include "Utils.hpp"
 
 #define LOG(sev)	LMS_LOG(SCROBBLING, sev) << "[listenbrainz] - "
 
-namespace StringUtils
-{
-	template<>
-	std::optional<std::chrono::seconds>
-	readAs(const std::string& str)
-	{
-		std::optional<std::chrono::seconds> res;
-
-		if (const std::optional<std::size_t> value {StringUtils::readAs<std::size_t>(str)})
-			res = std::chrono::seconds {*value};
-
-		return res;
-	}
-}
-
 namespace
 {
-	std::optional<UUID>
-	getListenBrainzToken(Database::Session& session, Database::IdType userId)
-	{
-		auto transaction {session.createSharedTransaction()};
-
-		const Database::User::pointer user {Database::User::getById(session, userId)};
-		if (!user)
-			return std::nullopt;
-
-		if (user->getScrobbler() != Database::Scrobbler::ListenBrainz)
-			return std::nullopt;
-
-		return user->getListenBrainzToken();
-	}
-
 	bool
 	canBeScrobbled(Database::Session& session, Database::IdType trackId, std::chrono::seconds duration)
 	{
@@ -164,252 +135,105 @@ namespace
 		res = Wt::Json::serialize(root);
 		return res;
 	}
-
-	template <typename T>
-	std::optional<T>
-	headerReadAs(const Wt::Http::Message& msg, std::string_view headerName)
-	{
-		std::optional<T> res;
-
-		if (const std::string* headerValue {msg.getHeader(std::string {headerName})})
-			res = StringUtils::readAs<T>(*headerValue);
-
-		return res;
-	}
 }
 
-namespace Scrobbling
+namespace Scrobbling::ListenBrainz
 {
-	static const std::string historyTracklistName {"__scrobbler_listenbrainz_history__"};
+	Scrobbler::Scrobbler(boost::asio::io_context& ioContext, Database::Db& db)
-
+		: _ioContext {ioContext}
-	ListenBrainzScrobbler::ListenBrainzScrobbler(Database::Db& db)
-		: _apiEndpoint {Service<IConfig>::get()->getString("listenbrainz-api-url", "https://api.listenbrainz.org/1/")}
 		, _db {db}
+		, _sendQueue {_ioContext, Service<IConfig>::get()->getString("listenbrainz-api-base-url", "https://api.listenbrainz.org")}
+		, _listensSynchronizer {_ioContext, db, _sendQueue}
 	{
-		LOG(INFO) << "Starting ListenBrainz scrobbler... API endpoint = '" << _apiEndpoint << "'";
+		LOG(INFO) << "Starting ListenBrainz scrobbler... API endpoint = '" << _sendQueue.getAPIBaseURL();
-
-		_client.done().connect([this](Wt::AsioWrapper::error_code ec, const Wt::Http::Message& msg)
-		{
-			onClientDone(ec, msg);
-		});
-
-		_ioService.setThreadCount(1);
-		_ioService.start();
 	}
 
-	ListenBrainzScrobbler::~ListenBrainzScrobbler()
+	Scrobbler::~Scrobbler()
 	{
-		_ioService.stop();
+		LOG(INFO) << "Stopped ListenBrainz scrobbler!";
-
-		LOG(INFO) << "Stopped ListenBrainz scrobbler";
 	}
 
 	void
-	ListenBrainzScrobbler::listenStarted(const Listen& listen)
+	Scrobbler::listenStarted(const Listen& listen)
 	{
-		_ioService.post([=]
+		enqueListen(listen, Wt::WDateTime {});
-		{
-			enqueListen(listen, Wt::WDateTime {});
-		});
 	}
 
 	void
-	ListenBrainzScrobbler::listenFinished(const Listen& listen, std::optional<std::chrono::seconds> duration)
+	Scrobbler::listenFinished(const Listen& listen, std::optional<std::chrono::seconds> duration)
 	{
 		if (duration && !canBeScrobbled(_db.getTLSSession(), listen.trackId, *duration))
 			return;
 
-		Listen timedListen {listen};
+		const Listen timedListen {listen};
 		const Wt::WDateTime now {Wt::WDateTime::currentDateTime()};
 
-		_ioService.post([=]
+		enqueListen(timedListen, now);
-		{
-			enqueListen(timedListen, now);
-		});
 	}
 
 	void
-	ListenBrainzScrobbler::addListen(const Listen& listen, const Wt::WDateTime& timePoint)
+	Scrobbler::addTimedListen(const TimedListen& listen)
 	{
-		assert(timePoint.isValid());
+		assert(listen.listenedAt.isValid());
-
+		enqueListen(listen, listen.listenedAt);
-		_ioService.post([=]
-		{
-			enqueListen(listen, timePoint);
-		});
 	}
 
-	Wt::Dbo::ptr<Database::TrackList>
+	Database::TrackList::pointer
-	ListenBrainzScrobbler::getListensTrackList(Database::Session& session, Wt::Dbo::ptr<Database::User> user)
+	Scrobbler::getListensTrackList(Database::Session& session, Database::User::pointer user)
 	{
-		return Database::TrackList::get(session, historyTracklistName, Database::TrackList::Type::Internal, user);
+		return Utils::getListensTrackList(session, user);
 	}
 
 	void
-	ListenBrainzScrobbler::enqueListen(const Listen& listen, const Wt::WDateTime& timePoint)
+	Scrobbler::enqueListen(const Listen& listen, const Wt::WDateTime& timePoint)
 	{
-		if (!timePoint.isValid())
+		std::optional<SendQueue::RequestData> requestData {createSubmitListenRequestData(listen, timePoint)};
-		{
+		if (!requestData)
-			// If we are currently throttled, just replace the entry if it has no timePoint
-			// in order to only report the newest track listened to
-			// If not throttled, just search past the next current first message as it is being sent
-
-			const std::size_t offset {_state == State::Throttled ? std::size_t {0} : std::size_t {1}};
-			if (_sendQueue.size() > offset)
-			{
-				_sendQueue.erase(std::remove_if(std::next(std::begin(_sendQueue), offset), std::end(_sendQueue),
-							[&](const QueuedListen& queuedListen) { return queuedListen.listen.userId == listen.userId && !queuedListen.timePoint.isValid(); }), std::end(_sendQueue));
-			}
-		}
-
-		_sendQueue.emplace_back(QueuedListen {listen, timePoint});
-
-		LOG(DEBUG) << "listen queue size = " << _sendQueue.size();
-
-		if (_state == State::Idle)
-			sendNextQueuedListen();
-	}
-
-	void
-	ListenBrainzScrobbler::sendNextQueuedListen()
-	{
-		assert(_state == State::Idle);
-
-		while (!_sendQueue.empty())
-		{
-			if (sendListen(_sendQueue.front().listen, _sendQueue.front().timePoint))
-			{
-				_state = State::Sending;
-				break;
-			}
-
-			_sendQueue.pop_front();
-		}
-	}
-
-	bool
-	ListenBrainzScrobbler::sendListen(const Listen& listen, const Wt::WDateTime& timePoint)
-	{
-		Database::Session& session {_db.getTLSSession()};
-
-		const std::optional<UUID> listenBrainzToken {getListenBrainzToken(session, listen.userId)};
-		if (!listenBrainzToken)
-			return false;
-
-		std::string payload {listenToJsonString(session, listen, timePoint, timePoint.isValid() ? "single" : "playing_now")};
-		if (payload.empty())
-		{
-			LOG(DEBUG) << "Cannot convert listen to json: skipping";
-			return false;
-		}
-
-		// now send this
-		Wt::Http::Message message;
-		message.addHeader("Authorization", "Token " + std::string {listenBrainzToken->getAsString()});
-		message.addHeader("Content-Type", "application/json");
-		message.addBodyText(payload);
-
-		const std::string endPoint {_apiEndpoint + "submit-listens"};
-		if (!_client.post(endPoint, message))
-		{
-			LOG(ERROR) << "Cannot post to '" << endPoint << "': invalid scheme or URL?";
-			return false;
-		}
-
-		LOG(DEBUG) << "Listen POST done to '" << endPoint << "'";
-		return true;
-	}
-
-	void
-	ListenBrainzScrobbler::onClientDone(Wt::AsioWrapper::error_code ec, const Wt::Http::Message& msg)
-	{
-		assert(!_sendQueue.empty());
-		QueuedListen& queuedListen {_sendQueue.front()};
-
-		_state = State::Idle;
-
-		LOG(DEBUG) << "POST done. status = " << msg.status() << ", msg = '" << msg.body() << "'";
-		if (ec)
-		{
-			LOG(ERROR) << "Retry " << queuedListen.retryCount << ", client error: '" << ec.message() << "'";
-			// may be a network error, try again later
-			if (++queuedListen.retryCount > _maxRetryCount)
-				_sendQueue.pop_front();
-
-			throttle(_defaultRetryWaitDuration);
 			return;
-		}
 
-		bool mustThrottle{};
+		SendQueue::Request submitListen {std::move(*requestData)};
-
+		if (timePoint.isValid())
-		switch (msg.status())
 		{
-			case 429:
+			submitListen.setPriority(SendQueue::Request::Priority::Normal);
-				mustThrottle = true;
+			submitListen.setOnSuccessFunc([=](std::string_view)
-				break;
+			{
-
+				_listensSynchronizer.saveListen(TimedListen {listen, timePoint});
-			case 200:
+			});
-				if (queuedListen.timePoint.isValid())
-					cacheListen(queuedListen.listen, queuedListen.timePoint);
-				_sendQueue.pop_front();
-				break;
-
-			default:
-				LOG(ERROR) << "Submit error: '" << msg.body() << "'";
-				_sendQueue.pop_front();
-				break;
-		}
-
-		const auto remainingCount {headerReadAs<std::size_t>(msg, "X-RateLimit-Remaining")};
-		LOG(DEBUG) << "Remaining messages = " << (remainingCount ? *remainingCount : 0);
-		if (mustThrottle || (remainingCount && *remainingCount == 0))
-		{
-			const auto waitDuration {headerReadAs<std::chrono::seconds>(msg, "X-RateLimit-Reset-In")};
-			throttle(waitDuration.value_or(_defaultRetryWaitDuration));
 		}
 		else
 		{
-			sendNextQueuedListen();
+			// We want "listen now" to appear as soon as possible
+			submitListen.setPriority(SendQueue::Request::Priority::High);
 		}
+
+		_sendQueue.enqueueRequest(std::move(submitListen));
 	}
 
-	void
+	std::optional<SendQueue::RequestData>
-	ListenBrainzScrobbler::throttle(std::chrono::seconds requestedDuration)
+	Scrobbler::createSubmitListenRequestData(const Listen& listen, const Wt::WDateTime& timePoint)
-	{
-		assert(_state == State::Idle);
-
-		const std::chrono::seconds duration {clamp(requestedDuration, _minRetryWaitDuration, _maxRetryWaitDuration)};
-		LOG(DEBUG) << "Throttling for " << duration.count() << " seconds";
-
-		_ioService.schedule(duration, [this]
-		{
-			_state = State::Idle;
-			sendNextQueuedListen();
-		});
-		_state = State::Throttled;
-	}
-
-	void
-	ListenBrainzScrobbler::cacheListen(const Listen& listen, const Wt::WDateTime& timePoint)
 	{
 		Database::Session& session {_db.getTLSSession()};
 
-		auto transaction {session.createUniqueTransaction()};
+		const std::optional<UUID> listenBrainzToken {Utils::getListenBrainzToken(session, listen.userId)};
+		if (!listenBrainzToken)
+			return std::nullopt;
 
-		const Database::User::pointer user {Database::User::getById(session, listen.userId)};
+		SendQueue::RequestData requestData;
-		if (!user)
+		requestData.endpoint = "/1/submit-listens";
-			return;
+		requestData.type = SendQueue::RequestData::Type::POST;
 
-		const Database::Track::pointer track {Database::Track::getById(session, listen.trackId)};
+		std::string bodyText {listenToJsonString(session, listen, timePoint, timePoint.isValid() ? "single" : "playing_now")};
-		if (!track)
+		if (bodyText.empty())
-			return;
+		{
+			LOG(DEBUG) << "Cannot convert listen to json: skipping";
+			return std::nullopt;
+		}
 
-		Database::TrackList::pointer tracklist {getListensTrackList(session, user)};
+		requestData.message.addBodyText(bodyText);
-		if (!tracklist)
+		requestData.message.addHeader("Authorization", "Token " + std::string {listenBrainzToken->getAsString()});
-			tracklist = Database::TrackList::create(session, historyTracklistName, Database::TrackList::Type::Internal, false, user);
+		requestData.message.addHeader("Content-Type", "application/json");
 
-		Database::TrackListEntry::create(session, track, getListensTrackList(session, user), timePoint);
+		return requestData;
 	}
-
+} // namespace Scrobbling::ListenBrainz
-} // Scrobbling
 

src/libs/scrobbling/impl/listenbrainz/ListenBrainzScrobbler.hpp

@@ -19,12 +19,12 @@
 
 #pragma once
 
-#include <deque>
+#include <optional>
-
+#include <boost/asio/io_context.hpp>
-#include <Wt/Http/Client.h>
-#include <Wt/WIOService.h>
 
 #include "IScrobbler.hpp"
+#include "ListensSynchronizer.hpp"
+#include "SendQueue.hpp"
 
 namespace Database
 {
@@ -33,59 +33,33 @@ namespace Database
 	class TrackList;
 }
 
-namespace Scrobbling
+namespace Scrobbling::ListenBrainz
 {
-	class ListenBrainzScrobbler final : public IScrobbler
+	class Scrobbler final : public IScrobbler
 	{
 		public:
-			ListenBrainzScrobbler(Database::Db& db);
+			Scrobbler(boost::asio::io_context& ioContext, Database::Db& db);
-			~ListenBrainzScrobbler();
+			~Scrobbler();
 
-			ListenBrainzScrobbler(const ListenBrainzScrobbler&) = delete;
+			Scrobbler(const Scrobbler&) = delete;
-			ListenBrainzScrobbler(const ListenBrainzScrobbler&&) = delete;
+			Scrobbler(const Scrobbler&&) = delete;
-			ListenBrainzScrobbler& operator=(const ListenBrainzScrobbler&) = delete;
+			Scrobbler& operator=(const Scrobbler&) = delete;
-			ListenBrainzScrobbler& operator=(const ListenBrainzScrobbler&&) = delete;
+			Scrobbler& operator=(const Scrobbler&&) = delete;
 
 		private:
 			void listenStarted(const Listen& listen) override;
 			void listenFinished(const Listen& listen, std::optional<std::chrono::seconds> duration) override;
-			void addListen(const Listen& listen, const Wt::WDateTime& timePoint) override;
+			void addTimedListen(const TimedListen& listen) override;
-
 			Wt::Dbo::ptr<Database::TrackList> getListensTrackList(Database::Session& session, Wt::Dbo::ptr<Database::User> user) override;
 
+			// Submit listens
 			void enqueListen(const Listen& listen, const Wt::WDateTime& timePoint);
-			void sendNextQueuedListen();
+			std::optional<SendQueue::RequestData> createSubmitListenRequestData(const Listen& listen, const Wt::WDateTime& timePoint);
-			bool sendListen(const Listen& listen, const Wt::WDateTime& timePoint);
-			void onClientDone(Wt::AsioWrapper::error_code ec, const Wt::Http::Message& msg);
-			void throttle(std::chrono::seconds duration);
-
-			void cacheListen(const Listen& listen, const Wt::WDateTime& timePoint);
-
-			enum class State
-			{
-				Idle,
-				Throttled,
-				Sending,
-			};
-			State						_state {State::Idle};
-
-			const std::string			_apiEndpoint;
-			const std::size_t			_maxRetryCount {2};
-			const std::chrono::seconds	_defaultRetryWaitDuration {30};
-			const std::chrono::seconds	_minRetryWaitDuration {1};
-			const std::chrono::seconds	_maxRetryWaitDuration {300};
 
+			boost::asio::io_context&	_ioContext;
 			Database::Db&				_db;
-			Wt::WIOService				_ioService;
+			SendQueue					_sendQueue;
-			Wt::Http::Client			_client {_ioService};
+			ListensSynchronizer			_listensSynchronizer;
-
-			struct QueuedListen
-			{
-				Listen			listen;
-				Wt::WDateTime	timePoint;
-				std::size_t		retryCount {};
-			};
-			std::deque<QueuedListen>	_sendQueue;
 	};
-} // Scrobbling
+} // Scrobbling::ListenBrainz
 

src/libs/scrobbling/impl/listenbrainz/ListensSynchronizer.cpp

@@ -0,0 +1,514 @@
+/*
+ * Copyright (C) 2021 Emeric Poupon
+ *
+ * This file is part of LMS.
+ *
+ * LMS is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * LMS is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with LMS.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "ListenBrainzScrobbler.hpp"
+
+#include <boost/asio/bind_executor.hpp>
+#include <Wt/Json/Array.h>
+#include <Wt/Json/Object.h>
+#include <Wt/Json/Value.h>
+#include <Wt/Json/Serializer.h>
+
+#include "database/Artist.hpp"
+#include "database/Db.hpp"
+#include "database/Release.hpp"
+#include "database/Session.hpp"
+#include "database/Track.hpp"
+#include "database/TrackList.hpp"
+#include "database/User.hpp"
+#include "utils/IConfig.hpp"
+#include "utils/Logger.hpp"
+#include "utils/Service.hpp"
+
+#include "Utils.hpp"
+
+#define LOG(sev)	LMS_LOG(SCROBBLING, sev) << "[listenbrainz Synchronizer] - "
+
+namespace
+{
+	using namespace Scrobbling::ListenBrainz;
+
+	SendQueue::RequestData
+	createValidateTokenRequestData(std::string_view authToken)
+	{
+		SendQueue::RequestData requestData;
+		requestData.type = SendQueue::RequestData::Type::GET;
+		requestData.endpoint = "/1/validate-token";
+		requestData.headers = { {"Authorization",  "Token " + std::string {authToken}} };
+
+		return requestData;
+	}
+
+	std::string
+	parseValidateToken(std::string_view msgBody)
+	{
+		std::string listenBrainzUserName;
+
+		Wt::Json::ParseError error;
+		Wt::Json::Object root;
+		if (!Wt::Json::parse(std::string {msgBody}, root, error))
+		{
+			LOG(ERROR) << "Cannot parse 'validate-token' result: " << error.what();
+			return listenBrainzUserName;
+		}
+
+		if (!root.get("valid").orIfNull(false))
+		{
+			LOG(INFO) << "Invalid listenbrainz user";
+			return listenBrainzUserName;
+		}
+
+		listenBrainzUserName = root.get("user_name").orIfNull("");
+		return listenBrainzUserName;
+	}
+
+	SendQueue::RequestData
+	createListenCountRequestData(std::string_view listenBrainzUserName)
+	{
+		LOG(DEBUG) << "Getting listen count for listenbrainz user '" << listenBrainzUserName << "'";
+
+		SendQueue::RequestData requestData;
+		requestData.type = SendQueue::RequestData::Type::GET;
+		requestData.endpoint = "/1/user/" + std::string {listenBrainzUserName} + "/listen-count";
+
+		return requestData;
+	}
+
+	std::optional<std::size_t>
+	parseListenCount(std::string_view msgBody)
+	{
+		try
+		{
+			Wt::Json::Object root;
+			Wt::Json::parse(std::string {msgBody}, root);
+
+			const Wt::Json::Object& payload {static_cast<Wt::Json::Object>(root.get("payload"))};
+			return static_cast<int>(payload.get("count"));
+		}
+		catch (const Wt::WException& e)
+		{
+			LOG(ERROR) << "Cannot parse listen count response: " << e.what();
+			return std::nullopt;
+		}
+	}
+
+	SendQueue::RequestData
+	createGetListensRequestData(std::string_view listenBrainzUserName, const Wt::WDateTime& maxDateTime)
+	{
+		LOG(DEBUG) << "Getting listens for listenbrainz user '" << listenBrainzUserName << "' with max_ts = " << maxDateTime.toString();
+
+		SendQueue::RequestData requestData;
+		requestData.type = SendQueue::RequestData::Type::GET;
+		requestData.endpoint = "/1/user/" + std::string {listenBrainzUserName} + "/listens?max_ts=" + std::to_string(maxDateTime.toTime_t());
+
+		return requestData;
+	}
+
+	Database::Track::pointer
+	tryMatchListen(Database::Session& session, const Wt::Json::Object& metadata)
+	{
+		Database::Track::pointer track;
+
+		// first try to get the associated track using MBIDs, and then fallback on names
+		if (metadata.type("additional_info") == Wt::Json::Type::Object)
+		{
+			const Wt::Json::Object& additionalInfo = metadata.get("additional_info");
+			if (std::optional<UUID> recordingMBID {UUID::fromString(additionalInfo.get("recording_mbid").orIfNull(""))})
+				track = Database::Track::getByRecordingMBID(session, *recordingMBID);
+		}
+
+		if (track)
+			return track;
+
+		// these fields are mandatory
+		const std::string trackName {static_cast<std::string>(metadata.get("track_name"))};
+		const std::string releaseName {static_cast<std::string>(metadata.get("release_name"))};
+
+		auto tracks {Database::Track::getByNameAndReleaseName(session, trackName, releaseName)};
+		if (tracks.size() > 1)
+		{
+			tracks.erase(std::remove_if(std::begin(tracks), std::end(tracks),
+				[&](const Database::Track::pointer track)
+				{
+					if (std::string artistName {metadata.get("artist_name").orIfNull("")}; !artistName.empty())
+					{
+						const auto& artists {track->getArtists({Database::TrackArtistLinkType::Artist})};
+						if (std::none_of(std::begin(artists), std::end(artists), [&](const Database::Artist::pointer& artist) { return artist->getName() == artistName; }))
+							return true;
+					}
+					if (metadata.type("additional_info") == Wt::Json::Type::Object)
+					{
+						const Wt::Json::Object& additionalInfo = metadata.get("additional_info");
+						if (track->getTrackNumber())
+						{
+							int otherTrackNumber {additionalInfo.get("tracknumber").orIfNull(-1)};
+							if (otherTrackNumber > 0 && static_cast<std::size_t>(otherTrackNumber) != *track->getTrackNumber())
+								return true;
+						}
+
+						if (auto releaseMBID {track->getRelease()->getMBID()})
+						{
+							if (std::optional<UUID> otherReleaseMBID {UUID::fromString(additionalInfo.get("release_mbid").orIfNull(""))})
+							{
+								if (otherReleaseMBID->getAsString() != releaseMBID->getAsString())
+									return true;
+							}
+						}
+					}
+
+					return false;
+				}), std::end(tracks));
+		}
+
+		if (tracks.size() == 1)
+			track = tracks.front();
+
+		return track;
+	}
+
+	struct ParseGetListensResult
+	{
+		Wt::WDateTime oldestEntry;
+		std::size_t listenCount{};
+		std::vector<Scrobbling::TimedListen> matchedListens;
+	};
+	ParseGetListensResult
+	parseGetListens(Database::Session& session, std::string_view msgBody, Database::IdType userId)
+	{
+		ParseGetListensResult result;
+
+		try
+		{
+			Wt::Json::Object root;
+			Wt::Json::parse(std::string {msgBody}, root);
+
+			const Wt::Json::Object& payload = root.get("payload");
+			const Wt::Json::Array& listens = payload.get("listens");
+
+			LOG(DEBUG) << "Got " << listens.size() << " listens";
+
+			if (listens.empty())
+				return result;
+
+			auto transaction {session.createSharedTransaction()};
+
+			for (const Wt::Json::Value& value : listens)
+			{
+				const Wt::Json::Object& listen = value;
+				const Wt::WDateTime listenedAt {Wt::WDateTime::fromTime_t(static_cast<int>(listen.get("listened_at")))};
+				const Wt::Json::Object& metadata = listen.get("track_metadata");
+
+				if (!listenedAt.isValid())
+				{
+					LOG(ERROR) << "bad listened_at field!";
+					continue;
+				}
+
+				result.listenCount++;
+				if (!result.oldestEntry.isValid())
+					result.oldestEntry = listenedAt;
+				else if (listenedAt < result.oldestEntry)
+					result.oldestEntry = listenedAt;
+
+				if (const Database::Track::pointer track {tryMatchListen(session, metadata)})
+					result.matchedListens.emplace_back(Scrobbling::TimedListen {userId, track.id(), listenedAt});
+			}
+		}
+		catch (const Wt::WException& error)
+		{
+			LOG(ERROR) << "Cannot parse 'get-listens' result: " << error.what();
+		}
+
+		return result;
+	}
+}
+
+namespace Scrobbling::ListenBrainz
+{
+	ListensSynchronizer::ListensSynchronizer(boost::asio::io_context& ioContext, Database::Db& db, SendQueue& sendQueue)
+		: _ioContext {ioContext}
+		, _db {db}
+		, _sendQueue {sendQueue}
+		, _maxSyncListenCount {Service<IConfig>::get()->getULong("listenbrainz-max-sync-listen-count", 1000)}
+		, _syncListensPeriod {Service<IConfig>::get()->getULong("listenbrainz-sync-listens-period-hours", 1)}
+	{
+		LOG(INFO) << "Starting Listens synchronizer, maxSyncListenCount = " << _maxSyncListenCount << ", _syncListensPeriod = " << _syncListensPeriod.count() << " hours";
+
+		scheduleGetListens(std::chrono::seconds {30});
+	}
+
+	void
+	ListensSynchronizer::saveListen(const TimedListen& listen)
+	{
+		_strand.dispatch([=]
+		{
+			Database::Session& session {_db.getTLSSession()};
+
+			auto transaction {session.createUniqueTransaction()};
+
+			const Database::User::pointer user {Database::User::getById(session, listen.userId)};
+			if (!user)
+				return;
+
+			const Database::Track::pointer track {Database::Track::getById(session, listen.trackId)};
+			if (!track)
+				return;
+
+			Database::TrackListEntry::create(session, track, Utils::getOrCreateListensTrackList(session, user), listen.listenedAt);
+
+			UserContext& context {getUserContext(listen.userId)};
+			if (context.listenCount)
+				(*context.listenCount)++;
+		});
+	}
+
+	ListensSynchronizer::UserContext&
+	ListensSynchronizer::getUserContext(Database::IdType userId)
+	{
+		auto itContext {_userContexts.find(userId)};
+		if (itContext == std::cend(_userContexts))
+		{
+			auto [itNewContext, inserted] {_userContexts.emplace(userId, userId)};
+			itContext = itNewContext;
+		}
+
+		return itContext->second;
+	}
+
+	bool
+	ListensSynchronizer::isFetching() const
+	{
+		return std::any_of(std::cbegin(_userContexts), std::cend(_userContexts), [](const auto& contextEntry)
+				{
+					const auto& [userId, context] {contextEntry};
+					return context.fetching;
+				});
+	}
+
+	void
+	ListensSynchronizer::scheduleGetListens(std::chrono::seconds fromNow)
+	{
+		if (_syncListensPeriod.count() == 0 || _maxSyncListenCount == 0)
+			return;
+
+		LOG(DEBUG) << "Scheduled sync in " << fromNow.count() << " seconds...";
+		_getListensTimer.expires_after(fromNow);
+		_getListensTimer.async_wait(boost::asio::bind_executor(_strand, [this] (const boost::system::error_code& ec)
+		{
+			if (ec == boost::asio::error::operation_aborted)
+			{
+				LOG(DEBUG) << "getListens aborted";
+				return;
+			}
+
+			startGetListens();
+		}));
+	}
+
+	void
+	ListensSynchronizer::startGetListens()
+	{
+		LOG(DEBUG) << "GetListens started!!!";
+
+		assert(!isFetching());
+
+		std::vector<Database::IdType> userIds;
+		{
+			Database::Session& session {_db.getTLSSession()};
+			auto transaction {session.createSharedTransaction()};
+			userIds = Database::User::getAllIds(_db.getTLSSession());
+		}
+
+		for (const Database::IdType userId : userIds)
+		{
+			if (Utils::getListenBrainzToken(_db.getTLSSession(), userId))
+				startGetListens(getUserContext(userId));
+		}
+
+		if (!isFetching())
+			scheduleGetListens(_syncListensPeriod);
+	}
+
+	void
+	ListensSynchronizer::startGetListens(UserContext& context)
+	{
+		context.fetching = true;
+		context.listenBrainzUserName = "";
+		context.maxDateTime = {};
+		context.fetchedListenCount = 0;
+		context.matchedListenCount = 0;
+		context.importedListenCount = 0;
+
+		enqueValidateToken(context);
+	}
+
+	void
+	ListensSynchronizer::onGetListensEnded(UserContext& context)
+	{
+		_strand.dispatch([this, &context]
+		{
+			LOG(DEBUG) << "Fetch done for user " << context.userId << ", fetched: " << context.fetchedListenCount << ", matched: " << context.matchedListenCount << ", imported: " << context.importedListenCount;
+			context.fetching = false;
+
+			if (!isFetching())
+				scheduleGetListens(_syncListensPeriod);
+		});
+	}
+
+	void
+	ListensSynchronizer::enqueValidateToken(UserContext& context)
+	{
+		assert(context.listenBrainzUserName.empty());
+
+		std::optional<SendQueue::RequestData> requestData {createValidateTokenRequestData(context.userId)};
+		if (!requestData)
+		{
+			onGetListensEnded(context);
+			return;
+		}
+
+		SendQueue::Request validateTokenRequest {std::move(*requestData)};
+		validateTokenRequest.setOnSuccessFunc([this, &context] (std::string_view msgBody)
+		{
+			context.listenBrainzUserName = parseValidateToken(msgBody);
+			if (context.listenBrainzUserName.empty())
+			{
+				onGetListensEnded(context);
+				return;
+			}
+			enqueGetListenCount(context);
+		});
+		validateTokenRequest.setOnFailureFunc([this, &context]
+		{
+			onGetListensEnded(context);
+		});
+
+		validateTokenRequest.setPriority(SendQueue::Request::Priority::Low);
+		_sendQueue.enqueueRequest(std::move(validateTokenRequest));
+	}
+
+	void
+	ListensSynchronizer::enqueGetListenCount(UserContext& context)
+	{
+		assert(!context.listenBrainzUserName.empty());
+
+		SendQueue::Request getListenCountRequest {createListenCountRequestData(context.listenBrainzUserName)};
+		getListenCountRequest.setOnSuccessFunc([=, &context] (std::string_view msgBody)
+		{
+			const auto listenCount = parseListenCount(msgBody);
+			if (listenCount)
+				LOG(DEBUG) << "Listen count for listenbrainz user '" << context.listenBrainzUserName << "' = " << *listenCount;
+
+			bool needSync {listenCount && (!context.listenCount || *context.listenCount != *listenCount)};
+			context.listenCount = listenCount;
+
+			if (!needSync)
+			{
+				onGetListensEnded(context);
+				return;
+			}
+
+			context.maxDateTime = Wt::WDateTime::currentDateTime();
+			enqueGetListens(context);
+		});
+		getListenCountRequest.setOnFailureFunc([this, &context]
+		{
+			onGetListensEnded(context);
+		});
+
+		getListenCountRequest.setPriority(SendQueue::Request::Priority::Low);
+		_sendQueue.enqueueRequest(std::move(getListenCountRequest));
+	}
+
+	void
+	ListensSynchronizer::enqueGetListens(UserContext& context)
+	{
+		assert(!context.listenBrainzUserName.empty());
+
+		SendQueue::Request getListensRequest {::createGetListensRequestData(context.listenBrainzUserName, context.maxDateTime)};
+		getListensRequest.setOnSuccessFunc([=, &context] (std::string_view msgBody)
+		{
+			processGetListensResponse(msgBody, context);
+			if (context.fetchedListenCount >= _maxSyncListenCount || !context.maxDateTime.isValid())
+			{
+				onGetListensEnded(context);
+				return;
+			}
+
+			enqueGetListens(context);
+		});
+		getListensRequest.setOnFailureFunc([=, &context]
+		{
+			onGetListensEnded(context);
+		});
+
+		getListensRequest.setPriority(SendQueue::Request::Priority::Low);
+		_sendQueue.enqueueRequest(std::move(getListensRequest));
+	}
+
+	std::optional<SendQueue::RequestData>
+	ListensSynchronizer::createValidateTokenRequestData(Database::IdType userId)
+	{
+		Database::Session& session {_db.getTLSSession()};
+
+		const std::optional<UUID> listenBrainzToken {Utils::getListenBrainzToken(session, userId)};
+		if (!listenBrainzToken)
+			return std::nullopt;
+
+		return ::createValidateTokenRequestData(listenBrainzToken->getAsString());
+	}
+
+	void
+	ListensSynchronizer::processGetListensResponse(std::string_view msgBody, UserContext& context)
+	{
+		Database::Session& session {_db.getTLSSession()};
+
+		const ParseGetListensResult parseResult {parseGetListens(session, msgBody, context.userId)};
+
+		context.fetchedListenCount += parseResult.listenCount;
+		context.matchedListenCount += parseResult.matchedListens.size();
+		context.maxDateTime = parseResult.oldestEntry;
+
+		if (parseResult.matchedListens.empty())
+			return;
+
+		auto transaction {session.createUniqueTransaction()};
+
+		Database::User::pointer user {Database::User::getById(session, context.userId)};
+		if (!user)
+			return;
+
+		Database::TrackList::pointer tracklist {Utils::getOrCreateListensTrackList(session, user)};
+
+		for (const TimedListen& listen : parseResult.matchedListens)
+		{
+			const Database::Track::pointer track {Database::Track::getById(session, listen.trackId)};
+			if (!track)
+				continue;
+
+			if (!tracklist->getEntryByTrackAndDateTime(track, listen.listenedAt))
+			{
+				context.importedListenCount++;
+				Database::TrackListEntry::create(session, track, tracklist, listen.listenedAt);
+			}
+		}
+	}
+
+} // namespace Scrobbling::ListenBrainz
+

src/libs/scrobbling/impl/listenbrainz/ListensSynchronizer.hpp

@@ -0,0 +1,98 @@
+/*
+ * Copyright (C) 2021 Emeric Poupon
+ *
+ * This file is part of LMS.
+ *
+ * LMS is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * LMS is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with LMS.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#pragma once
+
+#include <optional>
+#include <unordered_map>
+#include <boost/asio/io_context.hpp>
+#include <boost/asio/io_context_strand.hpp>
+#include <boost/asio/steady_timer.hpp>
+#include <Wt/Dbo/Dbo.h>
+
+#include "database/Types.hpp"
+#include "scrobbling/Listen.hpp"
+#include "SendQueue.hpp"
+
+namespace Database
+{
+	class Db;
+	class Session;
+	class TrackList;
+	class User;
+}
+
+namespace Scrobbling::ListenBrainz
+{
+	class ListensSynchronizer
+	{
+		public:
+			ListensSynchronizer(boost::asio::io_context& ioContext, Database::Db& db, SendQueue& sendQueue);
+
+			void saveListen(const TimedListen& listen);
+
+		private:
+			struct UserContext
+			{
+				UserContext(Database::IdType id) : userId {id} {}
+
+				UserContext(const UserContext&) = delete;
+				UserContext(UserContext&&) = delete;
+				UserContext& operator=(const UserContext&) = delete;
+				UserContext& operator=(UserContext&&) = delete;
+
+				const Database::IdType	userId;
+				bool					fetching {};
+				std::optional<std::size_t> listenCount {};
+
+				// resetted at each fetch
+				std::string		listenBrainzUserName; // need to be resolved first
+				Wt::WDateTime	maxDateTime;
+				std::size_t		fetchedListenCount{};
+				std::size_t		matchedListenCount{};
+				std::size_t		importedListenCount{};
+
+			};
+
+			UserContext& getUserContext(Database::IdType userId);
+			bool isFetching() const;
+			void scheduleGetListens(std::chrono::seconds fromNow);
+			void startGetListens();
+			void startGetListens(UserContext& context);
+			void onGetListensEnded(UserContext& context);
+			void enqueValidateToken(UserContext& context);
+			void enqueGetListenCount(UserContext& context);
+			void enqueGetListens(UserContext& context);
+			std::optional<SendQueue::RequestData>	createValidateTokenRequestData(Database::IdType userId);
+			std::optional<SendQueue::RequestData>	createGetListensRequestData(std::string_view listenBrainzUserName, const Wt::WDateTime& maxDateTime);
+			void									processGetListensResponse(std::string_view body, UserContext& context);
+
+			boost::asio::io_context&		_ioContext;
+			boost::asio::io_context::strand	_strand {_ioContext};
+			Database::Db&					_db;
+			SendQueue&						_sendQueue;
+			boost::asio::steady_timer		_getListensTimer {_ioContext};
+
+			std::unordered_map<Database::IdType, UserContext> _userContexts;
+
+			const std::size_t			_maxSyncListenCount;
+			const std::chrono::hours	_syncListensPeriod;
+	};
+} // Scrobbling::ListenBrainz
+

src/libs/scrobbling/impl/listenbrainz/SendQueue.cpp

@@ -0,0 +1,228 @@
+/*
+ * Copyright (C) 2021 Emeric Poupon
+ *
+ * This file is part of LMS.
+ *
+ * LMS is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * LMS is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with LMS.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "SendQueue.hpp"
+
+#include <boost/asio/bind_executor.hpp>
+
+#include "utils/Logger.hpp"
+#include "utils/String.hpp"
+
+#define LOG(sev)	LMS_LOG(SCROBBLING, sev) << "[listenbrainz SendQueue] - "
+
+namespace StringUtils
+{
+	template<>
+	std::optional<std::chrono::seconds>
+	readAs(const std::string& str)
+	{
+		std::optional<std::chrono::seconds> res;
+
+		if (const std::optional<std::size_t> value {StringUtils::readAs<std::size_t>(str)})
+			res = std::chrono::seconds {*value};
+
+		return res;
+	}
+}
+
+namespace
+{
+	template <typename T>
+	std::optional<T>
+	headerReadAs(const Wt::Http::Message& msg, std::string_view headerName)
+	{
+		std::optional<T> res;
+
+		if (const std::string* headerValue {msg.getHeader(std::string {headerName})})
+			res = StringUtils::readAs<T>(*headerValue);
+
+		return res;
+	}
+}
+
+namespace Scrobbling::ListenBrainz
+{
+	SendQueue::SendQueue(boost::asio::io_context& ioContext, std::string_view apiBaseURL)
+	: _ioContext {ioContext}
+	, _apiBaseURL {apiBaseURL}
+	{
+		_client.done().connect([this](Wt::AsioWrapper::error_code ec, const Wt::Http::Message& msg)
+		{
+			_strand.dispatch([=, msg = std::move(msg)]
+			{
+				onClientDone(ec, msg);
+			});
+		});
+	}
+
+	SendQueue::~SendQueue()
+	{
+		_client.abort();
+	}
+
+	void
+	SendQueue::enqueueRequest(Request request)
+	{
+		_strand.dispatch([this, request = std::move(request)]()
+		{
+			_sendQueue[request._priority].emplace_back(std::move(request));
+
+			if (_state == State::Idle)
+				sendNextQueuedRequest();
+		});
+	}
+
+	void
+	SendQueue::sendNextQueuedRequest()
+	{
+		assert(_state == State::Idle);
+
+		for (auto& [prio, requests] : _sendQueue)
+		{
+			LOG(DEBUG) << "Processing prio " << static_cast<int>(prio) << ", request count = " << requests.size();
+			while (!requests.empty())
+			{
+				Request request {std::move(requests.front())};
+				requests.pop_front();
+
+				if (!sendRequest(request._requestData))
+					continue;
+
+				_state = State::Sending;
+				_currentRequest = std::move(request);
+				return;
+			}
+		}
+	}
+
+	bool
+	SendQueue::sendRequest(const RequestData& requestData)
+	{
+		const std::string url {_apiBaseURL + requestData.endpoint};
+
+		LOG(DEBUG) << "Sending request type " << (requestData.type == RequestData::Type::GET ? "GET" : "POST") << " to url '" << url << "'";
+
+		bool res{};
+		switch (requestData.type)
+		{
+			case RequestData::Type::GET:
+				res = _client.get(url, requestData.headers);
+				break;
+			case RequestData::Type::POST:
+				res = _client.post(url, requestData.message);
+				break;
+		}
+
+		if (!res)
+			LOG(ERROR) << "Send failed, bad url or unsupported scheme?";
+
+		return res;
+	}
+
+	void
+	SendQueue::onClientDone(Wt::AsioWrapper::error_code ec, const Wt::Http::Message& msg)
+	{
+		if (ec == boost::asio::error::operation_aborted)
+		{
+			LOG(DEBUG) << "SendQueue: client aborted";
+			return;
+		}
+
+		assert(_currentRequest);
+		Request request {std::move(*_currentRequest)};
+		_state = State::Idle;
+
+		LOG(DEBUG) << "Client done. status = " << msg.status();
+		if (ec)
+		{
+			LOG(ERROR) << "Retry " << request._retryCount << ", client error: '" << ec.message() << "'";
+
+			// may be a network error, try again later
+			throttle(_defaultRetryWaitDuration);
+
+			if (request._retryCount++ < _maxRetryCount)
+			{
+				_sendQueue[request._priority].emplace_front(std::move(request));
+			}
+			else
+			{
+				LOG(ERROR) << "Too many retries, giving up operation and throttle";
+				if (request._onFailureFunc)
+					request._onFailureFunc();
+			}
+			return;
+		}
+
+		bool mustThrottle{};
+		if (msg.status() == 429)
+			_sendQueue[request._priority].emplace_front(std::move(request));
+
+		const auto remainingCount {headerReadAs<std::size_t>(msg, "X-RateLimit-Remaining")};
+		LOG(DEBUG) << "Remaining messages = " << (remainingCount ? *remainingCount : 0);
+		if (mustThrottle || (remainingCount && *remainingCount == 0))
+		{
+			const auto waitDuration {headerReadAs<std::chrono::seconds>(msg, "X-RateLimit-Reset-In")};
+			throttle(waitDuration.value_or(_defaultRetryWaitDuration));
+		}
+
+		if (!mustThrottle)
+		{
+			if (msg.status() == 200)
+			{
+				if (request._onSuccessFunc)
+					request._onSuccessFunc(msg.body());
+			}
+			else
+			{
+				LOG(ERROR) << "Send error: '" << msg.body() << "'";
+				if (request._onFailureFunc)
+					request._onFailureFunc();
+			}
+		}
+
+		if (_state == State::Idle)
+			sendNextQueuedRequest();
+	}
+
+	void
+	SendQueue::throttle(std::chrono::seconds requestedDuration)
+	{
+		assert(_state == State::Idle);
+
+		const std::chrono::seconds duration {clamp(requestedDuration, _minRetryWaitDuration, _maxRetryWaitDuration)};
+		LOG(DEBUG) << "Throttling for " << duration.count() << " seconds";
+
+		_throttleTimer.expires_after(duration);
+		_throttleTimer.async_wait([this](const boost::system::error_code& ec)
+		{
+			if (ec == boost::asio::error::operation_aborted)
+			{
+				LOG(DEBUG) << "SendQueue: throttle aborted";
+				return;
+			}
+
+			if (ec)
+				LOG(ERROR) << "async_wait failed:" << ec.message();
+
+			_state = State::Idle;
+			sendNextQueuedRequest();
+		});
+		_state = State::Throttled;
+	}
+} // namespace Scrobbling::ListenBrainz

src/libs/scrobbling/impl/listenbrainz/SendQueue.hpp

@@ -0,0 +1,118 @@
+/*
+ * Copyright (C) 2021 Emeric Poupon
+ *
+ * This file is part of LMS.
+ *
+ * LMS is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * LMS is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with LMS.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#pragma once
+
+#include <deque>
+
+#include <boost/asio/io_context.hpp>
+#include <boost/asio/io_context_strand.hpp>
+#include <boost/asio/steady_timer.hpp>
+
+#include <Wt/Http/Client.h>
+
+namespace Scrobbling::ListenBrainz
+{
+	class SendQueue
+	{
+		public:
+			SendQueue(boost::asio::io_context& ioContext, std::string_view apiBaseURL);
+			~SendQueue();
+
+			SendQueue(const SendQueue&) = delete;
+			SendQueue(const SendQueue&&) = delete;
+			SendQueue& operator=(const SendQueue&) = delete;
+			SendQueue& operator=(const SendQueue&&) = delete;
+
+			// generic queue operations
+			struct RequestData
+			{
+				enum class Type
+				{
+					GET,
+					POST,
+				};
+
+				Type type;
+				std::string endpoint;	// relative URL to the base API
+				std::vector<Wt::Http::Message::Header> headers; // used by GET
+				Wt::Http::Message message; // used by POST
+			};
+
+			class Request
+			{
+				public:
+
+					enum class Priority
+					{
+						High,
+						Normal,
+						Low,
+					};
+
+					Request(RequestData requestData) : _requestData {std::move(requestData)} {}
+
+					using OnSuccessFunc = std::function<void(std::string_view msgBody)>;
+					using OnFailureFunc = std::function<void()>;
+
+					void setOnSuccessFunc(OnSuccessFunc onSuccessFunc) { _onSuccessFunc = onSuccessFunc; }
+					void setOnFailureFunc(OnFailureFunc onFailureFunc) { _onFailureFunc = onFailureFunc; }
+					void setPriority(Priority priority) { _priority = priority; }
+
+				private:
+					friend class SendQueue;
+					RequestData	_requestData;
+					Priority _priority {Priority::Normal};
+					std::size_t	_retryCount {};
+					OnSuccessFunc _onSuccessFunc;
+					OnFailureFunc _onFailureFunc;
+			};
+
+			std::string_view getAPIBaseURL() const { return _apiBaseURL; }
+			void enqueueRequest(Request request);
+
+		private:
+			void sendNextQueuedRequest();
+			bool sendRequest(const RequestData& request);
+			void onClientDone(Wt::AsioWrapper::error_code ec, const Wt::Http::Message& msg);
+			void throttle(std::chrono::seconds duration);
+
+			const std::size_t			_maxRetryCount {2};
+			const std::chrono::seconds	_defaultRetryWaitDuration {30};
+			const std::chrono::seconds	_minRetryWaitDuration {1};
+			const std::chrono::seconds	_maxRetryWaitDuration {300};
+
+			enum class State
+			{
+				Idle,
+				Throttled,
+				Sending,
+			};
+			boost::asio::io_context&		_ioContext;
+			boost::asio::io_context::strand	_strand {_ioContext};
+			boost::asio::steady_timer		_throttleTimer {_ioContext};
+			std::string						_apiBaseURL;
+			State							_state {State::Idle};
+			Wt::Http::Client				_client {_ioContext};
+			std::map<Request::Priority, std::deque<Request>> _sendQueue;
+			std::optional<Request>			_currentRequest;
+	};
+
+} // namespace Scrobbling::ListenBrainz
+

src/libs/scrobbling/impl/listenbrainz/Utils.cpp

@@ -0,0 +1,63 @@
+/*
+ * Copyright (C) 2021 Emeric Poupon
+ *
+ * This file is part of LMS.
+ *
+ * LMS is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * LMS is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with LMS.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "Utils.hpp"
+
+#include <string_view>
+
+#include "database/Session.hpp"
+#include "database/TrackList.hpp"
+#include "database/User.hpp"
+
+static constexpr std::string_view historyTracklistName {"__scrobbler_listenbrainz_history__"};
+
+namespace Scrobbling::ListenBrainz::Utils
+{
+	std::optional<UUID>
+	getListenBrainzToken(Database::Session& session, Database::IdType userId)
+	{
+		auto transaction {session.createSharedTransaction()};
+
+		const Database::User::pointer user {Database::User::getById(session, userId)};
+		if (!user)
+			return std::nullopt;
+
+		if (user->getScrobbler() != Database::Scrobbler::ListenBrainz)
+			return std::nullopt;
+
+		return user->getListenBrainzToken();
+	}
+
+	Database::TrackList::pointer
+	getListensTrackList(Database::Session& session, Database::User::pointer user)
+	{
+		return Database::TrackList::get(session, historyTracklistName, Database::TrackList::Type::Internal, user);
+	}
+
+	Database::TrackList::pointer
+	getOrCreateListensTrackList(Database::Session& session, Database::User::pointer user)
+	{
+		Database::TrackList::pointer tracklist {getListensTrackList(session, user)};
+		if (!tracklist)
+			tracklist = Database::TrackList::create(session, historyTracklistName, Database::TrackList::Type::Internal, false, user);
+
+		return tracklist;
+	}
+
+}

src/libs/scrobbling/impl/listenbrainz/Utils.hpp

@@ -0,0 +1,38 @@
+/*
+ * Copyright (C) 2021 Emeric Poupon
+ *
+ * This file is part of LMS.
+ *
+ * LMS is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * LMS is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with LMS.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#pragma once
+
+#include <Wt/Dbo/ptr.h>
+#include "utils/UUID.hpp"
+#include "database/Types.hpp"
+
+namespace Database
+{
+	class Session;
+	class TrackList;
+	class User;
+}
+
+namespace Scrobbling::ListenBrainz::Utils
+{
+	std::optional<UUID>					getListenBrainzToken(Database::Session& session, Database::IdType userId);
+	Wt::Dbo::ptr<Database::TrackList>	getOrCreateListensTrackList(Database::Session& session, Wt::Dbo::ptr<Database::User> user);
+	Wt::Dbo::ptr<Database::TrackList>	getListensTrackList(Database::Session& session, Wt::Dbo::ptr<Database::User> user);
+}

src/libs/scrobbling/include/scrobbling/IScrobbling.hpp

@@ -19,6 +19,8 @@
 
 #pragma once
 
+#include <boost/asio/io_service.hpp>
+
 #include <chrono>
 #include <memory>
 #include <optional>
@@ -51,7 +53,7 @@ namespace Scrobbling
 			virtual void listenStarted(const Listen& listen) = 0;
 			virtual void listenFinished(const Listen& listen, std::optional<std::chrono::seconds> playedDuration = std::nullopt) = 0;
 
-			virtual void addListen(const Listen& listen, Wt::WDateTime timePoint) = 0;
+			virtual void addTimedListen(const TimedListen& listen) = 0;
 
 			// Stats
 			// From most recent to oldest
@@ -95,7 +97,7 @@ namespace Scrobbling
 																	bool& moreResults) = 0;
 	};
 
-	std::unique_ptr<IScrobbling> createScrobbling(Database::Db& db);
+	std::unique_ptr<IScrobbling> createScrobbling(boost::asio::io_service& ioService, Database::Db& db);
 
 } // ns Scrobbling
 

src/libs/scrobbling/include/scrobbling/Listen.hpp

@@ -19,6 +19,8 @@
 
 #pragma once
 
+#include <Wt/WDateTime.h>
+
 #include "database/Types.hpp"
 
 namespace Scrobbling
@@ -28,5 +30,10 @@ namespace Scrobbling
 		Database::IdType userId {};
 		Database::IdType trackId {};
 	};
+
+	struct TimedListen : public Listen
+	{
+		Wt::WDateTime listenedAt;
+	};
 } // ns Scrobbling
 

src/libs/subsonic/impl/SubsonicResource.cpp

@@ -1679,7 +1679,7 @@ handleScrobble(RequestContext& context)
 			{
 				const Database::IdType trackId {ids[i].value};
 				const unsigned long time {times[i]};
-				Service<Scrobbling::IScrobbling>::get()->addListen({context.userId, trackId}, Wt::WDateTime::fromTime_t(static_cast<std::time_t>(time / 1000)));
+				Service<Scrobbling::IScrobbling>::get()->addTimedListen({context.userId, trackId, Wt::WDateTime::fromTime_t(static_cast<std::time_t>(time / 1000))});
 			}
 		}
 	}

src/libs/utils/CMakeLists.txt

@@ -4,6 +4,7 @@ add_library(lmsutils SHARED
 	impl/ChildProcessManager.cpp
 	impl/Config.cpp
 	impl/FileResourceHandler.cpp
+	impl/IOContextRunner.cpp
 	impl/Logger.cpp
 	impl/NetAddress.cpp
 	impl/Path.cpp

src/libs/utils/impl/ChildProcessManager.cpp

@@ -25,42 +25,14 @@
 
 
 std::unique_ptr<IChildProcessManager>
-createChildProcessManager()
+createChildProcessManager(boost::asio::io_context& ioContext)
 {
-	return std::make_unique<ChildProcessManager>();
+	return std::make_unique<ChildProcessManager>(ioContext);
 }
 
-ChildProcessManager::ChildProcessManager()
+ChildProcessManager::ChildProcessManager(boost::asio::io_context& ioContext)
-: _work {boost::asio::make_work_guard(_ioContext)}
+: _ioContext {ioContext}
 {
-	start();
-}
-
-ChildProcessManager::~ChildProcessManager()
-{
-	stop();
-}
-
-void
-ChildProcessManager::start()
-{
-	LMS_LOG(CHILDPROCESS, INFO) << "Starting child process manager...";
-
-	_thread = std::make_unique<std::thread>([&]()
-	{
-		_ioContext.run();
-	});
-
-	LMS_LOG(CHILDPROCESS, INFO) << "Child process manager started!";
-}
-
-void
-ChildProcessManager::stop()
-{
-	LMS_LOG(CHILDPROCESS, INFO) << "Stopping child process manager";
-	_work.reset();
-	_thread->join();
-	LMS_LOG(CHILDPROCESS, INFO) << "Stopped child process manager";
 }
 
 std::unique_ptr<IChildProcess>

src/libs/utils/impl/ChildProcessManager.hpp

@@ -23,15 +23,14 @@
 #include <thread>
 
 #include <boost/asio/io_context.hpp>
-#include <boost/asio/executor_work_guard.hpp>
 
 #include "utils/IChildProcessManager.hpp"
 
 class ChildProcessManager : public IChildProcessManager
 {
 	public:
-		ChildProcessManager();
+		ChildProcessManager(boost::asio::io_context& ioContext);
-		~ChildProcessManager();
+		~ChildProcessManager() = default;
 
 		ChildProcessManager(const ChildProcessManager&) = delete;
 		ChildProcessManager(ChildProcessManager&&) = delete;
@@ -41,12 +40,7 @@ class ChildProcessManager : public IChildProcessManager
 	private:
 		std::unique_ptr<IChildProcess> spawnChildProcess(const std::filesystem::path& path, const IChildProcess::Args& args) override;
 
-		void start();
+		boost::asio::io_context&	_ioContext;
-		void stop();
-
-		boost::asio::io_context	_ioContext;
-		std::unique_ptr<std::thread> _thread;
-		boost::asio::executor_work_guard<boost::asio::io_context::executor_type> _work;
 };
 
 

src/libs/utils/impl/IOContextRunner.cpp

@@ -0,0 +1,49 @@
+/*
+ * Copyright (C) 2021 Emeric Poupon
+ *
+ * This file is part of LMS.
+ *
+ * LMS is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * LMS is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with LMS.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "utils/Logger.hpp"
+#include "utils/IOContextRunner.hpp"
+
+IOContextRunner::IOContextRunner(boost::asio::io_service& ioService, std::size_t threadCount)
+: _ioService {ioService}
+, _work {ioService}
+{
+	LMS_LOG(UTILS, INFO) << "Starting IO Context with " << threadCount << " threads...";
+	for (std::size_t i {}; i < threadCount; ++i)
+		_threads.emplace_back([&] { _ioService.run(); });
+}
+
+void
+IOContextRunner::stop()
+{
+	LMS_LOG(UTILS, INFO) << "Stopping IO Context";
+	_work.reset();
+	_ioService.stop();
+	LMS_LOG(UTILS, INFO) << "Stopped IO Context";
+}
+
+IOContextRunner::~IOContextRunner()
+{
+
+	stop();
+
+	for (std::thread& t : _threads)
+		t.join();
+
+}

src/libs/utils/include/utils/IChildProcessManager.hpp

@@ -20,10 +20,7 @@
 
 #include <filesystem>
 #include <memory>
-#pragma once
+#include <boost/asio/io_service.hpp>
-
-#include <filesystem>
-#include <memory>
 
 #include "IChildProcess.hpp"
 
@@ -35,6 +32,6 @@ class IChildProcessManager
 		virtual std::unique_ptr<IChildProcess> spawnChildProcess(const std::filesystem::path& path, const IChildProcess::Args& args) = 0;
 };
 
-std::unique_ptr<IChildProcessManager> createChildProcessManager();
+std::unique_ptr<IChildProcessManager> createChildProcessManager(boost::asio::io_service& ioService);
 
 

src/libs/utils/include/utils/IOContextRunner.hpp

@@ -0,0 +1,43 @@
+/*
+ * Copyright (C) 2021 Emeric Poupon
+ *
+ * This file is part of LMS.
+ *
+ * LMS is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * LMS is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with LMS.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#pragma once
+
+#include <optional>
+#include <thread>
+#include <boost/asio/io_service.hpp>
+
+class IOContextRunner
+{
+	public:
+		IOContextRunner(boost::asio::io_service& ioService, std::size_t threadCount);
+		~IOContextRunner();
+
+		IOContextRunner(const IOContextRunner&) = delete;
+		IOContextRunner(IOContextRunner&&) = delete;
+		IOContextRunner& operator=(const IOContextRunner&) = delete;
+		IOContextRunner& operator=(IOContextRunner&&) = delete;
+
+		void stop();
+
+	private:
+		boost::asio::io_service&		_ioService;
+		std::optional<boost::asio::io_service::work>	_work;
+		std::vector<std::thread>		_threads;
+};

src/lms/main.cpp

@@ -19,6 +19,7 @@
 
 #include <thread>
 
+#include <boost/asio/io_context.hpp>
 #include <boost/property_tree/xml_parser.hpp>
 
 #include <Wt/WServer.h>
@@ -38,6 +39,7 @@
 #include "ui/LmsApplicationManager.hpp"
 #include "utils/IChildProcessManager.hpp"
 #include "utils/IConfig.hpp"
+#include "utils/IOContextRunner.hpp"
 #include "utils/Service.hpp"
 #include "utils/String.hpp"
 #include "utils/WtLogger.hpp"
@@ -212,9 +214,12 @@ int main(int argc, char* argv[])
 			wtArgv[i] = wtServerArgs[i].c_str();
 		}
 
+		boost::asio::io_context ioContext; // ioContext used to dispatch all the services that are out of the Wt event loop
 		Wt::WServer server {argv[0]};
 		server.setServerConfiguration(wtServerArgs.size(), const_cast<char**>(&wtArgv[0]));
 
+		IOContextRunner ioContextRunner {ioContext, std::max<unsigned long>(2, std::thread::hardware_concurrency())};
+
 		// Initializing a connection pool to the database that will be shared along services
 		Database::Db database {config->getPath("working-dir") / "lms.db"};
 		{
@@ -226,7 +231,7 @@ int main(int argc, char* argv[])
 		UserInterface::LmsApplicationManager appManager;
 
 		// Service initialization order is important (reverse-order for deinit)
-		Service<IChildProcessManager> childProcessManagerService {createChildProcessManager()};
+		Service<IChildProcessManager> childProcessManagerService {createChildProcessManager(ioContext)};
 
 		Service<Auth::IAuthTokenService> authTokenService;
 		Service<Auth::IPasswordService> authPasswordService;
@@ -251,7 +256,7 @@ int main(int argc, char* argv[])
 				config->getULong("cover-max-file-size", 10) * 1000 * 1000,
 				config->getULong("cover-jpeg-quality", 75))};
 		Service<Recommendation::IEngine> recommendationEngineService {Recommendation::createEngine(database)};
-		Service<Scanner::IScanner> scannerService {Scanner::createScanner(database, *recommendationEngineService)};
+		Service<Scanner::IScanner> scannerService {Scanner::createScanner(/*ioContext,*/ database, *recommendationEngineService)};
 
 		scannerService->getEvents().scanComplete.connect([&]
 		{
@@ -260,7 +265,7 @@ int main(int argc, char* argv[])
 			coverArtService->flushCache();
 		});
 
-		Service<Scrobbling::IScrobbling> scrobblingService {Scrobbling::createScrobbling(database)};
+		Service<Scrobbling::IScrobbling> scrobblingService {Scrobbling::createScrobbling(ioContext, database)};
 
 		API::Subsonic::SubsonicResource subsonicResource {database};
 

src/test/database/DatabaseTest.cpp

@@ -481,6 +481,8 @@ testSingleTrackSingleRelease(Session& session)
 			auto transaction {session.createUniqueTransaction()};
 
 			track.get().modify()->setRelease(release.get());
+			track.get().modify()->setName("MyTrackName");
+			release.get().modify()->setName("MyReleaseName");
 		}
 
 		{
@@ -498,6 +500,23 @@ testSingleTrackSingleRelease(Session& session)
 			CHECK(track->getRelease());
 			CHECK(track->getRelease().id() == release.getId());
 		}
+
+		{
+			auto transaction {session.createUniqueTransaction()};
+			auto tracks {Track::getByNameAndReleaseName(session, "MyTrackName", "MyReleaseName")};
+			CHECK(tracks.size() == 1);
+			CHECK(tracks.front().id() == track.getId());
+		}
+		{
+			auto transaction {session.createUniqueTransaction()};
+			auto tracks {Track::getByNameAndReleaseName(session, "MyTrackName", "MyReleaseFoo")};
+			CHECK(tracks.size() == 0);
+		}
+		{
+			auto transaction {session.createUniqueTransaction()};
+			auto tracks {Track::getByNameAndReleaseName(session, "MyTrackFoo", "MyReleaseName")};
+			CHECK(tracks.size() == 0);
+		}
 	}
 
 	{