From 4177354afe26063d86b205f56a01aa66957ecc70 Mon Sep 17 00:00:00 2001 From: emeric Date: Sun, 30 Jan 2022 14:18:22 +0100 Subject: [PATCH] Send pending listens that have not been successfully sent --- src/libs/services/database/impl/Listen.cpp | 21 +- src/libs/services/database/impl/Migration.cpp | 13 +- src/libs/services/database/impl/User.cpp | 7 +- .../include/services/database/Listen.hpp | 21 +- .../include/services/database/User.hpp | 11 +- .../services/database/test/DatabaseTest.cpp | 4 +- src/libs/services/database/test/Listen.cpp | 29 ++- .../scrobbling/impl/ScrobblingService.cpp | 18 +- .../listenbrainz/ListenBrainzScrobbler.cpp | 19 +- .../listenbrainz/ListenBrainzScrobbler.hpp | 3 - .../impl/listenbrainz/ListensSynchronizer.cpp | 190 +++++++++++------- .../impl/listenbrainz/ListensSynchronizer.hpp | 27 +-- .../scrobbling/impl/listenbrainz/Utils.cpp | 8 - .../scrobbling/impl/listenbrainz/Utils.hpp | 2 +- src/libs/subsonic/impl/SubsonicResource.cpp | 2 +- src/libs/utils/impl/http/SendQueue.cpp | 6 +- src/libs/utils/include/utils/Logger.hpp | 3 +- src/lms/ui/admin/UsersView.cpp | 2 +- 18 files changed, 232 insertions(+), 154 deletions(-) diff --git a/src/libs/services/database/impl/Listen.cpp b/src/libs/services/database/impl/Listen.cpp index fde20d8c..c9a2cb45 100644 --- a/src/libs/services/database/impl/Listen.cpp +++ b/src/libs/services/database/impl/Listen.cpp @@ -155,17 +155,24 @@ namespace Database return session.getDboSession().find().where("id = ?").bind(id).resultValue(); } - RangeResults - Listen::find(Session& session, UserId userId, Scrobbler scrobbler, Range range) + RangeResults + Listen::find(Session& session, const FindParameters& parameters) { session.checkSharedLocked(); - auto query {session.getDboSession().find() - .where("user_id = ?").bind(userId) - .where("scrobbler = ?").bind(scrobbler) - .orderBy("date_time")}; + auto query {session.getDboSession().query("SELECT id FROM listen") + .orderBy("date_time")}; - return execQuery(query, range); + if (parameters.user.isValid()) + query.where("user_id = ?").bind(parameters.user); + + if (parameters.scrobbler) + query.where("scrobbler = ?").bind(*parameters.scrobbler); + + if (parameters.scrobblingState) + query.where("scrobbling_state = ?").bind(*parameters.scrobblingState); + + return execQuery(query, parameters.range); } Listen::pointer diff --git a/src/libs/services/database/impl/Migration.cpp b/src/libs/services/database/impl/Migration.cpp index 73c9e2c0..64198a88 100644 --- a/src/libs/services/database/impl/Migration.cpp +++ b/src/libs/services/database/impl/Migration.cpp @@ -75,6 +75,13 @@ namespace Database::Migration Db& _db; }; + static + std::string + dateTimeToDbFormat(const Wt::WDateTime& dateTime) + { + return dateTime.toString("yyyy'-'MM'-'dd'T'hh':'mm':'ss'.000'", false).toUTF8(); + } + static void migrateFromV5(Session& session) @@ -450,7 +457,7 @@ CREATE TABLE "starred_track" ( // Can't migrate using class mapping as mapping may evolve in the future // use time_t to avoid rounding issues later - const Wt::WDateTime now {Wt::WDateTime::fromTime_t(Wt::WDateTime::currentDateTime().toTime_t())}; + const std::string now {dateTimeToDbFormat(Wt::WDateTime::fromTime_t(Wt::WDateTime::currentDateTime().toTime_t()))}; std::map userScrobblers; auto getScrobbler {[&](IdType::ValueType userId) @@ -480,7 +487,7 @@ CREATE TABLE "starred_track" ( session.getDboSession().execute("INSERT INTO " + newTableName + " ('version', 'scrobbler', 'date_time', '" + colName + "', 'user_id') VALUES (?, ?, ?, ?, ?)") .bind(0) .bind(getScrobbler(userId)) - .bind(now.toString().toUTF8()) + .bind(now) .bind(entryId) .bind(userId); } @@ -523,7 +530,7 @@ CREATE TABLE "listen" ( { session.getDboSession().execute("INSERT INTO listen ('version', 'date_time', 'scrobbler', 'scrobbling_state', 'track_id', 'user_id') VALUES (?, ?, ?, ?, ?, ?)") .bind(0) - .bind(dateTime.toString().toUTF8()) + .bind(dateTimeToDbFormat(dateTime)) .bind(scrobbler) .bind(ScrobblingState::Synchronized) // consider sync is done to avoid duplicate submissions .bind(trackId) diff --git a/src/libs/services/database/impl/User.cpp b/src/libs/services/database/impl/User.cpp index a8abea66..31728e06 100644 --- a/src/libs/services/database/impl/User.cpp +++ b/src/libs/services/database/impl/User.cpp @@ -44,13 +44,16 @@ User::getCount(Session& session) } RangeResults -User::find(Session& session, Range range) +User::find(Session& session, const FindParameters& params) { session.checkSharedLocked(); auto query {session.getDboSession().query("SELECT id FROM user")}; - return execQuery(query, range); + if (params.scrobbler) + query.where("scrobbler = ?").bind(*params.scrobbler); + + return execQuery(query, params.range); } User::pointer diff --git a/src/libs/services/database/include/services/database/Listen.hpp b/src/libs/services/database/include/services/database/Listen.hpp index 80164cdc..edbb218b 100644 --- a/src/libs/services/database/include/services/database/Listen.hpp +++ b/src/libs/services/database/include/services/database/Listen.hpp @@ -44,11 +44,24 @@ class Listen : public Object Listen() = default; Listen(ObjectPtr user, ObjectPtr track, Scrobbler scrobbler, const Wt::WDateTime& dateTime); + struct FindParameters + { + UserId user; + std::optional scrobbler; + std::optional scrobblingState; + Range range; + + FindParameters& setUser(UserId _user) { user = _user; return *this; } + FindParameters& setScrobbler(Scrobbler _scrobbler) { scrobbler = _scrobbler; return *this; } + FindParameters& setScrobblingState(ScrobblingState _scrobblingState) { scrobblingState = _scrobblingState; return *this; } + FindParameters& setRange(Range _range) {range = _range; return *this; } + }; + // Accessors static std::size_t getCount(Session& session); static pointer find(Session& session, ListenId id); - static RangeResults find(Session& session, UserId userId, Scrobbler scrobbler, Range = {}); static pointer find(Session& session, UserId userId, TrackId trackId, Scrobbler scrobbler, const Wt::WDateTime& dateTime); + static RangeResults find(Session& session, const FindParameters& parameters); // Create static pointer create(Session& session, ObjectPtr user, ObjectPtr track, Scrobbler scrobbler, const Wt::WDateTime& dateTime); @@ -88,8 +101,10 @@ class Listen : public Object const std::vector& clusterIds, Range range = {}); - ScrobblingState getScrobblingState() const { return _scrobblingState; } - ObjectPtr getUser() const { return _user; } + ScrobblingState getScrobblingState() const { return _scrobblingState; } + ObjectPtr getUser() const { return _user; } + ObjectPtr getTrack() const { return _track; } + const Wt::WDateTime& getDateTime() const { return _dateTime; } void setScrobblingState(ScrobblingState state) { _scrobblingState = state; } diff --git a/src/libs/services/database/include/services/database/User.hpp b/src/libs/services/database/include/services/database/User.hpp index e42a2181..4a446296 100644 --- a/src/libs/services/database/include/services/database/User.hpp +++ b/src/libs/services/database/include/services/database/User.hpp @@ -45,6 +45,15 @@ class User : public Object std::string hash; }; + struct FindParameters + { + std::optional scrobbler; + Range range; + + FindParameters& setScrobbler(Scrobbler _scrobbler) { scrobbler = _scrobbler; return *this; } + FindParameters& setRange(Range _range) {range = _range; return *this; } + }; + static inline const std::size_t MinNameLength {3}; static inline const std::size_t MaxNameLength {15}; static inline const bool defaultSubsonicTranscodeEnable {true}; @@ -63,7 +72,7 @@ class User : public Object static std::size_t getCount(Session& session); static pointer find(Session& session, UserId id); static pointer find(Session& session, std::string_view loginName); - static RangeResults find(Session& session, Range range); + static RangeResults find(Session& session, const FindParameters& params); static pointer findDemoUser(Session& session); // accessors diff --git a/src/libs/services/database/test/DatabaseTest.cpp b/src/libs/services/database/test/DatabaseTest.cpp index f106e06a..918d519a 100644 --- a/src/libs/services/database/test/DatabaseTest.cpp +++ b/src/libs/services/database/test/DatabaseTest.cpp @@ -89,7 +89,7 @@ TEST_F(DatabaseFixture, SingleUser) { { auto transaction {session.createSharedTransaction()}; - EXPECT_TRUE(User::find(session, Range {}).results.empty()); + EXPECT_TRUE(User::find(session, User::FindParameters {}).results.empty()); EXPECT_EQ(User::getCount(session), 0); } @@ -98,7 +98,7 @@ TEST_F(DatabaseFixture, SingleUser) { auto transaction {session.createSharedTransaction()}; - EXPECT_EQ(User::find(session, Range {}).results.size(), 1); + EXPECT_EQ(User::find(session, User::FindParameters {}).results.size(), 1); EXPECT_EQ(User::getCount(session), 1); } } diff --git a/src/libs/services/database/test/Listen.cpp b/src/libs/services/database/test/Listen.cpp index 1b5b12b9..d43f7ad0 100644 --- a/src/libs/services/database/test/Listen.cpp +++ b/src/libs/services/database/test/Listen.cpp @@ -66,17 +66,28 @@ TEST_F(DatabaseFixture, Listen_get) { auto transaction {session.createSharedTransaction()}; - auto listens {Listen::find(session, user->getId(), Scrobbler::ListenBrainz)}; + auto listens {Listen::find(session, Listen::FindParameters{}.setUser(user->getId()).setScrobbler(Scrobbler::ListenBrainz))}; EXPECT_EQ(listens.results.size(), 0); } { auto transaction {session.createSharedTransaction()}; - auto listens {Listen::find(session, user->getId(), Scrobbler::Internal)}; - EXPECT_EQ(listens.moreResults, false); - ASSERT_EQ(listens.results.size(), 1); - EXPECT_EQ(listens.results.front()->getId(), listen->getId()); + { + auto listens {Listen::find(session, Listen::FindParameters{}.setUser(user->getId()).setScrobbler(Scrobbler::Internal))}; + EXPECT_EQ(listens.moreResults, false); + ASSERT_EQ(listens.results.size(), 1); + EXPECT_EQ(listens.results.front(), listen->getId()); + } + + { + auto listens {Listen::find(session, Listen::FindParameters{}.setUser(user->getId()).setScrobbler(Scrobbler::Internal).setScrobblingState(ScrobblingState::PendingAdd))}; + EXPECT_EQ(listens.results.size(), 1); + } + { + auto listens {Listen::find(session, Listen::FindParameters{}.setUser(user->getId()).setScrobbler(Scrobbler::Internal).setScrobblingState(ScrobblingState::Synchronized))}; + EXPECT_EQ(listens.results.size(), 0); + } } } @@ -91,11 +102,11 @@ TEST_F(DatabaseFixture, Listen_get_multi) { auto transaction {session.createSharedTransaction()}; - auto listens {Listen::find(session, user->getId(), Scrobbler::Internal)}; + auto listens {Listen::find(session, Listen::FindParameters{}.setUser(user->getId()).setScrobbler(Scrobbler::Internal))}; ASSERT_EQ(listens.results.size(), 3); - EXPECT_EQ(listens.results[0]->getId(), listen1.getId()); - EXPECT_EQ(listens.results[1]->getId(), listen2.getId()); - EXPECT_EQ(listens.results[2]->getId(), listen3.getId()); + EXPECT_EQ(listens.results[0], listen1.getId()); + EXPECT_EQ(listens.results[1], listen2.getId()); + EXPECT_EQ(listens.results[2], listen3.getId()); } } diff --git a/src/libs/services/scrobbling/impl/ScrobblingService.cpp b/src/libs/services/scrobbling/impl/ScrobblingService.cpp index a89ca613..ac24670d 100644 --- a/src/libs/services/scrobbling/impl/ScrobblingService.cpp +++ b/src/libs/services/scrobbling/impl/ScrobblingService.cpp @@ -105,7 +105,8 @@ namespace Scrobbling Session& session {_db.getTLSSession()}; auto transaction {session.createSharedTransaction()}; - return Database::Listen::getRecentArtists(session, userId, *scrobbler, clusterIds, linkType, range); + res = Database::Listen::getRecentArtists(session, userId, *scrobbler, clusterIds, linkType, range); + return res; } ScrobblingService::ReleaseContainer @@ -120,7 +121,8 @@ namespace Scrobbling Session& session {_db.getTLSSession()}; auto transaction {session.createSharedTransaction()}; - return Database::Listen::getRecentReleases(session, userId, *scrobbler, clusterIds, range); + res = Database::Listen::getRecentReleases(session, userId, *scrobbler, clusterIds, range); + return res; } ScrobblingService::TrackContainer @@ -135,7 +137,8 @@ namespace Scrobbling Session& session {_db.getTLSSession()}; auto transaction {session.createSharedTransaction()}; - return Database::Listen::getRecentTracks(session, userId, *scrobbler, clusterIds, range); + res = Database::Listen::getRecentTracks(session, userId, *scrobbler, clusterIds, range); + return res; } // Top @@ -151,7 +154,8 @@ namespace Scrobbling Session& session {_db.getTLSSession()}; auto transaction {session.createSharedTransaction()}; - return Database::Listen::getTopArtists(session, userId, *scrobbler, clusterIds, linkType, range); + res = Database::Listen::getTopArtists(session, userId, *scrobbler, clusterIds, linkType, range); + return res; } ScrobblingService::ReleaseContainer @@ -166,7 +170,8 @@ namespace Scrobbling Session& session {_db.getTLSSession()}; auto transaction {session.createSharedTransaction()}; - return Database::Listen::getTopReleases(session, userId, *scrobbler, clusterIds, range); + res = Database::Listen::getTopReleases(session, userId, *scrobbler, clusterIds, range); + return res; } ScrobblingService::TrackContainer @@ -181,7 +186,8 @@ namespace Scrobbling Session& session {_db.getTLSSession()}; auto transaction {session.createSharedTransaction()}; - return Database::Listen::getTopTracks(session, userId, *scrobbler, clusterIds, range); + res = Database::Listen::getTopTracks(session, userId, *scrobbler, clusterIds, range); + return res; } void diff --git a/src/libs/services/scrobbling/impl/listenbrainz/ListenBrainzScrobbler.cpp b/src/libs/services/scrobbling/impl/listenbrainz/ListenBrainzScrobbler.cpp index b0365e95..62a57ce0 100644 --- a/src/libs/services/scrobbling/impl/listenbrainz/ListenBrainzScrobbler.cpp +++ b/src/libs/services/scrobbling/impl/listenbrainz/ListenBrainzScrobbler.cpp @@ -69,7 +69,7 @@ namespace Scrobbling::ListenBrainz void Scrobbler::listenStarted(const Listen& listen) { - enqueListen(listen, Wt::WDateTime {}); + _listensSynchronizer.enqueListenNow(listen); } void @@ -78,23 +78,14 @@ namespace Scrobbling::ListenBrainz if (duration && !canBeScrobbled(_db.getTLSSession(), listen.trackId, *duration)) return; - const Listen timedListen {listen}; - const Wt::WDateTime now {Wt::WDateTime::currentDateTime()}; - - enqueListen(timedListen, now); + const TimedListen timedListen {listen, Wt::WDateTime::currentDateTime()}; + _listensSynchronizer.enqueListen(timedListen); } void - Scrobbler::addTimedListen(const TimedListen& listen) + Scrobbler::addTimedListen(const TimedListen& timedListen) { - assert(listen.listenedAt.isValid()); - enqueListen(listen, listen.listenedAt); - } - - void - Scrobbler::enqueListen(const Listen& listen, const Wt::WDateTime& timePoint) - { - _listensSynchronizer.enqueListen(listen, timePoint); + _listensSynchronizer.enqueListen(timedListen); } } // namespace Scrobbling::ListenBrainz diff --git a/src/libs/services/scrobbling/impl/listenbrainz/ListenBrainzScrobbler.hpp b/src/libs/services/scrobbling/impl/listenbrainz/ListenBrainzScrobbler.hpp index 7863ce74..4a7dc77c 100644 --- a/src/libs/services/scrobbling/impl/listenbrainz/ListenBrainzScrobbler.hpp +++ b/src/libs/services/scrobbling/impl/listenbrainz/ListenBrainzScrobbler.hpp @@ -50,9 +50,6 @@ namespace Scrobbling::ListenBrainz void listenFinished(const Listen& listen, std::optional duration) override; void addTimedListen(const TimedListen& listen) override; - // Submit listens - void enqueListen(const Listen& listen, const Wt::WDateTime& timePoint); - boost::asio::io_context& _ioContext; Database::Db& _db; std::string _baseAPIUrl; diff --git a/src/libs/services/scrobbling/impl/listenbrainz/ListensSynchronizer.cpp b/src/libs/services/scrobbling/impl/listenbrainz/ListensSynchronizer.cpp index 68c8d8e4..329a19bb 100644 --- a/src/libs/services/scrobbling/impl/listenbrainz/ListensSynchronizer.cpp +++ b/src/libs/services/scrobbling/impl/listenbrainz/ListensSynchronizer.cpp @@ -42,6 +42,7 @@ #include "Utils.hpp" #define LOG(sev) LMS_LOG(SCROBBLING, sev) << "[listenbrainz Synchronizer] - " +#define LOG_EX(sev) LMS_LOG_EX(Module::SCROBBLING, sev) << "[listenbrainz Synchronizer] - " namespace { @@ -173,6 +174,8 @@ namespace { using namespace Database; + //LOG(DEBUG) << "Trying to match track' " << Wt::Json::serialize(metadata) << "'"; + // first try to get the associated track using MBIDs, and then fallback on names if (metadata.type("additional_info") == Wt::Json::Type::Object) { @@ -302,7 +305,20 @@ namespace Scrobbling::ListenBrainz { LOG(INFO) << "Starting Listens synchronizer, maxSyncListenCount = " << _maxSyncListenCount << ", _syncListensPeriod = " << _syncListensPeriod.count() << " hours"; - scheduleGetListens(std::chrono::seconds {30}); + scheduleSync(std::chrono::seconds {30}); + } + + void + ListensSynchronizer::enqueListen(const TimedListen& listen) + { + assert(listen.listenedAt.isValid()); + enqueListen(listen, listen.listenedAt); + } + + void + ListensSynchronizer::enqueListenNow(const Listen& listen) + { + enqueListen(listen, {}); } void @@ -313,15 +329,22 @@ namespace Scrobbling::ListenBrainz if (timePoint.isValid()) { + const TimedListen timedListen {listen, timePoint}; // 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; + saveListen(timedListen, Database::ScrobblingState::PendingAdd); request.priority = Http::ClientRequestParameters::Priority::Normal; request.onSuccessFunc = [=](std::string_view) { - onListenSent(listenId); + _strand.dispatch([=] + { + if (saveListen(timedListen, Database::ScrobblingState::Synchronized)) + { + UserContext& context {getUserContext(listen.userId)}; + if (context.listenCount) + (*context.listenCount)++; + } + }); }; // on failure, this listen will be sent during the next sync } @@ -341,62 +364,93 @@ namespace Scrobbling::ListenBrainz const std::optional listenBrainzToken {Utils::getListenBrainzToken(_db.getTLSSession(), listen.userId)}; if (!listenBrainzToken) + { + LOG(DEBUG) << "No listenbrainz token found: skipping"; 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) + bool + ListensSynchronizer::saveListen(const TimedListen& listen, Database::ScrobblingState scrobblingState) { 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 {}; + Database::Listen::pointer dbListen {Database::Listen::find(session, listen.userId, listen.trackId, Database::Scrobbler::ListenBrainz, listen.listenedAt)}; + if (!dbListen) + { + const User::pointer user {User::find(session, listen.userId)}; + if (!user) + return false; - const User::pointer user {User::find(session, listen.userId)}; - if (!user) - return {}; + const Track::pointer track {Track::find(session, listen.trackId)}; + if (!track) + return false; - const Track::pointer track {Track::find(session, listen.trackId)}; - if (!track) - return {}; + dbListen = Database::Listen::create(session, user, track, Database::Scrobbler::ListenBrainz, listen.listenedAt); + dbListen.modify()->setScrobblingState(scrobblingState); - const auto dbListen {Database::Listen::create(session, user, track, Database::Scrobbler::ListenBrainz, listen.listenedAt)}; - assert(dbListen->getScrobblingState() == Database::ScrobblingState::PendingAdd); + LOG(DEBUG) << "LISTEN CREATED for user " << user->getLoginName() << ", track '" << track->getName() << "' AT " << listen.listenedAt.toString(); - return dbListen->getId(); + return true; + } + + if (dbListen->getScrobblingState() == scrobblingState) + return false; + + dbListen.modify()->setScrobblingState(scrobblingState); + return true; } void - ListensSynchronizer::onListenSent(Database::ListenId listenId) + ListensSynchronizer::enquePendingListens() { - _strand.dispatch([=] + std::vector pendingListens; + { Database::Session& session {_db.getTLSSession()}; + auto transaction {session.createUniqueTransaction()}; - if (Database::Listen::pointer listen {Database::Listen::find(session, listenId)}) - { - listen.modify()->setScrobblingState(Database::ScrobblingState::Synchronized); + Database::Listen::FindParameters params; + params.setScrobbler(Database::Scrobbler::ListenBrainz) + .setScrobblingState(Database::ScrobblingState::PendingAdd) + .setRange(Database::Range {0, 100}); // don't flood too much? - UserContext& context {getUserContext(listen->getUser()->getId())}; - if (context.listenCount) - (*context.listenCount)++; + const Database::RangeResults results {Database::Listen::find(session, params)}; + pendingListens.reserve(results.results.size()); + + for (Database::ListenId listenId : results.results) + { + const Database::Listen::pointer listen {Database::Listen::find(session, listenId)}; + + TimedListen timedListen; + timedListen.listenedAt = listen->getDateTime(); + timedListen.userId = listen->getUser()->getId(); + timedListen.trackId = listen->getTrack()->getId(); + + pendingListens.push_back(std::move(timedListen)); } - }); + } + + LOG(DEBUG) << "Queing " << pendingListens.size() << " pending listen"; + + for (const TimedListen& pendingListen : pendingListens) + enqueListen(pendingListen); } ListensSynchronizer::UserContext& ListensSynchronizer::getUserContext(Database::UserId userId) { + assert(_strand.running_in_this_thread()); + auto itContext {_userContexts.find(userId)}; if (itContext == std::cend(_userContexts)) { @@ -408,24 +462,24 @@ namespace Scrobbling::ListenBrainz } bool - ListensSynchronizer::isFetching() const + ListensSynchronizer::isSyncing() const { return std::any_of(std::cbegin(_userContexts), std::cend(_userContexts), [](const auto& contextEntry) { const auto& [userId, context] {contextEntry}; - return context.fetching; + return context.syncing; }); } void - ListensSynchronizer::scheduleGetListens(std::chrono::seconds fromNow) + ListensSynchronizer::scheduleSync(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) + _syncTimer.expires_after(fromNow); + _syncTimer.async_wait(boost::asio::bind_executor(_strand, [this] (const boost::system::error_code& ec) { if (ec == boost::asio::error::operation_aborted) { @@ -437,38 +491,37 @@ namespace Scrobbling::ListenBrainz throw Exception {"GetListens timer failure: " + std::string {ec.message()} }; } - startGetListens(); + startSync(); })); } void - ListensSynchronizer::startGetListens() + ListensSynchronizer::startSync() { - LOG(DEBUG) << "GetListens started!"; + LOG(DEBUG) << "Starting sync!"; - assert(!isFetching()); + assert(!isSyncing()); + + enquePendingListens(); Database::RangeResults userIds; { Database::Session& session {_db.getTLSSession()}; auto transaction {session.createSharedTransaction()}; - userIds = Database::User::find(_db.getTLSSession(), Database::Range {}); + userIds = Database::User::find(_db.getTLSSession(), Database::User::FindParameters{}.setScrobbler(Database::Scrobbler::ListenBrainz)); } for (const Database::UserId userId : userIds.results) - { - if (Utils::getListenBrainzToken(_db.getTLSSession(), userId)) - startGetListens(getUserContext(userId)); - } + startSync(getUserContext(userId)); - if (!isFetching()) - scheduleGetListens(_syncListensPeriod); + if (!isSyncing()) + scheduleSync(_syncListensPeriod); } void - ListensSynchronizer::startGetListens(UserContext& context) + ListensSynchronizer::startSync(UserContext& context) { - context.fetching = true; + context.syncing = true; context.listenBrainzUserName = ""; context.maxDateTime = {}; context.fetchedListenCount = 0; @@ -479,15 +532,15 @@ namespace Scrobbling::ListenBrainz } void - ListensSynchronizer::onGetListensEnded(UserContext& context) + ListensSynchronizer::onSyncEnded(UserContext& context) { _strand.dispatch([this, &context] { - LOG(DEBUG) << "Fetch done for user " << context.userId.getValue() << ", fetched: " << context.fetchedListenCount << ", matched: " << context.matchedListenCount << ", imported: " << context.importedListenCount; - context.fetching = false; + LOG_EX(context.importedListenCount > 0 ? Severity::INFO : Severity::DEBUG) << "Sync done for user '" << context.listenBrainzUserName << "', fetched: " << context.fetchedListenCount << ", matched: " << context.matchedListenCount << ", imported: " << context.importedListenCount; + context.syncing = false; - if (!isFetching()) - scheduleGetListens(_syncListensPeriod); + if (!isSyncing()) + scheduleSync(_syncListensPeriod); }); } @@ -499,7 +552,7 @@ namespace Scrobbling::ListenBrainz const std::optional listenBrainzToken {Utils::getListenBrainzToken(_db.getTLSSession(), context.userId)}; if (!listenBrainzToken) { - onGetListensEnded(context); + onSyncEnded(context); return; } @@ -512,14 +565,14 @@ namespace Scrobbling::ListenBrainz context.listenBrainzUserName = parseValidateToken(msgBody); if (context.listenBrainzUserName.empty()) { - onGetListensEnded(context); + onSyncEnded(context); return; } enqueGetListenCount(context); }; request.onFailureFunc = [this, &context] { - onGetListensEnded(context); + onSyncEnded(context); }; _client.sendGETRequest(std::move(request)); @@ -544,7 +597,7 @@ namespace Scrobbling::ListenBrainz if (!needSync) { - onGetListensEnded(context); + onSyncEnded(context); return; } @@ -553,7 +606,7 @@ namespace Scrobbling::ListenBrainz }; request.onFailureFunc = [this, &context] { - onGetListensEnded(context); + onSyncEnded(context); }; _client.sendGETRequest(std::move(request)); @@ -572,7 +625,7 @@ namespace Scrobbling::ListenBrainz processGetListensResponse(msgBody, context); if (context.fetchedListenCount >= _maxSyncListenCount || !context.maxDateTime.isValid()) { - onGetListensEnded(context); + onSyncEnded(context); return; } @@ -580,7 +633,7 @@ namespace Scrobbling::ListenBrainz }; request.onFailureFunc = [=, &context] { - onGetListensEnded(context); + onSyncEnded(context); }; _client.sendGETRequest(std::move(request)); @@ -597,27 +650,10 @@ namespace Scrobbling::ListenBrainz context.matchedListenCount += parseResult.matchedListens.size(); context.maxDateTime = parseResult.oldestEntry; - if (parseResult.matchedListens.empty()) - return; - - auto transaction {session.createUniqueTransaction()}; - - Database::User::pointer user {Database::User::find(session, context.userId)}; - if (!user) - return; - for (const TimedListen& listen : parseResult.matchedListens) { - const Database::Track::pointer track {Database::Track::find(session, listen.trackId)}; - if (!track) - continue; - - if (Database::Listen::find(session, listen.userId, listen.trackId, Database::Scrobbler::ListenBrainz, listen.listenedAt)) - continue; - - Database::Listen::create(session, user, track, Database::Scrobbler::ListenBrainz, listen.listenedAt); - context.importedListenCount++; + if (saveListen(listen, Database::ScrobblingState::Synchronized)) + context.importedListenCount++; } } } // namespace Scrobbling::ListenBrainz - diff --git a/src/libs/services/scrobbling/impl/listenbrainz/ListensSynchronizer.hpp b/src/libs/services/scrobbling/impl/listenbrainz/ListensSynchronizer.hpp index 258942a0..45fef878 100644 --- a/src/libs/services/scrobbling/impl/listenbrainz/ListensSynchronizer.hpp +++ b/src/libs/services/scrobbling/impl/listenbrainz/ListensSynchronizer.hpp @@ -50,11 +50,14 @@ namespace Scrobbling::ListenBrainz public: ListensSynchronizer(boost::asio::io_context& ioContext, Database::Db& db, Http::IClient& client); - void enqueListen(const Listen& listen, const Wt::WDateTime& timePoint); + void enqueListen(const TimedListen& listen); + void enqueListenNow(const Listen& listen); private: - Database::ListenId saveListen(const TimedListen& listen); - void onListenSent(Database::ListenId listenId); + void enqueListen(const Listen& listen, const Wt::WDateTime& timePoint); + bool saveListen(const TimedListen& listen, Database::ScrobblingState scrobblinState); + + void enquePendingListens(); struct UserContext { @@ -66,10 +69,10 @@ namespace Scrobbling::ListenBrainz UserContext& operator=(UserContext&&) = delete; const Database::UserId userId; - bool fetching {}; + bool syncing {}; std::optional listenCount {}; - // resetted at each fetch + // resetted at each sync std::string listenBrainzUserName; // need to be resolved first Wt::WDateTime maxDateTime; std::size_t fetchedListenCount{}; @@ -78,20 +81,20 @@ namespace Scrobbling::ListenBrainz }; UserContext& getUserContext(Database::UserId userId); - bool isFetching() const; - void scheduleGetListens(std::chrono::seconds fromNow); - void startGetListens(); - void startGetListens(UserContext& context); - void onGetListensEnded(UserContext& context); + bool isSyncing() const; + void scheduleSync(std::chrono::seconds fromNow); + void startSync(); + void startSync(UserContext& context); + void onSyncEnded(UserContext& context); void enqueValidateToken(UserContext& context); void enqueGetListenCount(UserContext& context); void enqueGetListens(UserContext& context); - void processGetListensResponse(std::string_view body, UserContext& context); + void processGetListensResponse(std::string_view body, UserContext& context); boost::asio::io_context& _ioContext; boost::asio::io_context::strand _strand {_ioContext}; Database::Db& _db; - boost::asio::steady_timer _getListensTimer {_ioContext}; + boost::asio::steady_timer _syncTimer {_ioContext}; Http::IClient& _client; std::unordered_map _userContexts; diff --git a/src/libs/services/scrobbling/impl/listenbrainz/Utils.cpp b/src/libs/services/scrobbling/impl/listenbrainz/Utils.cpp index e956370f..4e9acdf6 100644 --- a/src/libs/services/scrobbling/impl/listenbrainz/Utils.cpp +++ b/src/libs/services/scrobbling/impl/listenbrainz/Utils.cpp @@ -19,14 +19,9 @@ #include "Utils.hpp" -#include - #include "services/database/Session.hpp" -#include "services/database/TrackList.hpp" #include "services/database/User.hpp" -static constexpr std::string_view historyTracklistName {"__scrobbler_listenbrainz_history__"}; - namespace Scrobbling::ListenBrainz::Utils { std::optional @@ -38,9 +33,6 @@ namespace Scrobbling::ListenBrainz::Utils if (!user) return std::nullopt; - if (user->getScrobbler() != Database::Scrobbler::ListenBrainz) - return std::nullopt; - return user->getListenBrainzToken(); } } diff --git a/src/libs/services/scrobbling/impl/listenbrainz/Utils.hpp b/src/libs/services/scrobbling/impl/listenbrainz/Utils.hpp index 4a4897c3..47bd2a0e 100644 --- a/src/libs/services/scrobbling/impl/listenbrainz/Utils.hpp +++ b/src/libs/services/scrobbling/impl/listenbrainz/Utils.hpp @@ -32,5 +32,5 @@ namespace Database namespace Scrobbling::ListenBrainz::Utils { - std::optional getListenBrainzToken(Database::Session& session, Database::UserId userId); + std::optional getListenBrainzToken(Database::Session& session, Database::UserId userId); } diff --git a/src/libs/subsonic/impl/SubsonicResource.cpp b/src/libs/subsonic/impl/SubsonicResource.cpp index 54e53642..d8abd57b 100644 --- a/src/libs/subsonic/impl/SubsonicResource.cpp +++ b/src/libs/subsonic/impl/SubsonicResource.cpp @@ -1394,7 +1394,7 @@ handleGetUsersRequest(RequestContext& context) Response response {Response::createOkResponse(context.serverProtocolVersion)}; Response::Node& usersNode {response.createNode("users")}; - const auto userIds {User::find(context.dbSession, Range {})}; + const auto userIds {User::find(context.dbSession, User::FindParameters {})}; for (const UserId userId : userIds.results) { const User::pointer user {User::find(context.dbSession, userId)}; diff --git a/src/libs/utils/impl/http/SendQueue.cpp b/src/libs/utils/impl/http/SendQueue.cpp index d689179e..8d99fc72 100644 --- a/src/libs/utils/impl/http/SendQueue.cpp +++ b/src/libs/utils/impl/http/SendQueue.cpp @@ -26,7 +26,7 @@ #include "utils/Logger.hpp" #include "utils/String.hpp" -#define LOG(sev) LMS_LOG(SCROBBLING, sev) << "[listenbrainz SendQueue] - " +#define LOG(sev) LMS_LOG(SCROBBLING, sev) << "[Http SendQueue] - " namespace StringUtils { @@ -143,7 +143,7 @@ namespace Http { if (ec == boost::asio::error::operation_aborted) { - LOG(DEBUG) << "SendQueue: client aborted"; + LOG(DEBUG) << "Client aborted"; return; } @@ -228,7 +228,7 @@ namespace Http { if (ec == boost::asio::error::operation_aborted) { - LOG(DEBUG) << "SendQueue: throttle aborted"; + LOG(DEBUG) << "Throttle aborted"; return; } else if (ec) diff --git a/src/libs/utils/include/utils/Logger.hpp b/src/libs/utils/include/utils/Logger.hpp index 43b2f893..d7faf078 100644 --- a/src/libs/utils/include/utils/Logger.hpp +++ b/src/libs/utils/include/utils/Logger.hpp @@ -85,5 +85,6 @@ class Logger virtual void processLog(const Log& log) = 0; }; -#define LMS_LOG(module, severity) Log(Service::get(), Module::module, Severity::severity).getOstream() +#define LMS_LOG(module, severity) Log(Service::get(), Module::module, Severity::severity).getOstream() +#define LMS_LOG_EX(module, severity) Log(Service::get(), module, severity).getOstream() diff --git a/src/lms/ui/admin/UsersView.cpp b/src/lms/ui/admin/UsersView.cpp index a5daed2f..84f2c41b 100644 --- a/src/lms/ui/admin/UsersView.cpp +++ b/src/lms/ui/admin/UsersView.cpp @@ -71,7 +71,7 @@ UsersView::refreshView() auto transaction {LmsApp->getDbSession().createSharedTransaction()}; - for (const UserId userId : User::find(LmsApp->getDbSession(), Range {}).results) + for (const UserId userId : User::find(LmsApp->getDbSession(), User::FindParameters {}).results) { const User::pointer user {User::find(LmsApp->getDbSession(), userId)};