Weknora/dobotai-platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: address message audit review findings

Browse Source
This commit is contained in:
姚章浩
2026-07-21 10:52:59 +08:00
parent d857853ea3
commit 32b76b84cb
4 changed files with 124 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
db/migrations/000003_message_audit.up.sql
+1 -7
View File
@@ -17,9 +17,6 @@ CREATE TABLE IF NOT EXISTS channel_actors (
UNIQUE (tenant_key, channel, external_actor_id)
);
CREATE INDEX IF NOT EXISTS idx_channel_actors_lookup
ON channel_actors (tenant_key, channel, external_actor_id);
-- ============================================================
-- 2. Channel Conversations
-- ============================================================
@@ -36,9 +33,6 @@ CREATE TABLE IF NOT EXISTS channel_conversations (
UNIQUE (tenant_key, channel, external_conversation_id)
);
CREATE INDEX IF NOT EXISTS idx_channel_conversations_lookup
ON channel_conversations (tenant_key, channel, external_conversation_id);
-- ============================================================
-- 3. Messages
-- ============================================================
@@ -75,7 +69,7 @@ CREATE INDEX IF NOT EXISTS idx_messages_tenant_conv_created
-- Index for correlation lookups
CREATE INDEX IF NOT EXISTS idx_messages_correlation
ON messages (correlation_id);
ON messages (tenant_key, correlation_id);
-- ============================================================
-- 4. Extend turns table
db/migrations/000003_message_audit_contract_test.go
+15 -8
View File
@@ -197,15 +197,22 @@ func TestUp003MessagesPartialUniqueIndex(t *testing.T) {
// TestUp003MessagesIndexes checks required indexes on messages.
func TestUp003MessagesIndexes(t *testing.T) {
up := read003UpMigration(t)
upper := strings.ToUpper(up)
// Tenant + conversation + created_at
if !strings.Contains(upper, "TENANT_KEY") || !strings.Contains(upper, "CONVERSATION_ID") || !strings.Contains(upper, "CREATED_AT") {
t.Error("messages must have index on tenant_key, conversation_id, created_at")
conversationIndex := regexp.MustCompile(`(?is)CREATE\s+INDEX\s+IF\s+NOT\s+EXISTS\s+idx_messages_tenant_conv_created\s+ON\s+messages\s*\(\s*tenant_key\s*,\s*conversation_id\s*,\s*created_at\s+DESC\s*\)`)
if !conversationIndex.MatchString(up) {
t.Error("messages must have index on tenant_key, conversation_id, created_at DESC")
}
// correlation_id index
if !strings.Contains(upper, "CORRELATION_ID") {
t.Error("messages must have index on correlation_id")
correlationIndex := regexp.MustCompile(`(?is)CREATE\s+INDEX\s+IF\s+NOT\s+EXISTS\s+idx_messages_correlation\s+ON\s+messages\s*\(\s*tenant_key\s*,\s*correlation_id\s*\)`)
if !correlationIndex.MatchString(up) {
t.Error("messages correlation index must be tenant-scoped: (tenant_key, correlation_id)")
}
}
func TestUp003AvoidsRedundantLookupIndexes(t *testing.T) {
up := strings.ToLower(read003UpMigration(t))
for _, redundant := range []string{"idx_channel_actors_lookup", "idx_channel_conversations_lookup"} {
if strings.Contains(up, redundant) {
t.Errorf("migration must not create redundant index %s; UNIQUE already provides it", redundant)
}
}
}
internal/qa/service.go
+13 -1
View File
@@ -60,6 +60,7 @@ func (s *Service) AskEvents(ctx context.Context, request core.QARequest, observe
}
func (s *Service) ask(ctx context.Context, request core.QARequest, answerObserver weknora.AnswerObserver, eventObserver weknora.StreamObserver) (core.QAResult, error) {
request = normalizeRequestIdentity(request)
if err := validateRequest(request); err != nil {
return core.QAResult{}, err
}
@@ -205,6 +206,17 @@ func (s *Service) ask(ctx context.Context, request core.QARequest, answerObserve
WeKnoraRequestID: chat.RequestID, Answer: chat.Answer, References: chat.References}, nil
}
func normalizeRequestIdentity(request core.QARequest) core.QARequest {
request.TenantKey = strings.TrimSpace(request.TenantKey)
request.Channel = strings.TrimSpace(request.Channel)
request.ConversationType = core.ConversationType(strings.TrimSpace(string(request.ConversationType)))
request.ExternalChatID = strings.TrimSpace(request.ExternalChatID)
request.ExternalUserID = strings.TrimSpace(request.ExternalUserID)
request.ExternalMessageID = strings.TrimSpace(request.ExternalMessageID)
request.SenderName = strings.TrimSpace(request.SenderName)
return request
}
func normalizeMessageType(request core.QARequest) string {
messageType := strings.ToLower(strings.TrimSpace(request.MessageType))
if messageType == "" {
@@ -218,7 +230,7 @@ func normalizeMessageType(request core.QARequest) string {
return "image"
case "richtext", "rich_text":
return "rich_text"
case "text", "card", "feedback":
case "text":
return messageType
default:
return ""
internal/qa/service_test.go
+95
View File
@@ -3,6 +3,7 @@ package qa
import (
"context"
"errors"
"strings"
"sync"
"testing"
@@ -64,6 +65,17 @@ type fixedResolver struct{}
func (fixedResolver) Resolve(_, _ string) (string, error) { return "agent-1", nil }
type recordingResolver struct {
channel string
externalChatID string
}
func (r *recordingResolver) Resolve(channel, externalChatID string) (string, error) {
r.channel = channel
r.externalChatID = externalChatID
return "agent-1", nil
}
type fakeWeKnora struct {
mu sync.Mutex
sessions int
@@ -488,3 +500,86 @@ func TestAskFailureDoesNotGenerateAnswerMessageID(t *testing.T) {
t.Fatalf("failed turn must not generate answer message: %+v", finished)
}
}
func TestAskCanonicalizesStableIdentityBeforeIdempotency(t *testing.T) {
repo := newMemoryRepository()
client := &fakeWeKnora{}
resolver := &recordingResolver{}
service, err := New(repo, resolver, client, session.GroupShared)
if err != nil {
t.Fatal(err)
}
first := core.QARequest{
TenantKey: " tenant ", Channel: " dingtalk ", ConversationType: core.ConversationType(" direct "),
ExternalChatID: " dm-1 ", ExternalUserID: " user-1 ", ExternalMessageID: " message-trim ",
SenderName: " Alice ", MessageType: " Text ", Query: " preserve question whitespace ",
}
result, err := service.Ask(context.Background(), first)
if err != nil {
t.Fatalf("first Ask: %v", err)
}
second := first
second.TenantKey = "tenant"
second.Channel = "dingtalk"
second.ConversationType = core.ConversationDirect
second.ExternalChatID = "dm-1"
second.ExternalUserID = "user-1"
second.ExternalMessageID = "message-trim"
second.SenderName = "Alice"
replay, err := service.Ask(context.Background(), second)
if err != nil {
t.Fatalf("replay Ask: %v", err)
}
if !replay.Replayed || replay.TurnID != result.TurnID {
t.Fatalf("identity variants must replay the same turn: first=%+v replay=%+v", result, replay)
}
repo.mu.Lock()
begin := repo.beginTurns[0]
repo.mu.Unlock()
if begin.TenantKey != "tenant" || begin.Channel != "dingtalk" || begin.ConversationType != core.ConversationDirect ||
begin.ExternalChatID != "dm-1" || begin.ExternalUserID != "user-1" || begin.ExternalMessageID != "message-trim" ||
begin.SenderName != "Alice" {
t.Fatalf("stable identity was not canonicalized: %+v", begin)
}
if begin.Question != " preserve question whitespace " {
t.Fatalf("question content must not be identity-normalized: %q", begin.Question)
}
if resolver.channel != "dingtalk" || resolver.externalChatID != "dm-1" {
t.Fatalf("resolver received non-canonical identity: channel=%q chat=%q", resolver.channel, resolver.externalChatID)
}
client.mu.Lock()
defer client.mu.Unlock()
if len(client.requests) != 1 {
t.Fatalf("canonical duplicate must call WeKnora once, got %d", len(client.requests))
}
}
func TestAskRejectsNonQAMessageTypes(t *testing.T) {
for _, messageType := range []string{"card", "feedback"} {
t.Run(messageType, func(t *testing.T) {
repo := newMemoryRepository()
client := &fakeWeKnora{}
service, err := New(repo, fixedResolver{}, client, session.GroupShared)
if err != nil {
t.Fatal(err)
}
req := core.QARequest{
TenantKey: "tenant", Channel: "dingtalk", ConversationType: core.ConversationDirect,
ExternalUserID: "user-1", ExternalMessageID: "message-" + messageType,
MessageType: messageType, Query: "not a QA message",
}
if _, err = service.Ask(context.Background(), req); err == nil || !strings.Contains(err.Error(), "unsupported message_type") {
t.Fatalf("expected unsupported message_type error, got %v", err)
}
repo.mu.Lock()
beginCount := len(repo.beginTurns)
repo.mu.Unlock()
client.mu.Lock()
requestCount := len(client.requests)
client.mu.Unlock()
if beginCount != 0 || requestCount != 0 {
t.Fatalf("non-QA type must have zero DB/model side effects: begins=%d requests=%d", beginCount, requestCount)
}
})
}
}