Lobo_Corp/ToBePlayed-Core
Public Access
3
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(sync): short sync transactions, real Steam retries, no N+1

Browse Source
This commit is contained in:
lobo committed 2026-09-29 13:27:31 +02:00
1 parent 2a5fcc1c00
commit 2360beeb81
18 files changed
+536 -279
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/main/java/core_service/modules/account/repo/AccountCleanupRepository.java
+1 -1
View File
@@ -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<String> TABLES = List.of("user_game_status_change", "user_game_achievement",
static final List<String> 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");
src/main/java/core_service/modules/library/domain/UserSteamSync.java
+3
View File
@@ -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;
src/main/java/core_service/modules/library/repo/UserGameRepository.java
+2
View File
@@ -41,6 +41,8 @@ public interface UserGameRepository extends JpaRepository<UserGame, UserGameId>
List<UserGame> findByUserIdAndRemovedAtIsNull(Long userId);
List<UserGame> 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<Long> findActiveAppids(@Param("userId") Long userId, @Param("appids") Collection<Long> appids);
src/main/java/core_service/modules/library/repo/UserSteamSyncRepository.java
+28 -8
View File
@@ -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<UserSteamSync, Long> {
/**
* 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<UserSteamSync> 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.
src/main/java/core_service/modules/library/service/LibrarySyncService.java
+2 -2
View File
@@ -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)));
}
}
src/main/java/core_service/modules/library/service/LibrarySyncWorker.java
+101 -72
View File
@@ -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).
* <p>
* 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<SteamOwnedGamesResponse.Game> 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<SteamOwnedGamesResponse.Game> 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> 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<Long> correctedNames = gameCorrectionRepository.findCorrectedAppids(CorrectionField.NAME,
games.stream().map(SteamOwnedGamesResponse.Game::appid).filter(Objects::nonNull).map(Integer::longValue).toList());
List<Long> 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<Long> igdbAppids = new ArrayList<>(games.stream().filter(g -> g.appid() != null).map(g -> g.appid().longValue()).toList());
List<Long> 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<SteamOwnedGamesResponse.Game> ownedGames) {
Optional<List<SteamApiService.WishlistItem>> 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<SteamOwnedGamesResponse.Game> games, Instant now,
LocalDate playedOn) {
List<Long> 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<Long> correctedNames = gameCorrectionRepository.findCorrectedAppids(CorrectionField.NAME, appids);
Map<Long, SteamGame> knownGames = steamGameRepository.findAllById(appids).stream()
.collect(Collectors.toMap(SteamGame::getAppid, game -> game));
Map<Long, UserGame> userGames = userGameRepository.findByUserId(userId).stream()
.collect(Collectors.toMap(UserGame::getAppid, game -> game));
List<Long> 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<Long> owned = ownedGames.stream()
.filter(g -> g.appid() != null)
.map(g -> g.appid().longValue())
.collect(Collectors.toSet());
List<SteamApiService.WishlistItem> wanted = wishlist.get().stream()
.filter(item -> !owned.contains(item.appid()))
.toList();
Map<Long, SteamApiService.StoreItem> 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<SteamApiService.WishlistItem> wanted, Map<Long, SteamApiService.StoreItem> store) {}
private Optional<Wishlist> fetchWishlist(String steamId, List<SteamOwnedGamesResponse.Game> ownedGames) {
Set<Long> owned = ownedGames.stream().map(g -> g.appid().longValue()).collect(Collectors.toSet());
return steamApiService.getWishlist(steamId).map(items -> {
List<SteamApiService.WishlistItem> 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<Long, WishlistGame> 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<Long> correctedNames) {
private void upsertGame(SteamOwnedGamesResponse.Game g, SteamGame existing, Instant now, Set<Long> 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;
src/main/java/core_service/modules/steam_adapter/api/dto/response/SteamProfileAndGamesResponse.java
-9
View File
@@ -1,9 +0,0 @@
package core_service.modules.steam_adapter.api.dto.response;
import java.util.List;
public record SteamProfileAndGamesResponse(
SteamPlayerSummaryResponse.Player player,
List<SteamOwnedGamesResponse.Game> games
) {
}
src/main/java/core_service/modules/steam_adapter/service/SteamApiService.java
+81 -103
View File
@@ -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<SteamProfileAndGamesResponse> 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<SteamPlayerSummaryResponse.Player> 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<SteamPlayerSummaryResponse.Player> 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<SteamOwnedGamesResponse.Game> 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<List<WishlistItem>> 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<List<WishlistItem>> noWishlist(String steamId, Exception e) {
log.error("Error communicating with Steam API for the wishlist of SteamID: {}", steamId, e);
return Optional.empty();
}
/**
src/main/resources/application.yml
+7
View File
@@ -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:
src/main/resources/db/migration/V23__Sync_Started_At.sql
+3
View File
@@ -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;
src/test/java/core_service/modules/account/repo/AccountCleanupRepositoryTest.java
+9
View File
@@ -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<String> 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);
}
}
src/test/java/core_service/modules/library/repo/UserSteamSyncRepositoryTest.java
+17 -1
View File
@@ -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();
}
}
src/test/java/core_service/modules/library/service/LibrarySyncServiceTest.java
+1
View File
@@ -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));
}
src/test/java/core_service/modules/library/service/LibrarySyncWorkerConcurrencyTest.java
+140
View File
@@ -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<Void> 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<Void> 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<Void> 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<Void> 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);
}
}
src/test/java/core_service/modules/library/service/LibrarySyncWorkerTest.java
+61 -31
View File
@@ -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);
src/test/java/core_service/modules/steam_adapter/service/SteamApiServiceCacheTest.java
+60
View File
@@ -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);
}
}
src/test/java/core_service/modules/steam_adapter/service/SteamApiServiceTest.java
+15 -51
View File
@@ -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<SteamProfileAndGamesResponse> 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");
}
}