using System;
using System.IO;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Dapper;
using Microsoft.Data.Sqlite;
using Microsoft.Extensions.Hosting;
namespace Gott.Services;
///
/// Background-Service der Messages embedded:
/// 1. Beim Start: Backfill aller bestehenden Text-Messages
/// 2. Laufend: Neue Messages über Queue verarbeiten
///
public class EmbeddingBackgroundService : BackgroundService
{
readonly VectorSearchService _vectorSearch;
readonly EmbeddingService _embedding;
readonly string _dbConnectionString;
readonly ConcurrentQueue _queue = new();
readonly SemaphoreSlim _signal = new(0);
readonly string _dataPath;
volatile bool _reEmbedRequested;
public EmbeddingBackgroundService(VectorSearchService vectorSearch, EmbeddingService embedding, string dbPath)
{
_vectorSearch = vectorSearch;
_embedding = embedding;
_dbConnectionString = $"Data Source={dbPath}";
_dataPath = Path.GetDirectoryName(dbPath);
}
/// Triggers a full re-embedding of all data.
public void TriggerReEmbed()
{
_reEmbedRequested = true;
_signal.Release();
}
/// Enqueue eine neue Message zum Embedden.
public void Enqueue(long messageId, string chainId, string sessionId, string agentId, string role, string text)
{
_queue.Enqueue(new EmbeddingJob(messageId, chainId, sessionId, agentId, role, text));
_signal.Release();
}
protected override async Task ExecuteAsync(CancellationToken ct)
{
if (!_embedding.IsAvailable)
{
Console.WriteLine("[EmbeddingBG] Embedding model not available – service disabled.");
return;
}
// Kurz warten bis App vollständig gestartet
await Task.Delay(3000, ct);
// Backfill bestehender Messages
await BackfillAsync(ct);
// Memory-Dateien embedden
await BackfillMemoryFilesAsync(ct);
// Live-Embedding neuer Messages
Console.WriteLine("[EmbeddingBG] Waiting for new messages...");
while (!ct.IsCancellationRequested)
{
try
{
await _signal.WaitAsync(ct);
while (_queue.TryDequeue(out var job))
{
if (ct.IsCancellationRequested) break;
try
{
if (job.Role.StartsWith("memory:"))
{
// Memory-File-Embedding
var filePath = job.Role.Substring("memory:".Length);
await _vectorSearch.EmbedMemoryFileAsync(job.AgentId, filePath, job.Text);
}
else
{
await _vectorSearch.EmbedMessageAsync(job.MessageId, job.ChainId, job.SessionId,
job.AgentId, job.Role, job.Text);
}
}
catch (Exception ex)
{
Console.WriteLine($"[EmbeddingBG] Error embedding job: {ex.Message}");
}
}
// Check for re-embed request
if (_reEmbedRequested)
{
_reEmbedRequested = false;
Console.WriteLine("[EmbeddingBG] Re-embedding triggered...");
await _vectorSearch.ReEmbedAllAsync();
await BackfillAsync(ct);
await BackfillMemoryFilesAsync(ct);
Console.WriteLine("[EmbeddingBG] Re-embedding complete.");
}
}
catch (OperationCanceledException) { break; }
}
}
async Task BackfillAsync(CancellationToken ct)
{
Console.WriteLine("[EmbeddingBG] Starting backfill of existing messages...");
var existingIds = await _vectorSearch.GetEmbeddedMessageIdsAsync();
Console.WriteLine($"[EmbeddingBG] Already embedded: {existingIds.Count} messages");
using var conn = new SqliteConnection(_dbConnectionString);
await conn.OpenAsync(ct);
// Alle text-Messages laden (user + assistant, inkl. archivierte Sessions)
var messages = await conn.QueryAsync(@"
SELECT m.id, m.chain_id, c.session_id, c.agent_id, m.role, m.content, m.type
FROM messages m
JOIN chains c ON m.chain_id = c.chain_id
WHERE m.type = 'text'
AND m.role IN ('user', 'assistant')
AND m.content IS NOT NULL
AND LENGTH(m.content) >= 10
AND m.deleted_on IS NULL
ORDER BY m.id ASC
");
int total = 0;
int embedded = 0;
int skipped = 0;
foreach (var msg in messages)
{
if (ct.IsCancellationRequested) break;
total++;
if (existingIds.Contains(msg.Id))
{
skipped++;
continue;
}
try
{
// SessionId kann auch aus archive kommen
var sessionId = msg.SessionId ?? "";
await _vectorSearch.EmbedMessageAsync(msg.Id, msg.ChainId, sessionId,
msg.AgentId, msg.Role, msg.Content);
embedded++;
if (embedded % 100 == 0)
Console.WriteLine($"[EmbeddingBG] Backfill progress: {embedded} embedded, {skipped} skipped, {total} total");
}
catch (Exception ex)
{
Console.WriteLine($"[EmbeddingBG] Error backfilling message {msg.Id}: {ex.Message}");
}
}
Console.WriteLine($"[EmbeddingBG] Backfill complete: {embedded} new embeddings, {skipped} skipped, {total} total messages.");
}
async Task BackfillMemoryFilesAsync(CancellationToken ct)
{
Console.WriteLine("[EmbeddingBG] Starting memory file backfill...");
var agentsDir = Path.Combine(_dataPath, "agents");
if (!Directory.Exists(agentsDir))
{
Console.WriteLine("[EmbeddingBG] No agents directory found, skipping memory backfill.");
return;
}
int total = 0;
foreach (var agentDir in Directory.GetDirectories(agentsDir))
{
if (ct.IsCancellationRequested) break;
var agentId = Path.GetFileName(agentDir);
// Alle .md Dateien im Agent-Verzeichnis (rekursiv)
var mdFiles = Directory.GetFiles(agentDir, "*.md", SearchOption.AllDirectories);
foreach (var mdFile in mdFiles)
{
if (ct.IsCancellationRequested) break;
try
{
var text = await File.ReadAllTextAsync(mdFile, ct);
if (string.IsNullOrWhiteSpace(text) || text.Length < 10) continue;
var relativePath = Path.GetRelativePath(agentDir, mdFile);
await _vectorSearch.EmbedMemoryFileAsync(agentId, relativePath, text);
total++;
}
catch (Exception ex)
{
Console.WriteLine($"[EmbeddingBG] Error embedding memory file {mdFile}: {ex.Message}");
}
}
}
Console.WriteLine($"[EmbeddingBG] Memory file backfill complete: {total} files embedded.");
}
/// Embeddet eine einzelne Memory-Datei (nach memory_write).
public void EnqueueMemoryFile(string agentId, string filePath, string text)
{
_queue.Enqueue(new EmbeddingJob(0, "", "", agentId, "memory:" + filePath, text));
_signal.Release();
}
record EmbeddingJob(long MessageId, string ChainId, string SessionId, string AgentId, string Role, string Text);
class BackfillRow
{
public long Id { get; set; }
public string ChainId { get; set; }
public string SessionId { get; set; }
public string AgentId { get; set; }
public string Role { get; set; }
public string Content { get; set; }
public string Type { get; set; }
}
}