feat: landing page + logo, dev-mode banner + sync cap, non-blocking sync with progress splash
This commit is contained in:
@@ -26,6 +26,11 @@ public class GmailSyncOptions
|
||||
public int BackoffBaseMs { get; set; } = 500;
|
||||
/// <summary>Cron-like daily sync hour (UTC) for the scheduled worker.</summary>
|
||||
public int DailySyncHourUtc { get; set; } = 3;
|
||||
/// <summary>
|
||||
/// Cap on messages stored during a full sync. 0 = unlimited. In dev this is
|
||||
/// set to 1000 so the initial load pulls only the most recent emails.
|
||||
/// </summary>
|
||||
public int MaxMessages { get; set; } = 0;
|
||||
}
|
||||
|
||||
public class AiOptions
|
||||
|
||||
@@ -38,6 +38,11 @@ public static class DependencyInjection
|
||||
services.AddScoped<GmailClientFactory>();
|
||||
services.AddScoped<IGmailService, GmailApiService>();
|
||||
|
||||
// Background sync queue (singleton) + its worker.
|
||||
services.AddSingleton<SyncQueue>();
|
||||
services.AddSingleton<ISyncQueue>(sp => sp.GetRequiredService<SyncQueue>());
|
||||
services.AddHostedService<SyncQueueWorker>();
|
||||
|
||||
// Core services
|
||||
services.AddScoped<ISyncService, SyncService>();
|
||||
services.AddScoped<IAnalyticsService, AnalyticsService>();
|
||||
|
||||
@@ -0,0 +1,18 @@
|
||||
using System.Threading.Channels;
|
||||
using InboxIntel.Application.Abstractions;
|
||||
|
||||
namespace InboxIntel.Infrastructure.Sync;
|
||||
|
||||
/// <summary>
|
||||
/// In-process unbounded queue of sync jobs, backed by a Channel. Registered as
|
||||
/// a singleton; the controller writes, <see cref="SyncQueueWorker"/> reads.
|
||||
/// </summary>
|
||||
public sealed class SyncQueue : ISyncQueue
|
||||
{
|
||||
private readonly Channel<(Guid UserId, bool Full)> _channel =
|
||||
Channel.CreateUnbounded<(Guid, bool)>(new UnboundedChannelOptions { SingleReader = true });
|
||||
|
||||
public void Enqueue(Guid userId, bool fullSync) => _channel.Writer.TryWrite((userId, fullSync));
|
||||
|
||||
public ChannelReader<(Guid UserId, bool Full)> Reader => _channel.Reader;
|
||||
}
|
||||
@@ -0,0 +1,44 @@
|
||||
using InboxIntel.Application.Abstractions;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace InboxIntel.Infrastructure.Sync;
|
||||
|
||||
/// <summary>
|
||||
/// Drains the <see cref="SyncQueue"/> and runs each sync in its own DI scope,
|
||||
/// then refreshes analytics aggregates. Keeps sync work off the request thread.
|
||||
/// </summary>
|
||||
public class SyncQueueWorker : BackgroundService
|
||||
{
|
||||
private readonly SyncQueue _queue;
|
||||
private readonly IServiceScopeFactory _scopeFactory;
|
||||
private readonly ILogger<SyncQueueWorker> _logger;
|
||||
|
||||
public SyncQueueWorker(SyncQueue queue, IServiceScopeFactory scopeFactory, ILogger<SyncQueueWorker> logger)
|
||||
{
|
||||
_queue = queue;
|
||||
_scopeFactory = scopeFactory;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
await foreach (var (userId, full) in _queue.Reader.ReadAllAsync(stoppingToken))
|
||||
{
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var sync = scope.ServiceProvider.GetRequiredService<ISyncService>();
|
||||
var analytics = scope.ServiceProvider.GetRequiredService<IAnalyticsService>();
|
||||
try
|
||||
{
|
||||
if (full) await sync.RunFullSyncAsync(userId, stoppingToken);
|
||||
else await sync.RunIncrementalSyncAsync(userId, stoppingToken);
|
||||
await analytics.RefreshAggregatesAsync(userId, stoppingToken);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "Queued sync failed for user {UserId}", userId);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,10 +1,13 @@
|
||||
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;
|
||||
|
||||
@@ -19,17 +22,46 @@ public class SyncService : ISyncService
|
||||
private readonly AppDbContext _db;
|
||||
private readonly IGmailService _gmail;
|
||||
private readonly ILogger<SyncService> _logger;
|
||||
private readonly GmailSyncOptions _options;
|
||||
private readonly ISyncQueue _queue;
|
||||
|
||||
public SyncService(AppDbContext db, IGmailService gmail, ILogger<SyncService> logger)
|
||||
public SyncService(AppDbContext db, IGmailService gmail, ILogger<SyncService> logger, IOptions<GmailSyncOptions> options, ISyncQueue queue)
|
||||
{
|
||||
_db = db;
|
||||
_gmail = gmail;
|
||||
_logger = logger;
|
||||
_options = options.Value;
|
||||
_queue = queue;
|
||||
}
|
||||
|
||||
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<SyncStatus> GetStatusAsync(Guid userId, CancellationToken ct = default)
|
||||
=> (await GetOrCreateStateAsync(userId, ct)).Status;
|
||||
|
||||
public async Task<SyncProgressDto> GetProgressAsync(Guid userId, CancellationToken ct = default)
|
||||
{
|
||||
var s = await GetOrCreateStateAsync(userId, ct);
|
||||
var total = _options.MaxMessages > 0
|
||||
? Math.Min(s.TotalMessagesEstimate == 0 ? _options.MaxMessages : s.TotalMessagesEstimate, _options.MaxMessages)
|
||||
: s.TotalMessagesEstimate;
|
||||
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);
|
||||
@@ -43,12 +75,21 @@ public class SyncService : ISyncService
|
||||
{
|
||||
await SyncLabelsAsync(userId, ct);
|
||||
|
||||
// Dev cap: stop after MaxMessages (most recent first). 0 = unlimited.
|
||||
var maxMessages = _options.MaxMessages;
|
||||
var capReached = false;
|
||||
|
||||
string? pageToken = state.ResumePageToken; // resume support
|
||||
do
|
||||
{
|
||||
var page = await _gmail.ListMessageIdsAsync(userId, pageToken, ct);
|
||||
foreach (var messageId in page.MessageIds)
|
||||
{
|
||||
if (maxMessages > 0 && state.MessagesProcessed >= maxMessages)
|
||||
{
|
||||
capReached = true;
|
||||
break;
|
||||
}
|
||||
if (await _db.Emails.AnyAsync(e => e.UserId == userId && e.GmailMessageId == messageId, ct))
|
||||
continue;
|
||||
var detail = await _gmail.GetMessageAsync(userId, messageId, ct);
|
||||
@@ -58,10 +99,12 @@ public class SyncService : ISyncService
|
||||
|
||||
pageToken = page.NextPageToken;
|
||||
state.ResumePageToken = pageToken; // checkpoint
|
||||
state.TotalMessagesEstimate = page.ResultSizeEstimate;
|
||||
state.TotalMessagesEstimate = maxMessages > 0
|
||||
? Math.Min(page.ResultSizeEstimate, maxMessages)
|
||||
: page.ResultSizeEstimate;
|
||||
await _db.SaveChangesAsync(ct);
|
||||
}
|
||||
while (pageToken is not null && !ct.IsCancellationRequested);
|
||||
while (pageToken is not null && !capReached && !ct.IsCancellationRequested);
|
||||
|
||||
state.LastHistoryId = await _gmail.GetProfileHistoryIdAsync(userId, ct);
|
||||
state.ResumePageToken = null;
|
||||
|
||||
Reference in New Issue
Block a user