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 index 77af2fb26e..4057701109 100644 --- 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 getRecentlyChangedUsersOrThrow() throws ExternalAccountSynchronisationException { try { - List users = database.getRecentlyChangedRecords(); - log.info("{}Found {} users to synchronize with Mailjet", MAILJET, users.size()); + List 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 index 63db38df84..3a7b2016bd 100644 --- 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 getRecentlyChangedRecords() throws SegueDatabaseException; + List 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 index c78a5480cb..6adf9209ec 100644 --- 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 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 getRecentlyChangedRecords(final int limit) throws SegueDatabaseException { String query = "SELECT users.id, " + " external_accounts.provider_user_identifier, " + " users.email, " @@ -56,27 +74,28 @@ public List 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 index f716a4bebd..14e806632e 100644 --- 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 index 62feaf2534..4bd9705d84 100644 --- 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 index 74d8cf215e..4c79755040 100644 --- 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 result = persistenceManager.getRecentlyChangedRecords(); + List 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 result = persistenceManager.getRecentlyChangedRecords(); + List 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); }