Listenbrainz: save listen even if send failed

src/libs/services/database/include/services/database/Listen.hpp

@@ -24,15 +24,13 @@
 
 #include "services/database/ArtistId.hpp"
 #include "services/database/ClusterId.hpp"
-#include "services/database/IdType.hpp"
+#include "services/database/ListenId.hpp"
 #include "services/database/Object.hpp"
 #include "services/database/ReleaseId.hpp"
 #include "services/database/TrackId.hpp"
 #include "services/database/Types.hpp"
 #include "services/database/UserId.hpp"
 
-LMS_DECLARE_IDTYPE(ListenId)
-
 namespace Database
 {
 
@@ -90,6 +88,11 @@ class Listen : public Object<Listen, ListenId>
 														const std::vector<ClusterId>& clusterIds,
 														Range range = {});
 
+		ScrobblingState getScrobblingState() const { return _scrobblingState; }
+		ObjectPtr<User> getUser() const { return _user; }
+
+		void			setScrobblingState(ScrobblingState state) { _scrobblingState = state; }
+
 		template<class Action>
 		void persist(Action& a)
 		{

src/libs/services/database/include/services/database/ListenId.hpp

@@ -0,0 +1,25 @@
+/*
+ * Copyright (C) 2022 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 "services/database/IdType.hpp"
+
+LMS_DECLARE_IDTYPE(ListenId)
+

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

@@ -19,18 +19,9 @@
 
 #include "ListenBrainzScrobbler.hpp"
 
-#include <Wt/Json/Array.h>
-#include <Wt/Json/Object.h>
-#include <Wt/Json/Value.h>
-#include <Wt/Json/Serializer.h>
-
-#include "services/database/Artist.hpp"
 #include "services/database/Db.hpp"
-#include "services/database/Release.hpp"
 #include "services/database/Session.hpp"
 #include "services/database/Track.hpp"
-#include "services/database/TrackList.hpp"
-#include "services/database/User.hpp"
 #include "utils/IConfig.hpp"
 #include "utils/http/IClient.hpp"
 #include "utils/Logger.hpp"
@@ -56,86 +47,6 @@ namespace
 
 		return res;
 	}
-
-	std::optional<Wt::Json::Object>
-	listenToJsonPayload(Database::Session& session, const Scrobbling::Listen& listen, const Wt::WDateTime& timePoint)
-	{
-		auto transaction {session.createSharedTransaction()};
-
-		const Database::Track::pointer track {Database::Track::find(session, listen.trackId)};
-		if (!track)
-			return std::nullopt;
-
-		auto artists {track->getArtists({Database::TrackArtistLinkType::Artist})};
-		if (artists.empty())
-			artists = track->getArtists({Database::TrackArtistLinkType::ReleaseArtist});
-
-		if (artists.empty())
-		{
-			LOG(DEBUG) << "Track cannot be scrobbled since it does not have any artist";
-			return std::nullopt;
-		}
-
-		Wt::Json::Object additionalInfo;
-		additionalInfo["listening_from"] = "LMS";
-		if (track->getRelease())
-		{
-			if (auto MBID {track->getRelease()->getMBID()})
-				additionalInfo["release_mbid"] = Wt::Json::Value {std::string {MBID->getAsString()}};
-		}
-
-		{
-			Wt::Json::Array artistMBIDs;
-			for (const Database::Artist::pointer& artist : artists)
-			{
-				if (auto MBID {artist->getMBID()})
-					artistMBIDs.push_back(Wt::Json::Value {std::string {MBID->getAsString()}});
-			}
-
-			if (!artistMBIDs.empty())
-				additionalInfo["artist_mbids"] = std::move(artistMBIDs);
-		}
-
-		if (auto MBID {track->getTrackMBID()})
-			additionalInfo["track_mbid"] = Wt::Json::Value {std::string {MBID->getAsString()}};
-
-		if (auto MBID {track->getRecordingMBID()})
-			additionalInfo["recording_mbid"] = Wt::Json::Value {std::string {MBID->getAsString()}};
-
-		if (const std::optional<std::size_t> trackNumber {track->getTrackNumber()})
-			additionalInfo["tracknumber"] = Wt::Json::Value {static_cast<long long int>(*trackNumber)};
-
-		Wt::Json::Object trackMetadata;
-		trackMetadata["additional_info"] = std::move(additionalInfo);
-		trackMetadata["artist_name"] = Wt::Json::Value {artists.front()->getName()};
-		trackMetadata["track_name"] = Wt::Json::Value {track->getName()};
-		if (track->getRelease())
-			trackMetadata["release_name"] = Wt::Json::Value {track->getRelease()->getName()};
-
-		Wt::Json::Object payload;
-		payload["track_metadata"] = std::move(trackMetadata);
-		if (timePoint.isValid())
-			payload["listened_at"] = Wt::Json::Value {static_cast<long long int>(timePoint.toTime_t())};
-
-		return payload;
-	}
-
-	std::string
-	listenToJsonString(Database::Session& session, const Scrobbling::Listen& listen, const Wt::WDateTime& timePoint, std::string_view listenType)
-	{
-		std::string res;
-
-		std::optional<Wt::Json::Object> payload {listenToJsonPayload(session, listen, timePoint)};
-		if (!payload)
-			return res;
-
-		Wt::Json::Object root;
-		root["listen_type"] = Wt::Json::Value {std::string {listenType}};
-		root["payload"] = Wt::Json::Array {std::move(*payload)};
-
-		res = Wt::Json::serialize(root);
-		return res;
-	}
 }
 
 namespace Scrobbling::ListenBrainz
@@ -183,38 +94,7 @@ namespace Scrobbling::ListenBrainz
 	void
 	Scrobbler::enqueListen(const Listen& listen, const Wt::WDateTime& timePoint)
 	{
-		Http::ClientPOSTRequestParameters request;
+		_listensSynchronizer.enqueListen(listen, timePoint);
-		request.relativeUrl = "/1/submit-listens";
-
-		if (timePoint.isValid())
-		{
-			request.priority = Http::ClientRequestParameters::Priority::Normal;
-			request.onSuccessFunc = [=](std::string_view)
-			{
-				_listensSynchronizer.saveListen(TimedListen {listen, timePoint});
-			};
-		}
-		else
-		{
-			// We want "listen now" to appear as soon as possible
-			request.priority = Http::ClientRequestParameters::Priority::High;
-		}
-
-		std::string bodyText {listenToJsonString(_db.getTLSSession(), listen, timePoint, timePoint.isValid() ? "single" : "playing_now")};
-		if (bodyText.empty())
-		{
-			LOG(DEBUG) << "Cannot convert listen to json: skipping";
-			return;
-		}
-
-		const std::optional<UUID> listenBrainzToken {Utils::getListenBrainzToken(_db.getTLSSession(), listen.userId)};
-		if (!listenBrainzToken)
-			return;
-
-		request.message.addBodyText(bodyText);
-		request.message.addHeader("Authorization", "Token " + std::string {listenBrainzToken->getAsString()});
-		request.message.addHeader("Content-Type", "application/json");
-		_client->sendPOSTRequest(std::move(request));
 	}
 } // namespace Scrobbling::ListenBrainz
 

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

@@ -47,6 +47,86 @@ namespace
 {
 	using namespace Scrobbling::ListenBrainz;
 
+	std::optional<Wt::Json::Object>
+	listenToJsonPayload(Database::Session& session, const Scrobbling::Listen& listen, const Wt::WDateTime& timePoint)
+	{
+		auto transaction {session.createSharedTransaction()};
+
+		const Database::Track::pointer track {Database::Track::find(session, listen.trackId)};
+		if (!track)
+			return std::nullopt;
+
+		auto artists {track->getArtists({Database::TrackArtistLinkType::Artist})};
+		if (artists.empty())
+			artists = track->getArtists({Database::TrackArtistLinkType::ReleaseArtist});
+
+		if (artists.empty())
+		{
+			LOG(DEBUG) << "Track cannot be scrobbled since it does not have any artist";
+			return std::nullopt;
+		}
+
+		Wt::Json::Object additionalInfo;
+		additionalInfo["listening_from"] = "LMS";
+		if (track->getRelease())
+		{
+			if (auto MBID {track->getRelease()->getMBID()})
+				additionalInfo["release_mbid"] = Wt::Json::Value {std::string {MBID->getAsString()}};
+		}
+
+		{
+			Wt::Json::Array artistMBIDs;
+			for (const Database::Artist::pointer& artist : artists)
+			{
+				if (auto MBID {artist->getMBID()})
+					artistMBIDs.push_back(Wt::Json::Value {std::string {MBID->getAsString()}});
+			}
+
+			if (!artistMBIDs.empty())
+				additionalInfo["artist_mbids"] = std::move(artistMBIDs);
+		}
+
+		if (auto MBID {track->getTrackMBID()})
+			additionalInfo["track_mbid"] = Wt::Json::Value {std::string {MBID->getAsString()}};
+
+		if (auto MBID {track->getRecordingMBID()})
+			additionalInfo["recording_mbid"] = Wt::Json::Value {std::string {MBID->getAsString()}};
+
+		if (const std::optional<std::size_t> trackNumber {track->getTrackNumber()})
+			additionalInfo["tracknumber"] = Wt::Json::Value {static_cast<long long int>(*trackNumber)};
+
+		Wt::Json::Object trackMetadata;
+		trackMetadata["additional_info"] = std::move(additionalInfo);
+		trackMetadata["artist_name"] = Wt::Json::Value {artists.front()->getName()};
+		trackMetadata["track_name"] = Wt::Json::Value {track->getName()};
+		if (track->getRelease())
+			trackMetadata["release_name"] = Wt::Json::Value {track->getRelease()->getName()};
+
+		Wt::Json::Object payload;
+		payload["track_metadata"] = std::move(trackMetadata);
+		if (timePoint.isValid())
+			payload["listened_at"] = Wt::Json::Value {static_cast<long long int>(timePoint.toTime_t())};
+
+		return payload;
+	}
+
+	std::string
+	listenToJsonString(Database::Session& session, const Scrobbling::Listen& listen, const Wt::WDateTime& timePoint, std::string_view listenType)
+	{
+		std::string res;
+
+		std::optional<Wt::Json::Object> payload {listenToJsonPayload(session, listen, timePoint)};
+		if (!payload)
+			return res;
+
+		Wt::Json::Object root;
+		root["listen_type"] = Wt::Json::Value {std::string {listenType}};
+		root["payload"] = Wt::Json::Array {std::move(*payload)};
+
+		res = Wt::Json::serialize(root);
+		return res;
+	}
+
 	std::string
 	parseValidateToken(std::string_view msgBody)
 	{
@@ -226,30 +306,91 @@ namespace Scrobbling::ListenBrainz
 	}
 
 	void
+	ListensSynchronizer::enqueListen(const Listen& listen, const Wt::WDateTime& timePoint)
+	{
+		Http::ClientPOSTRequestParameters request;
+		request.relativeUrl = "/1/submit-listens";
+
+		if (timePoint.isValid())
+		{
+			// We want the listen to be sent again later in case of failure, so we just save it as pending send
+			const Database::ListenId listenId {saveListen(TimedListen {listen, timePoint})};
+			if (!listenId.isValid())
+				return;
+
+			request.priority = Http::ClientRequestParameters::Priority::Normal;
+			request.onSuccessFunc = [=](std::string_view)
+			{
+				onListenSent(listenId);
+			};
+			// on failure, this listen will be sent during the next sync
+		}
+		else
+		{
+			// We want "listen now" to appear as soon as possible
+			request.priority = Http::ClientRequestParameters::Priority::High;
+			// don't retry on failure
+		}
+
+		std::string bodyText {listenToJsonString(_db.getTLSSession(), listen, timePoint, timePoint.isValid() ? "single" : "playing_now")};
+		if (bodyText.empty())
+		{
+			LOG(DEBUG) << "Cannot convert listen to json: skipping";
+			return;
+		}
+
+		const std::optional<UUID> listenBrainzToken {Utils::getListenBrainzToken(_db.getTLSSession(), listen.userId)};
+		if (!listenBrainzToken)
+			return;
+
+		request.message.addBodyText(bodyText);
+		request.message.addHeader("Authorization", "Token " + std::string {listenBrainzToken->getAsString()});
+		request.message.addHeader("Content-Type", "application/json");
+		_client.sendPOSTRequest(std::move(request));
+
+	}
+
+	Database::ListenId
 	ListensSynchronizer::saveListen(const TimedListen& listen)
+	{
+		using namespace Database;
+
+		Session& session {_db.getTLSSession()};
+		auto transaction {session.createUniqueTransaction()};
+
+		if (Database::Listen::find(session, listen.userId, listen.trackId, Database::Scrobbler::ListenBrainz, listen.listenedAt))
+			return {};
+
+		const User::pointer user {User::find(session, listen.userId)};
+		if (!user)
+			return {};
+
+		const Track::pointer track {Track::find(session, listen.trackId)};
+		if (!track)
+			return {};
+
+		const auto dbListen {Database::Listen::create(session, user, track, Database::Scrobbler::ListenBrainz, listen.listenedAt)};
+		assert(dbListen->getScrobblingState() == Database::ScrobblingState::PendingAdd);
+
+		return dbListen->getId();
+	}
+
+	void
+	ListensSynchronizer::onListenSent(Database::ListenId listenId)
 	{
 		_strand.dispatch([=]
 		{
 			Database::Session& session {_db.getTLSSession()};
-
 			auto transaction {session.createUniqueTransaction()};
 
-			if (Database::Listen::find(session, listen.userId, listen.trackId, Database::Scrobbler::ListenBrainz, listen.listenedAt))
+			if (Database::Listen::pointer listen {Database::Listen::find(session, listenId)})
-				return;
+			{
+				listen.modify()->setScrobblingState(Database::ScrobblingState::Synchronized);
 
-			const Database::User::pointer user {Database::User::find(session, listen.userId)};
+				UserContext& context {getUserContext(listen->getUser()->getId())};
-			if (!user)
-				return;
-
-			const Database::Track::pointer track {Database::Track::find(session, listen.trackId)};
-			if (!track)
-				return;
-
-			Database::Listen::create(session, user, track, Database::Scrobbler::ListenBrainz, listen.listenedAt);
-
-			UserContext& context {getUserContext(listen.userId)};
 				if (context.listenCount)
 					(*context.listenCount)++;
+			}
 		});
 	}
 
@@ -303,7 +444,7 @@ namespace Scrobbling::ListenBrainz
 	void
 	ListensSynchronizer::startGetListens()
 	{
-		LOG(DEBUG) << "GetListens started!!!";
+		LOG(DEBUG) << "GetListens started!";
 
 		assert(!isFetching());
 

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

@@ -26,6 +26,8 @@
 #include <boost/asio/steady_timer.hpp>
 
 #include "services/database/Types.hpp"
+#include "services/database/ListenId.hpp"
+#include "services/database/UserId.hpp"
 #include "services/scrobbling/Listen.hpp"
 
 namespace Database
@@ -48,9 +50,12 @@ namespace Scrobbling::ListenBrainz
 		public:
 			ListensSynchronizer(boost::asio::io_context& ioContext, Database::Db& db, Http::IClient& client);
 
-			void saveListen(const TimedListen& listen);
+			void enqueListen(const Listen& listen, const Wt::WDateTime& timePoint);
 
 		private:
+			Database::ListenId	saveListen(const TimedListen& listen);
+			void				onListenSent(Database::ListenId listenId);
+
 			struct UserContext
 			{
 				UserContext(Database::UserId id) : userId {id} {}