gott/Services/EmbeddingBackgroundService.cs

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;

/// <summary>
/// Background-Service der Messages embedded:
/// 1. Beim Start: Backfill aller bestehenden Text-Messages
/// 2. Laufend: Neue Messages über Queue verarbeiten
/// </summary>
public class EmbeddingBackgroundService : BackgroundService
{
    readonly VectorSearchService _vectorSearch;
    readonly EmbeddingService _embedding;
    readonly string _dbConnectionString;
    readonly ConcurrentQueue<EmbeddingJob> _queue = new();
    readonly SemaphoreSlim _signal = new(0);
    readonly string _dataPath;

    public EmbeddingBackgroundService(VectorSearchService vectorSearch, EmbeddingService embedding, string dbPath)
    {
        _vectorSearch = vectorSearch;
        _embedding = embedding;
        _dbConnectionString = $"Data Source={dbPath}";
        _dataPath = Path.GetDirectoryName(dbPath);
    }

    /// <summary>Enqueue eine neue Message zum Embedden.</summary>
    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}");
                    }
                }
            }
            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<BackfillRow>(@"
            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.");
    }

    /// <summary>Embeddet eine einzelne Memory-Datei (nach memory_write).</summary>
    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; }
    }
}