Skip to content

Commit e6d5bfa

Browse files
thugharithughari
committed
updated caches and conflicts
1 parent 764e6f6 commit e6d5bfa

File tree

9 files changed

+253
-112
lines changed

9 files changed

+253
-112
lines changed

‎backend/src/main/java/com/thughari/jobtrackerpro/dto/EmailBatchItem.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,5 @@
11
package com.thughari.jobtrackerpro.dto;
22

3-
public record EmailBatchItem(String from, String subject, String body) {}
3+
import java.time.LocalDateTime;
4+
5+
public record EmailBatchItem(String from, String subject, String replyTo, String body, LocalDateTime receivedDate) {}

‎backend/src/main/java/com/thughari/jobtrackerpro/repo/UserRepository.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,8 @@
22

33
import com.thughari.jobtrackerpro.entity.User;
44

5+
import jakarta.transaction.Transactional;
6+
57
import org.springframework.cache.annotation.Cacheable;
68
import org.springframework.data.jpa.repository.JpaRepository;
79
import org.springframework.data.jpa.repository.Modifying;
@@ -19,10 +21,12 @@ public interface UserRepository extends JpaRepository<User, UUID> {
1921
Optional<User> findByEmail(String email);
2022

2123
@Modifying
24+
@Transactional
2225
@Query("UPDATE User u SET u.gmailSyncInProgress = true WHERE u.email = :email AND u.gmailSyncInProgress = false")
2326
int claimSyncLock(@Param("email") String email);
2427

2528
@Modifying
29+
@Transactional
2630
@Query("UPDATE User u SET u.gmailSyncInProgress = false WHERE u.email = :email")
2731
void releaseSyncLock(@Param("email") String email);
2832

‎backend/src/main/java/com/thughari/jobtrackerpro/service/GeminiExtractionService.java‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -113,6 +113,7 @@ private String buildBatchPrompt(List<EmailBatchItem> items) {
113113
emailListBuilder.append("""
114114
--- EMAIL INDEX: %d ---
115115
FROM: %s
116+
REPLY-TO: %s
116117
SUBJECT: %s
117118
BODY: %s
118119
@@ -122,6 +123,7 @@ private String buildBatchPrompt(List<EmailBatchItem> items) {
122123
""".formatted(
123124
i,
124125
item.from(),
126+
item.replyTo(),
125127
item.subject(),
126128
safeBody,
127129
buildUrlIndexList(urls)
@@ -288,6 +290,28 @@ private String buildBatchPrompt(List<EmailBatchItem> items) {
288290
careers@stripe.com → Stripe
289291
talent.wayfair.com → Wayfair
290292
293+
Emails may be sent by recruiting platforms such as: Naukri, Talent500, LinkedIn, Hired, Wellfound, Indeed, etc.
294+
295+
These platforms are NOT the hiring company.
296+
297+
If a recruiting platform is mentioned, identify the actual employer
298+
mentioned in the job description or company section.
299+
300+
Example:
301+
Email from: Talent500
302+
Job description mentions: Albertsons Companies
303+
304+
Correct company: Albertsons
305+
306+
If the email contains a Reply-To header, and the domain appears to be a company domain, prefer that domain over recruiting platforms.
307+
308+
Example:
309+
310+
From: messages.naukri.com
311+
Reply-To: recruiter@yupptv.com
312+
313+
Company = YuppTV
314+
291315
Ignore generic domains:
292316
gmail, yahoo, outlook, etc.
293317

‎backend/src/main/java/com/thughari/jobtrackerpro/service/GmailIntegrationService.java‎

Lines changed: 16 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -15,19 +15,21 @@
1515
import com.thughari.jobtrackerpro.exception.ResourceNotFoundException;
1616
import com.thughari.jobtrackerpro.interfaces.GeminiService;
1717
import com.thughari.jobtrackerpro.repo.UserRepository;
18+
import com.thughari.jobtrackerpro.util.CacheEvictService;
1819
import com.thughari.jobtrackerpro.util.UrlParser;
1920

2021
import jakarta.transaction.Transactional;
2122
import lombok.extern.slf4j.Slf4j;
2223
import org.springframework.beans.factory.annotation.Value;
23-
import org.springframework.cache.Cache;
24-
import org.springframework.cache.CacheManager;
2524
import org.springframework.cache.annotation.CacheEvict;
2625
import org.springframework.cache.annotation.Caching;
2726
import org.springframework.scheduling.annotation.Async;
2827
import org.springframework.stereotype.Service;
2928
import org.springframework.web.client.RestClient;
3029

30+
import java.time.Instant;
31+
import java.time.LocalDateTime;
32+
import java.time.ZoneOffset;
3133
import java.util.ArrayList;
3234
import java.util.List;
3335

@@ -39,7 +41,7 @@ public class GmailIntegrationService {
3941
private final GeminiService geminiService;
4042
private final JobService jobService;
4143
private final RestClient restClient;
42-
private final CacheManager cacheManager;
44+
private final CacheEvictService cacheEvictService;
4345
private final String APPLICATION_NAME = "JobTrackerPro";
4446

4547
private static final NetHttpTransport HTTP_TRANSPORT;
@@ -66,12 +68,12 @@ public class GmailIntegrationService {
6668
@Value("${app.google.pubsub-topic}")
6769
private String pubsubTopic;
6870

69-
public GmailIntegrationService(UserRepository userRepository, GeminiService geminiService, JobService jobService, CacheManager cacheManager) {
71+
public GmailIntegrationService(UserRepository userRepository, GeminiService geminiService, JobService jobService, CacheEvictService cacheEvictService) {
7072
this.userRepository = userRepository;
7173
this.geminiService = geminiService;
7274
this.jobService = jobService;
7375
this.restClient = RestClient.create();
74-
this.cacheManager = cacheManager;
76+
this.cacheEvictService = cacheEvictService;
7577
}
7678

7779
@Transactional
@@ -121,14 +123,14 @@ public void connectAndSetupPush(String authCode, String email) throws Exception
121123
}
122124

123125
@Async("taskExecutor")
124-
@Transactional
125126
public void initiateManualSync(String email) {
126127

127128
int updatedRows = userRepository.claimSyncLock(email);
128129

129130
if (updatedRows == 0) {
130131
return;
131132
}
133+
cacheEvictService.evictAllForUser(email);
132134
try {
133135
User user = userRepository.findByEmail(email)
134136
.orElseThrow(() -> new RuntimeException("User not connected to Gmail"));
@@ -145,15 +147,14 @@ public void initiateManualSync(String email) {
145147
Gmail service = createGmailClient(accessToken);
146148
String currentHistoryId = service.users().getProfile("me").execute().getHistoryId().toString();
147149

148-
user.setGmailHistoryId(currentHistoryId);
149-
userRepository.saveAndFlush(user);
150+
jobService.finalizeManualSync(email, currentHistoryId);
150151

151152
log.info("Manual sync finished for {}. Found {} jobs.", email, found);
152153
} catch (Exception e) {
153154
log.error("Manual sync failed for {}: {}", email, e.getMessage());
154155
} finally {
155156
userRepository.releaseSyncLock(email);
156-
evictUserCaches(email);
157+
cacheEvictService.evictAllForUser(email);
157158
}
158159
}
159160

@@ -198,18 +199,22 @@ public int scanInbox(String accessToken, String userEmail) {
198199
if (response.getMessages() != null) {
199200
for (Message msg : response.getMessages()) {
200201
Message fullMsg = service.users().messages().get("me", msg.getId()).setFormat("full").execute();
202+
long millisecondTimestamp = fullMsg.getInternalDate();
203+
LocalDateTime emailDate = LocalDateTime.ofInstant(
204+
Instant.ofEpochMilli(millisecondTimestamp), ZoneOffset.UTC);
201205

202-
String from = "", subj = "";
206+
String from = "", subj = "", replyTo="";
203207
if (fullMsg.getPayload().getHeaders() != null) {
204208
for (var h : fullMsg.getPayload().getHeaders()) {
205209
if ("From".equalsIgnoreCase(h.getName())) from = h.getValue();
206210
if ("Subject".equalsIgnoreCase(h.getName())) subj = h.getValue();
211+
if ("Reply-To".equalsIgnoreCase(h.getName())) replyTo = h.getValue();
207212
}
208213
}
209214

210215
if (!isSystemNoise(subj)) {
211216
String body = extractTextFromBody(fullMsg.getPayload());
212-
batchItems.add(new EmailBatchItem(from, subj, body));
217+
batchItems.add(new EmailBatchItem(from, subj, replyTo, body, emailDate));
213218
}
214219
}
215220
}
@@ -271,18 +276,6 @@ private void hydrateJobUrl(JobDTO job, List<List<String>> urlMaps) {
271276
}
272277
sanitizeUrl(job);
273278
}
274-
275-
private void evictUserCaches(String email) {
276-
Cache userCache = cacheManager.getCache("users");
277-
Cache entityCache = cacheManager.getCache("userEntities");
278-
Cache jobList = cacheManager.getCache("jobList");
279-
Cache jobDashboard = cacheManager.getCache("jobDashboard");
280-
281-
if (userCache != null) userCache.evict(email);
282-
if (entityCache != null) entityCache.evict(email);
283-
if (jobList != null) jobList.evict(email);
284-
if (jobDashboard != null) jobDashboard.evict(email);
285-
}
286279

287280
private void sanitizeUrl(JobDTO job) {
288281
if (job.getUrl() != null) {

‎backend/src/main/java/com/thughari/jobtrackerpro/service/GmailWebhookService.java‎

Lines changed: 32 additions & 64 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@
1010
import com.thughari.jobtrackerpro.entity.User;
1111
import com.thughari.jobtrackerpro.interfaces.GeminiService;
1212
import com.thughari.jobtrackerpro.repo.UserRepository;
13+
import com.thughari.jobtrackerpro.util.CacheEvictService;
1314
import com.thughari.jobtrackerpro.util.UrlParser;
1415

1516
import lombok.extern.slf4j.Slf4j;
@@ -21,6 +22,9 @@
2122
import org.springframework.transaction.annotation.Transactional;
2223

2324
import java.math.BigInteger;
25+
import java.time.Instant;
26+
import java.time.LocalDateTime;
27+
import java.time.ZoneOffset;
2428
import java.util.ArrayList;
2529
import java.util.Base64;
2630
import java.util.List;
@@ -32,30 +36,29 @@ public class GmailWebhookService {
3236
private final GeminiService geminiService;
3337
private final JobService jobService;
3438
private final UserRepository userRepository;
35-
private final CacheManager cacheManager;
39+
private final CacheEvictService cacheEvictService;
3640

3741
@Value("${spring.security.oauth2.client.registration.google.client-id}")
3842
private String clientId;
3943

4044
@Value("${spring.security.oauth2.client.registration.google.client-secret}")
4145
private String clientSecret;
4246

43-
public GmailWebhookService(GeminiService geminiService, JobService jobService, UserRepository userRepository, CacheManager cacheManager) {
47+
public GmailWebhookService(GeminiService geminiService, JobService jobService, UserRepository userRepository, CacheEvictService cacheEvictService) {
4448
this.geminiService = geminiService;
4549
this.jobService = jobService;
4650
this.userRepository = userRepository;
47-
this.cacheManager = cacheManager;
51+
this.cacheEvictService = cacheEvictService;
4852
}
4953

5054
@Async("taskExecutor")
51-
@Transactional
5255
public void processHistorySync(String userEmail) {
5356
final String email = userEmail.toLowerCase();
5457

5558
int updatedRows = userRepository.claimSyncLock(email);
56-
if (updatedRows == 0) return;
57-
58-
evictUserCaches(email);
59+
if (updatedRows == 0) return;
60+
61+
cacheEvictService.evictAllForUser(email);
5962

6063
try {
6164
User user = userRepository.findByEmail(email)
@@ -84,59 +87,19 @@ public void processHistorySync(String userEmail) {
8487

8588
List<EmailBatchItem> batchItems = collectMessages(service, historyResponse.getHistory());
8689

87-
if (!batchItems.isEmpty()) {log.info("Ingesting batch of {} emails via Gemini for {}", batchItems.size(), email);
88-
89-
List<List<String>> batchUrlLists = batchItems.stream()
90-
.map(item -> UrlParser.extractAndCleanUrls(item.body()))
91-
.toList();
92-
93-
List<JobDTO> extractedJobs = geminiService.extractJobsFromBatch(batchItems);
94-
95-
for (JobDTO job : extractedJobs) {
96-
Integer idx = job.getInputIndex();
97-
98-
if (idx != null && idx >= 0 && idx < batchUrlLists.size()) {
99-
List<String> originalUrls = batchUrlLists.get(idx);
100-
101-
if (job.getUrlIndex() != null && job.getUrlIndex() >= 0 && job.getUrlIndex() < originalUrls.size()) {
102-
job.setUrl(originalUrls.get(job.getUrlIndex()));
103-
}
104-
else if (job.getUrl() == null || job.getUrl().isEmpty()) {
105-
job.setUrl(originalUrls.stream()
106-
.filter(u -> {
107-
String lower = u.toLowerCase();
108-
return lower.contains("career") ||
109-
lower.contains("job") ||
110-
lower.contains("apply") ||
111-
lower.contains("/jobs/") ||
112-
lower.contains("/careers/");
113-
})
114-
.findFirst().orElse(""));
115-
}
116-
}
117-
118-
if (job.getUrl() != null) {
119-
String lower = job.getUrl().toLowerCase();
120-
if (lower.contains("unsubscribe") ||
121-
lower.contains("privacy") ||
122-
lower.contains("help") ||
123-
lower.contains("settings")) {
124-
job.setUrl("");
125-
}
126-
}
127-
128-
job.setUrlIndex(null);
129-
job.setInputIndex(null);
130-
131-
jobService.createOrUpdateJob(job, email);
132-
}
133-
}
134-
90+
if (!batchItems.isEmpty()) {
91+
92+
List<JobDTO> extractedJobs = geminiService.extractJobsFromBatch(batchItems);
93+
94+
log.info("Ingesting batch of {} emails via Gemini for {}", batchItems.size(), email);
95+
96+
jobService.saveBatchResults(email, batchItems, extractedJobs);
97+
}
13598
} catch (Exception e) {
13699
log.error("High-Performance Sync failed for {}: ", email, e);
137100
} finally {
138101
userRepository.releaseSyncLock(email);
139-
evictUserCaches(email);
102+
cacheEvictService.evictAllForUser(email);
140103
}
141104
}
142105

@@ -151,15 +114,20 @@ private List<EmailBatchItem> collectMessages(Gmail service, List<History> histor
151114
Message m = service.users().messages().get("me", added.getMessage().getId())
152115
.setFormat("full").execute();
153116

154-
String from = "", subj = "";
117+
long millisecondTimestamp = m.getInternalDate();
118+
LocalDateTime emailDate = LocalDateTime.ofInstant(
119+
Instant.ofEpochMilli(millisecondTimestamp), ZoneOffset.UTC);
120+
121+
String from = "", subj = "", replyTo="";
155122
for (var h : m.getPayload().getHeaders()) {
156123
if ("From".equalsIgnoreCase(h.getName())) from = h.getValue();
157124
if ("Subject".equalsIgnoreCase(h.getName())) subj = h.getValue();
125+
if ("Reply-To".equalsIgnoreCase(h.getName())) replyTo = h.getValue();
158126
}
159127

160128
if (!isSystemNoise(subj)) {
161129
String body = extractTextFromBody(m.getPayload());
162-
items.add(new EmailBatchItem(from, subj, body));
130+
items.add(new EmailBatchItem(from, subj, replyTo, body, emailDate));
163131
}
164132
} catch (Exception e) {
165133
log.warn("Failed to fetch message {}: {}", added.getMessage().getId(), e.getMessage());
@@ -197,12 +165,12 @@ private boolean isSystemNoise(String subject) {
197165
return s.contains("security alert") || s.contains("sign-in") || s.contains("verification code");
198166
}
199167

200-
private void evictUserCaches(String email) {
201-
Cache userCache = cacheManager.getCache("users");
202-
Cache entityCache = cacheManager.getCache("userEntities");
203-
if (userCache != null) userCache.evict(email);
204-
if (entityCache != null) entityCache.evict(email);
205-
}
168+
// private void evictUserCaches(String email) {
169+
// Cache userCache = cacheManager.getCache("users");
170+
// Cache entityCache = cacheManager.getCache("userEntities");
171+
// if (userCache != null) userCache.evict(email);
172+
// if (entityCache != null) entityCache.evict(email);
173+
// }
206174

207175
public String getFreshAccessToken(String refreshToken) throws Exception {
208176
return new GoogleRefreshTokenRequest(GoogleNetHttpTransport.newTrustedTransport(), GsonFactory.getDefaultInstance(),

0 commit comments

Comments
 (0)