Refactoring + Dokumentation: api.js komprimiert (745→267 Zeilen), SseHelper extrahiert, DatabaseService in Partial Classes aufgeteilt, unbenutztes CSS entfernt, ARCHITECTURE.md/ARCHITECTURE_QUICK.md/README.md aktualisiert (neue Features: Embedding, Remote, API-Keys, 16 Tools)
This commit is contained in:
Meleki committed
1 parent
8dc52c1337
commit
db1a7c7f64
18 files changed
+1701
-2007
No files matched your search
@@ -190,7 +190,7 @@ public class AnthropicService
|
||||
trimCount = TrimMessagesCleanly(messages, trimCount);
|
||||
if (trimCount > 0)
|
||||
{
|
||||
await TryWriteSse(response, "info", new { message = $"History zu lang – {trimCount} ältere Messages entfernt, Retry ({promptTrimRetries}/5)..." }, sseCt);
|
||||
await SseHelper.TryWriteSse(response, "info", new { message = $"History zu lang – {trimCount} ältere Messages entfernt, Retry ({promptTrimRetries}/5)..." }, sseCt);
|
||||
iterations--; // Iteration nicht zählen, war ja kein echter Versuch
|
||||
continue; // → nächste while-Iteration mit kürzerer History
|
||||
}
|
||||
@@ -202,9 +202,9 @@ public class AnthropicService
|
||||
Role = "system", Type = "error",
|
||||
Content = errMsg, IsError = true
|
||||
});
|
||||
await TryWriteSse(response, "error", new { error = errMsg }, sseCt);
|
||||
await SseHelper.TryWriteSse(response, "error", new { error = errMsg }, sseCt);
|
||||
await FinalizeChain(chain, usage, iterations, "error", sw);
|
||||
await TryWriteSse(response, "done", new { chainId = chain.ChainId }, sseCt);
|
||||
await SseHelper.TryWriteSse(response, "done", new { chainId = chain.ChainId }, sseCt);
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -323,7 +323,7 @@ public class AnthropicService
|
||||
|
||||
var finalStopReason = string.IsNullOrEmpty(stopReason) ? "unexpected_end" : stopReason;
|
||||
await FinalizeChain(chain, usage, iterations, finalStopReason, sw, streamErrorInfo);
|
||||
await TryWriteSse(response, "done", new
|
||||
await SseHelper.TryWriteSse(response, "done", new
|
||||
{
|
||||
model = modelId,
|
||||
inputTokens = usage.InputTokens,
|
||||
@@ -359,8 +359,8 @@ public class AnthropicService
|
||||
}
|
||||
|
||||
await FinalizeChain(chain, usage, iterations, "max_iterations", sw);
|
||||
await TryWriteSse(response, "error", new { error = "Max iterations erreicht" }, sseCt);
|
||||
await TryWriteSse(response, "done", new { chainId = chain.ChainId }, sseCt);
|
||||
await SseHelper.TryWriteSse(response, "error", new { error = "Max iterations erreicht" }, sseCt);
|
||||
await SseHelper.TryWriteSse(response, "done", new { chainId = chain.ChainId }, sseCt);
|
||||
}
|
||||
catch (OperationCanceledException) when (workCt.IsCancellationRequested || hardCt.IsCancellationRequested)
|
||||
{
|
||||
@@ -368,7 +368,7 @@ public class AnthropicService
|
||||
await FinalizeChain(chain, usage, iterations, "stopped", sw);
|
||||
// sseCt ist hier bereits cancelled (gelinkt mit workCt), daher CancellationToken.None
|
||||
// Der SSE-Stream zum Client ist noch offen – nur der Stop-Button hat ausgelöst
|
||||
await TryWriteSse(response, "done", new
|
||||
await SseHelper.TryWriteSse(response, "done", new
|
||||
{
|
||||
model = modelId,
|
||||
inputTokens = usage.InputTokens,
|
||||
@@ -389,7 +389,7 @@ public class AnthropicService
|
||||
// Wenn workCt/hardCt gecancelled ist, ist das ein regulärer Stop.
|
||||
await FinalizeChain(chain, usage, iterations, "stopped", sw);
|
||||
// sseCt ist hier bereits cancelled (gelinkt mit workCt), daher CancellationToken.None
|
||||
await TryWriteSse(response, "done", new
|
||||
await SseHelper.TryWriteSse(response, "done", new
|
||||
{
|
||||
model = modelId,
|
||||
inputTokens = usage.InputTokens,
|
||||
@@ -421,8 +421,8 @@ public class AnthropicService
|
||||
Content = errMsg, IsError = true
|
||||
});
|
||||
await FinalizeChain(chain, usage, iterations, "error", sw);
|
||||
await TryWriteSse(response, "error", new { error = errMsg }, sseCt);
|
||||
await TryWriteSse(response, "done", new { chainId = chain.ChainId }, sseCt);
|
||||
await SseHelper.TryWriteSse(response, "error", new { error = errMsg }, sseCt);
|
||||
await SseHelper.TryWriteSse(response, "done", new { chainId = chain.ChainId }, sseCt);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -455,7 +455,7 @@ public class AnthropicService
|
||||
// da der Worker noch nicht gestartet hat).
|
||||
cancellationToken.ThrowIfCancellationRequested();
|
||||
|
||||
await TryWriteSse(response, "preparer_start", new
|
||||
await SseHelper.TryWriteSse(response, "preparer_start", new
|
||||
{
|
||||
agentId = preparer.AgentId,
|
||||
agentName = preparer.Agent?.Name ?? "Vorbereiter-Agent"
|
||||
@@ -484,7 +484,7 @@ public class AnthropicService
|
||||
result.ShouldForward = false;
|
||||
result.ResponseText = questionText;
|
||||
|
||||
await TryWriteSse(response, "preparer_done", new
|
||||
await SseHelper.TryWriteSse(response, "preparer_done", new
|
||||
{
|
||||
agentName = preparer.Agent?.Name ?? "Vorbereiter-Agent",
|
||||
forwarded = false,
|
||||
@@ -504,7 +504,7 @@ public class AnthropicService
|
||||
result.ShouldForward = true;
|
||||
result.ForwardedPrompt = forwardedText;
|
||||
|
||||
await TryWriteSse(response, "preparer_done", new
|
||||
await SseHelper.TryWriteSse(response, "preparer_done", new
|
||||
{
|
||||
agentName = preparer.Agent?.Name ?? "Vorbereiter-Agent",
|
||||
forwarded = true
|
||||
@@ -545,7 +545,7 @@ public class AnthropicService
|
||||
workCt.ThrowIfCancellationRequested();
|
||||
|
||||
// SSE: Review-Phase Start
|
||||
await TryWriteSse(response, "review_start", new
|
||||
await SseHelper.TryWriteSse(response, "review_start", new
|
||||
{
|
||||
round = reviewRound + 1,
|
||||
maxRounds = review.MaxReviewIterations,
|
||||
@@ -576,7 +576,7 @@ public class AnthropicService
|
||||
result.HasFeedback = true;
|
||||
result.Feedback = $"[Review-Feedback von {review.Agent?.Name ?? "Review-Agent"} (Runde {reviewRound + 1}/{review.MaxReviewIterations})]:\n{feedbackText}";
|
||||
|
||||
await TryWriteSse(response, "review_feedback", new
|
||||
await SseHelper.TryWriteSse(response, "review_feedback", new
|
||||
{
|
||||
round = reviewRound + 1,
|
||||
maxRounds = review.MaxReviewIterations,
|
||||
@@ -587,7 +587,7 @@ public class AnthropicService
|
||||
else
|
||||
{
|
||||
// Default: Approved (kein [FEEDBACK]-Tag → alles OK)
|
||||
await TryWriteSse(response, "review_approved", new
|
||||
await SseHelper.TryWriteSse(response, "review_approved", new
|
||||
{
|
||||
round = reviewRound + 1,
|
||||
agentName = review.Agent?.Name ?? "Review-Agent"
|
||||
@@ -616,7 +616,7 @@ public class AnthropicService
|
||||
CancellationToken workCt,
|
||||
CancellationToken sseCt)
|
||||
{
|
||||
await TryWriteSse(response, "finalizer_start", new
|
||||
await SseHelper.TryWriteSse(response, "finalizer_start", new
|
||||
{
|
||||
agentId = finalizer.AgentId,
|
||||
agentName = finalizer.Agent?.Name ?? "Finalizer-Agent"
|
||||
@@ -632,7 +632,7 @@ public class AnthropicService
|
||||
baseIterations, "finalizer", finalizer.Agent?.Name ?? "Finalizer-Agent",
|
||||
workCt, sseCt);
|
||||
|
||||
await TryWriteSse(response, "finalizer_done", new
|
||||
await SseHelper.TryWriteSse(response, "finalizer_done", new
|
||||
{
|
||||
agentName = finalizer.Agent?.Name ?? "Finalizer-Agent"
|
||||
}, sseCt);
|
||||
@@ -828,7 +828,7 @@ public class AnthropicService
|
||||
trimCount = TrimMessagesCleanly(messages, trimCount);
|
||||
if (trimCount > 0)
|
||||
{
|
||||
await TryWriteSse(response, "info", new { message = $"{tag}: History zu lang – {trimCount} Messages entfernt, Retry..." }, sseCt);
|
||||
await SseHelper.TryWriteSse(response, "info", new { message = $"{tag}: History zu lang – {trimCount} Messages entfernt, Retry..." }, sseCt);
|
||||
iterations--;
|
||||
continue;
|
||||
}
|
||||
@@ -857,7 +857,7 @@ public class AnthropicService
|
||||
assistantContent.Add(new { type = "text", text });
|
||||
finalText.Append(text);
|
||||
|
||||
await TryWriteSse(response, $"{agentType}_text", new
|
||||
await SseHelper.TryWriteSse(response, $"{agentType}_text", new
|
||||
{
|
||||
text,
|
||||
agentName,
|
||||
@@ -882,7 +882,7 @@ public class AnthropicService
|
||||
|
||||
assistantContent.Add(new { type = "tool_use", id = toolId, name = toolName, input = toolInput });
|
||||
|
||||
await TryWriteSse(response, $"{agentType}_tool_use", new
|
||||
await SseHelper.TryWriteSse(response, $"{agentType}_tool_use", new
|
||||
{
|
||||
toolName,
|
||||
toolId,
|
||||
@@ -916,7 +916,7 @@ public class AnthropicService
|
||||
is_error = isError
|
||||
});
|
||||
|
||||
await TryWriteSse(response, $"{agentType}_tool_end", new
|
||||
await SseHelper.TryWriteSse(response, $"{agentType}_tool_end", new
|
||||
{
|
||||
toolName,
|
||||
toolId,
|
||||
@@ -1207,18 +1207,8 @@ public class AnthropicService
|
||||
catch { /* Logging darf nie die App crashen */ }
|
||||
}
|
||||
|
||||
private async Task FinalizeChain(ChainRecord chain, TokenUsage usage, int iterations, string stopReason, System.Diagnostics.Stopwatch sw, string streamErrorInfo = null)
|
||||
{
|
||||
chain.InputTokens = usage.InputTokens;
|
||||
chain.OutputTokens = usage.OutputTokens;
|
||||
chain.CacheCreationTokens = usage.CacheCreationTokens;
|
||||
chain.CacheReadTokens = usage.CacheReadTokens;
|
||||
chain.Iterations = iterations;
|
||||
chain.StopReason = stopReason;
|
||||
chain.StreamErrorInfo = streamErrorInfo;
|
||||
chain.DurationMs = sw.ElapsedMilliseconds;
|
||||
await _db.UpdateChain(chain);
|
||||
}
|
||||
private Task FinalizeChain(ChainRecord chain, TokenUsage usage, int iterations, string stopReason, System.Diagnostics.Stopwatch sw, string streamErrorInfo = null)
|
||||
=> SseHelper.FinalizeChain(_db, chain, usage, iterations, stopReason, sw, streamErrorInfo);
|
||||
|
||||
private async Task<(string StopReason, List<object> Content, List<ToolResult> Tools, string StreamErrorInfo)> ProcessStream(
|
||||
HttpResponse response,
|
||||
@@ -1287,7 +1277,7 @@ public class AnthropicService
|
||||
if (cbType == "text")
|
||||
{
|
||||
currentTextBlock.Clear();
|
||||
await TryWriteSse(response, "text_start", new { index = idx }, sseCt);
|
||||
await SseHelper.TryWriteSse(response, "text_start", new { index = idx }, sseCt);
|
||||
}
|
||||
else if (cbType == "tool_use")
|
||||
{
|
||||
@@ -1299,7 +1289,7 @@ public class AnthropicService
|
||||
currentToolId = cb.GetProperty("id").GetString() ?? "";
|
||||
currentToolName = cb.GetProperty("name").GetString() ?? "";
|
||||
currentToolInputJson.Clear();
|
||||
await TryWriteSse(response, "tool_start", new { name = currentToolName, id = currentToolId }, sseCt);
|
||||
await SseHelper.TryWriteSse(response, "tool_start", new { name = currentToolName, id = currentToolId }, sseCt);
|
||||
}
|
||||
else if (cbType == "server_tool_use")
|
||||
{
|
||||
@@ -1311,7 +1301,7 @@ public class AnthropicService
|
||||
currentToolId = cb.GetProperty("id").GetString() ?? "";
|
||||
currentToolName = cb.GetProperty("name").GetString() ?? "";
|
||||
currentToolInputJson.Clear();
|
||||
await TryWriteSse(response, "web_search_start", new { name = currentToolName, id = currentToolId }, sseCt);
|
||||
await SseHelper.TryWriteSse(response, "web_search_start", new { name = currentToolName, id = currentToolId }, sseCt);
|
||||
}
|
||||
else if (cbType == "web_search_tool_result")
|
||||
{
|
||||
@@ -1336,7 +1326,7 @@ public class AnthropicService
|
||||
{
|
||||
var text = delta.GetProperty("text").GetString() ?? "";
|
||||
currentTextBlock.Append(text);
|
||||
await TryWriteSse(response, "text_delta", new { text }, sseCt);
|
||||
await SseHelper.TryWriteSse(response, "text_delta", new { text }, sseCt);
|
||||
}
|
||||
else if (deltaType == "input_json_delta")
|
||||
{
|
||||
@@ -1357,7 +1347,7 @@ public class AnthropicService
|
||||
if (text.Length > 0)
|
||||
{
|
||||
assistantContent.Add(new { type = "text", text });
|
||||
await TryWriteSse(response, "text_end", new { }, sseCt);
|
||||
await SseHelper.TryWriteSse(response, "text_end", new { }, sseCt);
|
||||
}
|
||||
currentTextBlock.Clear();
|
||||
}
|
||||
@@ -1394,7 +1384,7 @@ public class AnthropicService
|
||||
}
|
||||
}
|
||||
|
||||
await TryWriteSse(response, "user_choice", new
|
||||
await SseHelper.TryWriteSse(response, "user_choice", new
|
||||
{
|
||||
toolCallId = currentToolId,
|
||||
question,
|
||||
@@ -1419,7 +1409,7 @@ public class AnthropicService
|
||||
workCt.ThrowIfCancellationRequested();
|
||||
executedTools.Add(new ToolResult(currentToolId, toolResult, isError));
|
||||
|
||||
await TryWriteSse(response, "tool_end", new
|
||||
await SseHelper.TryWriteSse(response, "tool_end", new
|
||||
{
|
||||
name = currentToolName,
|
||||
input = currentToolInputJson.ToString(),
|
||||
@@ -1447,7 +1437,7 @@ public class AnthropicService
|
||||
// Web-Search-Ergebnis von Anthropic – als Raw-JSON in assistantContent aufnehmen
|
||||
assistantContent.Add(webSearchResultBlocks[idx]);
|
||||
var sources = webSearchSources.TryGetValue(idx, out var s) ? s : new List<object>();
|
||||
await TryWriteSse(response, "web_search_end", new { id = currentToolId, sources }, sseCt);
|
||||
await SseHelper.TryWriteSse(response, "web_search_end", new { id = currentToolId, sources }, sseCt);
|
||||
}
|
||||
else if (bt.EndsWith("_tool_result") && serverToolResultBlocks.TryGetValue(idx, out var srvResultBlock))
|
||||
{
|
||||
@@ -1526,14 +1516,8 @@ public class AnthropicService
|
||||
|
||||
// ── Shared Helpers ──
|
||||
|
||||
internal const int MaxToolResultChars = 20_000;
|
||||
|
||||
internal static string TruncateToolResult(string result)
|
||||
{
|
||||
return result.Length > MaxToolResultChars
|
||||
? result[..MaxToolResultChars] + $"\n\n[... gekürzt, {result.Length - MaxToolResultChars} weitere Zeichen ...]"
|
||||
: result;
|
||||
}
|
||||
// TruncateToolResult → SseHelper.cs
|
||||
internal static string TruncateToolResult(string result) => SseHelper.TruncateToolResult(result);
|
||||
|
||||
/// <summary>
|
||||
/// Extrahiert den Text der letzten User-Message aus der Message-Liste.
|
||||
@@ -1697,34 +1681,5 @@ public class AnthropicService
|
||||
return client;
|
||||
}
|
||||
|
||||
public static async Task WriteSseEvent(HttpResponse response, string eventType, object data, CancellationToken ct = default)
|
||||
{
|
||||
ct.ThrowIfCancellationRequested();
|
||||
var json = JsonSerializer.Serialize(data);
|
||||
await response.WriteAsync($"event: {eventType}\ndata: {json}\n\n", ct);
|
||||
await response.Body.FlushAsync(ct);
|
||||
}
|
||||
|
||||
private static async Task TryWriteSse(HttpResponse response, string eventType, object data, CancellationToken ct)
|
||||
{
|
||||
try
|
||||
{
|
||||
if (!ct.IsCancellationRequested)
|
||||
await WriteSseEvent(response, eventType, data, ct);
|
||||
}
|
||||
catch
|
||||
{
|
||||
// Client bereits disconnected – ignorieren
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public record ToolResult(string Id, string Result, bool IsError);
|
||||
|
||||
public class TokenUsage
|
||||
{
|
||||
public long InputTokens { get; set; }
|
||||
public long OutputTokens { get; set; }
|
||||
public long CacheCreationTokens { get; set; }
|
||||
public long CacheReadTokens { get; set; }
|
||||
// WriteSseEvent, TryWriteSse, TokenUsage, ToolResult → SseHelper.cs
|
||||
}
|
||||
@@ -465,7 +465,7 @@ Für welche Plattform soll die App entwickelt werden? Web, Mobile oder Desktop?
|
||||
dict.TryGetValue("role", out var role) && role.ToString() == "user" &&
|
||||
dict.TryGetValue("content", out var existingContent) && existingContent is List<object> contentList)
|
||||
{
|
||||
contentList.Add(new { type = "tool_result", tool_use_id = msg.ToolCallId, content = AnthropicService.TruncateToolResult(msg.Content ?? ""), is_error = msg.IsError });
|
||||
contentList.Add(new { type = "tool_result", tool_use_id = msg.ToolCallId, content = SseHelper.TruncateToolResult(msg.Content ?? ""), is_error = msg.IsError });
|
||||
}
|
||||
else
|
||||
{
|
||||
@@ -474,7 +474,7 @@ Für welche Plattform soll die App entwickelt werden? Web, Mobile oder Desktop?
|
||||
["role"] = "user",
|
||||
["content"] = new List<object>
|
||||
{
|
||||
new { type = "tool_result", tool_use_id = msg.ToolCallId, content = AnthropicService.TruncateToolResult(msg.Content ?? ""), is_error = msg.IsError }
|
||||
new { type = "tool_result", tool_use_id = msg.ToolCallId, content = SseHelper.TruncateToolResult(msg.Content ?? ""), is_error = msg.IsError }
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@@ -0,0 +1,121 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Linq;
|
||||
using System.Threading.Tasks;
|
||||
using Dapper;
|
||||
using Gott.Models;
|
||||
|
||||
namespace Gott.Services;
|
||||
|
||||
public partial class DatabaseService
|
||||
{
|
||||
// ── Agents ──
|
||||
|
||||
public async Task<AgentRecord> CreateAgent(AgentRecord agent)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
var now = DateTime.UtcNow.ToString("o");
|
||||
agent.AgentId = Guid.NewGuid().ToString("N");
|
||||
agent.CreatedAt = now;
|
||||
agent.UpdatedAt = now;
|
||||
await conn.ExecuteAsync(@"
|
||||
INSERT INTO agents (agent_id, user_id, name, avatar, avatar_small, description, model, tool_choice, custom_commands, auto_commit, use_soul, use_identity, use_user_info, use_memory, use_auto_context, use_web_search, web_search_version, allow_user_choice, task_instruction, avatar_scale, avatar_offset_x, avatar_offset_y, is_deleted, created_at, updated_at)
|
||||
VALUES (@AgentId, @UserId, @Name, @Avatar, @AvatarSmall, @Description, @Model, @ToolChoice, @CustomCommands, @AutoCommit, @UseSoul, @UseIdentity, @UseUserInfo, @UseMemory, @UseAutoContext, @UseWebSearch, @WebSearchVersion, @AllowUserChoice, @TaskInstruction, @AvatarScale, @AvatarOffsetX, @AvatarOffsetY, 0, @CreatedAt, @UpdatedAt)
|
||||
", agent);
|
||||
return agent;
|
||||
}
|
||||
|
||||
public async Task<IEnumerable<AgentRecord>> GetAgents(string userId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
return await conn.QueryAsync<AgentRecord>(
|
||||
"SELECT * FROM agents WHERE user_id = @userId AND is_deleted = 0 ORDER BY name", new { userId });
|
||||
}
|
||||
|
||||
public async Task<IEnumerable<AgentRecord>> GetAvailableAgents(string userId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
return await conn.QueryAsync<AgentRecord>(
|
||||
"SELECT * FROM agents WHERE (user_id = @userId OR is_public = 1) AND is_deleted = 0 ORDER BY name", new { userId });
|
||||
}
|
||||
|
||||
public async Task<AgentRecord> GetAgent(string agentId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
return await conn.QuerySingleOrDefaultAsync<AgentRecord>(
|
||||
"SELECT * FROM agents WHERE agent_id = @agentId AND is_deleted = 0", new { agentId });
|
||||
}
|
||||
|
||||
public async Task UpdateAgent(AgentRecord agent)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
agent.UpdatedAt = DateTime.UtcNow.ToString("o");
|
||||
await conn.ExecuteAsync(@"
|
||||
UPDATE agents SET name = @Name, avatar = @Avatar, avatar_small = @AvatarSmall, description = @Description,
|
||||
model = @Model, tool_choice = @ToolChoice,
|
||||
custom_commands = @CustomCommands, auto_commit = @AutoCommit,
|
||||
use_soul = @UseSoul, use_identity = @UseIdentity, use_user_info = @UseUserInfo, use_memory = @UseMemory, use_auto_context = @UseAutoContext, use_web_search = @UseWebSearch, web_search_version = @WebSearchVersion,
|
||||
allow_user_choice = @AllowUserChoice, task_instruction = @TaskInstruction,
|
||||
avatar_scale = @AvatarScale, avatar_offset_x = @AvatarOffsetX, avatar_offset_y = @AvatarOffsetY,
|
||||
is_public = @IsPublic, updated_at = @UpdatedAt
|
||||
WHERE agent_id = @AgentId AND user_id = @UserId
|
||||
", agent);
|
||||
}
|
||||
|
||||
public async Task DeleteAgent(string agentId, string userId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
await conn.ExecuteAsync(
|
||||
"UPDATE agents SET is_deleted = 1, updated_at = @now WHERE agent_id = @agentId AND user_id = @userId",
|
||||
new { agentId, userId, now = DateTime.UtcNow.ToString("o") });
|
||||
}
|
||||
|
||||
|
||||
// ── Agent Skills ──
|
||||
|
||||
public async Task<IEnumerable<string>> GetAgentSkills(string agentId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
return await conn.QueryAsync<string>(
|
||||
"SELECT skill_slug FROM agent_skills WHERE agent_id = @agentId ORDER BY skill_slug",
|
||||
new { agentId });
|
||||
}
|
||||
|
||||
public async Task SetAgentSkills(string agentId, IEnumerable<string> skillSlugs)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
using var tx = conn.BeginTransaction();
|
||||
|
||||
await conn.ExecuteAsync("DELETE FROM agent_skills WHERE agent_id = @agentId", new { agentId }, tx);
|
||||
|
||||
var now = DateTime.UtcNow.ToString("o");
|
||||
foreach (var slug in skillSlugs)
|
||||
{
|
||||
await conn.ExecuteAsync(@"
|
||||
INSERT INTO agent_skills (agent_id, skill_slug, created_at)
|
||||
VALUES (@agentId, @slug, @now)",
|
||||
new { agentId, slug, now }, tx);
|
||||
}
|
||||
|
||||
tx.Commit();
|
||||
}
|
||||
|
||||
public async Task AddAgentSkill(string agentId, string skillSlug)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
var now = DateTime.UtcNow.ToString("o");
|
||||
await conn.ExecuteAsync(@"
|
||||
INSERT OR IGNORE INTO agent_skills (agent_id, skill_slug, created_at)
|
||||
VALUES (@agentId, @skillSlug, @now)",
|
||||
new { agentId, skillSlug, now });
|
||||
}
|
||||
|
||||
public async Task RemoveAgentSkill(string agentId, string skillSlug)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
await conn.ExecuteAsync(
|
||||
"DELETE FROM agent_skills WHERE agent_id = @agentId AND skill_slug = @skillSlug",
|
||||
new { agentId, skillSlug });
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,422 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.IO;
|
||||
using System.Linq;
|
||||
using System.Threading.Tasks;
|
||||
using Dapper;
|
||||
using Gott.Models;
|
||||
using Microsoft.Data.Sqlite;
|
||||
|
||||
namespace Gott.Services;
|
||||
|
||||
public partial class DatabaseService
|
||||
{
|
||||
// ── OpenClaw Import ──
|
||||
|
||||
public async Task CreateChainForImport(string chainId, string sessionId, string agentId, string model, int inputTokens, int outputTokens, int cacheCreationTokens, int cacheReadTokens, string createdAt)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
await conn.ExecuteAsync(@"
|
||||
INSERT INTO chains (chain_id, session_id, agent_id, model, input_tokens, output_tokens, cache_creation_tokens, cache_read_tokens, iterations, stop_reason, duration_ms, plan_mode, created_at)
|
||||
VALUES (@chainId, @sessionId, @agentId, @model, @inputTokens, @outputTokens, @cacheCreationTokens, @cacheReadTokens, 0, 'end_turn', 0, 0, @createdAt)
|
||||
", new { chainId, sessionId, agentId, model, inputTokens, outputTokens, cacheCreationTokens, cacheReadTokens, createdAt });
|
||||
}
|
||||
|
||||
public async Task CreateMessageForImport(string chainId, string role, string type, string content, string createdAt, string tag)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
await conn.ExecuteAsync(@"
|
||||
INSERT INTO messages (chain_id, role, type, content, is_error, iteration, attachment_count, tag, created_at)
|
||||
VALUES (@chainId, @role, @type, @content, 0, 0, 0, @tag, @createdAt)
|
||||
", new { chainId, role, type, content, tag, createdAt });
|
||||
}
|
||||
|
||||
public async Task DeleteOpenClawImport(string sessionId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
using var tx = conn.BeginTransaction();
|
||||
|
||||
// Get all chain_ids that belong to this session and have imported messages
|
||||
var importedChainIds = (await conn.QueryAsync<string>(@"
|
||||
SELECT DISTINCT c.chain_id
|
||||
FROM chains c
|
||||
INNER JOIN messages m ON m.chain_id = c.chain_id
|
||||
WHERE c.session_id = @sessionId AND m.tag = 'openclaw-import'
|
||||
", new { sessionId }, tx)).ToList();
|
||||
|
||||
if (importedChainIds.Count > 0)
|
||||
{
|
||||
// Delete messages for these chains
|
||||
await conn.ExecuteAsync(
|
||||
"DELETE FROM messages WHERE chain_id IN @chainIds",
|
||||
new { chainIds = importedChainIds }, tx);
|
||||
|
||||
// Delete the chains themselves
|
||||
await conn.ExecuteAsync(
|
||||
"DELETE FROM chains WHERE chain_id IN @chainIds",
|
||||
new { chainIds = importedChainIds }, tx);
|
||||
}
|
||||
|
||||
tx.Commit();
|
||||
}
|
||||
|
||||
|
||||
// ── Telegram Bots ──
|
||||
|
||||
private void MigrateTelegramBotsFromSessions(SqliteConnection conn)
|
||||
{
|
||||
// Prüfe ob es Sessions mit bot_token gibt, die noch kein telegram_bot_id haben
|
||||
var sessions = conn.Query<dynamic>(
|
||||
"SELECT session_id, user_id, name, telegram_bot_token, telegram_webhook_secret FROM sessions WHERE telegram_bot_token IS NOT NULL AND telegram_bot_token != '' AND (telegram_bot_id IS NULL OR telegram_bot_id = '')").ToList();
|
||||
|
||||
if (sessions.Count == 0) return;
|
||||
|
||||
var now = DateTime.UtcNow.ToString("o");
|
||||
foreach (var s in sessions)
|
||||
{
|
||||
string token = s.telegram_bot_token;
|
||||
string userId = s.user_id ?? "";
|
||||
string sessionName = s.name ?? "Unbenannt";
|
||||
string secret = s.telegram_webhook_secret ?? Guid.NewGuid().ToString("N");
|
||||
|
||||
// Prüfe ob für diesen Token bereits ein Bot existiert
|
||||
var existing = conn.QuerySingleOrDefault<dynamic>(
|
||||
"SELECT bot_id FROM telegram_bots WHERE bot_token = @token", new { token });
|
||||
|
||||
string botId;
|
||||
if (existing != null)
|
||||
{
|
||||
botId = existing.bot_id;
|
||||
}
|
||||
else
|
||||
{
|
||||
botId = Guid.NewGuid().ToString();
|
||||
conn.Execute(@"
|
||||
INSERT INTO telegram_bots (bot_id, user_id, name, bot_token, webhook_secret, created_at, updated_at)
|
||||
VALUES (@botId, @userId, @name, @token, @secret, @now, @now)",
|
||||
new { botId, userId, name = $"Bot von {sessionName}", token, secret, now });
|
||||
}
|
||||
|
||||
conn.Execute(
|
||||
"UPDATE sessions SET telegram_bot_id = @botId, updated_at = @now WHERE session_id = @sessionId",
|
||||
new { botId, sessionId = (string)s.session_id, now });
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<IEnumerable<TelegramBotRecord>> GetTelegramBots(string userId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
return await conn.QueryAsync<TelegramBotRecord>(
|
||||
"SELECT * FROM telegram_bots WHERE user_id = @userId ORDER BY created_at DESC",
|
||||
new { userId });
|
||||
}
|
||||
|
||||
public async Task<IEnumerable<TelegramBotRecord>> GetAvailableTelegramBots(string userId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
return await conn.QueryAsync<TelegramBotRecord>(
|
||||
"SELECT * FROM telegram_bots WHERE user_id = @userId OR is_public = 1 ORDER BY created_at DESC",
|
||||
new { userId });
|
||||
}
|
||||
|
||||
public async Task<TelegramBotRecord> GetTelegramBot(string botId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
return await conn.QuerySingleOrDefaultAsync<TelegramBotRecord>(
|
||||
"SELECT * FROM telegram_bots WHERE bot_id = @botId",
|
||||
new { botId });
|
||||
}
|
||||
|
||||
public async Task<TelegramBotRecord> CreateTelegramBot(string userId, string name, string botToken)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
var botId = Guid.NewGuid().ToString();
|
||||
var secret = Guid.NewGuid().ToString("N");
|
||||
var now = DateTime.UtcNow.ToString("o");
|
||||
await conn.ExecuteAsync(@"
|
||||
INSERT INTO telegram_bots (bot_id, user_id, name, bot_token, webhook_secret, created_at, updated_at)
|
||||
VALUES (@botId, @userId, @name, @botToken, @secret, @now, @now)",
|
||||
new { botId, userId, name, botToken, secret, now });
|
||||
return await GetTelegramBot(botId);
|
||||
}
|
||||
|
||||
public async Task UpdateTelegramBot(string botId, string name, string botToken, bool isPublic)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
var now = DateTime.UtcNow.ToString("o");
|
||||
await conn.ExecuteAsync(@"
|
||||
UPDATE telegram_bots SET name = @name, bot_token = @botToken, is_public = @isPublic, updated_at = @now
|
||||
WHERE bot_id = @botId",
|
||||
new { botId, name, botToken, isPublic, now });
|
||||
}
|
||||
|
||||
public async Task DeleteTelegramBot(string botId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
var now = DateTime.UtcNow.ToString("o");
|
||||
// Erst alle Session-Verknüpfungen entfernen
|
||||
await conn.ExecuteAsync(
|
||||
"UPDATE sessions SET telegram_bot_id = NULL, updated_at = @now WHERE telegram_bot_id = @botId",
|
||||
new { botId, now });
|
||||
await conn.ExecuteAsync("DELETE FROM telegram_bots WHERE bot_id = @botId", new { botId });
|
||||
}
|
||||
|
||||
public async Task<int> GetTelegramBotSessionCount(string botId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
return await conn.ExecuteScalarAsync<int>(
|
||||
"SELECT COUNT(*) FROM sessions WHERE telegram_bot_id = @botId AND is_deleted = 0",
|
||||
new { botId });
|
||||
}
|
||||
|
||||
public async Task SetSessionTelegramBot(string sessionId, string botId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
var now = DateTime.UtcNow.ToString("o");
|
||||
await conn.ExecuteAsync(
|
||||
"UPDATE sessions SET telegram_bot_id = @botId, updated_at = @now WHERE session_id = @sessionId",
|
||||
new { sessionId, botId = (object)botId ?? DBNull.Value, now });
|
||||
}
|
||||
|
||||
public async Task UpdateSessionTelegramChat(string sessionId, string chatId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
var now = DateTime.UtcNow.ToString("o");
|
||||
await conn.ExecuteAsync(@"
|
||||
UPDATE sessions SET telegram_chat_id = @chatId, updated_at = @now
|
||||
WHERE session_id = @sessionId",
|
||||
new { sessionId, chatId, now });
|
||||
}
|
||||
|
||||
public async Task<IEnumerable<SessionRecord>> GetSessionsByTelegramBot(string botId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
return await conn.QueryAsync<SessionRecord>(
|
||||
"SELECT * FROM sessions WHERE telegram_bot_id = @botId AND is_deleted = 0",
|
||||
new { botId });
|
||||
}
|
||||
|
||||
public async Task<SessionRecord> FindSessionForTelegramChat(string botId, long chatId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
var chatIdStr = chatId.ToString();
|
||||
return await conn.QuerySingleOrDefaultAsync<SessionRecord>(
|
||||
"SELECT * FROM sessions WHERE telegram_bot_id = @botId AND telegram_chat_id = @chatIdStr AND is_deleted = 0",
|
||||
new { botId, chatIdStr });
|
||||
}
|
||||
|
||||
public async Task<IEnumerable<SessionRecord>> FindUnlinkedSessionsForBot(string botId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
return await conn.QueryAsync<SessionRecord>(
|
||||
"SELECT * FROM sessions WHERE telegram_bot_id = @botId AND (telegram_chat_id IS NULL OR telegram_chat_id = '') AND is_deleted = 0",
|
||||
new { botId });
|
||||
}
|
||||
|
||||
// ── Cronjobs ──
|
||||
|
||||
public async Task<CronjobRecord> CreateCronjob(CronjobRecord cronjob)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
var now = DateTime.UtcNow.ToString("o");
|
||||
cronjob.CronjobId = Guid.NewGuid().ToString("N");
|
||||
cronjob.CreatedAt = now;
|
||||
cronjob.UpdatedAt = now;
|
||||
await conn.ExecuteAsync(@"
|
||||
INSERT INTO cronjobs (cronjob_id, user_id, session_id, name, cron_expression, prompt, notify_session_id, is_enabled, next_run_at, created_at, updated_at)
|
||||
VALUES (@CronjobId, @UserId, @SessionId, @Name, @CronExpression, @Prompt, @NotifySessionId, @IsEnabled, @NextRunAt, @CreatedAt, @UpdatedAt)
|
||||
", cronjob);
|
||||
return cronjob;
|
||||
}
|
||||
|
||||
public async Task<IEnumerable<CronjobRecord>> GetCronjobs(string userId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
return await conn.QueryAsync<CronjobRecord>(@"
|
||||
SELECT c.*, s.name AS session_name, ns.name AS notify_session_name
|
||||
FROM cronjobs c
|
||||
LEFT JOIN sessions s ON c.session_id = s.session_id
|
||||
LEFT JOIN sessions ns ON c.notify_session_id = ns.session_id
|
||||
WHERE c.user_id = @userId
|
||||
ORDER BY c.name",
|
||||
new { userId });
|
||||
}
|
||||
|
||||
public async Task<CronjobRecord> GetCronjob(string cronjobId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
return await conn.QuerySingleOrDefaultAsync<CronjobRecord>(@"
|
||||
SELECT c.*, s.name AS session_name, ns.name AS notify_session_name
|
||||
FROM cronjobs c
|
||||
LEFT JOIN sessions s ON c.session_id = s.session_id
|
||||
LEFT JOIN sessions ns ON c.notify_session_id = ns.session_id
|
||||
WHERE c.cronjob_id = @cronjobId",
|
||||
new { cronjobId });
|
||||
}
|
||||
|
||||
public async Task UpdateCronjob(CronjobRecord cronjob)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
cronjob.UpdatedAt = DateTime.UtcNow.ToString("o");
|
||||
await conn.ExecuteAsync(@"
|
||||
UPDATE cronjobs SET name = @Name, session_id = @SessionId, cron_expression = @CronExpression,
|
||||
prompt = @Prompt, notify_session_id = @NotifySessionId, is_enabled = @IsEnabled, next_run_at = @NextRunAt, updated_at = @UpdatedAt
|
||||
WHERE cronjob_id = @CronjobId AND user_id = @UserId",
|
||||
cronjob);
|
||||
}
|
||||
|
||||
public async Task DeleteCronjob(string cronjobId, string userId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
await conn.ExecuteAsync(
|
||||
"DELETE FROM cronjobs WHERE cronjob_id = @cronjobId AND user_id = @userId",
|
||||
new { cronjobId, userId });
|
||||
}
|
||||
|
||||
public async Task UpdateCronjobRun(string cronjobId, string status, string error, string nextRunAt)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
await conn.ExecuteAsync(@"
|
||||
UPDATE cronjobs SET last_run_at = @now, last_run_status = @status,
|
||||
last_run_error = @error, next_run_at = @nextRunAt, updated_at = @now
|
||||
WHERE cronjob_id = @cronjobId",
|
||||
new { cronjobId, status, error, nextRunAt, now = DateTime.UtcNow.ToString("o") });
|
||||
}
|
||||
|
||||
public async Task<IEnumerable<CronjobRecord>> GetDueCronjobs()
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
var now = DateTime.UtcNow.ToString("o");
|
||||
return await conn.QueryAsync<CronjobRecord>(@"
|
||||
SELECT c.*, s.name AS session_name, ns.name AS notify_session_name
|
||||
FROM cronjobs c
|
||||
LEFT JOIN sessions s ON c.session_id = s.session_id
|
||||
LEFT JOIN sessions ns ON c.notify_session_id = ns.session_id
|
||||
WHERE c.is_enabled = 1 AND c.next_run_at IS NOT NULL AND c.next_run_at <= @now",
|
||||
new { now });
|
||||
}
|
||||
|
||||
// ── API Keys ──
|
||||
|
||||
private void MigrateAnthropicTokenFromConfig(SqliteConnection conn)
|
||||
{
|
||||
// Nur migrieren wenn noch keine Anthropic-Keys existieren
|
||||
var count = conn.ExecuteScalar<int>("SELECT COUNT(*) FROM api_keys WHERE provider = 'anthropic'");
|
||||
if (count > 0) return;
|
||||
|
||||
// Token aus appsettings.Production.json lesen
|
||||
var configPath = Path.Combine(Path.GetDirectoryName(_connectionString.Replace("Data Source=", "")) ?? "", "..", "appsettings.Production.json");
|
||||
// Normaler Pfad: neben der DB-Datei im übergeordneten Ordner
|
||||
var basePath = Path.GetDirectoryName(_connectionString.Replace("Data Source=", ""));
|
||||
if (basePath != null)
|
||||
{
|
||||
var prodPath = Path.Combine(basePath, "..", "appsettings.Production.json");
|
||||
if (File.Exists(prodPath))
|
||||
{
|
||||
try
|
||||
{
|
||||
var json = File.ReadAllText(prodPath);
|
||||
using var doc = System.Text.Json.JsonDocument.Parse(json);
|
||||
if (doc.RootElement.TryGetProperty("AnthropicToken", out var tokenEl))
|
||||
{
|
||||
var token = tokenEl.GetString();
|
||||
if (!string.IsNullOrWhiteSpace(token))
|
||||
{
|
||||
var now = DateTime.UtcNow.ToString("o");
|
||||
conn.Execute(@"
|
||||
INSERT INTO api_keys (key_id, provider, name, api_key, is_active, created_at, updated_at)
|
||||
VALUES (@keyId, 'anthropic', 'Migriert aus Config', @token, 1, @now, @now)",
|
||||
new { keyId = Guid.NewGuid().ToString("N"), token, now });
|
||||
Console.WriteLine("[Migration] AnthropicToken aus appsettings.Production.json → api_keys migriert");
|
||||
}
|
||||
}
|
||||
}
|
||||
catch { /* Config nicht lesbar – ignorieren */ }
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public async Task<IEnumerable<ApiKeyRecord>> GetApiKeys()
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
return await conn.QueryAsync<ApiKeyRecord>(
|
||||
"SELECT * FROM api_keys ORDER BY provider, name");
|
||||
}
|
||||
|
||||
public async Task<IEnumerable<ApiKeyRecord>> GetApiKeysByProvider(string provider)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
return await conn.QueryAsync<ApiKeyRecord>(
|
||||
"SELECT * FROM api_keys WHERE provider = @provider ORDER BY name",
|
||||
new { provider });
|
||||
}
|
||||
|
||||
public async Task<ApiKeyRecord> GetActiveApiKey(string provider)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
return await conn.QuerySingleOrDefaultAsync<ApiKeyRecord>(
|
||||
"SELECT * FROM api_keys WHERE provider = @provider AND is_active = 1 LIMIT 1",
|
||||
new { provider });
|
||||
}
|
||||
|
||||
public async Task<ApiKeyRecord> GetApiKey(string keyId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
return await conn.QuerySingleOrDefaultAsync<ApiKeyRecord>(
|
||||
"SELECT * FROM api_keys WHERE key_id = @keyId",
|
||||
new { keyId });
|
||||
}
|
||||
|
||||
public async Task CreateApiKey(ApiKeyRecord record)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
await conn.ExecuteAsync(@"
|
||||
INSERT INTO api_keys (key_id, provider, name, api_key, is_active, created_at, updated_at)
|
||||
VALUES (@KeyId, @Provider, @Name, @ApiKey, @IsActive, @CreatedAt, @UpdatedAt)",
|
||||
record);
|
||||
}
|
||||
|
||||
public async Task UpdateApiKey(string keyId, string name, string apiKey = null)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
if (!string.IsNullOrWhiteSpace(apiKey))
|
||||
{
|
||||
await conn.ExecuteAsync(
|
||||
"UPDATE api_keys SET name = @name, api_key = @apiKey, updated_at = @now WHERE key_id = @keyId",
|
||||
new { keyId, name, apiKey, now = DateTime.UtcNow.ToString("o") });
|
||||
}
|
||||
else
|
||||
{
|
||||
await conn.ExecuteAsync(
|
||||
"UPDATE api_keys SET name = @name, updated_at = @now WHERE key_id = @keyId",
|
||||
new { keyId, name, now = DateTime.UtcNow.ToString("o") });
|
||||
}
|
||||
}
|
||||
|
||||
public async Task SetActiveApiKey(string keyId, string provider)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
// Erst alle deaktivieren
|
||||
await conn.ExecuteAsync(
|
||||
"UPDATE api_keys SET is_active = 0, updated_at = @now WHERE provider = @provider",
|
||||
new { provider, now = DateTime.UtcNow.ToString("o") });
|
||||
// Dann den gewählten aktivieren
|
||||
await conn.ExecuteAsync(
|
||||
"UPDATE api_keys SET is_active = 1, updated_at = @now WHERE key_id = @keyId",
|
||||
new { keyId, now = DateTime.UtcNow.ToString("o") });
|
||||
}
|
||||
|
||||
public async Task DeactivateApiKey(string keyId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
await conn.ExecuteAsync(
|
||||
"UPDATE api_keys SET is_active = 0, updated_at = @now WHERE key_id = @keyId",
|
||||
new { keyId, now = DateTime.UtcNow.ToString("o") });
|
||||
}
|
||||
|
||||
public async Task DeleteApiKey(string keyId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
await conn.ExecuteAsync("DELETE FROM api_keys WHERE key_id = @keyId", new { keyId });
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,439 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Linq;
|
||||
using System.Text.Json;
|
||||
using System.Threading.Tasks;
|
||||
using Dapper;
|
||||
using Gott.Models;
|
||||
|
||||
namespace Gott.Services;
|
||||
|
||||
public partial class DatabaseService
|
||||
{
|
||||
// ── Sessions ──
|
||||
|
||||
public async Task<SessionRecord> CreateSession(string userId, string name = null, string description = null, string image = null, string imageSmall = null, string agentId = null, string workspace = null, string context = null, string customCommands = null, double avatarScale = 1.0, double avatarOffsetX = 0, double avatarOffsetY = 0, string reviewAgentId = null, int maxReviewIterations = 3, string finalizerAgentId = null, string preparerAgentId = null)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
var now = DateTime.UtcNow.ToString("o");
|
||||
var session = new SessionRecord
|
||||
{
|
||||
SessionId = Guid.NewGuid().ToString("N"),
|
||||
UserId = userId,
|
||||
Name = name,
|
||||
Description = description,
|
||||
Image = image,
|
||||
ImageSmall = imageSmall,
|
||||
AgentId = agentId,
|
||||
Workspace = workspace,
|
||||
Context = context,
|
||||
CustomCommands = customCommands,
|
||||
AvatarScale = avatarScale,
|
||||
AvatarOffsetX = avatarOffsetX,
|
||||
AvatarOffsetY = avatarOffsetY,
|
||||
ReviewAgentId = reviewAgentId,
|
||||
MaxReviewIterations = maxReviewIterations,
|
||||
FinalizerAgentId = finalizerAgentId,
|
||||
PreparerAgentId = preparerAgentId,
|
||||
CreatedAt = now,
|
||||
UpdatedAt = now
|
||||
};
|
||||
await conn.ExecuteAsync(@"
|
||||
INSERT INTO sessions (session_id, user_id, name, description, image, image_small, agent_id, workspace, context, custom_commands, avatar_scale, avatar_offset_x, avatar_offset_y, review_agent_id, max_review_iterations, finalizer_agent_id, preparer_agent_id, created_at, updated_at)
|
||||
VALUES (@SessionId, @UserId, @Name, @Description, @Image, @ImageSmall, @AgentId, @Workspace, @Context, @CustomCommands, @AvatarScale, @AvatarOffsetX, @AvatarOffsetY, @ReviewAgentId, @MaxReviewIterations, @FinalizerAgentId, @PreparerAgentId, @CreatedAt, @UpdatedAt)
|
||||
", session);
|
||||
return session;
|
||||
}
|
||||
|
||||
public async Task<SessionRecord> GetSessionByName(string name)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
var sessions = await conn.QueryAsync<SessionRecord>(
|
||||
"SELECT * FROM sessions WHERE is_deleted = 0 AND LOWER(name) = LOWER(@name) ORDER BY updated_at DESC LIMIT 1",
|
||||
new { name });
|
||||
return sessions.FirstOrDefault();
|
||||
}
|
||||
|
||||
public async Task<List<SessionWithStats>> GetSessionsWithStats(string userId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
var sessions = await conn.QueryAsync<SessionRecord>(
|
||||
"SELECT * FROM sessions WHERE user_id = @userId AND is_deleted = 0 ORDER BY updated_at DESC", new { userId });
|
||||
|
||||
var results = new List<SessionWithStats>();
|
||||
foreach (var s in sessions)
|
||||
{
|
||||
var activeCount = await conn.ExecuteScalarAsync<int>(@"
|
||||
SELECT COUNT(*) FROM messages m
|
||||
JOIN chains c ON m.chain_id = c.chain_id
|
||||
WHERE c.session_id = @sessionId AND c.archive_id IS NULL AND m.type = 'text'",
|
||||
new { sessionId = s.SessionId });
|
||||
|
||||
var totalCount = await conn.ExecuteScalarAsync<int>(@"
|
||||
SELECT COUNT(*) FROM messages m
|
||||
JOIN chains c ON m.chain_id = c.chain_id
|
||||
WHERE c.session_id = @sessionId AND m.type = 'text'",
|
||||
new { sessionId = s.SessionId });
|
||||
|
||||
results.Add(new SessionWithStats
|
||||
{
|
||||
Session = s,
|
||||
MessageCount = activeCount,
|
||||
ArchiveMessageCount = totalCount - activeCount,
|
||||
TotalMessageCount = totalCount,
|
||||
TelegramConfigured = !string.IsNullOrEmpty(s.TelegramBotId),
|
||||
TelegramLinked = !string.IsNullOrEmpty(s.TelegramChatId)
|
||||
});
|
||||
}
|
||||
return results;
|
||||
}
|
||||
|
||||
public async Task<SessionRecord> GetSession(string sessionId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
return await conn.QuerySingleOrDefaultAsync<SessionRecord>(
|
||||
"SELECT * FROM sessions WHERE session_id = @sessionId AND is_deleted = 0", new { sessionId });
|
||||
}
|
||||
|
||||
public async Task UpdateSessionTimestamp(string sessionId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
await conn.ExecuteAsync(
|
||||
"UPDATE sessions SET updated_at = @now WHERE session_id = @sessionId",
|
||||
new { sessionId, now = DateTime.UtcNow.ToString("o") });
|
||||
}
|
||||
|
||||
public async Task UpdateSession(string sessionId, System.Text.Json.JsonElement body, ImageService imageService = null)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
var sets = new List<string> { "updated_at = @now" };
|
||||
var p = new Dapper.DynamicParameters();
|
||||
p.Add("sessionId", sessionId);
|
||||
p.Add("now", DateTime.UtcNow.ToString("o"));
|
||||
|
||||
if (body.TryGetProperty("name", out var n)) { sets.Add("name = @name"); p.Add("name", n.GetString()); }
|
||||
if (body.TryGetProperty("description", out var d)) { sets.Add("description = @desc"); p.Add("desc", d.GetString()); }
|
||||
if (body.TryGetProperty("image", out var img))
|
||||
{
|
||||
var imageData = img.GetString();
|
||||
if (imageService != null)
|
||||
{
|
||||
sets.Add("image_small = @imageSmall");
|
||||
p.Add("imageSmall", imageService.ResizeDataUrlSmall(imageData));
|
||||
imageData = imageService.ResizeDataUrl(imageData);
|
||||
}
|
||||
sets.Add("image = @image");
|
||||
p.Add("image", imageData);
|
||||
}
|
||||
if (body.TryGetProperty("agentId", out var a)) { sets.Add("agent_id = @agentId2"); p.Add("agentId2", a.GetString()); }
|
||||
if (body.TryGetProperty("workspace", out var w)) { sets.Add("workspace = @workspace"); p.Add("workspace", w.GetString()); }
|
||||
if (body.TryGetProperty("context", out var ctx)) { sets.Add("context = @context"); p.Add("context", ctx.GetString()); }
|
||||
if (body.TryGetProperty("customCommands", out var cc)) { sets.Add("custom_commands = @customCommands"); p.Add("customCommands", cc.ValueKind == System.Text.Json.JsonValueKind.Null ? null : cc.GetRawText()); }
|
||||
if (body.TryGetProperty("avatarScale", out var asc)) { sets.Add("avatar_scale = @avatarScale"); p.Add("avatarScale", asc.GetDouble()); }
|
||||
if (body.TryGetProperty("avatarOffsetX", out var aox)) { sets.Add("avatar_offset_x = @avatarOffsetX"); p.Add("avatarOffsetX", aox.GetDouble()); }
|
||||
if (body.TryGetProperty("avatarOffsetY", out var aoy)) { sets.Add("avatar_offset_y = @avatarOffsetY"); p.Add("avatarOffsetY", aoy.GetDouble()); }
|
||||
if (body.TryGetProperty("reviewAgentId", out var ra)) { sets.Add("review_agent_id = @reviewAgentId"); p.Add("reviewAgentId", ra.ValueKind == System.Text.Json.JsonValueKind.Null ? null : ra.GetString()); }
|
||||
if (body.TryGetProperty("maxReviewIterations", out var mri)) { sets.Add("max_review_iterations = @maxReviewIterations"); p.Add("maxReviewIterations", mri.GetInt32()); }
|
||||
if (body.TryGetProperty("finalizerAgentId", out var fa)) { sets.Add("finalizer_agent_id = @finalizerAgentId"); p.Add("finalizerAgentId", fa.ValueKind == System.Text.Json.JsonValueKind.Null ? null : fa.GetString()); }
|
||||
if (body.TryGetProperty("preparerAgentId", out var pa)) { sets.Add("preparer_agent_id = @preparerAgentId"); p.Add("preparerAgentId", pa.ValueKind == System.Text.Json.JsonValueKind.Null ? null : pa.GetString()); }
|
||||
|
||||
await conn.ExecuteAsync($"UPDATE sessions SET {string.Join(", ", sets)} WHERE session_id = @sessionId", p);
|
||||
}
|
||||
|
||||
public async Task DeleteSession(string sessionId, string userId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
await conn.ExecuteAsync(
|
||||
"UPDATE sessions SET is_deleted = 1, updated_at = @now WHERE session_id = @sessionId AND user_id = @userId",
|
||||
new { sessionId, userId, now = DateTime.UtcNow.ToString("o") });
|
||||
}
|
||||
|
||||
// ── Archives ──
|
||||
|
||||
public async Task<ArchiveRecord> CreateArchive(string sessionId, string name = null)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
|
||||
var chainCount = await conn.ExecuteScalarAsync<int>(
|
||||
"SELECT COUNT(*) FROM chains WHERE archive_id IS NULL AND session_id = @sessionId",
|
||||
new { sessionId });
|
||||
if (chainCount == 0) return null;
|
||||
|
||||
var messageCount = await conn.ExecuteScalarAsync<int>(@"
|
||||
SELECT COUNT(*) FROM messages
|
||||
WHERE type = 'text' AND chain_id IN (SELECT chain_id FROM chains WHERE archive_id IS NULL AND session_id = @sessionId)",
|
||||
new { sessionId });
|
||||
|
||||
var archive = new ArchiveRecord
|
||||
{
|
||||
ArchiveId = Guid.NewGuid().ToString("N"),
|
||||
SessionId = sessionId,
|
||||
Name = name,
|
||||
ChainCount = chainCount,
|
||||
MessageCount = messageCount,
|
||||
CreatedAt = DateTime.UtcNow.ToString("o")
|
||||
};
|
||||
|
||||
await conn.ExecuteAsync(@"
|
||||
INSERT INTO archives (archive_id, session_id, name, chain_count, message_count, created_at)
|
||||
VALUES (@ArchiveId, @SessionId, @Name, @ChainCount, @MessageCount, @CreatedAt)
|
||||
", archive);
|
||||
|
||||
await conn.ExecuteAsync(
|
||||
"UPDATE chains SET archive_id = @archiveId WHERE archive_id IS NULL AND session_id = @sessionId",
|
||||
new { archiveId = archive.ArchiveId, sessionId });
|
||||
|
||||
return archive;
|
||||
}
|
||||
|
||||
public async Task<IEnumerable<ArchiveRecord>> GetArchives(string sessionId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
return await conn.QueryAsync<ArchiveRecord>(
|
||||
"SELECT * FROM archives WHERE session_id = @sessionId ORDER BY id DESC",
|
||||
new { sessionId });
|
||||
}
|
||||
|
||||
public async Task<IEnumerable<ChainRecord>> GetArchiveChains(string archiveId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
return await conn.QueryAsync<ChainRecord>(
|
||||
"SELECT * FROM chains WHERE archive_id = @archiveId ORDER BY id",
|
||||
new { archiveId });
|
||||
}
|
||||
|
||||
// ── Chains ──
|
||||
|
||||
public async Task CreateChain(ChainRecord chain)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
chain.CreatedAt = DateTime.UtcNow.ToString("o");
|
||||
await conn.ExecuteAsync(@"
|
||||
INSERT INTO chains (chain_id, session_id, archive_id, agent_id, model, system_prompt, tool_choice, tool_names, input_tokens, output_tokens, iterations, stop_reason, duration_ms, plan_mode, created_at)
|
||||
VALUES (@ChainId, @SessionId, @ArchiveId, @AgentId, @Model, @SystemPrompt, @ToolChoice, @ToolNames, @InputTokens, @OutputTokens, @Iterations, @StopReason, @DurationMs, @PlanMode, @CreatedAt)
|
||||
", chain);
|
||||
}
|
||||
|
||||
public async Task UpdateChain(ChainRecord chain)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
await conn.ExecuteAsync(@"
|
||||
UPDATE chains SET
|
||||
input_tokens = @InputTokens,
|
||||
output_tokens = @OutputTokens,
|
||||
cache_creation_tokens = @CacheCreationTokens,
|
||||
cache_read_tokens = @CacheReadTokens,
|
||||
iterations = @Iterations,
|
||||
stop_reason = @StopReason,
|
||||
stream_error_info = @StreamErrorInfo,
|
||||
duration_ms = @DurationMs
|
||||
WHERE chain_id = @ChainId
|
||||
", chain);
|
||||
}
|
||||
|
||||
public async Task<ChainRecord> GetChain(string chainId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
return await conn.QuerySingleOrDefaultAsync<ChainRecord>(
|
||||
"SELECT * FROM chains WHERE chain_id = @chainId", new { chainId }
|
||||
);
|
||||
}
|
||||
|
||||
public async Task<IEnumerable<ChainRecord>> GetActiveChains(string sessionId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
return await conn.QueryAsync<ChainRecord>(
|
||||
"SELECT * FROM chains WHERE session_id = @sessionId AND archive_id IS NULL ORDER BY id DESC",
|
||||
new { sessionId });
|
||||
}
|
||||
|
||||
// ── Messages ──
|
||||
|
||||
public async Task<long> SaveMessage(MessageRecord record)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
record.CreatedAt = DateTime.UtcNow.ToString("o");
|
||||
return await conn.ExecuteScalarAsync<long>(@"
|
||||
INSERT INTO messages (chain_id, role, type, content, tool_name, tool_call_id, is_error, iteration, attachment_count, tag, thought_signature, created_at)
|
||||
VALUES (@ChainId, @Role, @Type, @Content, @ToolName, @ToolCallId, @IsError, @Iteration, @AttachmentCount, @Tag, @ThoughtSignature, @CreatedAt);
|
||||
SELECT last_insert_rowid();
|
||||
", record);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Speichert Message + Attachments in einer Transaktion (1x fsync statt N+1).
|
||||
/// </summary>
|
||||
public async Task<long> SaveMessageWithAttachments(MessageRecord record, List<AttachmentRecord> attachments)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
using var tx = conn.BeginTransaction();
|
||||
|
||||
record.CreatedAt = DateTime.UtcNow.ToString("o");
|
||||
record.AttachmentCount = attachments?.Count ?? 0;
|
||||
if (attachments != null && attachments.Count > 0)
|
||||
{
|
||||
var names = attachments.Select(a => a.FileName ?? "").ToArray();
|
||||
record.AttachmentNames = System.Text.Json.JsonSerializer.Serialize(names);
|
||||
}
|
||||
var messageId = await conn.ExecuteScalarAsync<long>(@"
|
||||
INSERT INTO messages (chain_id, role, type, content, tool_name, tool_call_id, is_error, iteration, attachment_count, attachment_names, tag, created_at)
|
||||
VALUES (@ChainId, @Role, @Type, @Content, @ToolName, @ToolCallId, @IsError, @Iteration, @AttachmentCount, @AttachmentNames, @Tag, @CreatedAt);
|
||||
SELECT last_insert_rowid();
|
||||
", record, tx);
|
||||
|
||||
if (attachments != null)
|
||||
{
|
||||
var now = DateTime.UtcNow.ToString("o");
|
||||
foreach (var att in attachments)
|
||||
{
|
||||
att.MessageId = messageId;
|
||||
att.CreatedAt = now;
|
||||
await conn.ExecuteAsync(@"
|
||||
INSERT INTO attachments (message_id, type, media_type, file_name, data, created_at)
|
||||
VALUES (@MessageId, @Type, @MediaType, @FileName, @Data, @CreatedAt)
|
||||
", att, tx);
|
||||
}
|
||||
}
|
||||
|
||||
tx.Commit();
|
||||
return messageId;
|
||||
}
|
||||
|
||||
|
||||
// ── Attachments ──
|
||||
|
||||
public async Task<IEnumerable<AttachmentRecord>> GetAttachments(long messageId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
return await conn.QueryAsync<AttachmentRecord>(
|
||||
"SELECT * FROM attachments WHERE message_id = @messageId ORDER BY id",
|
||||
new { messageId });
|
||||
}
|
||||
|
||||
public async Task<IEnumerable<AttachmentRecord>> GetAttachmentsForMessages(IEnumerable<long> messageIds)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
return await conn.QueryAsync<AttachmentRecord>(
|
||||
"SELECT * FROM attachments WHERE message_id IN @messageIds ORDER BY id",
|
||||
new { messageIds });
|
||||
}
|
||||
|
||||
public async Task<IEnumerable<MessageRecord>> GetChainMessages(string chainId, bool includeDeleted = false)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
var deletedFilter = includeDeleted ? "" : " AND deleted_on IS NULL";
|
||||
return await conn.QueryAsync<MessageRecord>(
|
||||
$"SELECT * FROM messages WHERE chain_id = @chainId{deletedFilter} ORDER BY id", new { chainId }
|
||||
);
|
||||
}
|
||||
|
||||
public async Task<int> GetActiveMessageCount(string sessionId, bool includeDeleted = false)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
var deletedFilter = includeDeleted ? "" : "AND m.deleted_on IS NULL";
|
||||
return await conn.ExecuteScalarAsync<int>($@"
|
||||
SELECT COUNT(*) FROM messages m
|
||||
JOIN chains c ON m.chain_id = c.chain_id
|
||||
WHERE c.archive_id IS NULL AND c.session_id = @sessionId AND m.type = 'text' {deletedFilter}",
|
||||
new { sessionId });
|
||||
}
|
||||
|
||||
public async Task<List<MessageRecord>> GetLatestActiveMessages(string sessionId, int limit, bool includeDeleted = false)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
var deletedFilter = includeDeleted ? "" : "AND m.deleted_on IS NULL";
|
||||
var messages = await conn.QueryAsync<MessageRecord>($@"
|
||||
SELECT m.*, (CASE WHEN m.tag = 'openclaw-import' THEN m.id - 10000000000 ELSE m.id END) AS sort_key FROM messages m
|
||||
JOIN chains c ON m.chain_id = c.chain_id
|
||||
WHERE c.archive_id IS NULL AND c.session_id = @sessionId {deletedFilter}
|
||||
ORDER BY sort_key DESC LIMIT @limit",
|
||||
new { sessionId, limit });
|
||||
return messages.Reverse().ToList();
|
||||
}
|
||||
|
||||
public async Task<int> GetActiveMessageCountBeforeSortKey(string sessionId, long beforeSortKey, bool includeDeleted = false)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
var deletedFilter = includeDeleted ? "" : "AND m.deleted_on IS NULL";
|
||||
return await conn.ExecuteScalarAsync<int>($@"
|
||||
SELECT COUNT(*) FROM messages m
|
||||
JOIN chains c ON m.chain_id = c.chain_id
|
||||
WHERE (CASE WHEN m.tag = 'openclaw-import' THEN m.id - 10000000000 ELSE m.id END) < @beforeSortKey AND c.archive_id IS NULL AND c.session_id = @sessionId AND m.type = 'text' {deletedFilter}",
|
||||
new { sessionId, beforeSortKey });
|
||||
}
|
||||
|
||||
public async Task<List<MessageRecord>> GetActiveMessagesBeforeSortKey(string sessionId, long beforeSortKey, int limit, bool includeDeleted = false)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
var deletedFilter = includeDeleted ? "" : "AND m.deleted_on IS NULL";
|
||||
var messages = await conn.QueryAsync<MessageRecord>($@"
|
||||
SELECT m.*, (CASE WHEN m.tag = 'openclaw-import' THEN m.id - 10000000000 ELSE m.id END) AS sort_key FROM messages m
|
||||
JOIN chains c ON m.chain_id = c.chain_id
|
||||
WHERE (CASE WHEN m.tag = 'openclaw-import' THEN m.id - 10000000000 ELSE m.id END) < @beforeSortKey AND c.archive_id IS NULL AND c.session_id = @sessionId {deletedFilter}
|
||||
ORDER BY sort_key DESC LIMIT @limit",
|
||||
new { sessionId, beforeSortKey, limit });
|
||||
return messages.Reverse().ToList();
|
||||
}
|
||||
|
||||
public async Task<IEnumerable<MessageRecord>> GetActiveMessagesAfterSortKey(string sessionId, long afterSortKey, int limit = 200, bool includeDeleted = false)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
var deletedFilter = includeDeleted ? "" : "AND m.deleted_on IS NULL";
|
||||
return await conn.QueryAsync<MessageRecord>($@"
|
||||
SELECT m.*, (CASE WHEN m.tag = 'openclaw-import' THEN m.id - 10000000000 ELSE m.id END) AS sort_key FROM messages m
|
||||
JOIN chains c ON m.chain_id = c.chain_id
|
||||
WHERE (CASE WHEN m.tag = 'openclaw-import' THEN m.id - 10000000000 ELSE m.id END) > @afterSortKey AND c.archive_id IS NULL AND c.session_id = @sessionId {deletedFilter}
|
||||
ORDER BY sort_key LIMIT @limit",
|
||||
new { sessionId, afterSortKey, limit });
|
||||
}
|
||||
|
||||
// ── Message Marking ──
|
||||
|
||||
public async Task<bool> ToggleMessageMark(long messageId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
await conn.ExecuteAsync(
|
||||
"UPDATE messages SET is_marked = CASE WHEN is_marked = 1 THEN 0 ELSE 1 END WHERE id = @messageId",
|
||||
new { messageId });
|
||||
return await conn.ExecuteScalarAsync<bool>(
|
||||
"SELECT is_marked FROM messages WHERE id = @messageId",
|
||||
new { messageId });
|
||||
}
|
||||
|
||||
// ── Message Deletion ──
|
||||
|
||||
public async Task<string> ToggleMessageDelete(long messageId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
var current = await conn.ExecuteScalarAsync<string>(
|
||||
"SELECT deleted_on FROM messages WHERE id = @messageId", new { messageId });
|
||||
string newValue;
|
||||
if (current == null)
|
||||
{
|
||||
newValue = DateTime.UtcNow.ToString("o");
|
||||
await conn.ExecuteAsync(
|
||||
"UPDATE messages SET deleted_on = @newValue WHERE id = @messageId",
|
||||
new { messageId, newValue });
|
||||
}
|
||||
else
|
||||
{
|
||||
newValue = null;
|
||||
await conn.ExecuteAsync(
|
||||
"UPDATE messages SET deleted_on = NULL WHERE id = @messageId",
|
||||
new { messageId });
|
||||
}
|
||||
return newValue;
|
||||
}
|
||||
|
||||
public async Task<IEnumerable<MessageRecord>> GetMarkedMessages(string sessionId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
return await conn.QueryAsync<MessageRecord>(@"
|
||||
SELECT m.* FROM messages m
|
||||
JOIN chains c ON m.chain_id = c.chain_id
|
||||
WHERE c.session_id = @sessionId AND m.is_marked = 1
|
||||
ORDER BY m.id",
|
||||
new { sessionId });
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,190 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Threading.Tasks;
|
||||
using Dapper;
|
||||
using Gott.Models;
|
||||
|
||||
namespace Gott.Services;
|
||||
|
||||
public partial class DatabaseService
|
||||
{
|
||||
// ── Shared Messages ──
|
||||
|
||||
public async Task<SharedMessageRecord> CreateSharedMessage(long messageId, string userId, bool includeContext, string expiresAt = null)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
var record = new SharedMessageRecord
|
||||
{
|
||||
ShareToken = Guid.NewGuid().ToString("N"),
|
||||
MessageId = messageId,
|
||||
UserId = userId,
|
||||
IncludeContext = includeContext,
|
||||
ExpiresAt = expiresAt,
|
||||
CreatedAt = DateTime.UtcNow.ToString("o")
|
||||
};
|
||||
await conn.ExecuteAsync(@"
|
||||
INSERT INTO shared_messages (share_token, message_id, user_id, include_context, expires_at, created_at)
|
||||
VALUES (@ShareToken, @MessageId, @UserId, @IncludeContext, @ExpiresAt, @CreatedAt)
|
||||
", record);
|
||||
return record;
|
||||
}
|
||||
|
||||
public async Task<SharedMessageRecord> GetSharedMessage(string shareToken)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
return await conn.QuerySingleOrDefaultAsync<SharedMessageRecord>(
|
||||
"SELECT * FROM shared_messages WHERE share_token = @shareToken", new { shareToken });
|
||||
}
|
||||
|
||||
public async Task<IEnumerable<SharedMessageRecord>> GetSharedMessagesByUser(string userId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
return await conn.QueryAsync<SharedMessageRecord>(
|
||||
"SELECT * FROM shared_messages WHERE user_id = @userId ORDER BY created_at DESC", new { userId });
|
||||
}
|
||||
|
||||
public async Task DeleteSharedMessage(string shareToken, string userId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
await conn.ExecuteAsync(
|
||||
"DELETE FROM shared_messages WHERE share_token = @shareToken AND user_id = @userId",
|
||||
new { shareToken, userId });
|
||||
}
|
||||
|
||||
public async Task<MessageRecord> GetMessage(long messageId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
return await conn.QuerySingleOrDefaultAsync<MessageRecord>(
|
||||
"SELECT * FROM messages WHERE id = @messageId", new { messageId });
|
||||
}
|
||||
|
||||
public async Task<List<MessageRecord>> GetChainMessagesUpTo(string chainId, long messageId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
var result = await conn.QueryAsync<MessageRecord>(
|
||||
"SELECT * FROM messages WHERE chain_id = @chainId AND id <= @messageId ORDER BY id ASC",
|
||||
new { chainId, messageId });
|
||||
return result.AsList();
|
||||
}
|
||||
|
||||
|
||||
// ── Users ──
|
||||
|
||||
public void EnsureDefaultUser(AuthService auth)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
var count = conn.ExecuteScalar<int>("SELECT COUNT(*) FROM users");
|
||||
string adminUserId;
|
||||
if (count == 0)
|
||||
{
|
||||
adminUserId = Guid.NewGuid().ToString("N");
|
||||
var now = DateTime.UtcNow.ToString("o");
|
||||
conn.Execute(@"
|
||||
INSERT INTO users (user_id, username, password_hash, display_name, is_admin, created_at, updated_at)
|
||||
VALUES (@userId, @username, @hash, @displayName, 1, @now, @now)",
|
||||
new
|
||||
{
|
||||
userId = adminUserId,
|
||||
username = "admin",
|
||||
hash = auth.HashPassword("admin"),
|
||||
displayName = "Admin",
|
||||
now
|
||||
});
|
||||
}
|
||||
else
|
||||
{
|
||||
adminUserId = conn.ExecuteScalar<string>("SELECT user_id FROM users ORDER BY id LIMIT 1");
|
||||
}
|
||||
|
||||
// Bestehende Sessions/Agents ohne User dem ersten User zuweisen
|
||||
conn.Execute("UPDATE sessions SET user_id = @userId WHERE user_id IS NULL", new { userId = adminUserId });
|
||||
conn.Execute("UPDATE agents SET user_id = @userId WHERE user_id IS NULL", new { userId = adminUserId });
|
||||
|
||||
// Erster User ist immer Admin
|
||||
conn.Execute("UPDATE users SET is_admin = 1 WHERE user_id = @userId AND is_admin = 0", new { userId = adminUserId });
|
||||
}
|
||||
|
||||
public async Task<UserRecord> GetUserByUsername(string username)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
return await conn.QuerySingleOrDefaultAsync<UserRecord>(
|
||||
"SELECT * FROM users WHERE username = @username COLLATE NOCASE",
|
||||
new { username });
|
||||
}
|
||||
|
||||
public async Task<UserRecord> GetUser(string userId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
return await conn.QuerySingleOrDefaultAsync<UserRecord>(
|
||||
"SELECT * FROM users WHERE user_id = @userId",
|
||||
new { userId });
|
||||
}
|
||||
|
||||
public async Task<IEnumerable<UserRecord>> GetUsers()
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
return await conn.QueryAsync<UserRecord>(
|
||||
"SELECT * FROM users ORDER BY username");
|
||||
}
|
||||
|
||||
public async Task UpdateUserPassword(string userId, string passwordHash)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
await conn.ExecuteAsync(
|
||||
"UPDATE users SET password_hash = @passwordHash, updated_at = @now WHERE user_id = @userId",
|
||||
new { userId, passwordHash, now = DateTime.UtcNow.ToString("o") });
|
||||
}
|
||||
|
||||
public async Task UpdateUserProfile(string userId, string displayName, string persona)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
await conn.ExecuteAsync(
|
||||
"UPDATE users SET display_name = @displayName, persona = @persona, updated_at = @now WHERE user_id = @userId",
|
||||
new { userId, displayName, persona, now = DateTime.UtcNow.ToString("o") });
|
||||
}
|
||||
|
||||
public async Task<string> GetUserSettings(string userId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
return await conn.QuerySingleOrDefaultAsync<string>(
|
||||
"SELECT settings FROM users WHERE user_id = @userId",
|
||||
new { userId });
|
||||
}
|
||||
|
||||
public async Task UpdateUserSettings(string userId, string settings)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
await conn.ExecuteAsync(
|
||||
"UPDATE users SET settings = @settings, updated_at = @now WHERE user_id = @userId",
|
||||
new { userId, settings, now = DateTime.UtcNow.ToString("o") });
|
||||
}
|
||||
|
||||
public async Task<UserRecord> CreateUser(string username, string passwordHash, string displayName, bool isAdmin, string defaultPath = null)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
var now = DateTime.UtcNow.ToString("o");
|
||||
var userId = Guid.NewGuid().ToString("N");
|
||||
await conn.ExecuteAsync(@"
|
||||
INSERT INTO users (user_id, username, password_hash, display_name, is_admin, default_path, created_at, updated_at)
|
||||
VALUES (@userId, @username, @passwordHash, @displayName, @isAdmin, @defaultPath, @now, @now)",
|
||||
new { userId, username, passwordHash, displayName, isAdmin, defaultPath, now });
|
||||
return await GetUser(userId);
|
||||
}
|
||||
|
||||
public async Task UpdateUserAdmin(string userId, string username, string displayName, bool isAdmin, string defaultPath)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
await conn.ExecuteAsync(
|
||||
"UPDATE users SET username = @username, display_name = @displayName, is_admin = @isAdmin, default_path = @defaultPath, updated_at = @now WHERE user_id = @userId",
|
||||
new { userId, username, displayName, isAdmin, defaultPath, now = DateTime.UtcNow.ToString("o") });
|
||||
}
|
||||
|
||||
public async Task DeleteUser(string userId)
|
||||
{
|
||||
using var conn = CreateConnection();
|
||||
await conn.ExecuteAsync("DELETE FROM tokens WHERE user_id = @userId", new { userId });
|
||||
await conn.ExecuteAsync("DELETE FROM shared_messages WHERE user_id = @userId", new { userId });
|
||||
await conn.ExecuteAsync("DELETE FROM users WHERE user_id = @userId", new { userId });
|
||||
}
|
||||
|
||||
}
|
||||
+1
-1119
File diff suppressed because it is too large.
Load diff
+18
-35
@@ -332,8 +332,8 @@ public class OpenAiService
|
||||
Role = "system", Type = "error",
|
||||
Content = errMsg, IsError = true
|
||||
});
|
||||
await TryWriteSse(response, "error", new { error = errMsg }, sseCt);
|
||||
await TryWriteSse(response, "done", new { chainId }, sseCt);
|
||||
await SseHelper.TryWriteSse(response, "error", new { error = errMsg }, sseCt);
|
||||
await SseHelper.TryWriteSse(response, "done", new { chainId }, sseCt);
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -399,9 +399,9 @@ public class OpenAiService
|
||||
Role = "system", Type = "error",
|
||||
Content = errMsg, IsError = true
|
||||
});
|
||||
await TryWriteSse(response, "error", new { error = errMsg }, sseCt);
|
||||
await SseHelper.TryWriteSse(response, "error", new { error = errMsg }, sseCt);
|
||||
await FinalizeChain(chain, usage, iteration, "error", sw);
|
||||
await TryWriteSse(response, "done", new { chainId }, sseCt);
|
||||
await SseHelper.TryWriteSse(response, "done", new { chainId }, sseCt);
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -452,11 +452,11 @@ public class OpenAiService
|
||||
{
|
||||
if (!textStarted)
|
||||
{
|
||||
await TryWriteSse(response, "text_start", new { index = 0 }, sseCt);
|
||||
await SseHelper.TryWriteSse(response, "text_start", new { index = 0 }, sseCt);
|
||||
textStarted = true;
|
||||
}
|
||||
fullText.Append(chunk);
|
||||
await TryWriteSse(response, "text_delta", new { text = chunk }, sseCt);
|
||||
await SseHelper.TryWriteSse(response, "text_delta", new { text = chunk }, sseCt);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -544,7 +544,7 @@ public class OpenAiService
|
||||
|
||||
// Text-End senden falls Text gestreamt wurde
|
||||
if (textStarted)
|
||||
await TryWriteSse(response, "text_end", new { }, sseCt);
|
||||
await SseHelper.TryWriteSse(response, "text_end", new { }, sseCt);
|
||||
|
||||
// Text in DB speichern
|
||||
var finalText = fullText.ToString();
|
||||
@@ -562,7 +562,7 @@ public class OpenAiService
|
||||
if (toolCallAccumulators.Count == 0)
|
||||
{
|
||||
await FinalizeChain(chain, usage, iteration, stopReason, sw);
|
||||
await TryWriteSse(response, "done", new
|
||||
await SseHelper.TryWriteSse(response, "done", new
|
||||
{
|
||||
model = modelString,
|
||||
inputTokens = usage.InputTokens,
|
||||
@@ -625,7 +625,7 @@ public class OpenAiService
|
||||
foreach (var (_, (tcId, tcName, tcArgs, _)) in toolCallAccumulators)
|
||||
{
|
||||
// Tool-Start SSE
|
||||
await TryWriteSse(response, "tool_start", new { name = tcName, id = tcId }, sseCt);
|
||||
await SseHelper.TryWriteSse(response, "tool_start", new { name = tcName, id = tcId }, sseCt);
|
||||
|
||||
// Tool ausführen
|
||||
JsonElement inputEl;
|
||||
@@ -636,7 +636,7 @@ public class OpenAiService
|
||||
var isError = toolResult.StartsWith("Fehler:") || toolResult.StartsWith("Error:");
|
||||
|
||||
// Tool-End SSE
|
||||
await TryWriteSse(response, "tool_end", new
|
||||
await SseHelper.TryWriteSse(response, "tool_end", new
|
||||
{
|
||||
name = tcName,
|
||||
id = tcId,
|
||||
@@ -668,14 +668,14 @@ public class OpenAiService
|
||||
}
|
||||
|
||||
// Max Iterations erreicht
|
||||
await TryWriteSse(response, "error", new { error = "Max iterations erreicht" }, sseCt);
|
||||
await SseHelper.TryWriteSse(response, "error", new { error = "Max iterations erreicht" }, sseCt);
|
||||
await FinalizeChain(chain, usage, iteration, "max_iterations", sw);
|
||||
await TryWriteSse(response, "done", new { chainId }, sseCt);
|
||||
await SseHelper.TryWriteSse(response, "done", new { chainId }, sseCt);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
await FinalizeChain(chain, usage, iteration, "stopped", sw);
|
||||
await TryWriteSse(response, "done", new
|
||||
await SseHelper.TryWriteSse(response, "done", new
|
||||
{
|
||||
model = modelString,
|
||||
iterations = iteration,
|
||||
@@ -693,9 +693,9 @@ public class OpenAiService
|
||||
Role = "system", Type = "error",
|
||||
Content = errMsg, IsError = true
|
||||
});
|
||||
await TryWriteSse(response, "error", new { error = errMsg }, sseCt);
|
||||
await SseHelper.TryWriteSse(response, "error", new { error = errMsg }, sseCt);
|
||||
await FinalizeChain(chain, usage, iteration, "error", sw);
|
||||
await TryWriteSse(response, "done", new { chainId }, sseCt);
|
||||
await SseHelper.TryWriteSse(response, "done", new { chainId }, sseCt);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -797,17 +797,8 @@ public class OpenAiService
|
||||
return null;
|
||||
}
|
||||
|
||||
private async Task FinalizeChain(ChainRecord chain, TokenUsage usage, int iterations, string stopReason, System.Diagnostics.Stopwatch sw)
|
||||
{
|
||||
chain.InputTokens = usage.InputTokens;
|
||||
chain.OutputTokens = usage.OutputTokens;
|
||||
chain.CacheCreationTokens = usage.CacheCreationTokens;
|
||||
chain.CacheReadTokens = usage.CacheReadTokens;
|
||||
chain.Iterations = iterations;
|
||||
chain.StopReason = stopReason;
|
||||
chain.DurationMs = sw.ElapsedMilliseconds;
|
||||
await _db.UpdateChain(chain);
|
||||
}
|
||||
private Task FinalizeChain(ChainRecord chain, TokenUsage usage, int iterations, string stopReason, System.Diagnostics.Stopwatch sw)
|
||||
=> SseHelper.FinalizeChain(_db, chain, usage, iterations, stopReason, sw);
|
||||
|
||||
private HttpClient CreateClient(string apiKey)
|
||||
{
|
||||
@@ -819,13 +810,5 @@ public class OpenAiService
|
||||
return client;
|
||||
}
|
||||
|
||||
private static async Task TryWriteSse(HttpResponse response, string eventType, object data, CancellationToken ct)
|
||||
{
|
||||
try
|
||||
{
|
||||
if (!ct.IsCancellationRequested)
|
||||
await AnthropicService.WriteSseEvent(response, eventType, data, ct);
|
||||
}
|
||||
catch { }
|
||||
}
|
||||
// TryWriteSse → SseHelper.cs
|
||||
}
|
||||
@@ -0,0 +1,85 @@
|
||||
using System.Diagnostics;
|
||||
using System.Text.Json;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using Gott.Models;
|
||||
using Microsoft.AspNetCore.Http;
|
||||
|
||||
namespace Gott.Services;
|
||||
|
||||
/// <summary>
|
||||
/// Shared SSE-Streaming-Utilities für AnthropicService und OpenAiService.
|
||||
/// </summary>
|
||||
public static class SseHelper
|
||||
{
|
||||
/// <summary>
|
||||
/// Schreibt ein SSE-Event (wirft bei CancellationToken).
|
||||
/// </summary>
|
||||
public static async Task WriteSseEvent(HttpResponse response, string eventType, object data, CancellationToken ct = default)
|
||||
{
|
||||
ct.ThrowIfCancellationRequested();
|
||||
var json = JsonSerializer.Serialize(data);
|
||||
await response.WriteAsync($"event: {eventType}\ndata: {json}\n\n", ct);
|
||||
await response.Body.FlushAsync(ct);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Versucht ein SSE-Event zu schreiben, ignoriert Fehler (Client disconnected).
|
||||
/// </summary>
|
||||
public static async Task TryWriteSse(HttpResponse response, string eventType, object data, CancellationToken ct)
|
||||
{
|
||||
try
|
||||
{
|
||||
if (!ct.IsCancellationRequested)
|
||||
await WriteSseEvent(response, eventType, data, ct);
|
||||
}
|
||||
catch
|
||||
{
|
||||
// Client bereits disconnected – ignorieren
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Finalisiert eine Chain in der Datenbank (Token-Counts, Stopgrund, Dauer).
|
||||
/// </summary>
|
||||
public static async Task FinalizeChain(DatabaseService db, ChainRecord chain, TokenUsage usage, int iterations, string stopReason, Stopwatch sw, string streamErrorInfo = null)
|
||||
{
|
||||
chain.InputTokens = usage.InputTokens;
|
||||
chain.OutputTokens = usage.OutputTokens;
|
||||
chain.CacheCreationTokens = usage.CacheCreationTokens;
|
||||
chain.CacheReadTokens = usage.CacheReadTokens;
|
||||
chain.Iterations = iterations;
|
||||
chain.StopReason = stopReason;
|
||||
chain.StreamErrorInfo = streamErrorInfo;
|
||||
chain.DurationMs = sw.ElapsedMilliseconds;
|
||||
await db.UpdateChain(chain);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Kürzt Tool-Results auf max. Zeichenzahl.
|
||||
/// </summary>
|
||||
public const int MaxToolResultChars = 20_000;
|
||||
|
||||
public static string TruncateToolResult(string result)
|
||||
{
|
||||
return result.Length > MaxToolResultChars
|
||||
? result[..MaxToolResultChars] + $"\n\n[... gekürzt, {result.Length - MaxToolResultChars} weitere Zeichen ...]"
|
||||
: result;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Token-Zähler für eine Chat-Session (Input, Output, Cache).
|
||||
/// </summary>
|
||||
public class TokenUsage
|
||||
{
|
||||
public long InputTokens { get; set; }
|
||||
public long OutputTokens { get; set; }
|
||||
public long CacheCreationTokens { get; set; }
|
||||
public long CacheReadTokens { get; set; }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Ergebnis einer Tool-Ausführung.
|
||||
/// </summary>
|
||||
public record ToolResult(string Id, string Result, bool IsError);
|
||||
Reference in new issue
Block a user