using InboxIntel.Application.Abstractions; using InboxIntel.Infrastructure.Persistence; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; namespace InboxIntel.Infrastructure.Ai; /// /// Fills Email.Embedding (pgvector) for semantic search, in small background batches /// so interactive requests are never starved (per docs/discovery/06: embeddings are the /// small always-on model; the batch pause keeps VRAM/CPU pressure low). Exits immediately /// when the embedding provider is unavailable (AI disabled / Ollama down) — semantic search /// simply stays dormant and lexical search is unaffected. /// public class EmbeddingBackfillWorker : BackgroundService { private const int BatchSize = 32; private static readonly TimeSpan BatchPause = TimeSpan.FromSeconds(2); private static readonly TimeSpan IdleRescan = TimeSpan.FromMinutes(15); private readonly IServiceScopeFactory _scopeFactory; private readonly ILogger _logger; public EmbeddingBackfillWorker(IServiceScopeFactory scopeFactory, ILogger logger) { _scopeFactory = scopeFactory; _logger = logger; } protected override async Task ExecuteAsync(CancellationToken stoppingToken) { // Provider availability is fixed by configuration for the process lifetime. using (var probe = _scopeFactory.CreateScope()) { if (!probe.ServiceProvider.GetRequiredService().IsAvailable) { _logger.LogDebug("EmbeddingBackfillWorker idle: no embedding provider (AI disabled)."); return; } } _logger.LogInformation("EmbeddingBackfillWorker started (batch {Batch}, pause {Pause}s)", BatchSize, BatchPause.TotalSeconds); while (!stoppingToken.IsCancellationRequested) { int processed; try { processed = await ProcessBatchAsync(stoppingToken); } catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) { break; } catch (Exception ex) { // Ollama hiccups must never crash the host; back off and retry. _logger.LogWarning(ex, "Embedding batch failed; retrying after idle pause."); processed = 0; } await Task.Delay(processed > 0 ? BatchPause : IdleRescan, stoppingToken); } } /// Embeds one batch. Public-ish (internal) for direct testing. internal async Task ProcessBatchAsync(CancellationToken ct) { using var scope = _scopeFactory.CreateScope(); var db = scope.ServiceProvider.GetRequiredService(); var embeddings = scope.ServiceProvider.GetRequiredService(); var batch = await db.Emails .Where(e => e.Embedding == null) .OrderByDescending(e => e.SentAtUtc) // newest mail becomes searchable first .Take(BatchSize) .ToListAsync(ct); if (batch.Count == 0) return 0; // Subject + snippet is the semantic core; bodies are noisy (signatures, quoting) // and slow to embed. Truncate defensively to keep well inside the model context. var texts = batch .Select(e => Truncate($"{e.Subject}\n{e.Snippet ?? e.BodyText}", 2000)) .ToList(); var vectors = await embeddings.EmbedBatchAsync(texts, ct); if (vectors.Count != batch.Count) { _logger.LogWarning("Embedding batch returned {Got} vectors for {Want} emails; skipping batch.", vectors.Count, batch.Count); return 0; } for (var i = 0; i < batch.Count; i++) { if (vectors[i].Length == 0) continue; // provider soft-failure for one item batch[i].Embedding = new Pgvector.Vector(vectors[i]); } await db.SaveChangesAsync(ct); _logger.LogDebug("Embedded {Count} emails", batch.Count); return batch.Count; } private static string Truncate(string s, int max) => s.Length <= max ? s : s[..max]; }