isaaccomputerscience · mariusmarin-dev · Aug 13, 2026 · Aug 4, 2026 · Aug 6, 2026 · Aug 7, 2026
diff --git a/src/main/java/uk/ac/cam/cl/dtg/segue/api/managers/ExternalAccountManager.java b/src/main/java/uk/ac/cam/cl/dtg/segue/api/managers/ExternalAccountManager.java
@@ -40,6 +40,7 @@ public class ExternalAccountManager implements IExternalAccountManager {
   private static final Logger log = LoggerFactory.getLogger(ExternalAccountManager.class);
   private static final String MAILJET = "MAILJET - ";
   private static final int BULK_BATCH_SIZE = 100;
+  private static final int MAX_USERS_PER_SYNC = 1000;
   private static final int RATE_LIMIT_RETRY_SLEEP_MS = 10000; // 10 seconds
   private static final int JOB_POLL_INTERVAL_MS = 5000;  // 5 seconds between polls
   private static final int JOB_POLL_MAX_ATTEMPTS = 60;   // max 5 minutes (60 × 5s)
@@ -105,8 +106,14 @@ public synchronized SyncResult synchroniseChangedUsers() throws ExternalAccountS
   private List<UserExternalAccountChanges> getRecentlyChangedUsersOrThrow()
       throws ExternalAccountSynchronisationException {
     try {
-      List<UserExternalAccountChanges> users = database.getRecentlyChangedRecords();
-      log.info("{}Found {} users to synchronize with Mailjet", MAILJET, users.size());
+      List<UserExternalAccountChanges> users = database.getRecentlyChangedRecords(MAX_USERS_PER_SYNC);
+      if (users.size() >= MAX_USERS_PER_SYNC) {
+        int totalUsersToSync = database.countRecentlyChangedRecords();
+        log.info("{}Found {} users to synchronize with Mailjet; only the first {} will be processed this run",
+            MAILJET, totalUsersToSync, MAX_USERS_PER_SYNC);
+      } else {
+        log.info("{}Found {} users to synchronize with Mailjet", MAILJET, users.size());
+      }
       return users;
     } catch (SegueDatabaseException e) {
       throw new ExternalAccountSynchronisationException("Failed to retrieve users for synchronization"

diff --git a/src/main/java/uk/ac/cam/cl/dtg/segue/dao/users/IExternalAccountDataManager.java b/src/main/java/uk/ac/cam/cl/dtg/segue/dao/users/IExternalAccountDataManager.java
@@ -22,7 +22,9 @@
 
 public interface IExternalAccountDataManager {
 
-  List<UserExternalAccountChanges> getRecentlyChangedRecords() throws SegueDatabaseException;
+  List<UserExternalAccountChanges> getRecentlyChangedRecords(int limit) throws SegueDatabaseException;
+
+  int countRecentlyChangedRecords() throws SegueDatabaseException;
 
   void updateProviderLastUpdated(Long userId) throws SegueDatabaseException;
 

diff --git a/src/main/java/uk/ac/cam/cl/dtg/segue/dao/users/PgExternalAccountPersistenceManager.java b/src/main/java/uk/ac/cam/cl/dtg/segue/dao/users/PgExternalAccountPersistenceManager.java
@@ -41,10 +41,28 @@ public PgExternalAccountPersistenceManager(final PostgresSqlDb database) {
     this.database = database;
   }
 
+  // IMPORTANT: registered_contexts is JSONB[] (array of JSONB objects) in PostgreSQL
+  // We use array_to_json() to convert it to proper JSON that Java can parse
+  private static final String RECENTLY_CHANGED_USERS_FROM_WHERE_CLAUSE =
+      "FROM users "
+          + "    LEFT OUTER JOIN user_preferences AS news_prefs "
+          + "        ON users.id = news_prefs.user_id "
+          + "        AND news_prefs.preference_type = 'EMAIL_PREFERENCE' "
+          + "        AND news_prefs.preference_name = 'NEWS_AND_UPDATES' "
+          + "    LEFT OUTER JOIN user_preferences AS events_prefs "
+          + "        ON users.id = events_prefs.user_id "
+          + "        AND events_prefs.preference_type = 'EMAIL_PREFERENCE' "
+          + "        AND events_prefs.preference_name = 'EVENTS' "
+          + "    LEFT OUTER JOIN external_accounts "
+          + "        ON users.id = external_accounts.user_id "
+          + "        AND external_accounts.provider_name = 'MailJet' "
+          + "WHERE (users.last_updated >= external_accounts.provider_last_updated "
+          + "       OR news_prefs.last_updated >= external_accounts.provider_last_updated "
+          + "       OR events_prefs.last_updated >= external_accounts.provider_last_updated "
+          + "       OR external_accounts.provider_last_updated IS NULL) ";
+
   @Override
-  public List<UserExternalAccountChanges> getRecentlyChangedRecords() throws SegueDatabaseException {
-    // IMPORTANT: registered_contexts is JSONB[] (array of JSONB objects) in PostgreSQL
-    // We use array_to_json() to convert it to proper JSON that Java can parse
+  public List<UserExternalAccountChanges> getRecentlyChangedRecords(final int limit) throws SegueDatabaseException {
     String query = "SELECT users.id, "
         + "       external_accounts.provider_user_identifier, "
         + "       users.email, "
@@ -56,27 +74,28 @@ public List<UserExternalAccountChanges> getRecentlyChangedRecords() throws Segue
         + "       news_prefs.preference_value AS news_emails, "
         + "       events_prefs.preference_value AS events_emails, "
         + "       external_accounts.provider_last_updated "
-        + "FROM users "
-        + "    LEFT OUTER JOIN user_preferences AS news_prefs "
-        + "        ON users.id = news_prefs.user_id "
-        + "        AND news_prefs.preference_type = 'EMAIL_PREFERENCE' "
-        + "        AND news_prefs.preference_name = 'NEWS_AND_UPDATES' "
-        + "    LEFT OUTER JOIN user_preferences AS events_prefs "
-        + "        ON users.id = events_prefs.user_id "
-        + "        AND events_prefs.preference_type = 'EMAIL_PREFERENCE' "
-        + "        AND events_prefs.preference_name = 'EVENTS' "
-        + "    LEFT OUTER JOIN external_accounts "
-        + "        ON users.id = external_accounts.user_id "
-        + "        AND external_accounts.provider_name = 'MailJet' "
-        + "WHERE (users.last_updated >= external_accounts.provider_last_updated "
-        + "       OR news_prefs.last_updated >= external_accounts.provider_last_updated "
-        + "       OR events_prefs.last_updated >= external_accounts.provider_last_updated "
-        + "       OR external_accounts.provider_last_updated IS NULL) "
-        + "ORDER BY users.id";
+        + RECENTLY_CHANGED_USERS_FROM_WHERE_CLAUSE
+        + "ORDER BY users.id "
+        + "LIMIT " + limit;
 
     return executeQueryAndBuildUserRecords(query);
   }
 
+  @Override
+  public int countRecentlyChangedRecords() throws SegueDatabaseException {
+    String query = "SELECT COUNT(*) " + RECENTLY_CHANGED_USERS_FROM_WHERE_CLAUSE;
+
+    try (Connection conn = database.getDatabaseConnection();
+         PreparedStatement pst = conn.prepareStatement(query);
+         ResultSet results = pst.executeQuery()
+    ) {
+      return results.next() ? results.getInt(1) : 0;
+    } catch (SQLException e) {
+      throw new SegueDatabaseException("Database error while counting recently changed records: "
+          + e.getMessage(), e);
+    }
+  }
+
   /**
    * Execute query and build user records list.
    * Extracted to reduce nesting complexity.

diff --git a/src/main/java/uk/ac/cam/cl/dtg/util/email/MailJetApiClientWrapper.java b/src/main/java/uk/ac/cam/cl/dtg/util/email/MailJetApiClientWrapper.java
@@ -46,6 +46,7 @@ public class MailJetApiClientWrapper {
   private static final int BULK_BATCH_SIZE = 100;
   private static final String ACTION = "Action";
   private static final String LIST_ID = "ListID";
+  private static final int API_CALL_DELAY_MS = 500;
 
   private final MailjetClient mailjetClient;
   private final String newsListId;
@@ -93,6 +94,7 @@ public JSONObject getAccountByIdOrEmail(final String mailjetIdOrEmail) throws Ma
 
     try {
       MailjetRequest request = new MailjetRequest(Contact.resource, mailjetIdOrEmail);
+      rateLimitDelay();
       MailjetResponse response = mailjetClient.get(request);
 
       if (response.getStatus() == 404) {
@@ -138,6 +140,7 @@ public void permanentlyDeleteAccountById(final String mailjetId) throws MailjetE
 
     try {
       MailjetRequest request = new MailjetRequest(Contacts.resource, mailjetId);
+      rateLimitDelay();
       MailjetResponse response = mailjetClient.delete(request);
 
       if (response.getStatus() == 204 || response.getStatus() == 200) {
@@ -202,6 +205,7 @@ public String addNewUserOrGetUserIfExists(final String email) throws MailjetExce
   private String createNewMailjetAccount(String normalizedEmail) throws MailjetException, JSONException {
     try {
       MailjetRequest request = new MailjetRequest(Contact.resource).property(Contact.EMAIL, normalizedEmail);
+      rateLimitDelay();
       MailjetResponse response = mailjetClient.post(request);
 
       if (response.getStatus() == 201 || response.getStatus() == 200) {
@@ -287,6 +291,7 @@ public void updateUserProperties(final String mailjetId, final String firstName,
                       .put(PROPERTY_VALUE_KEY, emailVerificationStatus != null ? emailVerificationStatus : ""))
               .put(new JSONObject().put("Name", "stage").put(PROPERTY_VALUE_KEY, stage != null ? stage : "unknown")));
 
+      rateLimitDelay();
       MailjetResponse response = mailjetClient.put(request);
 
       if (response.getStatus() == 200 && response.getTotal() == 1) {
@@ -346,6 +351,7 @@ ContactManagecontactslists.CONTACTSLISTS, new JSONArray().put(
                     new JSONObject().put(ContactslistImportList.LISTID, eventsListId)
                   .put(ContactslistImportList.ACTION, eventsEmails.getValue())));
 
+      rateLimitDelay();
       MailjetResponse response = mailjetClient.post(request);
 
       if (response.getStatus() == 201 && response.getTotal() == 1) {
@@ -378,6 +384,17 @@ ContactManagecontactslists.CONTACTSLISTS, new JSONArray().put(
     }
   }
 
+  /**
+   * Pause before an outgoing Mailjet API call to stay under their rate limit.
+   */
+  private void rateLimitDelay() {
+    try {
+      Thread.sleep(API_CALL_DELAY_MS);
+    } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+    }
+  }
+
   /**
    * Check if exception is a 404 not found error.
    */
@@ -499,6 +516,7 @@ private String submitBulkSyncRequest(final MailjetRequest request,
                                         final MailJetSubscriptionAction newsAction,
                                         final MailJetSubscriptionAction eventsAction)
       throws MailjetException {
+    rateLimitDelay();
     MailjetResponse response = mailjetClient.post(request);
     int status = response.getStatus();
 
@@ -529,6 +547,7 @@ public JobStatus getBulkJobStatus(final String jobId) throws MailjetException {
 
     try {
       MailjetRequest request = new MailjetRequest(ContactManagemanycontacts.resource, Long.parseLong(jobId));
+      rateLimitDelay();
       MailjetResponse response = mailjetClient.get(request);
 
       if (response.getStatus() != 200) {

diff --git a/src/test/java/uk/ac/cam/cl/dtg/segue/api/managers/ExternalAccountManagerTest.java b/src/test/java/uk/ac/cam/cl/dtg/segue/api/managers/ExternalAccountManagerTest.java
@@ -1,5 +1,6 @@
 package uk.ac.cam.cl.dtg.segue.api.managers;
 
+import static org.easymock.EasyMock.anyInt;
 import static org.easymock.EasyMock.anyObject;
 import static org.easymock.EasyMock.createMock;
 import static org.easymock.EasyMock.eq;
@@ -58,7 +59,7 @@ void synchroniseChangedUsers_WithBulkUsers_ShouldSubmitAndPoll()
           )
       );
 
-      expect(mockDatabase.getRecentlyChangedRecords()).andReturn(changedUsers);
+      expect(mockDatabase.getRecentlyChangedRecords(anyInt())).andReturn(changedUsers);
       expect(mockMailjetApi.bulkSyncUsers(anyObject(), eq(MailJetSubscriptionAction.FORCE_SUBSCRIBE),
           eq(MailJetSubscriptionAction.FORCE_SUBSCRIBE))).andReturn("job123");
       // Job completes successfully with no errors
@@ -93,7 +94,7 @@ void synchroniseChangedUsers_WithMixedSubscriptionPreferences_ShouldGroupByPrefe
           )
       );
 
-      expect(mockDatabase.getRecentlyChangedRecords()).andReturn(changedUsers);
+      expect(mockDatabase.getRecentlyChangedRecords(anyInt())).andReturn(changedUsers);
       // First bulk call for group (FORCE_SUBSCRIBE, FORCE_SUBSCRIBE)
       expect(mockMailjetApi.bulkSyncUsers(anyObject(), eq(MailJetSubscriptionAction.FORCE_SUBSCRIBE),
           eq(MailJetSubscriptionAction.FORCE_SUBSCRIBE))).andReturn("job123");
@@ -130,7 +131,7 @@ void synchroniseChangedUsers_WithDeletedUser_ShouldDeleteIndividually()
           )
       );
 
-      expect(mockDatabase.getRecentlyChangedRecords()).andReturn(changedUsers);
+      expect(mockDatabase.getRecentlyChangedRecords(anyInt())).andReturn(changedUsers);
       mockMailjetApi.permanentlyDeleteAccountById("mailjetId123");
       expectLastCall();
       mockDatabase.updateExternalAccount(1L, null);
@@ -161,7 +162,7 @@ void synchroniseChangedUsers_WithDeliveryFailedUser_ShouldGroupAsRemove()
           )
       );
 
-      expect(mockDatabase.getRecentlyChangedRecords()).andReturn(changedUsers);
+      expect(mockDatabase.getRecentlyChangedRecords(anyInt())).andReturn(changedUsers);
       // Delivery failed users should call with REMOVE for both news and events
       expect(mockMailjetApi.bulkSyncUsers(anyObject(), eq(MailJetSubscriptionAction.REMOVE),
           eq(MailJetSubscriptionAction.REMOVE))).andReturn("job123");
@@ -184,7 +185,7 @@ void synchroniseChangedUsers_WithDeliveryFailedUser_ShouldGroupAsRemove()
     void synchroniseChangedUsers_WithEmptyUserList_ShouldReturnWithoutError()
         throws SegueDatabaseException, ExternalAccountSynchronisationException {
       // Arrange
-      expect(mockDatabase.getRecentlyChangedRecords()).andReturn(List.of());
+      expect(mockDatabase.getRecentlyChangedRecords(anyInt())).andReturn(List.of());
 
       replay(mockDatabase, mockMailjetApi);
 
@@ -199,7 +200,7 @@ void synchroniseChangedUsers_WithEmptyUserList_ShouldReturnWithoutError()
     @Test
     void synchroniseChangedUsers_WithDatabaseException_ShouldThrow() throws SegueDatabaseException {
       // Arrange
-      expect(mockDatabase.getRecentlyChangedRecords())
+      expect(mockDatabase.getRecentlyChangedRecords(anyInt()))
           .andThrow(new SegueDatabaseException("Database error"));
 
       replay(mockDatabase);
@@ -222,7 +223,7 @@ void synchroniseChangedUsers_WithJobErrorsButUserDataCorrect_ShouldMarkAsSynced(
           )
       );
 
-      expect(mockDatabase.getRecentlyChangedRecords()).andReturn(changedUsers);
+      expect(mockDatabase.getRecentlyChangedRecords(anyInt())).andReturn(changedUsers);
       expect(mockMailjetApi.bulkSyncUsers(anyObject(), anyObject(), anyObject()))
           .andReturn("job123");
       // Job has 1 error
@@ -261,7 +262,7 @@ void synchroniseChangedUsers_WithJobErrorsAndUserNotFound_ShouldNotSyncUser()
           )
       );
 
-      expect(mockDatabase.getRecentlyChangedRecords()).andReturn(changedUsers);
+      expect(mockDatabase.getRecentlyChangedRecords(anyInt())).andReturn(changedUsers);
       expect(mockMailjetApi.bulkSyncUsers(anyObject(), anyObject(), anyObject()))
           .andReturn("job123");
       // Job has 1 error
@@ -293,7 +294,7 @@ void synchroniseChangedUsers_WithMailjetException_ShouldLogAndContinue()
           )
       );
 
-      expect(mockDatabase.getRecentlyChangedRecords()).andReturn(changedUsers);
+      expect(mockDatabase.getRecentlyChangedRecords(anyInt())).andReturn(changedUsers);
       expect(mockMailjetApi.bulkSyncUsers(anyObject(), anyObject(), anyObject()))
           .andThrow(new MailjetException("Mailjet error"));
 
@@ -317,7 +318,7 @@ void synchroniseChangedUsers_WithCommunicationException_ShouldThrow()
           )
       );
 
-      expect(mockDatabase.getRecentlyChangedRecords()).andReturn(changedUsers);
+      expect(mockDatabase.getRecentlyChangedRecords(anyInt())).andReturn(changedUsers);
       expect(mockMailjetApi.bulkSyncUsers(anyObject(), anyObject(), anyObject()))
           .andThrow(new MailjetClientCommunicationException("Communication error"));
 
@@ -341,7 +342,7 @@ void synchroniseChangedUsers_WithRateLimitException_ShouldThrow()
           )
       );
 
-      expect(mockDatabase.getRecentlyChangedRecords()).andReturn(changedUsers);
+      expect(mockDatabase.getRecentlyChangedRecords(anyInt())).andReturn(changedUsers);
       expect(mockMailjetApi.bulkSyncUsers(anyObject(), anyObject(), anyObject()))
           .andThrow(new MailjetRateLimitException("Rate limit exceeded"));
 
@@ -368,7 +369,7 @@ void synchroniseChangedUsers_WithSingleRateLimitDuringPolling_ShouldContinuePoll
           )
       );
 
-      expect(mockDatabase.getRecentlyChangedRecords()).andReturn(changedUsers);
+      expect(mockDatabase.getRecentlyChangedRecords(anyInt())).andReturn(changedUsers);
       expect(mockMailjetApi.bulkSyncUsers(anyObject(), anyObject(), anyObject()))
           .andReturn("job123");
       // First poll hits rate limit, second succeeds
@@ -400,7 +401,7 @@ void synchroniseChangedUsers_WithRepeatedRateLimitDuringPolling_ShouldFailFast()
           )
       );
 
-      expect(mockDatabase.getRecentlyChangedRecords()).andReturn(changedUsers);
+      expect(mockDatabase.getRecentlyChangedRecords(anyInt())).andReturn(changedUsers);
       expect(mockMailjetApi.bulkSyncUsers(anyObject(), anyObject(), anyObject()))
           .andReturn("job123");
       // Both polls hit rate limit

diff --git a/src/test/java/uk/ac/cam/cl/dtg/segue/dao/users/PgExternalAccountPersistenceManagerTest.java b/src/test/java/uk/ac/cam/cl/dtg/segue/dao/users/PgExternalAccountPersistenceManagerTest.java
@@ -75,7 +75,7 @@ void getRecentlyChangedRecords_WithValidData_ShouldReturnUserList() throws Excep
       replay(mockDatabase, mockConnection, mockPreparedStatement, mockResultSet);
 
       // Act
-      List<UserExternalAccountChanges> result = persistenceManager.getRecentlyChangedRecords();
+      List<UserExternalAccountChanges> result = persistenceManager.getRecentlyChangedRecords(5000);
 
       // Assert
       verify(mockDatabase, mockConnection, mockPreparedStatement, mockResultSet);
@@ -105,7 +105,7 @@ void getRecentlyChangedRecords_WithEmptyResults_ShouldReturnEmptyList() throws E
       replay(mockDatabase, mockConnection, mockPreparedStatement, mockResultSet);
 
       // Act
-      List<UserExternalAccountChanges> result = persistenceManager.getRecentlyChangedRecords();
+      List<UserExternalAccountChanges> result = persistenceManager.getRecentlyChangedRecords(5000);
 
       // Assert
       verify(mockDatabase, mockConnection, mockPreparedStatement, mockResultSet);
@@ -121,7 +121,7 @@ void getRecentlyChangedRecords_WithDatabaseError_ShouldThrowException() throws E
 
       // Act & Assert
       assertThrows(SegueDatabaseException.class,
-          () -> persistenceManager.getRecentlyChangedRecords());
+          () -> persistenceManager.getRecentlyChangedRecords(5000));
 
       verify(mockDatabase);
     }