feat(account): tell Core about Steam ID and block changes

Sent on the task executor after the commit, and every user again in the nightly cleanup. A failed call only warns.

src/main/java/auth_service/modules/account/model/CoreAccountState.java

@@ -0,0 +1,7 @@
+package auth_service.modules.account.model;
+
+/**
+ * One user as Core copies it.
+ */
+public record CoreAccountState(long userId, Long steamId, boolean blocked) {
+}

src/main/java/auth_service/modules/account/repo/AccountRepository.java

@@ -1,6 +1,8 @@
 package auth_service.modules.account.repo;
 
+import auth_service.modules.account.model.CoreAccountState;
 import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.core.RowMapper;
 import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
 import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
 import org.springframework.stereotype.Repository;
@@ -56,6 +58,20 @@ public class AccountRepository {
                 Timestamp.from(cutoff));
     }
 
+    private static final RowMapper<CoreAccountState> STATE_ROW = (rs, row) ->
+            new CoreAccountState(rs.getLong("id"), rs.getObject("steam_id", Long.class), !rs.getBoolean("enabled"));
+
+    public Optional<CoreAccountState> findState(long userId) {
+        return jdbc.query("SELECT id, steam_id, enabled FROM users WHERE id = ?", STATE_ROW, userId).stream().findFirst();
+    }
+
+    /**
+     * Every user, also the flagged ones.
+     */
+    public List<CoreAccountState> findAllStates() {
+        return jdbc.query("SELECT id, steam_id, enabled FROM users ORDER BY id", STATE_ROW);
+    }
+
     /**
      * Also the flagged users: their rows in Core are no orphans yet.
      */

src/main/java/auth_service/modules/account/service/AccountCleanupService.java

@@ -12,8 +12,8 @@ import java.time.Instant;
 import java.util.List;
 
 /**
- * Nightly, after Core's sync: tells Core who is flagged, removes users flagged longer than the retention (Core
- * first, then Auth), and lets Core remove rows of users Auth does not know.
+ * Nightly, after Core's sync: tells Core who is flagged and every user's Steam ID and block, removes users flagged
+ * longer than the retention (Core first, then Auth), and lets Core remove rows of users Auth does not know.
  */
 @Service
 @Slf4j
@@ -41,10 +41,19 @@ public class AccountCleanupService {
             log.warn("Account cleanup stopped: Core did not take the flagged users: {}", e.getMessage());
             return;
         }
+        sendStates();
         purgeDue();
         pruneOrphans();
     }
 
+    private void sendStates() {
+        try {
+            coreAccountService.replaceStates(accountRepository.findAllStates());
+        } catch (RuntimeException e) {
+            log.warn("Account cleanup: Core did not take the Steam IDs and blocks: {}", e.getMessage());
+        }
+    }
+
     private void purgeDue() {
         List<Long> due = accountRepository.findFlaggedBefore(Instant.now().minus(retention));
         if (due.isEmpty()) {

src/main/java/auth_service/modules/account/service/CoreAccountService.java

@@ -1,5 +1,6 @@
 package auth_service.modules.account.service;
 
+import auth_service.modules.account.model.CoreAccountState;
 import auth_service.modules.account.model.CoreCleanupResult;
 import auth_service.shared.security.ServiceTokenIssuer;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -52,6 +53,19 @@ public class CoreAccountService {
                 .body(Map.of("userIds", userIds)).retrieve().toBodilessEntity();
     }
 
+    public void applyStates(List<CoreAccountState> states) {
+        restClient.post().uri("/internal/accounts/states").headers(this::authorize)
+                .body(Map.of("accounts", states)).retrieve().toBodilessEntity();
+    }
+
+    /**
+     * After this, Core blocks exactly the blocked users of the list.
+     */
+    public void replaceStates(List<CoreAccountState> states) {
+        restClient.put().uri("/internal/accounts/states").headers(this::authorize)
+                .body(Map.of("accounts", states)).retrieve().toBodilessEntity();
+    }
+
     /**
      * @param dryRun Core only counts; Core also counts when its own dry run is on
      */

src/main/java/auth_service/modules/account/service/CoreAccountStatePublisher.java

@@ -0,0 +1,55 @@
+package auth_service.modules.account.service;
+
+import auth_service.modules.account.repo.AccountRepository;
+import auth_service.shared.account.AccountStatePublisher;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.support.TransactionSynchronization;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
+
+import java.util.List;
+import java.util.concurrent.Executor;
+
+/**
+ * Sends on the task executor: in afterCommit the transaction still holds its database connection, and a slow Core
+ * (up to 13 s) would keep it from the pool.
+ */
+@Service
+@Slf4j
+public class CoreAccountStatePublisher implements AccountStatePublisher {
+
+    private final AccountRepository accountRepository;
+    private final CoreAccountService coreAccountService;
+    private final Executor executor;
+
+    public CoreAccountStatePublisher(AccountRepository accountRepository, CoreAccountService coreAccountService,
+                                     @Qualifier("applicationTaskExecutor") Executor executor) {
+        this.accountRepository = accountRepository;
+        this.coreAccountService = coreAccountService;
+        this.executor = executor;
+    }
+
+    @Override
+    public void publish(long userId) {
+        if (TransactionSynchronizationManager.isSynchronizationActive()) {
+            TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
+                @Override
+                public void afterCommit() {
+                    executor.execute(() -> send(userId));
+                }
+            });
+        } else {
+            executor.execute(() -> send(userId));
+        }
+    }
+
+    private void send(long userId) {
+        try {
+            accountRepository.findState(userId).ifPresent(state -> coreAccountService.applyStates(List.of(state)));
+        } catch (RuntimeException e) {
+            log.warn("Core did not take the state of user {}, the nightly cleanup sends it again: {}", userId,
+                    e.getMessage());
+        }
+    }
+}

src/main/java/auth_service/modules/admin/service/AdminUserService.java

@@ -6,6 +6,7 @@ import auth_service.modules.admin.model.AdminUserPage;
 import auth_service.modules.admin.model.TargetState;
 import auth_service.modules.admin.repo.AdminUserRepository;
 import auth_service.shared.account.AccountDeleter;
+import auth_service.shared.account.AccountStatePublisher;
 import auth_service.shared.audit.AdminAuditLog;
 import auth_service.shared.security.RequiresRole;
 import auth_service.shared.security.Role;
@@ -29,6 +30,7 @@ public class AdminUserService {
     private final AdminUserRepository adminUserRepository;
     private final AdminAuditLog adminAuditLog;
     private final AccountDeleter accountDeleter;
+    private final AccountStatePublisher accountStatePublisher;
 
     public AdminUserPage findUsers(String query, int page, int size) {
         String search = query == null ? "" : query.trim();
@@ -48,6 +50,7 @@ public class AdminUserService {
             throw new AdminActionRefusedException("ALREADY_DELETED", "The account is already deleted");
         }
         adminUserRepository.setBlocked(targetUserId, blocked, Instant.now());
+        accountStatePublisher.publish(targetUserId);
         adminAuditLog.record(actorUserId, blocked ? "BLOCK" : "UNBLOCK", targetUserId, null);
         log.info("Admin {} {} user {}", actorUserId, blocked ? "blocked" : "unblocked", targetUserId);
     }

src/main/java/auth_service/modules/user/service/UserService.java

@@ -10,6 +10,7 @@ import auth_service.modules.user.model.SteamLinkOutcome;
 import auth_service.modules.user.model.User;
 import auth_service.modules.user.model.UserSettings;
 import auth_service.modules.user.persistence.UserRepository;
+import auth_service.shared.account.AccountStatePublisher;
 import auth_service.shared.security.RequiresRole;
 import auth_service.shared.security.Role;
 import auth_service.shared.security.TokenGenerator;
@@ -31,6 +32,7 @@ public class UserService {
 
     private final UserRepository userRepository;
     private final TokenGenerator tokenGenerator;
+    private final AccountStatePublisher accountStatePublisher;
 
     @RequiresRole(Role.USER)
     @Cacheable(value = "users", key = "#userId")
@@ -74,8 +76,10 @@ public class UserService {
             }
             log.info("Steam ID taken from user {}, who had typed it in, for user {}", holder.get().userId(), userId);
             userRepository.clearSteamId(holder.get().userId());
+            accountStatePublisher.publish(holder.get().userId());
         }
         userRepository.setVerifiedSteamId(userId, steamId);
+        accountStatePublisher.publish(userId);
         log.info("Steam ID verified for user {}", userId);
         return SteamLinkOutcome.LINKED;
     }
@@ -89,6 +93,7 @@ public class UserService {
             @CacheEvict(value = "publicStats", allEntries = true)})
     public String unlinkSteamId(long userId, Role role) {
         userRepository.clearSteamId(userId);
+        accountStatePublisher.publish(userId);
         log.info("Steam ID unlinked for user {}", userId);
         return tokenGenerator.generateToken(String.valueOf(userId), null, role);
     }

src/main/java/auth_service/shared/account/AccountStatePublisher.java

@@ -0,0 +1,9 @@
+package auth_service.shared.account;
+
+/**
+ * Port for a change Core copies: the Steam ID or the block of a user. Sends after the commit; a failure only warns,
+ * because the nightly cleanup sends every user again.
+ */
+public interface AccountStatePublisher {
+    void publish(long userId);
+}

src/test/java/auth_service/modules/account/repo/AccountRepositoryTest.java

@@ -1,5 +1,6 @@
 package auth_service.modules.account.repo;
 
+import auth_service.modules.account.model.CoreAccountState;
 import auth_service.shared.test.AbstractPersistenceTest;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -49,6 +50,16 @@ class AccountRepositoryTest extends AbstractPersistenceTest {
         bob = insertUser("bob@example.com");
     }
 
+    @Test
+    void findStates_CarryTheSteamIdAndTheBlock() {
+        jdbc.update("update users set enabled = false, steam_id = 76561198000000001 where id = ?", alice);
+
+        assertThat(repository.findState(alice)).contains(new CoreAccountState(alice, 76561198000000001L, true));
+        assertThat(repository.findAllStates()).containsExactly(new CoreAccountState(alice, 76561198000000001L, true),
+                new CoreAccountState(bob, null, false));
+        assertThat(repository.findState(-1L)).isEmpty();
+    }
+
     @Test
     void flagDeleted_SetsTheFlagAndRevokesOnlyThatUsersTokens() {
         assertThat(repository.flagDeleted(alice, NOW)).isEqualTo(1);

src/test/java/auth_service/modules/account/service/AccountCleanupServiceTest.java

@@ -1,5 +1,6 @@
 package auth_service.modules.account.service;
 
+import auth_service.modules.account.model.CoreAccountState;
 import auth_service.modules.account.model.CoreCleanupResult;
 import auth_service.modules.account.repo.AccountRepository;
 import auth_service.shared.test.AbstractUnitTest;
@@ -24,6 +25,8 @@ import static org.mockito.Mockito.when;
 
 class AccountCleanupServiceTest extends AbstractUnitTest {
 
+    private static final List<CoreAccountState> STATES = List.of(new CoreAccountState(1L, 76561198000000001L, true));
+
     private static final Duration RETENTION = Duration.ofDays(30);
 
     @Mock
@@ -45,6 +48,7 @@ class AccountCleanupServiceTest extends AbstractUnitTest {
         lenient().when(accountRepository.findFlaggedIds()).thenReturn(List.of(7L, 8L));
         lenient().when(accountRepository.findFlaggedBefore(any())).thenReturn(List.of(7L));
         lenient().when(accountRepository.findAllIds()).thenReturn(List.of(1L, 7L, 8L));
+        lenient().when(accountRepository.findAllStates()).thenReturn(STATES);
     }
 
     @Test
@@ -55,9 +59,22 @@ class AccountCleanupServiceTest extends AbstractUnitTest {
         service(false).runNightly();
 
         verify(coreAccountService).replaceDeleted(List.of(7L, 8L));
+        verify(coreAccountService).replaceStates(STATES);
         verify(accountRepository).deleteUsers(List.of(7L));
     }
 
+    @Test
+    void runNightly_AFailedStateCallStillPurgesAndPrunes() {
+        doThrow(new RestClientException("Core too old")).when(coreAccountService).replaceStates(any());
+        when(coreAccountService.purge(eq(List.of(7L)), anyBoolean())).thenReturn(core(false));
+        when(coreAccountService.prune(eq(List.of(1L, 7L, 8L)), anyBoolean())).thenReturn(core(false));
+
+        service(false).runNightly();
+
+        verify(accountRepository).deleteUsers(List.of(7L));
+        verify(coreAccountService).prune(List.of(1L, 7L, 8L), false);
+    }
+
     @Test
     void runNightly_InDryRunAsksCoreForADryRunAndDeletesNothingInAuth() {
         when(coreAccountService.purge(eq(List.of(7L)), anyBoolean())).thenReturn(core(true));

src/test/java/auth_service/modules/account/service/CoreAccountServiceTest.java

@@ -1,5 +1,6 @@
 package auth_service.modules.account.service;
 
+import auth_service.modules.account.model.CoreAccountState;
 import auth_service.modules.account.model.CoreCleanupResult;
 import auth_service.shared.security.ServiceTokenIssuer;
 import auth_service.shared.test.AbstractUnitTest;
@@ -65,6 +66,17 @@ class CoreAccountServiceTest extends AbstractUnitTest {
         core.verify();
     }
 
+    @Test
+    void applyAndReplaceStates_SendTheContractBodies() {
+        expect("applyStates");
+        expect("replaceStates");
+
+        service.applyStates(List.of(new CoreAccountState(42L, 76561198000000001L, true)));
+        service.replaceStates(List.of(new CoreAccountState(42L, null, false)));
+
+        core.verify();
+    }
+
     @Test
     void purgeAndPrune_ReadTheContractResult() {
         expect("purge");

src/test/java/auth_service/modules/account/service/CoreAccountStatePublisherTest.java

@@ -0,0 +1,82 @@
+package auth_service.modules.account.service;
+
+import auth_service.modules.account.model.CoreAccountState;
+import auth_service.modules.account.repo.AccountRepository;
+import auth_service.shared.test.AbstractUnitTest;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mock;
+import org.springframework.transaction.support.TransactionSynchronization;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
+import org.springframework.web.client.RestClientException;
+
+import java.util.List;
+import java.util.Optional;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class CoreAccountStatePublisherTest extends AbstractUnitTest {
+
+    private static final CoreAccountState STATE = new CoreAccountState(7L, 76561198000000001L, true);
+
+    @Mock
+    private AccountRepository accountRepository;
+
+    @Mock
+    private CoreAccountService coreAccountService;
+
+    private CoreAccountStatePublisher publisher;
+
+    // Runs each task at once, so the test sees the call.
+    @BeforeEach
+    void setUp() {
+        publisher = new CoreAccountStatePublisher(accountRepository, coreAccountService, Runnable::run);
+    }
+
+    @Test
+    void withoutATransaction_SendsAtOnce() {
+        when(accountRepository.findState(7L)).thenReturn(Optional.of(STATE));
+
+        publisher.publish(7L);
+
+        verify(coreAccountService).applyStates(List.of(STATE));
+    }
+
+    @Test
+    void inATransaction_SendsOnlyAfterTheCommit() {
+        when(accountRepository.findState(7L)).thenReturn(Optional.of(STATE));
+        TransactionSynchronizationManager.initSynchronization();
+        try {
+            publisher.publish(7L);
+            verify(coreAccountService, never()).applyStates(any());
+
+            TransactionSynchronizationManager.getSynchronizations().forEach(TransactionSynchronization::afterCommit);
+        } finally {
+            TransactionSynchronizationManager.clearSynchronization();
+        }
+
+        verify(coreAccountService).applyStates(List.of(STATE));
+    }
+
+    @Test
+    void coreDown_OnlyWarns() {
+        when(accountRepository.findState(7L)).thenReturn(Optional.of(STATE));
+        doThrow(new RestClientException("Core down")).when(coreAccountService).applyStates(any());
+
+        assertDoesNotThrow(() -> publisher.publish(7L));
+    }
+
+    @Test
+    void unknownUser_SendsNothing() {
+        when(accountRepository.findState(7L)).thenReturn(Optional.empty());
+
+        publisher.publish(7L);
+
+        verify(coreAccountService, never()).applyStates(any());
+    }
+}

src/test/java/auth_service/modules/admin/service/AdminUserServiceTest.java

@@ -5,6 +5,7 @@ import auth_service.modules.admin.exception.TargetUserNotFoundException;
 import auth_service.modules.admin.model.TargetState;
 import auth_service.modules.admin.repo.AdminUserRepository;
 import auth_service.shared.account.AccountDeleter;
+import auth_service.shared.account.AccountStatePublisher;
 import auth_service.shared.audit.AdminAuditLog;
 import auth_service.shared.security.Role;
 import auth_service.shared.test.AbstractUnitTest;
@@ -39,11 +40,14 @@ class AdminUserServiceTest extends AbstractUnitTest {
     @Mock
     private AccountDeleter accountDeleter;
 
+    @Mock
+    private AccountStatePublisher accountStatePublisher;
+
     private AdminUserService service;
 
     @BeforeEach
     void setUp() {
-        service = new AdminUserService(adminUserRepository, adminAuditLog, accountDeleter);
+        service = new AdminUserService(adminUserRepository, adminAuditLog, accountDeleter, accountStatePublisher);
     }
 
     private void target(Role role, boolean deleted) {
@@ -67,6 +71,7 @@ class AdminUserServiceTest extends AbstractUnitTest {
 
         verify(adminUserRepository).setBlocked(eq(TARGET), eq(true), any());
         verify(adminAuditLog).record(ADMIN, "BLOCK", TARGET, null);
+        verify(accountStatePublisher).publish(TARGET);
     }
 
     @Test

src/test/java/auth_service/modules/user/service/UserServiceTest.java

@@ -10,6 +10,7 @@ import auth_service.modules.user.model.SteamLinkOutcome;
 import auth_service.modules.user.model.User;
 import auth_service.modules.user.model.UserSettings;
 import auth_service.modules.user.persistence.UserRepository;
+import auth_service.shared.account.AccountStatePublisher;
 import auth_service.shared.security.TokenGenerator;
 import auth_service.shared.test.AbstractUnitTest;
 import org.junit.jupiter.api.Test;
@@ -41,6 +42,9 @@ class UserServiceTest extends AbstractUnitTest {
     @Mock
     private TokenGenerator tokenGenerator;
 
+    @Mock
+    private AccountStatePublisher accountStatePublisher;
+
     @InjectMocks
     private UserService userService;
 
@@ -124,6 +128,7 @@ class UserServiceTest extends AbstractUnitTest {
         assertEquals(SteamLinkOutcome.LINKED, userService.linkVerifiedSteamId(1L, STEAM_ID));
 
         verify(userRepository).setVerifiedSteamId(1L, STEAM_ID);
+        verify(accountStatePublisher).publish(1L);
     }
 
     @Test
@@ -145,6 +150,8 @@ class UserServiceTest extends AbstractUnitTest {
         InOrder order = inOrder(userRepository);
         order.verify(userRepository).clearSteamId(2L);
         order.verify(userRepository).setVerifiedSteamId(1L, STEAM_ID);
+        verify(accountStatePublisher).publish(2L);
+        verify(accountStatePublisher).publish(1L);
     }
 
     @Test
@@ -155,6 +162,7 @@ class UserServiceTest extends AbstractUnitTest {
 
         verify(userRepository, never()).clearSteamId(anyLong());
         verify(userRepository, never()).setVerifiedSteamId(anyLong(), anyLong());
+        verifyNoInteractions(accountStatePublisher);
     }
 
     @Test
@@ -164,6 +172,7 @@ class UserServiceTest extends AbstractUnitTest {
         assertEquals("token-without-steam", userService.unlinkSteamId(1L, Role.USER));
 
         verify(userRepository).clearSteamId(1L);
+        verify(accountStatePublisher).publish(1L);
     }
 
     @Test

src/test/resources/contract/auth-core-contract.json

@@ -55,6 +55,16 @@
       "path": "/internal/accounts/deleted",
       "request": {"userIds": []}
     },
+    "applyStates": {
+      "method": "POST",
+      "path": "/internal/accounts/states",
+      "request": {"accounts": [{"userId": 42, "steamId": 76561198000000001, "blocked": true}]}
+    },
+    "replaceStates": {
+      "method": "PUT",
+      "path": "/internal/accounts/states",
+      "request": {"accounts": [{"userId": 42, "steamId": null, "blocked": false}]}
+    },
     "purge": {
       "method": "POST",
       "path": "/internal/accounts/purge",