From 2360beeb810fb87533ba39b19e875b5997d8ab7a Mon Sep 17 00:00:00 2001 From: lobo Date: Tue, 29 Sep 2026 13:27:31 +0200 Subject: [PATCH] fix(sync): short sync transactions, real Steam retries, no N+1 --- readme.md | 6 +- .../repo/AccountCleanupRepository.java | 2 +- .../modules/library/domain/UserSteamSync.java | 3 + .../library/repo/UserGameRepository.java | 2 + .../library/repo/UserSteamSyncRepository.java | 36 +++- .../library/service/LibrarySyncService.java | 4 +- .../library/service/LibrarySyncWorker.java | 173 +++++++++------- .../SteamProfileAndGamesResponse.java | 9 - .../service/SteamApiService.java | 184 ++++++++---------- src/main/resources/application.yml | 7 + .../db/migration/V23__Sync_Started_At.sql | 3 + .../repo/AccountCleanupRepositoryTest.java | 9 + .../repo/UserSteamSyncRepositoryTest.java | 18 +- .../service/LibrarySyncServiceTest.java | 1 + .../LibrarySyncWorkerConcurrencyTest.java | 140 +++++++++++++ .../service/LibrarySyncWorkerTest.java | 92 ++++++--- .../service/SteamApiServiceCacheTest.java | 60 ++++++ .../service/SteamApiServiceTest.java | 66 ++----- 18 files changed, 536 insertions(+), 279 deletions(-) delete mode 100644 src/main/java/core_service/modules/steam_adapter/api/dto/response/SteamProfileAndGamesResponse.java create mode 100644 src/main/resources/db/migration/V23__Sync_Started_At.sql create mode 100644 src/test/java/core_service/modules/library/service/LibrarySyncWorkerConcurrencyTest.java diff --git a/readme.md b/readme.md index 5656125..997bb20 100644 --- a/readme.md +++ b/readme.md @@ -14,7 +14,11 @@ To Be Played is split into three repositories: ## Features - **Steam sync**: owned games, playtime and achievements through the Steam Web API. Manual sync plus a nightly - sync. Resilience4j retries and rate-limits every Steam call; Caffeine caches game counts and player summaries. + sync. Resilience4j rate-limits every Steam call and tries the owned games up to three times (never again on a 4xx; + the wishlist and the calls a page waits for once); Caffeine caches game counts and player summaries. A sync commits its claim (`RUNNING`, with + `sync_started_at`), its writes and its finish apart, and calls Steam and IGDB outside any transaction: a second + sync sees `RUNNING` and skips, a failed one stores `FAILED`. A run older than `app.library.running-stale-minutes` + can be taken over; its own finish then changes nothing. - **Backlog**: status (`BACKLOG`, `PLAYING`, `FINISHED`, `COMPLETED` = 100 %, `DROPPED`), rating 1–5, notes, priority, filters and sorting. "Pile of Shame" summary with a random suggestion of what to play. - **Picker**: up to 10 random backlog games for a time budget (IGDB normal time to beat) and a mood, HIGH priority diff --git a/src/main/java/core_service/modules/account/repo/AccountCleanupRepository.java b/src/main/java/core_service/modules/account/repo/AccountCleanupRepository.java index 5df0ed8..f412cf7 100644 --- a/src/main/java/core_service/modules/account/repo/AccountCleanupRepository.java +++ b/src/main/java/core_service/modules/account/repo/AccountCleanupRepository.java @@ -19,7 +19,7 @@ import java.util.Map; public class AccountCleanupRepository { // Children before user_game (user_game_status_change would also go by its cascade), the tombstones last. - private static final List TABLES = List.of("user_game_status_change", "user_game_achievement", + static final List TABLES = List.of("user_game_status_change", "user_game_achievement", "user_game_achievement_meta", "user_game_playtime_day", "wishlist_game", "backlog_snapshot", "user_steam_sync", "user_game", "game_import", "deleted_user"); diff --git a/src/main/java/core_service/modules/library/domain/UserSteamSync.java b/src/main/java/core_service/modules/library/domain/UserSteamSync.java index c329e79..dbb3cf3 100644 --- a/src/main/java/core_service/modules/library/domain/UserSteamSync.java +++ b/src/main/java/core_service/modules/library/domain/UserSteamSync.java @@ -36,6 +36,9 @@ public class UserSteamSync { @Column(name = "last_sync_status", length = 16) private Status lastSyncStatus; + @Column(name = "sync_started_at") + private Instant syncStartedAt; + @Column(name = "history_steam_id", length = 32) private String historySteamId; diff --git a/src/main/java/core_service/modules/library/repo/UserGameRepository.java b/src/main/java/core_service/modules/library/repo/UserGameRepository.java index dd3d7cd..02ead69 100644 --- a/src/main/java/core_service/modules/library/repo/UserGameRepository.java +++ b/src/main/java/core_service/modules/library/repo/UserGameRepository.java @@ -41,6 +41,8 @@ public interface UserGameRepository extends JpaRepository List findByUserIdAndRemovedAtIsNull(Long userId); + List findByUserId(Long userId); + @Query("select ug.appid from UserGame ug where ug.userId = :userId and ug.removedAt is null and ug.appid in :appids") List findActiveAppids(@Param("userId") Long userId, @Param("appids") Collection appids); diff --git a/src/main/java/core_service/modules/library/repo/UserSteamSyncRepository.java b/src/main/java/core_service/modules/library/repo/UserSteamSyncRepository.java index ee1a600..db4e4e6 100644 --- a/src/main/java/core_service/modules/library/repo/UserSteamSyncRepository.java +++ b/src/main/java/core_service/modules/library/repo/UserSteamSyncRepository.java @@ -1,7 +1,9 @@ package core_service.modules.library.repo; import core_service.modules.library.domain.UserSteamSync; +import jakarta.persistence.LockModeType; import org.springframework.data.jpa.repository.JpaRepository; +import org.springframework.data.jpa.repository.Lock; import org.springframework.data.jpa.repository.Modifying; import org.springframework.data.jpa.repository.Query; import org.springframework.data.repository.query.Param; @@ -9,40 +11,58 @@ import org.springframework.transaction.annotation.Transactional; import java.time.Instant; import java.util.List; +import java.util.Optional; public interface UserSteamSyncRepository extends JpaRepository { + /** + * The row before a claim, locked until the claim commits: a run that claims after another one finished sees + * that run's Steam ID, not an older one. + */ + @Lock(LockModeType.PESSIMISTIC_WRITE) + @Query("select s from UserSteamSync s where s.userId = :userId") + Optional findForClaim(@Param("userId") Long userId); + /** * Atomic claim-run upsert. Returns 1 if this caller successfully claimed the run * (row created, or existing row was not RUNNING / RUNNING marker stale). * Returns 0 if another sync is already in flight, or the user is deleted (table deleted_user) — caller must skip. + * + * @param startedAt the run's key for {@link #markFinished}, in microseconds as Postgres stores it */ @Modifying @Query(value = """ - INSERT INTO user_steam_sync (user_id, steam_id, last_sync_status) - SELECT :userId, :steamId, 'RUNNING' + INSERT INTO user_steam_sync (user_id, steam_id, last_sync_status, sync_started_at) + SELECT :userId, :steamId, 'RUNNING', :startedAt WHERE NOT EXISTS (SELECT 1 FROM deleted_user WHERE user_id = :userId) ON CONFLICT (user_id) DO UPDATE SET steam_id = EXCLUDED.steam_id, - last_sync_status = 'RUNNING' + last_sync_status = 'RUNNING', + sync_started_at = EXCLUDED.sync_started_at WHERE user_steam_sync.last_sync_status <> 'RUNNING' - OR user_steam_sync.last_synced_at IS NULL - OR user_steam_sync.last_synced_at < :runningStaleBefore + OR user_steam_sync.sync_started_at IS NULL + OR user_steam_sync.sync_started_at < :runningStaleBefore """, nativeQuery = true) int tryClaimRun(@Param("userId") Long userId, @Param("steamId") String steamId, + @Param("startedAt") Instant startedAt, @Param("runningStaleBefore") Instant runningStaleBefore); + /** + * @return 0 when another run took over the stale claim; that run's RUNNING stays + */ @Modifying @Query(value = """ UPDATE user_steam_sync SET last_synced_at = :now, last_sync_status = :status WHERE user_id = :userId + AND sync_started_at = :startedAt """, nativeQuery = true) - void markFinished(@Param("userId") Long userId, - @Param("status") String status, - @Param("now") Instant now); + int markFinished(@Param("userId") Long userId, + @Param("startedAt") Instant startedAt, + @Param("status") String status, + @Param("now") Instant now); /** * history_since changes only with the account: the games of that first sync are no "added" in the pile trend. diff --git a/src/main/java/core_service/modules/library/service/LibrarySyncService.java b/src/main/java/core_service/modules/library/service/LibrarySyncService.java index 55cbd58..d33842d 100644 --- a/src/main/java/core_service/modules/library/service/LibrarySyncService.java +++ b/src/main/java/core_service/modules/library/service/LibrarySyncService.java @@ -145,7 +145,7 @@ public class LibrarySyncService { private boolean isAlreadyRunning(UserSteamSync sync) { if (sync.getLastSyncStatus() != UserSteamSync.Status.RUNNING) return false; - if (sync.getLastSyncedAt() == null) return true; - return sync.getLastSyncedAt().isAfter(Instant.now().minus(Duration.ofMinutes(runningStaleMinutes))); + return sync.getSyncStartedAt() != null + && sync.getSyncStartedAt().isAfter(Instant.now().minus(Duration.ofMinutes(runningStaleMinutes))); } } diff --git a/src/main/java/core_service/modules/library/service/LibrarySyncWorker.java b/src/main/java/core_service/modules/library/service/LibrarySyncWorker.java index e17dc95..0c40751 100644 --- a/src/main/java/core_service/modules/library/service/LibrarySyncWorker.java +++ b/src/main/java/core_service/modules/library/service/LibrarySyncWorker.java @@ -4,7 +4,6 @@ import core_service.modules.library.domain.CorrectionField; import core_service.modules.library.domain.GameSource; import core_service.modules.library.domain.SteamGame; import core_service.modules.library.domain.UserGame; -import core_service.modules.library.domain.UserGameId; import core_service.modules.library.domain.UserSteamSync; import core_service.modules.library.domain.WishlistGame; import core_service.modules.library.repo.GameCorrectionRepository; @@ -21,17 +20,18 @@ import org.springframework.beans.factory.annotation.Value; import org.springframework.context.ApplicationEventPublisher; import org.springframework.scheduling.annotation.Async; import org.springframework.stereotype.Service; -import org.springframework.transaction.annotation.Transactional; +import org.springframework.transaction.support.TransactionTemplate; import java.time.Duration; import java.time.Instant; import java.time.LocalDate; import java.time.ZoneId; +import java.time.temporal.ChronoUnit; import java.util.ArrayList; import java.util.HashSet; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; -import java.util.Objects; import java.util.Optional; import java.util.Set; import java.util.stream.Collectors; @@ -39,9 +39,11 @@ import java.util.stream.Collectors; import core_service.modules.library.domain.UserSteamSync.Status; /** - * Worker bean. Runs the actual Steam fetch + DB upserts. - * Separate bean so Spring proxy applies @Async and @Transactional - * (self-invocation from LibrarySyncService would bypass both). + * Worker bean. Runs the actual Steam fetch + DB upserts. Separate bean so the Spring proxy applies @Async + * (self-invocation from LibrarySyncService would bypass it). + *

+ * The claim, the writes and the finish each commit on their own; the Steam and IGDB calls run outside any + * transaction. So other requests see RUNNING, and FAILED is stored also after a database error. */ @Service @RequiredArgsConstructor @@ -58,6 +60,7 @@ public class LibrarySyncWorker { private final TimeToBeatService timeToBeatService; private final MoodService moodService; private final ApplicationEventPublisher eventPublisher; + private final TransactionTemplate transaction; @Value("${app.library.running-stale-minutes:5}") private long runningStaleMinutes; @@ -66,7 +69,6 @@ public class LibrarySyncWorker { private ZoneId zone; @Async - @Transactional public void runSync(Long userId, String steamId) { sync(userId, steamId); } @@ -75,101 +77,129 @@ public class LibrarySyncWorker { * Same as {@link #runSync}, but on the caller's thread. The nightly job uses it to sync one account after * the other instead of starting all at once against the Steam rate limit and the connection pool. */ - @Transactional public void runSyncNow(Long userId, String steamId) { sync(userId, steamId); } + private record Claim(String historySteamId) {} + private void sync(Long userId, String steamId) { Instant runningStaleBefore = Instant.now().minus(Duration.ofMinutes(runningStaleMinutes)); - UserSteamSync before = userSteamSyncRepository.findById(userId).orElse(null); - String previousSteamId = before != null ? before.getSteamId() : null; - String historySteamId = before != null ? before.getHistorySteamId() : null; - int claimed = userSteamSyncRepository.tryClaimRun(userId, steamId, runningStaleBefore); - if (claimed == 0) { + Instant syncStartedAt = Instant.now().truncatedTo(ChronoUnit.MICROS); + Claim claim = transaction.execute(tx -> { + UserSteamSync before = userSteamSyncRepository.findForClaim(userId).orElse(null); + if (userSteamSyncRepository.tryClaimRun(userId, steamId, syncStartedAt, runningStaleBefore) == 0) { + return null; + } + String previousSteamId = before != null ? before.getSteamId() : null; + if (previousSteamId != null && !previousSteamId.equals(steamId)) { + // The old games belong to the previous Steam account, also when Steam fails now. The upserts + // restore the games this account also owns. + userGameRepository.softDeleteMissingSince(userId, syncStartedAt, Instant.now()); + wishlistGameRepository.deleteByUserId(userId); + } + return new Claim(before != null ? before.getHistorySteamId() : null); + }); + if (claim == null) { log.info("Sync skipped for user {} — another run already in flight, or the account is deleted", userId); return; } - Instant syncStartedAt = Instant.now(); Status finalStatus = Status.FAILED; try { - boolean steamIdChanged = previousSteamId != null && !previousSteamId.equals(steamId); // Growth needs a successful earlier sync of this account; else all its playtime would count as today. - boolean recordPlaytime = steamId.equals(historySteamId); - LocalDate today = LocalDate.ofInstant(syncStartedAt, zone); - if (steamIdChanged) { - // The old games belong to the previous Steam account. The upserts below restore the games this account also owns. - userGameRepository.softDeleteMissingSince(userId, syncStartedAt, Instant.now()); - wishlistGameRepository.deleteByUserId(userId); - } - List games = steamApiService.getOwnedGames(steamId); + LocalDate playedOn = steamId.equals(claim.historySteamId()) ? LocalDate.ofInstant(syncStartedAt, zone) : null; + // Distinct: the maps in saveGames are loaded once, so a second line of a game would be a new game again. + List games = steamApiService.getOwnedGames(steamId).stream() + .filter(g -> g.appid() != null) + .collect(Collectors.toMap(SteamOwnedGamesResponse.Game::appid, g -> g, (first, second) -> first, + LinkedHashMap::new)) + .values().stream().toList(); log.info("Sync user={} fetched {} games from Steam", userId, games.size()); // Before the empty check: an account without games can still have a public wishlist. - syncWishlist(userId, steamId, games); - if (games.isEmpty()) { - // Steam errors and a private "Game details" setting also end here. A soft-delete would wipe the library. - log.warn("Sync user={} got no games for SteamID {}", userId, steamId); + Optional wishlist = fetchWishlist(steamId, games); + boolean saved = transaction.execute(tx -> { + wishlist.ifPresentOrElse(found -> saveWishlist(userId, found), + () -> log.warn("Sync user={} kept the old wishlist, Steam did not answer", userId)); + if (games.isEmpty()) { + // A private "Game details" setting ends here. A soft-delete would wipe the library. + log.warn("Sync user={} got no games for SteamID {}", userId, steamId); + return false; + } + saveGames(userId, steamId, games, syncStartedAt, playedOn); + return true; + }); + if (!saved) { return; } - - // setSourceName keeps them anyway; skipping saves an update and a flush per corrected game and sync. - Set correctedNames = gameCorrectionRepository.findCorrectedAppids(CorrectionField.NAME, - games.stream().map(SteamOwnedGamesResponse.Game::appid).filter(Objects::nonNull).map(Integer::longValue).toList()); - List played = new ArrayList<>(); - for (SteamOwnedGamesResponse.Game g : games) { - if (g.appid() == null) continue; - upsertGame(g, syncStartedAt, correctedNames); - if (upsertUserGame(userId, g, syncStartedAt, recordPlaytime ? today : null)) { - played.add(g.appid().longValue()); - } - } - if (!played.isEmpty()) { - userGameRepository.playFromPileIfUntouched(userId, played); - } - - int removed = userGameRepository.softDeleteMissingSince(userId, syncStartedAt, Instant.now()); - if (removed > 0) { - log.info("Sync user={} soft-deleted {} games no longer owned", userId, removed); - } - userSteamSyncRepository.markHistoryBaseline(userId, steamId); // Also the games added by hand: here a failed refresh from the add is asked again. - List igdbAppids = new ArrayList<>(games.stream().filter(g -> g.appid() != null).map(g -> g.appid().longValue()).toList()); + List igdbAppids = new ArrayList<>(games.stream().map(g -> g.appid().longValue()).toList()); igdbAppids.addAll(userGameRepository.findActiveManualAppids(userId)); igdbAppids.removeAll(new HashSet<>(userGameRepository.findHiddenAppids(userId))); timeToBeatService.refresh(igdbAppids); moodService.refresh(igdbAppids); - eventPublisher.publishEvent(new LibrarySyncedEvent(userId, steamId)); finalStatus = Status.OK; } catch (Exception e) { log.error("Sync failed for user {}", userId, e); } finally { - userSteamSyncRepository.markFinished(userId, finalStatus.name(), Instant.now()); + Status status = finalStatus; + transaction.executeWithoutResult(tx -> { + if (userSteamSyncRepository.markFinished(userId, syncStartedAt, status.name(), Instant.now()) == 0) { + log.warn("Sync user={} ran past running-stale-minutes; the run that took over sets the status", userId); + } else if (status == Status.OK) { + // The listeners run after this commit. + eventPublisher.publishEvent(new LibrarySyncedEvent(userId, steamId)); + } + }); } } - private void syncWishlist(Long userId, String steamId, List ownedGames) { - Optional> wishlist = steamApiService.getWishlist(steamId); - if (wishlist.isEmpty()) { - log.warn("Sync user={} kept the old wishlist, Steam did not answer", userId); - return; + private void saveGames(Long userId, String steamId, List games, Instant now, + LocalDate playedOn) { + List appids = games.stream().map(g -> g.appid().longValue()).toList(); + // setSourceName keeps them anyway; skipping saves an update and a flush per corrected game and sync. + Set correctedNames = gameCorrectionRepository.findCorrectedAppids(CorrectionField.NAME, appids); + Map knownGames = steamGameRepository.findAllById(appids).stream() + .collect(Collectors.toMap(SteamGame::getAppid, game -> game)); + Map userGames = userGameRepository.findByUserId(userId).stream() + .collect(Collectors.toMap(UserGame::getAppid, game -> game)); + List played = new ArrayList<>(); + for (SteamOwnedGamesResponse.Game g : games) { + Long appid = g.appid().longValue(); + upsertGame(g, knownGames.get(appid), now, correctedNames); + if (upsertUserGame(userId, g, userGames.get(appid), now, playedOn)) { + played.add(appid); + } } - Set owned = ownedGames.stream() - .filter(g -> g.appid() != null) - .map(g -> g.appid().longValue()) - .collect(Collectors.toSet()); - List wanted = wishlist.get().stream() - .filter(item -> !owned.contains(item.appid())) - .toList(); - Map store = wanted.isEmpty() - ? Map.of() - : steamApiService.getStoreItems(wanted.stream().map(SteamApiService.WishlistItem::appid).toList()); + if (!played.isEmpty()) { + userGameRepository.playFromPileIfUntouched(userId, played); + } + int removed = userGameRepository.softDeleteMissingSince(userId, now, Instant.now()); + if (removed > 0) { + log.info("Sync user={} soft-deleted {} games no longer owned", userId, removed); + } + userSteamSyncRepository.markHistoryBaseline(userId, steamId); + } + + private record Wishlist(List wanted, Map store) {} + + private Optional fetchWishlist(String steamId, List ownedGames) { + Set owned = ownedGames.stream().map(g -> g.appid().longValue()).collect(Collectors.toSet()); + return steamApiService.getWishlist(steamId).map(items -> { + List wanted = items.stream().filter(item -> !owned.contains(item.appid())).toList(); + return new Wishlist(wanted, wanted.isEmpty() + ? Map.of() + : steamApiService.getStoreItems(wanted.stream().map(SteamApiService.WishlistItem::appid).toList())); + }); + } + + private void saveWishlist(Long userId, Wishlist wishlist) { // A failed store batch leaves apps out of `store`; they keep the name and price of the last sync. Map previous = wishlistGameRepository.findForUser(userId).stream() .collect(Collectors.toMap(WishlistGame::getAppid, w -> w)); wishlistGameRepository.deleteByUserId(userId); - wishlistGameRepository.saveAll(wanted.stream().map(item -> { - SteamApiService.StoreItem info = store.get(item.appid()); + wishlistGameRepository.saveAll(wishlist.wanted().stream().map(item -> { + SteamApiService.StoreItem info = wishlist.store().get(item.appid()); if (info != null) { return new WishlistGame(userId, item.appid(), info.name(), item.addedAt(), info.price(), info.originalPrice(), info.discountPct()); @@ -182,9 +212,8 @@ public class LibrarySyncWorker { }).toList()); } - private void upsertGame(SteamOwnedGamesResponse.Game g, Instant now, Set correctedNames) { + private void upsertGame(SteamOwnedGamesResponse.Game g, SteamGame existing, Instant now, Set correctedNames) { Long appid = g.appid().longValue(); - SteamGame existing = steamGameRepository.findById(appid).orElse(null); // An unvetted app can come without a name, and the name column is NOT NULL. boolean hasName = g.name() != null && !g.name().isBlank(); if (existing == null) { @@ -202,9 +231,9 @@ public class LibrarySyncWorker { * @param playedOn the day that gets the playtime growth, null to record none * @return true when the game was played since the last sync */ - private boolean upsertUserGame(Long userId, SteamOwnedGamesResponse.Game g, Instant now, LocalDate playedOn) { + private boolean upsertUserGame(Long userId, SteamOwnedGamesResponse.Game g, UserGame existing, Instant now, + LocalDate playedOn) { Long appid = g.appid().longValue(); - UserGame existing = userGameRepository.findById(new UserGameId(userId, appid)).orElse(null); int playtime = g.playtimeForever() != null ? g.playtimeForever() : 0; // Hours typed by hand are no Steam playtime, so the first Steam value of a game added by hand is a baseline. boolean takesOverManualGame = existing != null && existing.getSource() == GameSource.MANUAL; diff --git a/src/main/java/core_service/modules/steam_adapter/api/dto/response/SteamProfileAndGamesResponse.java b/src/main/java/core_service/modules/steam_adapter/api/dto/response/SteamProfileAndGamesResponse.java deleted file mode 100644 index 314d9ba..0000000 --- a/src/main/java/core_service/modules/steam_adapter/api/dto/response/SteamProfileAndGamesResponse.java +++ /dev/null @@ -1,9 +0,0 @@ -package core_service.modules.steam_adapter.api.dto.response; - -import java.util.List; - -public record SteamProfileAndGamesResponse( - SteamPlayerSummaryResponse.Player player, - List games -) { -} diff --git a/src/main/java/core_service/modules/steam_adapter/service/SteamApiService.java b/src/main/java/core_service/modules/steam_adapter/service/SteamApiService.java index b7de47b..6fc66d3 100644 --- a/src/main/java/core_service/modules/steam_adapter/service/SteamApiService.java +++ b/src/main/java/core_service/modules/steam_adapter/service/SteamApiService.java @@ -4,7 +4,6 @@ import core_service.modules.steam_adapter.api.dto.response.SteamGlobalAchievemen import core_service.modules.steam_adapter.api.dto.response.SteamOwnedGamesResponse; import core_service.modules.steam_adapter.api.dto.response.SteamPlayerAchievementsResponse; import core_service.modules.steam_adapter.api.dto.response.SteamPlayerSummaryResponse; -import core_service.modules.steam_adapter.api.dto.response.SteamProfileAndGamesResponse; import core_service.modules.steam_adapter.api.dto.response.SteamSchemaForGameResponse; import core_service.modules.steam_adapter.api.dto.response.SteamStoreItemsResponse; import core_service.modules.steam_adapter.api.dto.response.SteamWishlistResponse; @@ -54,119 +53,98 @@ public class SteamApiService { @Cacheable(value = "steamGameCount", key = "#steamId") @RateLimiter(name = "steamApi") - @Retry(name = "steamApi") + @Retry(name = "steamApiOnce", fallbackMethod = "noGamesCount") public int getOwnedGamesCount(String steamId) { if (apiKey == null || apiKey.isEmpty() || apiKey.equals("${STEAM_API_KEY}")) { log.error("Steam API Key is not configured correctly!"); return 0; } - try { - log.info("Fetching owned games count for SteamID: {} using API Key starting with: {}", - steamId, apiKey.substring(0, Math.min(apiKey.length(), 4))); + SteamOwnedGamesResponse response = steamRestClient.get() + .uri(uriBuilder -> uriBuilder + .path("/IPlayerService/GetOwnedGames/v1/") + .queryParam("key", apiKey) + .queryParam("steamid", steamId) + .queryParam("include_appinfo", "true") + .queryParam("include_played_free_games", "true") + .queryParam("skip_unvetted_apps", 0) + .queryParam("format", "json") + .build()) + .retrieve() + .body(SteamOwnedGamesResponse.class); - SteamOwnedGamesResponse response = steamRestClient.get() - .uri(uriBuilder -> uriBuilder - .path("/IPlayerService/GetOwnedGames/v1/") - .queryParam("key", apiKey) - .queryParam("steamid", steamId) - .queryParam("include_appinfo", "true") - .queryParam("include_played_free_games", "true") - .queryParam("skip_unvetted_apps", 0) - .queryParam("format", "json") - .build()) - .retrieve() - .body(SteamOwnedGamesResponse.class); - - log.info("Steam API Response object for {}: {}", steamId, response); - - if (response != null && response.response() != null) { - Integer count = response.response().gameCount(); - if (count == null) { - if (response.response().games() != null) { - int size = response.response().games().size(); - log.info("game_count was null, but games list has {} entries for SteamID: {}", size, steamId); - return size; - } - log.warn("Steam API returned an empty response object for SteamID: {}. Check if the ID is correct and 'Game details' are Public.", steamId); - return 0; + if (response != null && response.response() != null) { + Integer count = response.response().gameCount(); + if (count == null) { + if (response.response().games() != null) { + int size = response.response().games().size(); + log.info("game_count was null, but games list has {} entries for SteamID: {}", size, steamId); + return size; } - log.info("Successfully fetched {} games for SteamID: {}", count, steamId); - return count; + log.warn("Steam API returned an empty response object for SteamID: {}. Check if the ID is correct and 'Game details' are Public.", steamId); + return 0; } - - } catch (Exception e) { - log.error("Error communicating with Steam API for game count of SteamID: {}", steamId, e); + log.info("Successfully fetched {} games for SteamID: {}", count, steamId); + return count; } - return 0; } - public Optional getFullPlayerProfile(String steamId) { - var playerOpt = getPlayerSummary(steamId); - if (playerOpt.isEmpty()) { - return Optional.empty(); - } - - var games = getOwnedGames(steamId); - return Optional.of(new SteamProfileAndGamesResponse(playerOpt.get(), games)); + // The fallbacks run after the last try; the callers keep their "Steam failed" rule. + private int noGamesCount(String steamId, Exception e) { + log.error("Error communicating with Steam API for game count of SteamID: {}", steamId, e); + return 0; } // Asked on every page load for the avatar in the header. @Cacheable(value = "steamPlayerSummary", key = "#steamId", unless = "#result == null") @RateLimiter(name = "steamApi") - @Retry(name = "steamApi") + @Retry(name = "steamApiOnce", fallbackMethod = "noPlayerSummary") public Optional getPlayerSummary(String steamId) { - try { - log.info("Fetching Steam profile for SteamID: {}", steamId); + log.info("Fetching Steam profile for SteamID: {}", steamId); - SteamPlayerSummaryResponse response = steamRestClient.get() - .uri(uriBuilder -> uriBuilder - .path("/ISteamUser/GetPlayerSummaries/v0002/") - .queryParam("key", apiKey) - .queryParam("steamids", steamId) - .build()) - .retrieve() - .body(SteamPlayerSummaryResponse.class); + SteamPlayerSummaryResponse response = steamRestClient.get() + .uri(uriBuilder -> uriBuilder + .path("/ISteamUser/GetPlayerSummaries/v0002/") + .queryParam("key", apiKey) + .queryParam("steamids", steamId) + .build()) + .retrieve() + .body(SteamPlayerSummaryResponse.class); - if (response != null && response.response() != null && !response.response().players().isEmpty()) { - return Optional.of(response.response().players().getFirst()); - } - - } catch (Exception e) { - log.error("Error communicating with Steam API for SteamID: {}", steamId, e); + if (response != null && response.response() != null && !response.response().players().isEmpty()) { + return Optional.of(response.response().players().getFirst()); } + return Optional.empty(); + } + private Optional noPlayerSummary(String steamId, Exception e) { + log.error("Error communicating with Steam API for SteamID: {}", steamId, e); return Optional.empty(); } @RateLimiter(name = "steamApi") + // No fallback: after the last try the sync stores FAILED. @Retry(name = "steamApi") public List getOwnedGames(String steamId) { - try { - log.info("Fetching owned games for SteamID: {}", steamId); + log.info("Fetching owned games for SteamID: {}", steamId); - SteamOwnedGamesResponse response = steamRestClient.get() - .uri(uriBuilder -> uriBuilder - .path("/IPlayerService/GetOwnedGames/v1/") - .queryParam("key", apiKey) - .queryParam("steamid", steamId) - .queryParam("include_appinfo", 1) - .queryParam("include_played_free_games", 1) - // Else Steam hides some owned apps, e.g. a game from a developer license. - .queryParam("skip_unvetted_apps", 0) - .build()) - .retrieve() - .body(SteamOwnedGamesResponse.class); + SteamOwnedGamesResponse response = steamRestClient.get() + .uri(uriBuilder -> uriBuilder + .path("/IPlayerService/GetOwnedGames/v1/") + .queryParam("key", apiKey) + .queryParam("steamid", steamId) + .queryParam("include_appinfo", 1) + .queryParam("include_played_free_games", 1) + // Else Steam hides some owned apps, e.g. a game from a developer license. + .queryParam("skip_unvetted_apps", 0) + .build()) + .retrieve() + .body(SteamOwnedGamesResponse.class); - if (response != null && response.response() != null && response.response().games() != null) { - return response.response().games(); - } - - } catch (Exception e) { - log.error("Error communicating with Steam API for games of SteamID: {}", steamId, e); + if (response != null && response.response() != null && response.response().games() != null) { + return response.response().games(); } - return Collections.emptyList(); } @@ -183,29 +161,29 @@ public class SteamApiService { * @return empty on a Steam error, so the caller can keep the list it has */ @RateLimiter(name = "steamApi") - @Retry(name = "steamApi") + @Retry(name = "steamApiOnce", fallbackMethod = "noWishlist") public Optional> getWishlist(String steamId) { - try { - SteamWishlistResponse response = steamRestClient.get() - .uri(uriBuilder -> uriBuilder - .path("/IWishlistService/GetWishlist/v1/") - .queryParam("key", apiKey) - .queryParam("steamid", steamId) - .build()) - .retrieve() - .body(SteamWishlistResponse.class); - if (response == null || response.response() == null || response.response().items() == null) { - return Optional.of(List.of()); - } - return Optional.of(response.response().items().stream() - .filter(item -> item.appid() != null) - .map(item -> new WishlistItem(item.appid(), - item.dateAdded() != null ? Instant.ofEpochSecond(item.dateAdded()) : null)) - .toList()); - } catch (Exception e) { - log.error("Error communicating with Steam API for the wishlist of SteamID: {}", steamId, e); - return Optional.empty(); + SteamWishlistResponse response = steamRestClient.get() + .uri(uriBuilder -> uriBuilder + .path("/IWishlistService/GetWishlist/v1/") + .queryParam("key", apiKey) + .queryParam("steamid", steamId) + .build()) + .retrieve() + .body(SteamWishlistResponse.class); + if (response == null || response.response() == null || response.response().items() == null) { + return Optional.of(List.of()); } + return Optional.of(response.response().items().stream() + .filter(item -> item.appid() != null) + .map(item -> new WishlistItem(item.appid(), + item.dateAdded() != null ? Instant.ofEpochSecond(item.dateAdded()) : null)) + .toList()); + } + + private Optional> noWishlist(String steamId, Exception e) { + log.error("Error communicating with Steam API for the wishlist of SteamID: {}", steamId, e); + return Optional.empty(); } /** diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index f2fcddd..b204674 100644 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -87,6 +87,13 @@ resilience4j: retry-exceptions: - org.springframework.web.client.RestClientException - org.springframework.web.client.ResourceAccessException + # A 4xx (a refused key, a private list) is the same on the next try. + ignore-exceptions: + - org.springframework.web.client.HttpClientErrorException + # Calls a page waits for, and the optional wishlist: one try, else a Steam outage holds every page (or the + # sync) for three read timeouts. + steamApiOnce: + max-attempts: 1 ratelimiter: instances: steamApi: diff --git a/src/main/resources/db/migration/V23__Sync_Started_At.sql b/src/main/resources/db/migration/V23__Sync_Started_At.sql new file mode 100644 index 0000000..646c790 --- /dev/null +++ b/src/main/resources/db/migration/V23__Sync_Started_At.sql @@ -0,0 +1,3 @@ +-- The claim of a sync is committed on its own now; a RUNNING row is stale by the start of its run, not by the end of +-- the last one. +ALTER TABLE user_steam_sync ADD COLUMN sync_started_at TIMESTAMPTZ; diff --git a/src/test/java/core_service/modules/account/repo/AccountCleanupRepositoryTest.java b/src/test/java/core_service/modules/account/repo/AccountCleanupRepositoryTest.java index bbe80b2..61d917a 100644 --- a/src/test/java/core_service/modules/account/repo/AccountCleanupRepositoryTest.java +++ b/src/test/java/core_service/modules/account/repo/AccountCleanupRepositoryTest.java @@ -114,4 +114,13 @@ class AccountCleanupRepositoryTest extends AbstractPersistenceTest { assertThat(repository.findOrphans(List.of(2L))).containsExactly(1L, 7L); assertThat(repository.findOrphans(List.of(1L, 2L, 7L))).isEmpty(); } + + // A new table with user rows must be in the purge list, or "Delete my account" leaves its rows behind. + @Test + void tables_CoverEveryTableWithAUserId() { + List withUserId = jdbc.queryForList("select table_name from information_schema.columns " + + "where table_schema = 'public' and column_name = 'user_id' order by table_name", String.class); + + assertThat(AccountCleanupRepository.TABLES).containsExactlyInAnyOrderElementsOf(withUserId); + } } diff --git a/src/test/java/core_service/modules/library/repo/UserSteamSyncRepositoryTest.java b/src/test/java/core_service/modules/library/repo/UserSteamSyncRepositoryTest.java index 5c38227..ca4c5e0 100644 --- a/src/test/java/core_service/modules/library/repo/UserSteamSyncRepositoryTest.java +++ b/src/test/java/core_service/modules/library/repo/UserSteamSyncRepositoryTest.java @@ -21,7 +21,7 @@ class UserSteamSyncRepositoryTest extends AbstractPersistenceTest { private JdbcTemplate jdbc; private int claim(long userId) { - return repository.tryClaimRun(userId, STEAM_ID, Instant.now().minusSeconds(600)); + return repository.tryClaimRun(userId, STEAM_ID, Instant.now(), Instant.now().minusSeconds(600)); } @Test @@ -44,4 +44,20 @@ class UserSteamSyncRepositoryTest extends AbstractPersistenceTest { assertThat(claim(3L)).isZero(); } + + @Test + void tryClaimRun_RunningRowStartedLongAgo_IsClaimedEvenWhenTheLastSyncEndedRecently() { + jdbc.update("insert into user_steam_sync (user_id, steam_id, last_synced_at, last_sync_status, sync_started_at) " + + "values (4, ?, now(), 'RUNNING', now() - interval '20 minutes')", STEAM_ID); + + assertThat(claim(4L)).isEqualTo(1); + } + + @Test + void tryClaimRun_RunningRowStartedJustNow_IsRefusedEvenWhenTheLastSyncEndedLongAgo() { + jdbc.update("insert into user_steam_sync (user_id, steam_id, last_synced_at, last_sync_status, sync_started_at) " + + "values (5, ?, now() - interval '1 day', 'RUNNING', now())", STEAM_ID); + + assertThat(claim(5L)).isZero(); + } } diff --git a/src/test/java/core_service/modules/library/service/LibrarySyncServiceTest.java b/src/test/java/core_service/modules/library/service/LibrarySyncServiceTest.java index 34bae71..9b6f4f9 100644 --- a/src/test/java/core_service/modules/library/service/LibrarySyncServiceTest.java +++ b/src/test/java/core_service/modules/library/service/LibrarySyncServiceTest.java @@ -63,6 +63,7 @@ class LibrarySyncServiceTest extends AbstractUnitTest { UserSteamSync sync = new UserSteamSync(USER_ID, steamId); sync.setLastSyncStatus(status); sync.setLastSyncedAt(Instant.now().minus(age)); + sync.setSyncStartedAt(Instant.now().minus(age)); when(userSteamSyncRepository.findById(USER_ID)).thenReturn(Optional.of(sync)); } diff --git a/src/test/java/core_service/modules/library/service/LibrarySyncWorkerConcurrencyTest.java b/src/test/java/core_service/modules/library/service/LibrarySyncWorkerConcurrencyTest.java new file mode 100644 index 0000000..099af0b --- /dev/null +++ b/src/test/java/core_service/modules/library/service/LibrarySyncWorkerConcurrencyTest.java @@ -0,0 +1,140 @@ +package core_service.modules.library.service; + +import core_service.modules.library.domain.UserSteamSync; +import core_service.modules.library.repo.UserSteamSyncRepository; +import core_service.modules.steam_adapter.api.dto.response.SteamOwnedGamesResponse; +import core_service.modules.steam_adapter.service.SteamApiService; +import core_service.shared.test.AbstractIntegrationTest; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.test.context.bean.override.mockito.MockitoBean; + +import java.util.List; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * The claim and the finish of a sync are committed on their own, so other requests see RUNNING and FAILED. + */ +class LibrarySyncWorkerConcurrencyTest extends AbstractIntegrationTest { + + private static final long USER_ID = 4711L; + private static final long APPID = 991220L; + private static final String STEAM_ID = "76561198000004711"; + + @Autowired + private LibrarySyncWorker worker; + + @Autowired + private UserSteamSyncRepository userSteamSyncRepository; + + @Autowired + private JdbcTemplate jdbcTemplate; + + @MockitoBean + private SteamApiService steamApiService; + + @BeforeEach + @AfterEach + void clean() { + jdbcTemplate.update("DELETE FROM user_game_playtime_day WHERE user_id = ?", USER_ID); + jdbcTemplate.update("DELETE FROM user_game WHERE user_id = ?", USER_ID); + jdbcTemplate.update("DELETE FROM wishlist_game WHERE user_id = ?", USER_ID); + jdbcTemplate.update("DELETE FROM user_steam_sync WHERE user_id = ?", USER_ID); + jdbcTemplate.update("DELETE FROM steam_game WHERE appid = ?", APPID); + } + + @Test + void runSyncNow_SecondSyncWhileTheFirstWaitsForSteam_IsSkipped() throws Exception { + // Last synced yesterday: the running check must not take that as a stale run. + jdbcTemplate.update("INSERT INTO user_steam_sync (user_id, steam_id, last_synced_at, last_sync_status) " + + "VALUES (?, ?, NOW() - INTERVAL '1 day', 'OK')", USER_ID, STEAM_ID); + CountDownLatch steamAsked = new CountDownLatch(1); + CountDownLatch secondDone = new CountDownLatch(1); + when(steamApiService.getWishlist(STEAM_ID)).thenReturn(Optional.empty()); + when(steamApiService.getOwnedGames(STEAM_ID)).thenAnswer(invocation -> { + steamAsked.countDown(); + secondDone.await(5, TimeUnit.SECONDS); + return List.of(game("icon")); + }); + + CompletableFuture first = CompletableFuture.runAsync(() -> worker.runSyncNow(USER_ID, STEAM_ID)); + assertThat(steamAsked.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(userSteamSyncRepository.findById(USER_ID).orElseThrow().getLastSyncStatus()) + .isEqualTo(UserSteamSync.Status.RUNNING); + CompletableFuture second = CompletableFuture.runAsync(() -> { + worker.runSyncNow(USER_ID, STEAM_ID); + secondDone.countDown(); + }); + CompletableFuture.allOf(first, second).get(30, TimeUnit.SECONDS); + + verify(steamApiService, times(1)).getOwnedGames(STEAM_ID); + assertThat(userSteamSyncRepository.findById(USER_ID).orElseThrow().getLastSyncStatus()) + .isEqualTo(UserSteamSync.Status.OK); + } + + @Test + void runSyncNow_DatabaseErrorWhileSaving_StoresFailed() { + when(steamApiService.getWishlist(STEAM_ID)).thenReturn(Optional.empty()); + // img_icon_url is VARCHAR(512). + when(steamApiService.getOwnedGames(STEAM_ID)).thenReturn(List.of(game("x".repeat(600)))); + + worker.runSyncNow(USER_ID, STEAM_ID); + + UserSteamSync row = userSteamSyncRepository.findById(USER_ID).orElseThrow(); + assertThat(row.getLastSyncStatus()).isEqualTo(UserSteamSync.Status.FAILED); + assertThat(jdbcTemplate.queryForObject("SELECT count(*) FROM user_game WHERE user_id = ?", Integer.class, USER_ID)) + .isZero(); + } + + @Test + void runSyncNow_StaleRunFinishesWhileTheRunThatTookOverStillRuns_RowStaysRunning() throws Exception { + CountDownLatch firstAsked = new CountDownLatch(1); + CountDownLatch secondAsked = new CountDownLatch(1); + CountDownLatch releaseFirst = new CountDownLatch(1); + CountDownLatch releaseSecond = new CountDownLatch(1); + AtomicInteger calls = new AtomicInteger(); + when(steamApiService.getWishlist(STEAM_ID)).thenReturn(Optional.empty()); + when(steamApiService.getOwnedGames(STEAM_ID)).thenAnswer(invocation -> { + if (calls.incrementAndGet() == 1) { + firstAsked.countDown(); + releaseFirst.await(10, TimeUnit.SECONDS); + } else { + secondAsked.countDown(); + releaseSecond.await(10, TimeUnit.SECONDS); + } + return List.of(game("icon")); + }); + + CompletableFuture first = CompletableFuture.runAsync(() -> worker.runSyncNow(USER_ID, STEAM_ID)); + assertThat(firstAsked.await(10, TimeUnit.SECONDS)).isTrue(); + // The first run hangs past running-stale-minutes, so the next sync takes over. + jdbcTemplate.update("UPDATE user_steam_sync SET sync_started_at = NOW() - INTERVAL '10 minutes' WHERE user_id = ?", + USER_ID); + CompletableFuture second = CompletableFuture.runAsync(() -> worker.runSyncNow(USER_ID, STEAM_ID)); + assertThat(secondAsked.await(10, TimeUnit.SECONDS)).isTrue(); + releaseFirst.countDown(); + first.get(30, TimeUnit.SECONDS); + UserSteamSync.Status whileSecondRuns = userSteamSyncRepository.findById(USER_ID).orElseThrow().getLastSyncStatus(); + releaseSecond.countDown(); + second.get(30, TimeUnit.SECONDS); + + // OK here lets a third sync claim and run next to the second one. + assertThat(whileSecondRuns).isEqualTo(UserSteamSync.Status.RUNNING); + } + + private static SteamOwnedGamesResponse.Game game(String iconUrl) { + return new SteamOwnedGamesResponse.Game((int) APPID, "Half-Life 2", 60, iconUrl, true, 0); + } +} diff --git a/src/test/java/core_service/modules/library/service/LibrarySyncWorkerTest.java b/src/test/java/core_service/modules/library/service/LibrarySyncWorkerTest.java index 158f14c..2258aba 100644 --- a/src/test/java/core_service/modules/library/service/LibrarySyncWorkerTest.java +++ b/src/test/java/core_service/modules/library/service/LibrarySyncWorkerTest.java @@ -5,7 +5,6 @@ import core_service.modules.library.domain.GameSource; import core_service.modules.library.domain.GameStatus; import core_service.modules.library.domain.SteamGame; import core_service.modules.library.domain.UserGame; -import core_service.modules.library.domain.UserGameId; import core_service.modules.library.domain.UserSteamSync; import core_service.modules.library.domain.WishlistGame; import core_service.modules.library.repo.GameCorrectionRepository; @@ -25,8 +24,11 @@ import org.mockito.ArgumentCaptor; import org.mockito.InOrder; import org.mockito.InjectMocks; import org.mockito.Mock; +import org.mockito.Spy; import org.springframework.context.ApplicationEventPublisher; import org.springframework.test.util.ReflectionTestUtils; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.support.TransactionTemplate; import java.time.Instant; import java.time.LocalDate; @@ -47,6 +49,8 @@ import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.argThat; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.inOrder; +import static org.mockito.Mockito.lenient; +import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.verifyNoInteractions; @@ -71,9 +75,9 @@ class LibrarySyncWorkerTest extends AbstractUnitTest { */ private void steamReturns(SteamOwnedGamesResponse.Game game, String previousSteamId) { if (previousSteamId != null) { - when(userSteamSyncRepository.findById(USER_ID)).thenReturn(Optional.of(syncRow(previousSteamId, previousSteamId))); + when(userSteamSyncRepository.findForClaim(USER_ID)).thenReturn(Optional.of(syncRow(previousSteamId, previousSteamId))); } - when(userSteamSyncRepository.tryClaimRun(eq(USER_ID), eq(STEAM_ID), any())).thenReturn(1); + when(userSteamSyncRepository.tryClaimRun(eq(USER_ID), eq(STEAM_ID), any(), any())).thenReturn(1); when(steamApiService.getOwnedGames(STEAM_ID)).thenReturn(List.of(game)); } @@ -85,7 +89,7 @@ class LibrarySyncWorkerTest extends AbstractUnitTest { private UserGame owned(int playtime) { UserGame existing = new UserGame(USER_ID, 220L, playtime, Instant.EPOCH, Instant.EPOCH); - when(userGameRepository.findById(new UserGameId(USER_ID, 220L))).thenReturn(Optional.of(existing)); + when(userGameRepository.findByUserId(USER_ID)).thenReturn(List.of(existing)); return existing; } @@ -119,69 +123,83 @@ class LibrarySyncWorkerTest extends AbstractUnitTest { @Mock private ApplicationEventPublisher eventPublisher; + @Spy + private TransactionTemplate transaction = new TransactionTemplate(mock(PlatformTransactionManager.class)); + @InjectMocks private LibrarySyncWorker worker; @BeforeEach void setUp() { ReflectionTestUtils.setField(worker, "zone", ZoneId.of("Europe/Berlin")); + lenient().when(userSteamSyncRepository.markFinished(eq(USER_ID), any(), anyString(), any())).thenReturn(1); + } + + @Test + void runSync_AnotherRunTookOverTheStaleClaim_PublishesNothing() { + steamReturns(game(100, 0), null); + when(userSteamSyncRepository.markFinished(eq(USER_ID), any(), eq("OK"), any())).thenReturn(0); + + worker.runSync(USER_ID, STEAM_ID); + + verifyNoInteractions(eventPublisher); } @Test void runSync_EmptySteamResult_MarksFailedAndKeepsLibrary() { - when(userSteamSyncRepository.tryClaimRun(eq(USER_ID), eq(STEAM_ID), any())).thenReturn(1); + when(userSteamSyncRepository.tryClaimRun(eq(USER_ID), eq(STEAM_ID), any(), any())).thenReturn(1); when(steamApiService.getOwnedGames(STEAM_ID)).thenReturn(List.of()); worker.runSync(USER_ID, STEAM_ID); verify(userGameRepository, never()).softDeleteMissingSince(anyLong(), any(), any()); - verify(userSteamSyncRepository).markFinished(eq(USER_ID), eq("FAILED"), any()); + verify(userSteamSyncRepository).markFinished(eq(USER_ID), any(), eq("FAILED"), any()); verifyNoInteractions(eventPublisher); } @Test void runSync_EmptyResultForNewSteamId_RemovesOldLibrary() { - when(userSteamSyncRepository.findById(USER_ID)) + when(userSteamSyncRepository.findForClaim(USER_ID)) .thenReturn(Optional.of(new UserSteamSync(USER_ID, "76561198005679790"))); - when(userSteamSyncRepository.tryClaimRun(eq(USER_ID), eq(STEAM_ID), any())).thenReturn(1); + when(userSteamSyncRepository.tryClaimRun(eq(USER_ID), eq(STEAM_ID), any(), any())).thenReturn(1); when(steamApiService.getOwnedGames(STEAM_ID)).thenReturn(List.of()); worker.runSync(USER_ID, STEAM_ID); verify(userGameRepository).softDeleteMissingSince(eq(USER_ID), any(), any()); - verify(userSteamSyncRepository).markFinished(eq(USER_ID), eq("FAILED"), any()); + verify(userSteamSyncRepository).markFinished(eq(USER_ID), any(), eq("FAILED"), any()); } @Test void runSync_SteamErrorForNewSteamId_RemovesOldLibrary() { - when(userSteamSyncRepository.findById(USER_ID)) + when(userSteamSyncRepository.findForClaim(USER_ID)) .thenReturn(Optional.of(new UserSteamSync(USER_ID, "76561198005679790"))); - when(userSteamSyncRepository.tryClaimRun(eq(USER_ID), eq(STEAM_ID), any())).thenReturn(1); + when(userSteamSyncRepository.tryClaimRun(eq(USER_ID), eq(STEAM_ID), any(), any())).thenReturn(1); when(steamApiService.getOwnedGames(STEAM_ID)).thenThrow(new IllegalStateException("Steam down")); worker.runSync(USER_ID, STEAM_ID); verify(userGameRepository).softDeleteMissingSince(eq(USER_ID), any(), any()); - verify(userSteamSyncRepository).markFinished(eq(USER_ID), eq("FAILED"), any()); + verify(userSteamSyncRepository).markFinished(eq(USER_ID), any(), eq("FAILED"), any()); } @Test void runSync_WithGames_MarksOk() { - when(userSteamSyncRepository.tryClaimRun(eq(USER_ID), eq(STEAM_ID), any())).thenReturn(1); + when(userSteamSyncRepository.tryClaimRun(eq(USER_ID), eq(STEAM_ID), any(), any())).thenReturn(1); when(steamApiService.getOwnedGames(STEAM_ID)).thenReturn(List.of( game(0, 0))); worker.runSync(USER_ID, STEAM_ID); verify(userGameRepository).softDeleteMissingSince(eq(USER_ID), any(), any()); - verify(userSteamSyncRepository).markFinished(eq(USER_ID), eq("OK"), any()); + verify(userSteamSyncRepository).markFinished(eq(USER_ID), any(), eq("OK"), any()); verify(eventPublisher).publishEvent(new LibrarySyncedEvent(USER_ID, STEAM_ID)); } @ParameterizedTest @CsvSource({"0, BACKLOG", "1, PLAYING"}) void runSync_NewGame_GetsStatusFromPlaytime(int playtime, GameStatus expected) { - when(userSteamSyncRepository.tryClaimRun(eq(USER_ID), eq(STEAM_ID), any())).thenReturn(1); + when(userSteamSyncRepository.tryClaimRun(eq(USER_ID), eq(STEAM_ID), any(), any())).thenReturn(1); when(steamApiService.getOwnedGames(STEAM_ID)).thenReturn(List.of( game(playtime, 0))); @@ -197,7 +215,7 @@ class LibrarySyncWorkerTest extends AbstractUnitTest { UserGame manual = UserGame.manual(USER_ID, 220L, "PS5", GameStatus.FINISHED, 60, Instant.EPOCH); manual.setRating(5); manual.setNotes("Played on the couch"); - when(userGameRepository.findById(new UserGameId(USER_ID, 220L))).thenReturn(Optional.of(manual)); + when(userGameRepository.findByUserId(USER_ID)).thenReturn(List.of(manual)); steamReturns(game(90, 30), STEAM_ID); worker.runSync(USER_ID, STEAM_ID); @@ -213,7 +231,7 @@ class LibrarySyncWorkerTest extends AbstractUnitTest { @Test void runSync_GameAddedByHandIsNowOnSteam_TheHandTypedHoursAreNoPlaytimeGrowth() { UserGame manual = UserGame.manual(USER_ID, 220L, "PS5", GameStatus.PLAYING, 60, Instant.EPOCH); - when(userGameRepository.findById(new UserGameId(USER_ID, 220L))).thenReturn(Optional.of(manual)); + when(userGameRepository.findByUserId(USER_ID)).thenReturn(List.of(manual)); steamReturns(game(90, 30), STEAM_ID); worker.runSync(USER_ID, STEAM_ID); @@ -225,8 +243,8 @@ class LibrarySyncWorkerTest extends AbstractUnitTest { void runSync_ExistingGame_KeepsStatus() { UserGame existing = new UserGame(USER_ID, 220L, 0, Instant.EPOCH, Instant.EPOCH); existing.setStatus(GameStatus.DROPPED); - when(userGameRepository.findById(new UserGameId(USER_ID, 220L))).thenReturn(Optional.of(existing)); - when(userSteamSyncRepository.tryClaimRun(eq(USER_ID), eq(STEAM_ID), any())).thenReturn(1); + when(userGameRepository.findByUserId(USER_ID)).thenReturn(List.of(existing)); + when(userSteamSyncRepository.tryClaimRun(eq(USER_ID), eq(STEAM_ID), any(), any())).thenReturn(1); when(steamApiService.getOwnedGames(STEAM_ID)).thenReturn(List.of( game(90, 0))); @@ -236,6 +254,18 @@ class LibrarySyncWorkerTest extends AbstractUnitTest { assertEquals(90, existing.getPlaytimeMinutes()); } + @Test + void runSync_SteamListsAGameTwice_CountsItOnce() { + when(userSteamSyncRepository.findForClaim(USER_ID)).thenReturn(Optional.of(syncRow(STEAM_ID, STEAM_ID))); + when(userSteamSyncRepository.tryClaimRun(eq(USER_ID), eq(STEAM_ID), any(), any())).thenReturn(1); + when(steamApiService.getOwnedGames(STEAM_ID)).thenReturn(List.of(game(600, 120), game(600, 120))); + + worker.runSync(USER_ID, STEAM_ID); + + verify(userGameRepository).save(any(UserGame.class)); + verify(playtimeDayRepository).addMinutes(USER_ID, 220L, today(), 120); + } + @Test void runSync_PlaytimeGrew_RecordsTheDifferenceForToday() { owned(100); @@ -358,10 +388,10 @@ class LibrarySyncWorkerTest extends AbstractUnitTest { @Test void runSync_FailedFirstSyncOfNewSteamId_NextSyncIsStillOnlyTheBaseline() { - when(userSteamSyncRepository.findById(USER_ID)).thenReturn( + when(userSteamSyncRepository.findForClaim(USER_ID)).thenReturn( Optional.of(syncRow(OTHER_STEAM_ID, OTHER_STEAM_ID)), Optional.of(syncRow(STEAM_ID, OTHER_STEAM_ID))); - when(userSteamSyncRepository.tryClaimRun(eq(USER_ID), eq(STEAM_ID), any())).thenReturn(1); + when(userSteamSyncRepository.tryClaimRun(eq(USER_ID), eq(STEAM_ID), any(), any())).thenReturn(1); when(steamApiService.getOwnedGames(STEAM_ID)).thenReturn(List.of(), List.of(game(5000, 60))); worker.runSync(USER_ID, STEAM_ID); @@ -425,19 +455,19 @@ class LibrarySyncWorkerTest extends AbstractUnitTest { worker.runSync(USER_ID, STEAM_ID); verify(wishlistGameRepository, never()).deleteByUserId(anyLong()); - verify(userSteamSyncRepository).markFinished(eq(USER_ID), eq("OK"), any()); + verify(userSteamSyncRepository).markFinished(eq(USER_ID), any(), eq("OK"), any()); } @Test void runSync_NoOwnedGames_StillSyncsTheWishlist() { - when(userSteamSyncRepository.tryClaimRun(eq(USER_ID), eq(STEAM_ID), any())).thenReturn(1); + when(userSteamSyncRepository.tryClaimRun(eq(USER_ID), eq(STEAM_ID), any(), any())).thenReturn(1); when(steamApiService.getOwnedGames(STEAM_ID)).thenReturn(List.of()); when(steamApiService.getWishlist(STEAM_ID)).thenReturn(Optional.of(List.of(new SteamApiService.WishlistItem(1257610L, null)))); worker.runSync(USER_ID, STEAM_ID); verify(wishlistGameRepository).saveAll(anyList()); - verify(userSteamSyncRepository).markFinished(eq(USER_ID), eq("FAILED"), any()); + verify(userSteamSyncRepository).markFinished(eq(USER_ID), any(), eq("FAILED"), any()); } @Test @@ -469,8 +499,8 @@ class LibrarySyncWorkerTest extends AbstractUnitTest { @Test void runSync_SteamIdChanged_LibraryFailed_RemovesTheOldAccountsWishlist() { - when(userSteamSyncRepository.findById(USER_ID)).thenReturn(Optional.of(syncRow(OTHER_STEAM_ID, OTHER_STEAM_ID))); - when(userSteamSyncRepository.tryClaimRun(eq(USER_ID), eq(STEAM_ID), any())).thenReturn(1); + when(userSteamSyncRepository.findForClaim(USER_ID)).thenReturn(Optional.of(syncRow(OTHER_STEAM_ID, OTHER_STEAM_ID))); + when(userSteamSyncRepository.tryClaimRun(eq(USER_ID), eq(STEAM_ID), any(), any())).thenReturn(1); when(steamApiService.getOwnedGames(STEAM_ID)).thenReturn(List.of()); worker.runSync(USER_ID, STEAM_ID); @@ -525,7 +555,7 @@ class LibrarySyncWorkerTest extends AbstractUnitTest { @Test void runSync_GameWithoutName_GetsTheAppidAsName() { - when(userSteamSyncRepository.tryClaimRun(eq(USER_ID), eq(STEAM_ID), any())).thenReturn(1); + when(userSteamSyncRepository.tryClaimRun(eq(USER_ID), eq(STEAM_ID), any(), any())).thenReturn(1); when(steamApiService.getOwnedGames(STEAM_ID)).thenReturn(List.of( new SteamOwnedGamesResponse.Game(2000001, null, 12, null, null, null), game(90, 0))); @@ -539,8 +569,8 @@ class LibrarySyncWorkerTest extends AbstractUnitTest { @Test void runSync_GameWithoutName_KeepsTheStoredName() { SteamGame stored = new SteamGame(2000001L, "El Mucho", null, Instant.EPOCH); - when(userSteamSyncRepository.tryClaimRun(eq(USER_ID), eq(STEAM_ID), any())).thenReturn(1); - when(steamGameRepository.findById(2000001L)).thenReturn(Optional.of(stored)); + when(userSteamSyncRepository.tryClaimRun(eq(USER_ID), eq(STEAM_ID), any(), any())).thenReturn(1); + when(steamGameRepository.findAllById(List.of(2000001L))).thenReturn(List.of(stored)); when(steamApiService.getOwnedGames(STEAM_ID)).thenReturn(List.of( new SteamOwnedGamesResponse.Game(2000001, " ", 12, null, null, null))); @@ -552,7 +582,7 @@ class LibrarySyncWorkerTest extends AbstractUnitTest { @Test void runSync_NewName_GoesThroughTheGuardedUpdate() { SteamGame stored = new SteamGame(220L, "Half-Life 2: Corrected", null, Instant.EPOCH); - when(steamGameRepository.findById(220L)).thenReturn(Optional.of(stored)); + when(steamGameRepository.findAllById(List.of(220L))).thenReturn(List.of(stored)); steamReturns(game(10, 0), null); worker.runSync(USER_ID, STEAM_ID); @@ -565,7 +595,7 @@ class LibrarySyncWorkerTest extends AbstractUnitTest { @Test void runSync_CorrectedName_SendsNoUpdate() { - when(steamGameRepository.findById(220L)).thenReturn(Optional.of(new SteamGame(220L, "HL2 (corrected)", null, Instant.EPOCH))); + when(steamGameRepository.findAllById(List.of(220L))).thenReturn(List.of(new SteamGame(220L, "HL2 (corrected)", null, Instant.EPOCH))); when(gameCorrectionRepository.findCorrectedAppids(eq(CorrectionField.NAME), any())).thenReturn(Set.of(220L)); steamReturns(game(10, 0), null); diff --git a/src/test/java/core_service/modules/steam_adapter/service/SteamApiServiceCacheTest.java b/src/test/java/core_service/modules/steam_adapter/service/SteamApiServiceCacheTest.java index c9c0d22..5eca018 100644 --- a/src/test/java/core_service/modules/steam_adapter/service/SteamApiServiceCacheTest.java +++ b/src/test/java/core_service/modules/steam_adapter/service/SteamApiServiceCacheTest.java @@ -1,6 +1,8 @@ package core_service.modules.steam_adapter.service; +import core_service.modules.steam_adapter.api.dto.response.SteamOwnedGamesResponse; import core_service.modules.steam_adapter.api.dto.response.SteamPlayerSummaryResponse; +import core_service.modules.steam_adapter.api.dto.response.SteamWishlistResponse; import core_service.shared.test.AbstractIntegrationTest; import io.github.resilience4j.ratelimiter.RateLimiter; import io.github.resilience4j.ratelimiter.RateLimiterRegistry; @@ -10,6 +12,10 @@ import org.mockito.Answers; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.cache.CacheManager; import org.springframework.test.context.bean.override.mockito.MockitoBean; +import org.springframework.http.HttpHeaders; +import org.springframework.http.HttpStatus; +import org.springframework.web.client.HttpClientErrorException; +import org.springframework.web.client.ResourceAccessException; import org.springframework.web.client.RestClient; import java.time.Duration; @@ -18,6 +24,7 @@ import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Function; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.when; @@ -98,4 +105,57 @@ class SteamApiServiceCacheTest extends AbstractIntegrationTest { assertThat(steamApiService.getPlayerSummary(STEAM_ID)).isPresent(); assertThat(steamCalls).hasValue(1); } + + @Test + void getOwnedGames_TransientSteamError_IsTriedAgain() { + SteamOwnedGamesResponse games = new SteamOwnedGamesResponse(new SteamOwnedGamesResponse.OwnedGamesResponse(1, + List.of(new SteamOwnedGamesResponse.Game(220, "Half-Life 2", 60, null, true, 0)))); + when(steamRestClient.get().uri(any(Function.class)).retrieve().body(SteamOwnedGamesResponse.class)) + .thenAnswer(invocation -> { + if (steamCalls.incrementAndGet() == 1) { + throw new ResourceAccessException("Read timed out"); + } + return games; + }); + + assertThat(steamApiService.getOwnedGames(STEAM_ID)).hasSize(1); + assertThat(steamCalls).hasValue(2); + } + + @Test + void getOwnedGames_SteamDownForEveryTry_Throws() { + when(steamRestClient.get().uri(any(Function.class)).retrieve().body(SteamOwnedGamesResponse.class)) + .thenAnswer(invocation -> { + steamCalls.incrementAndGet(); + throw new ResourceAccessException("Read timed out"); + }); + + assertThatThrownBy(() -> steamApiService.getOwnedGames(STEAM_ID)).isInstanceOf(ResourceAccessException.class); + assertThat(steamCalls).hasValue(3); + } + + @Test + void getOwnedGames_SteamRefusesTheKey_IsNotTriedAgain() { + when(steamRestClient.get().uri(any(Function.class)).retrieve().body(SteamOwnedGamesResponse.class)) + .thenAnswer(invocation -> { + steamCalls.incrementAndGet(); + throw HttpClientErrorException.create(HttpStatus.FORBIDDEN, "Forbidden", HttpHeaders.EMPTY, null, null); + }); + + assertThatThrownBy(() -> steamApiService.getOwnedGames(STEAM_ID)).isInstanceOf(HttpClientErrorException.class); + assertThat(steamCalls).hasValue(1); + } + + @Test + void getWishlist_SteamDownForEveryTry_IsEmptySoTheCallerKeepsTheOldList() { + when(steamRestClient.get().uri(any(Function.class)).retrieve().body(SteamWishlistResponse.class)) + .thenAnswer(invocation -> { + steamCalls.incrementAndGet(); + throw new ResourceAccessException("Read timed out"); + }); + + assertThat(steamApiService.getWishlist(STEAM_ID)).isEmpty(); + // One try: the wishlist is optional and must not hold up the sync of the library. + assertThat(steamCalls).hasValue(1); + } } diff --git a/src/test/java/core_service/modules/steam_adapter/service/SteamApiServiceTest.java b/src/test/java/core_service/modules/steam_adapter/service/SteamApiServiceTest.java index ce7dec1..b5b054d 100644 --- a/src/test/java/core_service/modules/steam_adapter/service/SteamApiServiceTest.java +++ b/src/test/java/core_service/modules/steam_adapter/service/SteamApiServiceTest.java @@ -3,7 +3,6 @@ package core_service.modules.steam_adapter.service; import core_service.modules.steam_adapter.api.dto.response.SteamOwnedGamesResponse; import core_service.modules.steam_adapter.api.dto.response.SteamPlayerAchievementsResponse; import core_service.modules.steam_adapter.api.dto.response.SteamPlayerSummaryResponse; -import core_service.modules.steam_adapter.api.dto.response.SteamProfileAndGamesResponse; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; @@ -21,6 +20,7 @@ import java.util.Map; import java.util.Optional; import java.util.stream.LongStream; +import static org.springframework.test.web.client.response.MockRestResponseCreators.withException; import static org.hamcrest.CoreMatchers.allOf; import static org.hamcrest.CoreMatchers.containsString; import static org.hamcrest.CoreMatchers.not; @@ -172,54 +172,6 @@ class SteamApiServiceTest { assertThat(result.getFirst().hasCommunityVisibleStats()).isNull(); } - @Test - void getFullPlayerProfile_shouldReturnCombinedResult() { - String steamId = "76561197960435530"; - String summaryJson = """ - { - "response": { - "players": [ - { - "steamid": "76561197960435530", - "personaname": "TestUser", - "profileurl": "url", - "avatarfull": "avatar" - } - ] - } - } - """; - String gamesJson = """ - { - "response": { - "game_count": 1, - "games": [ - { - "appid": 10, - "name": "Counter-Strike", - "playtime_forever": 32, - "img_icon_url": "icon_url", - "has_community_visible_stats": true - } - ] - } - } - """; - - server.expect(requestTo(containsString("/ISteamUser/GetPlayerSummaries/v0002/"))) - .andRespond(withSuccess(summaryJson, MediaType.APPLICATION_JSON)); - - server.expect(requestTo(containsString("/IPlayerService/GetOwnedGames/v1/"))) - .andRespond(withSuccess(gamesJson, MediaType.APPLICATION_JSON)); - - Optional result = steamApiService.getFullPlayerProfile(steamId); - - assertThat(result).isPresent(); - assertThat(result.get().player().personaname()).isEqualTo("TestUser"); - assertThat(result.get().games()).hasSize(1); - assertThat(result.get().games().getFirst().name()).isEqualTo("Counter-Strike"); - } - @Test void getWishlist_ReturnsItemsWithTheDateAdded() { server.expect(requestTo(containsString("/IWishlistService/GetWishlist/v1/"))) @@ -241,12 +193,14 @@ class SteamApiServiceTest { assertThat(steamApiService.getWishlist("76561197960435530")).contains(List.of()); } + // Without the resilience4j proxy the error reaches the caller; with it the fallback makes it empty + // (SteamApiServiceCacheTest). @Test - void getWishlist_SteamError_IsEmptySoTheCallerKeepsTheOldList() { + void getWishlist_SteamError_IsThrownForTheFallback() { server.expect(requestTo(containsString("/IWishlistService/GetWishlist/v1/"))) .andRespond(withServerError()); - assertThat(steamApiService.getWishlist("76561197960435530")).isEmpty(); + assertThatThrownBy(() -> steamApiService.getWishlist("76561197960435530")).isInstanceOf(RestClientException.class); } @Test @@ -408,4 +362,14 @@ class SteamApiServiceTest { assertThat(steamApiService.getGlobalAchievementPercentages(620)).isEmpty(); assertThat(steamApiService.getGlobalAchievementPercentages(620)).isEmpty(); } + + // The key is a query parameter; Spring 7 leaves the query out of an I/O error, and it must stay so in our logs. + @Test + void getOwnedGames_Timeout_HasNoApiKeyInItsMessage() { + server.expect(requestTo(containsString("/IPlayerService/GetOwnedGames/v1/"))) + .andRespond(withException(new java.net.SocketTimeoutException("Read timed out"))); + + assertThatThrownBy(() -> steamApiService.getOwnedGames("76561198000000001")) + .hasMessageContaining("Read timed out").hasMessageNotContaining("test-api-key"); + } }