chore: init project
This commit is contained in:
cesnimda
@@ -0,0 +1,74 @@
|
||||
using InboxIntel.Application.Abstractions;
|
||||
using InboxIntel.Infrastructure.Configuration;
|
||||
using InboxIntel.Infrastructure.Persistence;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Options;
|
||||
|
||||
namespace InboxIntel.Infrastructure.Sync;
|
||||
|
||||
/// <summary>
|
||||
/// Background worker that runs a daily incremental sync for every user and
|
||||
/// refreshes analytics aggregates. Non-blocking: it runs in its own scope and
|
||||
/// never touches the request pipeline. Failures are logged and retried on the
|
||||
/// next tick rather than crashing the host.
|
||||
/// </summary>
|
||||
public class GmailSyncWorker : BackgroundService
|
||||
{
|
||||
private readonly IServiceScopeFactory _scopeFactory;
|
||||
private readonly GmailSyncOptions _options;
|
||||
private readonly ILogger<GmailSyncWorker> _logger;
|
||||
|
||||
public GmailSyncWorker(IServiceScopeFactory scopeFactory, IOptions<GmailSyncOptions> options, ILogger<GmailSyncWorker> logger)
|
||||
{
|
||||
_scopeFactory = scopeFactory;
|
||||
_options = options.Value;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
_logger.LogInformation("GmailSyncWorker started; daily sync hour = {Hour}:00 UTC", _options.DailySyncHourUtc);
|
||||
|
||||
while (!stoppingToken.IsCancellationRequested)
|
||||
{
|
||||
try
|
||||
{
|
||||
var now = DateTimeOffset.UtcNow;
|
||||
if (now.Hour == _options.DailySyncHourUtc)
|
||||
await RunForAllUsersAsync(stoppingToken);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "GmailSyncWorker tick failed");
|
||||
}
|
||||
|
||||
// Re-evaluate hourly. A production deployment may swap this for Hangfire/cron.
|
||||
await Task.Delay(TimeSpan.FromHours(1), stoppingToken);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task RunForAllUsersAsync(CancellationToken ct)
|
||||
{
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var db = scope.ServiceProvider.GetRequiredService<AppDbContext>();
|
||||
var sync = scope.ServiceProvider.GetRequiredService<ISyncService>();
|
||||
var analytics = scope.ServiceProvider.GetRequiredService<IAnalyticsService>();
|
||||
|
||||
var userIds = await db.Users.Select(u => u.Id).ToListAsync(ct);
|
||||
foreach (var userId in userIds)
|
||||
{
|
||||
try
|
||||
{
|
||||
await sync.RunIncrementalSyncAsync(userId, ct);
|
||||
await analytics.RefreshAggregatesAsync(userId, ct);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "Scheduled sync failed for user {UserId}", userId);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,28 @@
|
||||
using InboxIntel.Application.Abstractions;
|
||||
using InboxIntel.Domain.Enums;
|
||||
|
||||
namespace InboxIntel.Infrastructure.Sync;
|
||||
|
||||
/// <summary>
|
||||
/// Fast, dependency-free first-pass classifier applied during sync. The AI
|
||||
/// layer can later refine these labels, but this guarantees every email has a
|
||||
/// sensible category even when AI is disabled.
|
||||
/// </summary>
|
||||
public static class HeuristicClassifier
|
||||
{
|
||||
private static readonly string[] FinanceHints = { "invoice", "receipt", "payment", "statement", "bank", "transaction", "billing" };
|
||||
private static readonly string[] SocialHints = { "facebook", "twitter", "linkedin", "instagram", "tiktok" };
|
||||
private static readonly string[] NoReplyHints = { "noreply", "no-reply", "donotreply", "newsletter", "mailer", "notifications" };
|
||||
|
||||
public static EmailCategory Classify(GmailMessageDetail d)
|
||||
{
|
||||
var subject = (d.Subject ?? string.Empty).ToLowerInvariant();
|
||||
var from = d.FromAddress.ToLowerInvariant();
|
||||
|
||||
if (d.HasListUnsubscribe) return EmailCategory.Newsletter;
|
||||
if (FinanceHints.Any(h => subject.Contains(h))) return EmailCategory.Finance;
|
||||
if (SocialHints.Any(h => from.Contains(h))) return EmailCategory.Social;
|
||||
if (NoReplyHints.Any(h => from.Contains(h))) return EmailCategory.Notification;
|
||||
return EmailCategory.Personal;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,246 @@
|
||||
using InboxIntel.Application.Abstractions;
|
||||
using InboxIntel.Domain.Entities;
|
||||
using InboxIntel.Domain.Enums;
|
||||
using InboxIntel.Infrastructure.Gmail;
|
||||
using InboxIntel.Infrastructure.Persistence;
|
||||
using Microsoft.EntityFrameworkCore;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace InboxIntel.Infrastructure.Sync;
|
||||
|
||||
/// <summary>
|
||||
/// Orchestrates Gmail synchronisation. Full sync pages through every message;
|
||||
/// incremental sync replays the Gmail history feed since the last historyId.
|
||||
/// Progress is checkpointed to <see cref="SyncState"/> so an interrupted run
|
||||
/// resumes from its last page token instead of restarting.
|
||||
/// </summary>
|
||||
public class SyncService : ISyncService
|
||||
{
|
||||
private readonly AppDbContext _db;
|
||||
private readonly IGmailService _gmail;
|
||||
private readonly ILogger<SyncService> _logger;
|
||||
|
||||
public SyncService(AppDbContext db, IGmailService gmail, ILogger<SyncService> logger)
|
||||
{
|
||||
_db = db;
|
||||
_gmail = gmail;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
public async Task<SyncStatus> GetStatusAsync(Guid userId, CancellationToken ct = default)
|
||||
=> (await GetOrCreateStateAsync(userId, ct)).Status;
|
||||
|
||||
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);
|
||||
|
||||
string? pageToken = state.ResumePageToken; // resume support
|
||||
do
|
||||
{
|
||||
var page = await _gmail.ListMessageIdsAsync(userId, pageToken, ct);
|
||||
foreach (var messageId in page.MessageIds)
|
||||
{
|
||||
if (await _db.Emails.AnyAsync(e => e.UserId == userId && e.GmailMessageId == messageId, ct))
|
||||
continue;
|
||||
var detail = await _gmail.GetMessageAsync(userId, messageId, ct);
|
||||
await UpsertMessageAsync(userId, detail, ct);
|
||||
state.MessagesProcessed++;
|
||||
}
|
||||
|
||||
pageToken = page.NextPageToken;
|
||||
state.ResumePageToken = pageToken; // checkpoint
|
||||
state.TotalMessagesEstimate = page.ResultSizeEstimate;
|
||||
await _db.SaveChangesAsync(ct);
|
||||
}
|
||||
while (pageToken is not null && !ct.IsCancellationRequested);
|
||||
|
||||
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);
|
||||
state.Status = SyncStatus.Failed;
|
||||
state.ConsecutiveFailures++;
|
||||
state.LastError = ex.Message;
|
||||
await _db.SaveChangesAsync(ct);
|
||||
}
|
||||
|
||||
private async Task<SyncState> 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);
|
||||
}
|
||||
|
||||
/// <summary>Resolves domain/sender/thread, then inserts the email and attachment metadata.</summary>
|
||||
private async Task UpsertMessageAsync(Guid userId, GmailMessageDetail d, CancellationToken ct)
|
||||
{
|
||||
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 email = new Email
|
||||
{
|
||||
UserId = userId,
|
||||
GmailMessageId = d.GmailMessageId,
|
||||
ThreadId = thread.Id,
|
||||
SenderId = sender.Id,
|
||||
Subject = d.Subject,
|
||||
Snippet = d.Snippet,
|
||||
BodyText = d.BodyText,
|
||||
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"),
|
||||
HasAttachments = d.HasAttachments,
|
||||
HasListUnsubscribe = d.HasListUnsubscribe,
|
||||
ListUnsubscribeRaw = d.ListUnsubscribeRaw,
|
||||
SupportsOneClickUnsubscribe = d.SupportsOneClickUnsubscribe,
|
||||
Category = HeuristicClassifier.Classify(d)
|
||||
};
|
||||
_db.Emails.Add(email);
|
||||
|
||||
foreach (var (fileName, mime, size, attId) in d.Attachments)
|
||||
{
|
||||
_db.Attachments.Add(new Attachment
|
||||
{
|
||||
UserId = userId, EmailId = email.Id, FileName = fileName,
|
||||
MimeType = mime, 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;
|
||||
}
|
||||
|
||||
private async Task<Sender> ResolveSenderAsync(Guid userId, string address, string? displayName, CancellationToken ct)
|
||||
{
|
||||
var sender = await _db.Senders.FirstOrDefaultAsync(s => s.UserId == userId && s.Address == address, ct);
|
||||
if (sender is not null) return sender;
|
||||
|
||||
var domainName = address.Contains('@') ? address[(address.IndexOf('@') + 1)..] : "unknown";
|
||||
var domain = 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 = displayName, Domain = domain, DomainId = domain.Id };
|
||||
_db.Senders.Add(sender);
|
||||
return sender;
|
||||
}
|
||||
|
||||
private async Task<MailThread> ResolveThreadAsync(Guid userId, string gmailThreadId, string? subject, string? snippet, DateTimeOffset sentAt, CancellationToken ct)
|
||||
{
|
||||
var thread = 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 = subject,
|
||||
Snippet = snippet, FirstMessageUtc = sentAt
|
||||
};
|
||||
_db.Threads.Add(thread);
|
||||
return thread;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user