using InboxIntel.Application.Abstractions;
using InboxIntel.Application.DTOs;
using InboxIntel.Domain.Entities;
using InboxIntel.Domain.Enums;
using InboxIntel.Infrastructure.Configuration;
using InboxIntel.Infrastructure.Gmail;
using InboxIntel.Infrastructure.Persistence;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
namespace InboxIntel.Infrastructure.Sync;
///
/// Orchestrates Gmail synchronisation. Full sync pages through every message;
/// incremental sync replays the Gmail history feed since the last historyId.
/// Progress is checkpointed to so an interrupted run
/// resumes from its last page token instead of restarting.
///
public class SyncService : ISyncService
{
private readonly AppDbContext _db;
private readonly IGmailService _gmail;
private readonly ILogger _logger;
private readonly GmailSyncOptions _options;
private readonly ISyncQueue _queue;
private readonly IAiService _ai;
public SyncService(AppDbContext db, IGmailService gmail, ILogger logger, IOptions options, ISyncQueue queue, IAiService ai)
{
_db = db;
_gmail = gmail;
_logger = logger;
_options = options.Value;
_queue = queue;
_ai = ai;
}
public async Task QueueSyncAsync(Guid userId, bool fullSync, CancellationToken ct = default)
{
// Flip to Running synchronously so the UI splash shows immediately,
// then hand the actual work to the background worker.
var state = await GetOrCreateStateAsync(userId, ct);
state.Status = SyncStatus.Running;
state.LastSyncType = fullSync ? SyncType.Full : SyncType.Incremental;
state.StartedUtc = DateTimeOffset.UtcNow;
state.LastError = null;
if (fullSync) state.MessagesProcessed = 0;
await _db.SaveChangesAsync(ct);
_queue.Enqueue(userId, fullSync);
}
public async Task GetStatusAsync(Guid userId, CancellationToken ct = default)
=> (await GetOrCreateStateAsync(userId, ct)).Status;
public async Task GetProgressAsync(Guid userId, CancellationToken ct = default)
{
var s = await GetOrCreateStateAsync(userId, ct);
// Total is unknown (0 -> indeterminate bar) until Gmail returns the real
// count; then it's the mailbox total, capped by MaxMessages in dev.
var total = s.TotalMessagesEstimate;
if (_options.MaxMessages > 0 && total > 0)
total = Math.Min(total, _options.MaxMessages);
return new SyncProgressDto(
s.Status.ToString(), s.MessagesProcessed, total,
s.Status == SyncStatus.Running, s.LastSuccessfulSyncUtc, s.LastError);
}
public async Task RunFullSyncAsync(Guid userId, CancellationToken ct = default)
{
var state = await GetOrCreateStateAsync(userId, ct);
state.Status = SyncStatus.Running;
state.LastSyncType = SyncType.Full;
state.StartedUtc = DateTimeOffset.UtcNow;
state.LastError = null;
await _db.SaveChangesAsync(ct);
try
{
await SyncLabelsAsync(userId, ct);
// Dev cap: stop after MaxMessages (most recent first). 0 = unlimited.
var maxMessages = _options.MaxMessages;
// Phase 1 — enumerate message ids (cheap, ids only) to get an ACCURATE
// total. Gmail's resultSizeEstimate is unreliable, so we count instead.
var ids = new List();
string? listToken = null;
do
{
var page = await _gmail.ListMessageIdsAsync(userId, listToken, ct);
foreach (var id in page.MessageIds)
{
ids.Add(id);
if (maxMessages > 0 && ids.Count >= maxMessages) break;
}
listToken = page.NextPageToken;
}
while (listToken is not null && (maxMessages == 0 || ids.Count < maxMessages) && !ct.IsCancellationRequested);
state.TotalMessagesEstimate = ids.Count; // the true target (capped in dev)
await _db.SaveChangesAsync(ct);
// Phase 2 — fetch bodies in chunks, in parallel, skipping ones we already
// have. The DbContext is not thread-safe, so upserts run sequentially.
const int chunkSize = 100;
for (var i = 0; i < ids.Count && !ct.IsCancellationRequested; i += chunkSize)
{
var chunk = ids.GetRange(i, Math.Min(chunkSize, ids.Count - i));
var have = (await _db.Emails
.Where(e => e.UserId == userId && chunk.Contains(e.GmailMessageId))
.Select(e => e.GmailMessageId).ToListAsync(ct)).ToHashSet();
state.MessagesProcessed += have.Count; // already-synced count toward progress
var toFetch = chunk.Where(id => !have.Contains(id)).ToList();
var details = await FetchDetailsParallelAsync(userId, toFetch, ct);
foreach (var detail in details)
{
await UpsertMessageAsync(userId, detail, ct);
state.MessagesProcessed++;
if (state.MessagesProcessed % 25 == 0)
await _db.SaveChangesAsync(ct); // smooth progress for the splash
}
await _db.SaveChangesAsync(ct);
}
state.LastHistoryId = await _gmail.GetProfileHistoryIdAsync(userId, ct);
state.ResumePageToken = null;
state.Status = SyncStatus.Completed;
state.CompletedUtc = DateTimeOffset.UtcNow;
state.LastSuccessfulSyncUtc = DateTimeOffset.UtcNow;
state.ConsecutiveFailures = 0;
await _db.SaveChangesAsync(ct);
}
catch (Exception ex)
{
await MarkFailedAsync(state, ex, ct);
throw;
}
}
public async Task RunIncrementalSyncAsync(Guid userId, CancellationToken ct = default)
{
var state = await GetOrCreateStateAsync(userId, ct);
if (string.IsNullOrEmpty(state.LastHistoryId))
{
await RunFullSyncAsync(userId, ct); // no watermark yet -> full sync
return;
}
state.Status = SyncStatus.Running;
state.LastSyncType = SyncType.Incremental;
state.StartedUtc = DateTimeOffset.UtcNow;
await _db.SaveChangesAsync(ct);
try
{
string? pageToken = null;
string? newHistoryId = state.LastHistoryId;
do
{
var page = await _gmail.ListHistoryAsync(userId, state.LastHistoryId!, pageToken, ct);
foreach (var id in page.ChangedMessageIds.Distinct())
{
var detail = await _gmail.GetMessageAsync(userId, id, ct);
await UpsertMessageAsync(userId, detail, ct);
}
foreach (var id in page.DeletedMessageIds.Distinct())
{
var existing = await _db.Emails.FirstOrDefaultAsync(e => e.UserId == userId && e.GmailMessageId == id, ct);
if (existing is not null) _db.Emails.Remove(existing);
}
if (page.NewHistoryId is not null) newHistoryId = page.NewHistoryId;
pageToken = page.NextPageToken;
await _db.SaveChangesAsync(ct);
}
while (pageToken is not null && !ct.IsCancellationRequested);
state.LastHistoryId = newHistoryId;
state.Status = SyncStatus.Completed;
state.CompletedUtc = DateTimeOffset.UtcNow;
state.LastSuccessfulSyncUtc = DateTimeOffset.UtcNow;
state.ConsecutiveFailures = 0;
await _db.SaveChangesAsync(ct);
}
catch (Exception ex)
{
await MarkFailedAsync(state, ex, ct);
throw;
}
}
private async Task MarkFailedAsync(SyncState state, Exception ex, CancellationToken ct)
{
_logger.LogError(ex, "Sync failed for user {UserId}", state.UserId);
// Discard the failed batch's pending changes; otherwise saving the
// failure status re-attempts the same bad inserts and throws again.
_db.ChangeTracker.Clear();
var fresh = await _db.SyncStates.FirstOrDefaultAsync(s => s.UserId == state.UserId, ct);
if (fresh is null) return;
fresh.Status = SyncStatus.Failed;
fresh.ConsecutiveFailures++;
fresh.LastError = ex.Message;
await _db.SaveChangesAsync(ct);
}
private async Task GetOrCreateStateAsync(Guid userId, CancellationToken ct)
{
var state = await _db.SyncStates.FirstOrDefaultAsync(s => s.UserId == userId, ct);
if (state is null)
{
state = new SyncState { UserId = userId };
_db.SyncStates.Add(state);
await _db.SaveChangesAsync(ct);
}
return state;
}
private async Task SyncLabelsAsync(Guid userId, CancellationToken ct)
{
var remote = await _gmail.ListLabelsAsync(userId, ct);
foreach (var label in remote)
{
var existing = await _db.Labels.FirstOrDefaultAsync(l => l.UserId == userId && l.GmailLabelId == label.GmailLabelId, ct);
if (existing is null) _db.Labels.Add(label);
else { existing.Name = label.Name; existing.Type = label.Type; }
}
await _db.SaveChangesAsync(ct);
}
// Per-sync-run cache of the user's GmailLabelId -> local Label.Id, so email/label linkage
// costs no extra query per message. Populated lazily; labels are synced before messages.
private Dictionary? _labelCache;
private Guid _labelCacheUserId;
private async Task> GetLabelMapAsync(Guid userId, CancellationToken ct)
{
if (_labelCache is null || _labelCacheUserId != userId)
{
_labelCache = await _db.Labels
.Where(l => l.UserId == userId)
.ToDictionaryAsync(l => l.GmailLabelId, l => l.Id, ct);
_labelCacheUserId = userId;
}
return _labelCache;
}
/// Resolves domain/sender/thread, then upserts the email, its labels, and
/// attachment metadata. Idempotent: re-syncing a message replaces the prior row (and its
/// labels) rather than duplicating it.
private async Task UpsertMessageAsync(Guid userId, GmailMessageDetail d, CancellationToken ct)
{
// True upsert: drop any existing copy first so labels/flags re-populate cleanly and
// an incrementally-changed message can't be duplicated. EmailLabels/Attachments cascade.
var prior = await _db.Emails.FirstOrDefaultAsync(
e => e.UserId == userId && e.GmailMessageId == d.GmailMessageId, ct);
if (prior is not null) _db.Emails.Remove(prior);
var sender = await ResolveSenderAsync(userId, d.FromAddress, d.FromDisplayName, ct);
var thread = await ResolveThreadAsync(userId, d.GmailThreadId, d.Subject, d.Snippet, d.SentAtUtc, ct);
var category = HeuristicClassifier.Classify(d);
// The heuristic rule set falls back to Personal when nothing more specific
// matches; if AI is enabled, give it a shot at a better category.
if (category == EmailCategory.Personal && _ai.IsEnabled)
category = await _ai.ClassifyFallbackAsync(d.Subject, d.FromAddress, d.Snippet, ct);
var email = new Email
{
UserId = userId,
GmailMessageId = d.GmailMessageId,
ThreadId = thread.Id,
SenderId = sender.Id,
Subject = Trunc(d.Subject, 1024),
Snippet = Trunc(d.Snippet, 2048),
BodyText = d.BodyText, // unlimited (text column)
SentAtUtc = d.SentAtUtc,
ReceivedAtUtc = d.SentAtUtc,
SizeEstimateBytes = d.SizeEstimateBytes,
IsUnread = d.IsUnread,
IsInInbox = d.LabelIds.Contains("INBOX"),
IsStarred = d.LabelIds.Contains("STARRED"),
IsImportant = d.LabelIds.Contains("IMPORTANT"),
IsTrashed = d.LabelIds.Contains("TRASH"),
HasAttachments = d.HasAttachments,
HasListUnsubscribe = d.HasListUnsubscribe,
ListUnsubscribeRaw = Trunc(d.ListUnsubscribeRaw, 2048),
SupportsOneClickUnsubscribe = d.SupportsOneClickUnsubscribe,
Category = category
};
_db.Emails.Add(email);
// Link the email to its Gmail labels (system + user) so label-based folders — Sent,
// Spam, and "Unlabelled" (no user label) — resolve correctly. Previously no EmailLabel
// rows were ever created, so every label folder was empty and every email looked unlabelled.
var labelMap = await GetLabelMapAsync(userId, ct);
foreach (var gmailLabelId in d.LabelIds.Distinct())
if (labelMap.TryGetValue(gmailLabelId, out var localLabelId))
_db.Set().Add(new EmailLabel { EmailId = email.Id, LabelId = localLabelId });
foreach (var (fileName, mime, size, attId) in d.Attachments)
{
_db.Attachments.Add(new Attachment
{
UserId = userId,
EmailId = email.Id,
FileName = Trunc(fileName, 512) ?? string.Empty,
MimeType = Trunc(mime, 255),
SizeBytes = size,
GmailAttachmentId = attId
});
}
// Maintain sender rollups for fast grouping.
sender.EmailCount++;
if (d.IsUnread) sender.UnreadCount++;
sender.TotalSizeBytes += d.SizeEstimateBytes;
sender.LastReceivedUtc = d.SentAtUtc;
if (d.HasListUnsubscribe) sender.HasUnsubscribe = true;
thread.MessageCount++;
thread.LastMessageUtc = d.SentAtUtc;
}
///
/// Fetches message bodies concurrently (bounded by MaxParallelism) to speed
/// up the sync. Results only — the caller persists them sequentially.
///
private async Task> FetchDetailsParallelAsync(Guid userId, List ids, CancellationToken ct)
{
if (ids.Count == 0) return new List();
using var sem = new SemaphoreSlim(Math.Max(1, _options.MaxParallelism));
var tasks = ids.Select(async id =>
{
await sem.WaitAsync(ct);
try { return await _gmail.GetMessageAsync(userId, id, ct); }
finally { sem.Release(); }
});
var results = await Task.WhenAll(tasks);
return results.ToList();
}
private async Task ResolveSenderAsync(Guid userId, string address, string? displayName, CancellationToken ct)
{
// Check the in-memory tracker first so senders/domains added earlier in
// this batch (not yet saved) are reused instead of duplicated.
var sender = _db.Senders.Local.FirstOrDefault(s => s.UserId == userId && s.Address == address)
?? await _db.Senders.FirstOrDefaultAsync(s => s.UserId == userId && s.Address == address, ct);
if (sender is not null) return sender;
address = Trunc(address, 320)!;
var rawDomain = address.Contains('@') ? address[(address.IndexOf('@') + 1)..] : "unknown";
var domainName = Trunc(rawDomain, 255)!;
var domain = _db.Domains.Local.FirstOrDefault(x => x.UserId == userId && x.Name == domainName)
?? await _db.Domains.FirstOrDefaultAsync(x => x.UserId == userId && x.Name == domainName, ct);
if (domain is null)
{
domain = new MailDomain { UserId = userId, Name = domainName };
_db.Domains.Add(domain);
}
domain.EmailCount++;
sender = new Sender { UserId = userId, Address = address, DisplayName = Trunc(displayName, 255), Domain = domain, DomainId = domain.Id };
_db.Senders.Add(sender);
return sender;
}
/// Truncates a string to a column's max length so no email can overflow it.
private static string? Trunc(string? value, int max)
=> value is null ? null : value.Length <= max ? value : value[..max];
private async Task ResolveThreadAsync(Guid userId, string gmailThreadId, string? subject, string? snippet, DateTimeOffset sentAt, CancellationToken ct)
{
var thread = _db.Threads.Local.FirstOrDefault(t => t.UserId == userId && t.GmailThreadId == gmailThreadId)
?? await _db.Threads.FirstOrDefaultAsync(t => t.UserId == userId && t.GmailThreadId == gmailThreadId, ct);
if (thread is not null) return thread;
thread = new MailThread
{
UserId = userId,
GmailThreadId = gmailThreadId,
Subject = Trunc(subject, 1024),
Snippet = Trunc(snippet, 2048),
FirstMessageUtc = sentAt
};
_db.Threads.Add(thread);
return thread;
}
}