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 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, string multiModelMode = null, string panelAgentIds = null, int moaLayers = 2) { 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, MultiModelMode = multiModelMode, PanelAgentIds = panelAgentIds, MoaLayers = moaLayers, 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, multi_model_mode, panel_agent_ids, moa_layers, created_at, updated_at) VALUES (@SessionId, @UserId, @Name, @Description, @Image, @ImageSmall, @AgentId, @Workspace, @Context, @CustomCommands, @AvatarScale, @AvatarOffsetX, @AvatarOffsetY, @ReviewAgentId, @MaxReviewIterations, @FinalizerAgentId, @PreparerAgentId, @MultiModelMode, @PanelAgentIds, @MoaLayers, @CreatedAt, @UpdatedAt) ", session); return session; } public async Task GetSessionByName(string name) { using var conn = CreateConnection(); var sessions = await conn.QueryAsync( "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> GetSessionsWithStats(string userId) { using var conn = CreateConnection(); var sessions = await conn.QueryAsync(@" SELECT s.* FROM sessions s LEFT JOIN ( SELECT c.session_id, MAX(m.created_at) AS last_message_at FROM chains c JOIN messages m ON m.chain_id = c.chain_id WHERE m.type = 'text' GROUP BY c.session_id ) lm ON lm.session_id = s.session_id WHERE s.user_id = @userId AND s.is_deleted = 0 ORDER BY COALESCE(lm.last_message_at, s.updated_at) DESC", new { userId }); var results = new List(); foreach (var s in sessions) { var activeCount = await conn.ExecuteScalarAsync(@" 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(@" 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 GetSession(string sessionId) { using var conn = CreateConnection(); return await conn.QuerySingleOrDefaultAsync( "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, bool allowUserReassign = false) { using var conn = CreateConnection(); var sets = new List { "updated_at = @now" }; var p = new Dapper.DynamicParameters(); p.Add("sessionId", sessionId); p.Add("now", DateTime.UtcNow.ToString("o")); // Session einem anderen Benutzer zuweisen – nur wenn explizit erlaubt (Admin). // Leerer/null userId wird ignoriert (eine Session muss immer einen Besitzer haben). if (allowUserReassign && body.TryGetProperty("userId", out var uid) && uid.ValueKind == System.Text.Json.JsonValueKind.String && !string.IsNullOrWhiteSpace(uid.GetString())) { sets.Add("user_id = @newUserId"); p.Add("newUserId", uid.GetString()); } 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()); } if (body.TryGetProperty("multiModelMode", out var mmm)) { sets.Add("multi_model_mode = @multiModelMode"); p.Add("multiModelMode", mmm.ValueKind == System.Text.Json.JsonValueKind.Null ? null : mmm.GetString()); } if (body.TryGetProperty("panelAgentIds", out var pai)) { sets.Add("panel_agent_ids = @panelAgentIds"); p.Add("panelAgentIds", pai.ValueKind == System.Text.Json.JsonValueKind.Null ? null : pai.GetString()); } if (body.TryGetProperty("moaLayers", out var mol) && mol.ValueKind == System.Text.Json.JsonValueKind.Number) { sets.Add("moa_layers = @moaLayers"); p.Add("moaLayers", mol.GetInt32()); } if (body.TryGetProperty("unityMcpUrl", out var umu)) { sets.Add("unity_mcp_url = @unityMcpUrl"); p.Add("unityMcpUrl", umu.ValueKind == System.Text.Json.JsonValueKind.Null ? null : umu.GetString()); } if (body.TryGetProperty("unityMcpToken", out var umt)) { sets.Add("unity_mcp_token = @unityMcpToken"); p.Add("unityMcpToken", umt.ValueKind == System.Text.Json.JsonValueKind.Null ? null : umt.GetString()); } if (body.TryGetProperty("mcpServers", out var mcps)) { sets.Add("mcp_servers = @mcpServers"); p.Add("mcpServers", mcps.ValueKind == System.Text.Json.JsonValueKind.Null ? null : mcps.GetRawText()); } 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 CreateArchive(string sessionId, string name = null) { using var conn = CreateConnection(); var chainCount = await conn.ExecuteScalarAsync( "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(@" 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> GetArchives(string sessionId) { using var conn = CreateConnection(); return await conn.QueryAsync( "SELECT * FROM archives WHERE session_id = @sessionId ORDER BY id DESC", new { sessionId }); } public async Task> GetArchiveChains(string archiveId) { using var conn = CreateConnection(); return await conn.QueryAsync( "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, effort, created_at) VALUES (@ChainId, @SessionId, @ArchiveId, @AgentId, @Model, @SystemPrompt, @ToolChoice, @ToolNames, @InputTokens, @OutputTokens, @Iterations, @StopReason, @DurationMs, @PlanMode, @Effort, @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, workflow_info = @WorkflowInfo, duration_ms = @DurationMs WHERE chain_id = @ChainId ", chain); } public async Task GetChain(string chainId) { using var conn = CreateConnection(); return await conn.QuerySingleOrDefaultAsync( "SELECT * FROM chains WHERE chain_id = @chainId", new { chainId } ); } public async Task> GetActiveChains(string sessionId) { using var conn = CreateConnection(); return await conn.QueryAsync( "SELECT * FROM chains WHERE session_id = @sessionId AND archive_id IS NULL ORDER BY id DESC", new { sessionId }); } // ── Messages ── public async Task SaveMessage(MessageRecord record) { using var conn = CreateConnection(); record.CreatedAt = DateTime.UtcNow.ToString("o"); return await conn.ExecuteScalarAsync(@" 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); } /// /// Speichert Message + Attachments in einer Transaktion (1x fsync statt N+1). /// public async Task SaveMessageWithAttachments(MessageRecord record, List 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(@" 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> GetAttachments(long messageId) { using var conn = CreateConnection(); return await conn.QueryAsync( "SELECT * FROM attachments WHERE message_id = @messageId ORDER BY id", new { messageId }); } public async Task> GetAttachmentsForMessages(IEnumerable messageIds) { using var conn = CreateConnection(); return await conn.QueryAsync( "SELECT * FROM attachments WHERE message_id IN @messageIds ORDER BY id", new { messageIds }); } public async Task> GetChainMessages(string chainId, bool includeDeleted = false) { using var conn = CreateConnection(); var deletedFilter = includeDeleted ? "" : " AND deleted_on IS NULL"; return await conn.QueryAsync( $"SELECT * FROM messages WHERE chain_id = @chainId{deletedFilter} ORDER BY id", new { chainId } ); } public async Task GetActiveMessageCount(string sessionId, bool includeDeleted = false) { using var conn = CreateConnection(); var deletedFilter = includeDeleted ? "" : "AND m.deleted_on IS NULL"; return await conn.ExecuteScalarAsync($@" 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> 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($@" 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 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($@" 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> 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($@" 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> 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($@" 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 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( "SELECT is_marked FROM messages WHERE id = @messageId", new { messageId }); } // ── Message Deletion ── public async Task ToggleMessageDelete(long messageId) { using var conn = CreateConnection(); var current = await conn.ExecuteScalarAsync( "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> GetMarkedMessages(string sessionId) { using var conn = CreateConnection(); return await conn.QueryAsync(@" 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 }); } }