Compare commits
4 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 38560c5f41 | |||
| f0d3d5b50d | |||
| 56e3bf4686 | |||
| 257a55a3f0 |
@@ -1,6 +1,9 @@
|
||||
# DobotAI configuration
|
||||
DOBOTAI_LISTEN_ADDR=127.0.0.1:8090
|
||||
DOBOTAI_INGRESS_TOKEN=replace-me
|
||||
# Optional local-only audit API. Use a distinct secret from DOBOTAI_INGRESS_TOKEN.
|
||||
# When set, DOBOTAI_TENANT_KEY must be set and the listen address must be loopback-only.
|
||||
DOBOTAI_ADMIN_TOKEN=
|
||||
DOBOTAI_GROUP_SESSION_POLICY=shared
|
||||
DOBOTAI_AGENT_BINDINGS={"dingtalk:*":"replace-with-agent-id"}
|
||||
|
||||
|
||||
+42
-6
@@ -12,7 +12,9 @@ import (
|
||||
"time"
|
||||
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/agentbinding"
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/audit"
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/channels/dingtalk"
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/feedback"
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/httpapi"
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/qa"
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/session"
|
||||
@@ -49,7 +51,15 @@ func main() {
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
api, err := httpapi.New(qaService, os.Getenv("DOBOTAI_INGRESS_TOKEN"))
|
||||
address := strings.TrimSpace(os.Getenv("DOBOTAI_LISTEN_ADDR"))
|
||||
if address == "" {
|
||||
address = "127.0.0.1:8090"
|
||||
}
|
||||
auditOptions, err := buildAuditOptions(store, os.Getenv("DOBOTAI_TENANT_KEY"), os.Getenv("DOBOTAI_ADMIN_TOKEN"), address)
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
api, err := httpapi.NewWithOptions(qaService, os.Getenv("DOBOTAI_INGRESS_TOKEN"), auditOptions)
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
@@ -64,6 +74,14 @@ func main() {
|
||||
}
|
||||
var streamClient *dingtalk.StreamClient
|
||||
if dingCfg.Enabled() {
|
||||
feedbackService, err := feedback.NewService(dingCfg.TenantKey, "dingtalk", store)
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
feedbackHandler, err := dingtalk.NewCardFeedbackHandler(feedbackService)
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
sender, err := dingtalk.NewWebhookSender(nil)
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
@@ -89,7 +107,7 @@ func main() {
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
streamClient, err = dingtalk.NewStreamClient(dingCfg, handler)
|
||||
streamClient, err = dingtalk.NewStreamClient(dingCfg, handler, feedbackHandler)
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
@@ -99,10 +117,6 @@ func main() {
|
||||
defer streamClient.Close()
|
||||
log.Printf("DingTalk Stream ingress started for tenant %s", dingCfg.TenantKey)
|
||||
}
|
||||
address := strings.TrimSpace(os.Getenv("DOBOTAI_LISTEN_ADDR"))
|
||||
if address == "" {
|
||||
address = "127.0.0.1:8090"
|
||||
}
|
||||
server := &http.Server{Addr: address, Handler: api.Handler(), ReadHeaderTimeout: 5 * time.Second, ReadTimeout: 15 * time.Second, WriteTimeout: 130 * time.Second, IdleTimeout: 60 * time.Second}
|
||||
go func() {
|
||||
log.Printf("DobotAI ingress listening on %s", address)
|
||||
@@ -118,3 +132,25 @@ func main() {
|
||||
log.Printf("HTTP shutdown: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func buildAuditOptions(repo audit.TurnRepository, tenantKey, adminToken, address string) (httpapi.Options, error) {
|
||||
adminToken = strings.TrimSpace(adminToken)
|
||||
if adminToken == "" {
|
||||
return httpapi.Options{}, nil
|
||||
}
|
||||
if repo == nil {
|
||||
return httpapi.Options{}, errors.New("audit repository is required when DOBOTAI_ADMIN_TOKEN is set")
|
||||
}
|
||||
tenantKey = strings.TrimSpace(tenantKey)
|
||||
if tenantKey == "" {
|
||||
return httpapi.Options{}, errors.New("DOBOTAI_TENANT_KEY is required when DOBOTAI_ADMIN_TOKEN is set")
|
||||
}
|
||||
if err := httpapi.ValidateLoopbackListenAddress(address); err != nil {
|
||||
return httpapi.Options{}, err
|
||||
}
|
||||
service, err := audit.NewService(tenantKey, repo)
|
||||
if err != nil {
|
||||
return httpapi.Options{}, err
|
||||
}
|
||||
return httpapi.Options{AuditService: service, AdminToken: adminToken}, nil
|
||||
}
|
||||
|
||||
@@ -0,0 +1,87 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/audit"
|
||||
)
|
||||
|
||||
type auditRepoStub struct {
|
||||
tenant string
|
||||
}
|
||||
|
||||
func (r *auditRepoStub) ListTurns(_ context.Context, tenant string, _ audit.ListQuery) ([]audit.TurnSummary, error) {
|
||||
r.tenant = tenant
|
||||
return nil, nil
|
||||
}
|
||||
func (r *auditRepoStub) GetTurn(context.Context, string, string) (audit.TurnDetail, error) {
|
||||
return audit.TurnDetail{}, audit.ErrNotFound
|
||||
}
|
||||
|
||||
func TestBuildAuditOptionsDisabledIsBackwardCompatible(t *testing.T) {
|
||||
opts, err := buildAuditOptions(nil, "", "", "0.0.0.0:8090")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if opts.AuditService != nil || opts.AdminToken != "" {
|
||||
t.Fatalf("options=%+v", opts)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuildAuditOptionsRequiresTenantRepositoryAndLoopback(t *testing.T) {
|
||||
repo := &auditRepoStub{}
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
repo audit.TurnRepository
|
||||
tenant string
|
||||
address string
|
||||
}{
|
||||
{name: "missing repository", tenant: "tenant", address: "127.0.0.1:8090"},
|
||||
{name: "missing tenant", repo: repo, address: "127.0.0.1:8090"},
|
||||
{name: "public listener", repo: repo, tenant: "tenant", address: "0.0.0.0:8090"},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
if _, err := buildAuditOptions(tc.repo, tc.tenant, "admin-secret", tc.address); err == nil {
|
||||
t.Fatal("expected configuration error")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuildAuditOptionsFixesServerSideTenant(t *testing.T) {
|
||||
repo := &auditRepoStub{}
|
||||
opts, err := buildAuditOptions(repo, " tenant-a ", " admin-secret ", "127.0.0.1:8090")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if opts.AdminToken != "admin-secret" || opts.AuditService == nil {
|
||||
t.Fatalf("options=%+v", opts)
|
||||
}
|
||||
if _, err := opts.AuditService.ListTurns(context.Background(), audit.ListTurnsRequest{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if repo.tenant != "tenant-a" {
|
||||
t.Fatalf("tenant=%q", repo.tenant)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMainWiresFeedbackIntoDingTalkStream(t *testing.T) {
|
||||
data, err := os.ReadFile("main.go")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
source := string(data)
|
||||
for _, fragment := range []string{
|
||||
`"gitea.zheromiku.top/Weknora/dobotai-platform/internal/feedback"`,
|
||||
`feedback.NewService(dingCfg.TenantKey, "dingtalk", store)`,
|
||||
`dingtalk.NewCardFeedbackHandler(feedbackService)`,
|
||||
`dingtalk.NewStreamClient(dingCfg, handler, feedbackHandler)`,
|
||||
} {
|
||||
if !strings.Contains(source, fragment) {
|
||||
t.Errorf("main feedback wiring missing %q", fragment)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
-- 000005_answer_feedback down migration
|
||||
-- Drop in reverse order: indexes first, then table.
|
||||
|
||||
DROP INDEX IF EXISTS idx_messages_card_provider_unique;
|
||||
DROP INDEX IF EXISTS idx_answer_feedback_tenant_reason_updated;
|
||||
DROP INDEX IF EXISTS idx_answer_feedback_tenant_value_updated;
|
||||
DROP TABLE IF EXISTS answer_feedback;
|
||||
@@ -0,0 +1,32 @@
|
||||
-- 000005_answer_feedback: user feedback on answers via DingTalk card
|
||||
-- Each (tenant_key, channel, answer_message_id, actor_id) holds the current
|
||||
-- feedback state: up or down with optional reason_code for down.
|
||||
-- Depends on: 000004_delivery_audit (and its 000003 message-audit baseline).
|
||||
|
||||
CREATE TABLE IF NOT EXISTS answer_feedback (
|
||||
id TEXT PRIMARY KEY,
|
||||
tenant_key TEXT NOT NULL,
|
||||
channel TEXT NOT NULL,
|
||||
answer_message_id TEXT NOT NULL REFERENCES messages(id),
|
||||
turn_id TEXT NOT NULL REFERENCES turns(id),
|
||||
actor_id TEXT NOT NULL REFERENCES channel_actors(id),
|
||||
value TEXT NOT NULL CHECK (value IN ('up', 'down')),
|
||||
reason_code TEXT NOT NULL DEFAULT '' CHECK (reason_code IN ('', 'incorrect', 'not_resolved', 'irrelevant', 'unclear', 'outdated', 'other')),
|
||||
source TEXT NOT NULL CHECK (source IN ('dingtalk_card')),
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
||||
UNIQUE (tenant_key, channel, answer_message_id, actor_id),
|
||||
CHECK (value != 'up' OR reason_code = '')
|
||||
);
|
||||
|
||||
-- Queries that filter by value or reason_code within a tenant use these indexes.
|
||||
CREATE INDEX IF NOT EXISTS idx_answer_feedback_tenant_value_updated
|
||||
ON answer_feedback (tenant_key, value, updated_at DESC);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_answer_feedback_tenant_reason_updated
|
||||
ON answer_feedback (tenant_key, reason_code, updated_at DESC);
|
||||
|
||||
-- A DingTalk outTrackId must identify exactly one delivered card answer.
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS idx_messages_card_provider_unique
|
||||
ON messages (tenant_key, channel, provider_message_id)
|
||||
WHERE provider_message_id != '' AND delivery_method = 'card';
|
||||
@@ -0,0 +1,296 @@
|
||||
// Package migrations_test contains source-level contract tests for the
|
||||
// 000005_answer_feedback migration. No real database connection is used.
|
||||
package migrations_test
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"regexp"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func read005UpMigration(t *testing.T) string {
|
||||
t.Helper()
|
||||
data, err := os.ReadFile(filepath.Join(".", "000005_answer_feedback.up.sql"))
|
||||
if err != nil {
|
||||
t.Fatalf("read up migration: %v", err)
|
||||
}
|
||||
return string(data)
|
||||
}
|
||||
|
||||
func read005DownMigration(t *testing.T) string {
|
||||
t.Helper()
|
||||
data, err := os.ReadFile(filepath.Join(".", "000005_answer_feedback.down.sql"))
|
||||
if err != nil {
|
||||
t.Fatalf("read down migration: %v", err)
|
||||
}
|
||||
return string(data)
|
||||
}
|
||||
|
||||
// --- RED: These tests will fail until the migration files exist ---
|
||||
|
||||
// TestUp005CreatesAnswerFeedbackTable ensures the table is created with IF NOT EXISTS.
|
||||
func TestUp005CreatesAnswerFeedbackTable(t *testing.T) {
|
||||
up := read005UpMigration(t)
|
||||
upper := strings.ToUpper(up)
|
||||
|
||||
if !strings.Contains(upper, "CREATE TABLE IF NOT EXISTS ANSWER_FEEDBACK") {
|
||||
t.Error("up migration must create answer_feedback table with IF NOT EXISTS")
|
||||
}
|
||||
}
|
||||
|
||||
// TestUp005AnswerFeedbackColumns ensures all required columns are present.
|
||||
func TestUp005AnswerFeedbackColumns(t *testing.T) {
|
||||
up := read005UpMigration(t)
|
||||
upper := strings.ToUpper(up)
|
||||
|
||||
required := []string{
|
||||
"ID TEXT PRIMARY KEY",
|
||||
"TENANT_KEY TEXT NOT NULL",
|
||||
"CHANNEL TEXT NOT NULL",
|
||||
"ANSWER_MESSAGE_ID TEXT NOT NULL",
|
||||
"TURN_ID TEXT NOT NULL",
|
||||
"ACTOR_ID TEXT NOT NULL",
|
||||
"VALUE TEXT NOT NULL",
|
||||
"REASON_CODE TEXT NOT NULL DEFAULT ''",
|
||||
"SOURCE TEXT NOT NULL",
|
||||
"CREATED_AT TIMESTAMPTZ NOT NULL DEFAULT NOW()",
|
||||
"UPDATED_AT TIMESTAMPTZ NOT NULL DEFAULT NOW()",
|
||||
}
|
||||
for _, col := range required {
|
||||
if !strings.Contains(upper, col) {
|
||||
t.Errorf("up migration must contain column %q", col)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestUp005ValueCheck ensures value is constrained to 'up' or 'down'.
|
||||
func TestUp005ValueCheck(t *testing.T) {
|
||||
up := read005UpMigration(t)
|
||||
upper := strings.ToUpper(up)
|
||||
|
||||
if !strings.Contains(upper, "VALUE IN ('UP'") || !strings.Contains(upper, "'DOWN'") {
|
||||
t.Error("value must be CHECK constrained to 'up' and 'down'")
|
||||
}
|
||||
}
|
||||
|
||||
// TestUp005ReasonCodeCheck ensures reason_code is constrained to allowed values.
|
||||
func TestUp005ReasonCodeCheck(t *testing.T) {
|
||||
up := read005UpMigration(t)
|
||||
upper := strings.ToUpper(up)
|
||||
|
||||
allowed := []string{"''", "INCORRECT", "NOT_RESOLVED", "IRRELEVANT", "UNCLEAR", "OUTDATED", "OTHER"}
|
||||
for _, r := range allowed {
|
||||
if !strings.Contains(upper, r) {
|
||||
t.Errorf("reason_code CHECK must include %q", r)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestUp005SourceCheck ensures source is constrained to 'dingtalk_card'.
|
||||
func TestUp005SourceCheck(t *testing.T) {
|
||||
up := read005UpMigration(t)
|
||||
upper := strings.ToUpper(up)
|
||||
|
||||
if !strings.Contains(upper, "DINGTALK_CARD") {
|
||||
t.Error("source must be CHECK constrained to 'dingtalk_card'")
|
||||
}
|
||||
}
|
||||
|
||||
// TestUp005UpValueMustHaveNoReason ensures CHECK that value='up' implies reason_code=”.
|
||||
func TestUp005UpValueMustHaveNoReason(t *testing.T) {
|
||||
up := read005UpMigration(t)
|
||||
upper := strings.ToUpper(up)
|
||||
|
||||
// Must have a CHECK that when value='up', reason_code must be empty
|
||||
hasCheck := strings.Contains(upper, "VALUE = 'UP'") && strings.Contains(upper, "REASON_CODE = ''")
|
||||
// Also check for the inverse: "VALUE != 'UP' OR REASON_CODE = ''"
|
||||
hasOr := strings.Contains(upper, "OR") && strings.Contains(upper, "REASON_CODE = ''")
|
||||
if !hasCheck && !hasOr {
|
||||
t.Error("must have CHECK ensuring value='up' implies reason_code=''")
|
||||
}
|
||||
}
|
||||
|
||||
// TestUp005ForeignKeyConstraints ensures FK constraints on answer_message_id, turn_id, actor_id.
|
||||
func TestUp005ForeignKeyConstraints(t *testing.T) {
|
||||
up := read005UpMigration(t)
|
||||
upper := strings.ToUpper(up)
|
||||
|
||||
// Must reference messages(id), turns(id), channel_actors(id)
|
||||
if !strings.Contains(upper, "REFERENCES MESSAGES(ID)") {
|
||||
t.Error("answer_message_id must have FK REFERENCES messages(id)")
|
||||
}
|
||||
if !strings.Contains(upper, "REFERENCES TURNS(ID)") {
|
||||
t.Error("turn_id must have FK REFERENCES turns(id)")
|
||||
}
|
||||
if !strings.Contains(upper, "REFERENCES CHANNEL_ACTORS(ID)") {
|
||||
t.Error("actor_id must have FK REFERENCES channel_actors(id)")
|
||||
}
|
||||
}
|
||||
|
||||
// TestUp005UniqueConstraint ensures UNIQUE on (tenant_key, channel, answer_message_id, actor_id).
|
||||
func TestUp005UniqueConstraint(t *testing.T) {
|
||||
up := read005UpMigration(t)
|
||||
upper := strings.ToUpper(up)
|
||||
|
||||
hasUnique := strings.Contains(upper, "UNIQUE") &&
|
||||
strings.Contains(upper, "TENANT_KEY") &&
|
||||
strings.Contains(upper, "CHANNEL") &&
|
||||
strings.Contains(upper, "ANSWER_MESSAGE_ID") &&
|
||||
strings.Contains(upper, "ACTOR_ID")
|
||||
if !hasUnique {
|
||||
t.Error("must have UNIQUE constraint on (tenant_key, channel, answer_message_id, actor_id)")
|
||||
}
|
||||
}
|
||||
|
||||
// TestUp005Indexes ensures required indexes are created.
|
||||
func TestUp005Indexes(t *testing.T) {
|
||||
up := read005UpMigration(t)
|
||||
upper := strings.ToUpper(up)
|
||||
|
||||
// Index on (tenant_key, value, updated_at DESC)
|
||||
hasValueIdx := strings.Contains(upper, "TENANT_KEY") &&
|
||||
strings.Contains(upper, "VALUE") &&
|
||||
strings.Contains(upper, "UPDATED_AT") &&
|
||||
strings.Contains(upper, "DESC")
|
||||
if !hasValueIdx {
|
||||
t.Error("must have index on (tenant_key, value, updated_at DESC)")
|
||||
}
|
||||
|
||||
// Index on (tenant_key, reason_code, updated_at DESC)
|
||||
hasReasonIdx := strings.Contains(upper, "TENANT_KEY") &&
|
||||
strings.Contains(upper, "REASON_CODE") &&
|
||||
strings.Contains(upper, "UPDATED_AT") &&
|
||||
strings.Contains(upper, "DESC")
|
||||
if !hasReasonIdx {
|
||||
t.Error("must have index on (tenant_key, reason_code, updated_at DESC)")
|
||||
}
|
||||
|
||||
// Must use CREATE INDEX IF NOT EXISTS
|
||||
if got := countMatchesRe(up, `(?i)CREATE\s+INDEX\s+IF\s+NOT\s+EXISTS`); got < 2 {
|
||||
t.Errorf("expected at least 2 CREATE INDEX IF NOT EXISTS, got %d", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestUp005UpMigrationUsesIfNotExists ensures idempotency.
|
||||
func TestUp005UpMigrationUsesIfNotExists(t *testing.T) {
|
||||
up := read005UpMigration(t)
|
||||
upper := strings.ToUpper(up)
|
||||
|
||||
if strings.Contains(upper, "CREATE TABLE") && !strings.Contains(upper, "CREATE TABLE IF NOT EXISTS") {
|
||||
t.Error("CREATE TABLE must use IF NOT EXISTS for idempotency")
|
||||
}
|
||||
|
||||
// Must have exactly 1 CREATE TABLE IF NOT EXISTS
|
||||
if got := countMatchesRe(up, `(?i)CREATE\s+TABLE\s+IF\s+NOT\s+EXISTS`); got != 1 {
|
||||
t.Errorf("expected 1 CREATE TABLE IF NOT EXISTS, got %d", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestUp005CreatesUniqueCardProviderIndex(t *testing.T) {
|
||||
up := strings.ToUpper(read005UpMigration(t))
|
||||
for _, fragment := range []string{
|
||||
"CREATE UNIQUE INDEX IF NOT EXISTS IDX_MESSAGES_CARD_PROVIDER_UNIQUE",
|
||||
"ON MESSAGES (TENANT_KEY, CHANNEL, PROVIDER_MESSAGE_ID)",
|
||||
"PROVIDER_MESSAGE_ID != ''",
|
||||
"DELIVERY_METHOD = 'CARD'",
|
||||
} {
|
||||
if !strings.Contains(up, fragment) {
|
||||
t.Errorf("provider binding index missing %q", fragment)
|
||||
}
|
||||
}
|
||||
down := strings.ToUpper(read005DownMigration(t))
|
||||
if !strings.Contains(down, "DROP INDEX IF EXISTS IDX_MESSAGES_CARD_PROVIDER_UNIQUE") {
|
||||
t.Error("down migration must remove the B2 provider binding index")
|
||||
}
|
||||
}
|
||||
|
||||
// TestUp005NoForbiddenColumns ensures no secrets/protocol columns.
|
||||
func TestUp005NoForbiddenColumns(t *testing.T) {
|
||||
up := read005UpMigration(t)
|
||||
upper := strings.ToUpper(up)
|
||||
|
||||
forbidden := []string{
|
||||
"WEBHOOK", "TOKEN", "RAW_CALLBACK", "RAW_PAYLOAD",
|
||||
"CONTENT", "CALLBACK", "EXTENSION", "CORP_ID",
|
||||
"SESSIONWEBHOOK", "DOWNLOAD_CODE", "DATAURI", "DATA_URI",
|
||||
"RAW_RESPONSE", "RAW_ERROR", "PROVIDER_RESPONSE", "API_RESPONSE",
|
||||
}
|
||||
for _, col := range forbidden {
|
||||
if strings.Contains(upper, col) {
|
||||
t.Errorf("migration must not contain forbidden column %q", col)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestUp005DoesNotModifyBaselineTables ensures migration only touches answer_feedback.
|
||||
func TestUp005DoesNotModifyBaselineTables(t *testing.T) {
|
||||
up := read005UpMigration(t)
|
||||
upper := strings.ToUpper(up)
|
||||
|
||||
// Must not ALTER other tables
|
||||
baseline := []string{"MESSAGES", "TURNS", "CHANNEL_ACTORS", "CHANNEL_CONVERSATIONS", "SESSION_MAPPINGS"}
|
||||
for _, tbl := range baseline {
|
||||
if strings.Contains(upper, "ALTER TABLE "+tbl) {
|
||||
t.Errorf("migration must not ALTER TABLE %s", tbl)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestDown005DropsAnswerFeedbackTable ensures down drops only the new table and indexes.
|
||||
func TestDown005DropsAnswerFeedbackTable(t *testing.T) {
|
||||
down := read005DownMigration(t)
|
||||
upper := strings.ToUpper(down)
|
||||
|
||||
if !strings.Contains(upper, "DROP TABLE IF EXISTS ANSWER_FEEDBACK") {
|
||||
t.Error("down migration must drop answer_feedback with IF EXISTS")
|
||||
}
|
||||
}
|
||||
|
||||
// TestDown005DoesNotDropBaselineTables ensures down does not touch B1 tables.
|
||||
func TestDown005DoesNotDropBaselineTables(t *testing.T) {
|
||||
down := read005DownMigration(t)
|
||||
upper := strings.ToUpper(down)
|
||||
|
||||
baseline := []string{"MESSAGES", "TURNS", "CHANNEL_ACTORS", "CHANNEL_CONVERSATIONS", "SESSION_MAPPINGS"}
|
||||
for _, tbl := range baseline {
|
||||
if strings.Contains(upper, "DROP TABLE IF EXISTS "+tbl) {
|
||||
t.Errorf("down migration must not drop baseline table %s", tbl)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestDown005DoesNotDropBaselineColumns ensures down does not drop baseline columns.
|
||||
func TestDown005DoesNotDropBaselineColumns(t *testing.T) {
|
||||
down := read005DownMigration(t)
|
||||
upper := strings.ToUpper(down)
|
||||
|
||||
if strings.Contains(upper, "DROP COLUMN") || strings.Contains(upper, "ALTER TABLE") {
|
||||
t.Error("down migration must not drop/alter columns — only drop the answer_feedback table and indexes")
|
||||
}
|
||||
}
|
||||
|
||||
// TestUp005NoReservedWords ensures no bare reserved words as column names.
|
||||
func TestUp005NoReservedWords(t *testing.T) {
|
||||
up := read005UpMigration(t)
|
||||
|
||||
// Check for bare "references" as column name (not in FK context)
|
||||
bareRefs := regexp.MustCompile(`(?im)^\s*REFERENCES\s+`)
|
||||
if bareRefs.MatchString(up) {
|
||||
lines := strings.Split(up, "\n")
|
||||
for _, line := range lines {
|
||||
trimmed := strings.TrimSpace(line)
|
||||
upperLine := strings.ToUpper(trimmed)
|
||||
if strings.HasPrefix(upperLine, "REFERENCES ") && !strings.Contains(upperLine, "FOREIGN KEY") {
|
||||
t.Errorf("reserved word 'references' used as column name: %s", trimmed)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// --- Helpers ---
|
||||
|
||||
func countMatchesRe(haystack, pattern string) int {
|
||||
return len(regexp.MustCompile(pattern).FindAllString(haystack, -1))
|
||||
}
|
||||
@@ -0,0 +1,138 @@
|
||||
package audit
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"time"
|
||||
)
|
||||
|
||||
var (
|
||||
ErrNotFound = errors.New("turn not found")
|
||||
ErrInvalidArgument = errors.New("invalid audit query")
|
||||
)
|
||||
|
||||
// TurnStatus represents the status of a turn.
|
||||
type TurnStatus string
|
||||
|
||||
const (
|
||||
StatusProcessing TurnStatus = "processing"
|
||||
StatusSucceeded TurnStatus = "succeeded"
|
||||
StatusFailed TurnStatus = "failed"
|
||||
StatusCancelled TurnStatus = "cancelled"
|
||||
)
|
||||
|
||||
// TurnSummary is a lightweight representation of a turn for list views.
|
||||
// It must never contain full question/answer text, error_message, WeKnora
|
||||
// session/request IDs, or provider_message_id.
|
||||
type TurnSummary struct {
|
||||
ID string `json:"id"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
Channel string `json:"channel"`
|
||||
Status TurnStatus `json:"status"`
|
||||
ConversationType string `json:"conversation_type"`
|
||||
ConversationTitle string `json:"conversation_title,omitempty"`
|
||||
ActorExternalID string `json:"actor_external_id,omitempty"`
|
||||
ActorDisplayName string `json:"actor_display_name,omitempty"`
|
||||
QuestionSummary string `json:"question_summary"`
|
||||
AnswerSummary string `json:"answer_summary"`
|
||||
ReferenceCount int `json:"reference_count"`
|
||||
AnswerDeliveryStatus string `json:"answer_delivery_status,omitempty"`
|
||||
AnswerDeliveryMethod string `json:"answer_delivery_method,omitempty"`
|
||||
FeedbackUpCount int `json:"feedback_up_count"`
|
||||
FeedbackDownCount int `json:"feedback_down_count"`
|
||||
FeedbackUnresolvedCount int `json:"feedback_unresolved_count"`
|
||||
}
|
||||
|
||||
// SafeReference is a reference with only safe fields exposed.
|
||||
// Content and Raw are never included.
|
||||
type SafeReference struct {
|
||||
ID string `json:"id"`
|
||||
KnowledgeID string `json:"knowledge_id"`
|
||||
KnowledgeTitle string `json:"knowledge_title"`
|
||||
Score float64 `json:"score"`
|
||||
}
|
||||
|
||||
// SafeMedia is a media audit entry with only safe metadata.
|
||||
type SafeMedia struct {
|
||||
Type string `json:"type"`
|
||||
MIMEType string `json:"mime_type"`
|
||||
SizeBytes int64 `json:"size_bytes"`
|
||||
Width int `json:"width"`
|
||||
Height int `json:"height"`
|
||||
}
|
||||
|
||||
// MessageMeta provides request/answer message metadata.
|
||||
type MessageMeta struct {
|
||||
ID string `json:"id"`
|
||||
Direction string `json:"direction"`
|
||||
Role string `json:"role"`
|
||||
MessageType string `json:"message_type"`
|
||||
Status string `json:"status"`
|
||||
ActorName string `json:"actor_name,omitempty"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
DeliveredAt *time.Time `json:"delivered_at,omitempty"`
|
||||
DeliveryMethod string `json:"delivery_method,omitempty"`
|
||||
FailureCode string `json:"failure_code,omitempty"`
|
||||
}
|
||||
|
||||
// AnswerFeedback is the safe current feedback state for one actor. Provider
|
||||
// tracking IDs, internal IDs and raw callback fields are intentionally absent.
|
||||
type AnswerFeedback struct {
|
||||
ActorExternalID string `json:"actor_external_id"`
|
||||
ActorDisplayName string `json:"actor_display_name,omitempty"`
|
||||
Value string `json:"value"`
|
||||
ReasonCode string `json:"reason_code"`
|
||||
UpdatedAt time.Time `json:"updated_at"`
|
||||
}
|
||||
|
||||
// TurnDetail is the full detail view of a turn. It must never include
|
||||
// error_message, WeKnora session/request ID, provider_message_id,
|
||||
// sessionWebhook, tokens, raw API/error, or Reference.Content/Raw.
|
||||
type TurnDetail struct {
|
||||
ID string `json:"id"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
Channel string `json:"channel"`
|
||||
Status TurnStatus `json:"status"`
|
||||
ConversationType string `json:"conversation_type"`
|
||||
ConversationTitle string `json:"conversation_title,omitempty"`
|
||||
ActorExternalID string `json:"actor_external_id,omitempty"`
|
||||
ActorDisplayName string `json:"actor_display_name,omitempty"`
|
||||
Question string `json:"question"`
|
||||
Answer string `json:"answer"`
|
||||
References []SafeReference `json:"references,omitempty"`
|
||||
Media []SafeMedia `json:"media,omitempty"`
|
||||
RequestMessage *MessageMeta `json:"request_message,omitempty"`
|
||||
AnswerMessage *MessageMeta `json:"answer_message,omitempty"`
|
||||
ReferenceCount int `json:"reference_count"`
|
||||
AnswerDeliveryStatus string `json:"answer_delivery_status,omitempty"`
|
||||
AnswerDeliveryMethod string `json:"answer_delivery_method,omitempty"`
|
||||
Feedback []AnswerFeedback `json:"feedback,omitempty"`
|
||||
}
|
||||
|
||||
// ListTurnsRequest is the input for listing turns.
|
||||
type ListTurnsRequest struct {
|
||||
Limit int
|
||||
Cursor string
|
||||
Channel string
|
||||
Status TurnStatus
|
||||
}
|
||||
|
||||
// ListTurnsResult is the output of a list turns query.
|
||||
type ListTurnsResult struct {
|
||||
Items []TurnSummary `json:"items"`
|
||||
NextCursor string `json:"next_cursor,omitempty"`
|
||||
}
|
||||
|
||||
type ListQuery struct {
|
||||
Limit int
|
||||
Channel string
|
||||
Status TurnStatus
|
||||
BeforeCreatedAt *time.Time
|
||||
BeforeID string
|
||||
}
|
||||
|
||||
// TurnRepository is the data access interface for audit queries.
|
||||
type TurnRepository interface {
|
||||
ListTurns(ctx context.Context, tenantKey string, query ListQuery) ([]TurnSummary, error)
|
||||
GetTurn(ctx context.Context, tenantKey, turnID string) (TurnDetail, error)
|
||||
}
|
||||
@@ -0,0 +1,54 @@
|
||||
package audit
|
||||
|
||||
import (
|
||||
"encoding/base64"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
const maxCursorLength = 512
|
||||
|
||||
type cursorData struct {
|
||||
CreatedAt time.Time `json:"t"`
|
||||
ID string `json:"id"`
|
||||
}
|
||||
|
||||
func encodeCursor(createdAt time.Time, id string) string {
|
||||
data := cursorData{
|
||||
CreatedAt: createdAt,
|
||||
ID: id,
|
||||
}
|
||||
payload, err := json.Marshal(data)
|
||||
if err != nil {
|
||||
// This should never happen for a simple struct.
|
||||
panic("encodeCursor marshal: " + err.Error())
|
||||
}
|
||||
return base64.RawURLEncoding.EncodeToString(payload)
|
||||
}
|
||||
|
||||
func encodeCursorString(payload string) string {
|
||||
return base64.RawURLEncoding.EncodeToString([]byte(payload))
|
||||
}
|
||||
|
||||
func decodeCursor(cursor string) (cursorData, error) {
|
||||
if strings.TrimSpace(cursor) == "" {
|
||||
return cursorData{}, errors.New("empty cursor")
|
||||
}
|
||||
if len(cursor) > maxCursorLength {
|
||||
return cursorData{}, errors.New("cursor too long")
|
||||
}
|
||||
payload, err := base64.RawURLEncoding.DecodeString(cursor)
|
||||
if err != nil {
|
||||
return cursorData{}, errors.New("invalid cursor encoding")
|
||||
}
|
||||
var data cursorData
|
||||
if err := json.Unmarshal(payload, &data); err != nil {
|
||||
return cursorData{}, errors.New("invalid cursor payload")
|
||||
}
|
||||
if data.CreatedAt.IsZero() || data.ID == "" {
|
||||
return cursorData{}, errors.New("incomplete cursor data")
|
||||
}
|
||||
return data, nil
|
||||
}
|
||||
@@ -0,0 +1,63 @@
|
||||
package audit
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestCursorRoundTrip(t *testing.T) {
|
||||
now := time.Date(2026, 7, 21, 12, 0, 0, 0, time.UTC)
|
||||
cursor := encodeCursor(now, "turn-abc-123")
|
||||
decoded, err := decodeCursor(cursor)
|
||||
if err != nil {
|
||||
t.Fatalf("decodeCursor: %v", err)
|
||||
}
|
||||
if !decoded.CreatedAt.Equal(now) {
|
||||
t.Fatalf("CreatedAt=%v, want %v", decoded.CreatedAt, now)
|
||||
}
|
||||
if decoded.ID != "turn-abc-123" {
|
||||
t.Fatalf("ID=%q, want %q", decoded.ID, "turn-abc-123")
|
||||
}
|
||||
}
|
||||
|
||||
func TestDecodeCursorRejectsEmpty(t *testing.T) {
|
||||
_, err := decodeCursor("")
|
||||
if err == nil {
|
||||
t.Fatal("expected error for empty cursor")
|
||||
}
|
||||
}
|
||||
|
||||
func TestDecodeCursorRejectsInvalidBase64(t *testing.T) {
|
||||
_, err := decodeCursor("!!!not-valid-base64!!!")
|
||||
if err == nil {
|
||||
t.Fatal("expected error for invalid base64")
|
||||
}
|
||||
}
|
||||
|
||||
func TestDecodeCursorRejectsTruncated(t *testing.T) {
|
||||
// Create a valid cursor then truncate it
|
||||
cursor := encodeCursor(time.Now().UTC(), "id")
|
||||
truncated := cursor[:len(cursor)-3]
|
||||
_, err := decodeCursor(truncated)
|
||||
if err == nil {
|
||||
t.Fatal("expected error for truncated cursor")
|
||||
}
|
||||
}
|
||||
|
||||
func TestDecodeCursorRejectsTooLong(t *testing.T) {
|
||||
// 1024 bytes should exceed max cursor length
|
||||
_, err := decodeCursor(strings.Repeat("A", 1024))
|
||||
if err == nil {
|
||||
t.Fatal("expected error for too-long cursor")
|
||||
}
|
||||
}
|
||||
|
||||
func TestDecodeCursorRejectsMissingFields(t *testing.T) {
|
||||
// Invalid JSON payload (not a valid cursor struct)
|
||||
// We create a cursor-like string that decodes but has missing fields
|
||||
_, err := decodeCursor(encodeCursorString("{}"))
|
||||
if err == nil {
|
||||
t.Fatal("expected error for missing fields in cursor")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,43 @@
|
||||
package audit
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestFeedbackProjectionJSONUsesOnlySafeFields(t *testing.T) {
|
||||
updated := time.Date(2026, 7, 21, 15, 0, 0, 0, time.UTC)
|
||||
detail := TurnDetail{Feedback: []AnswerFeedback{{
|
||||
ActorExternalID: "user-1", ActorDisplayName: "Alice", Value: "down", ReasonCode: "not_resolved", UpdatedAt: updated,
|
||||
}}}
|
||||
raw, err := json.Marshal(detail)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
body := string(raw)
|
||||
for _, want := range []string{`"feedback"`, `"actor_external_id":"user-1"`, `"actor_display_name":"Alice"`, `"value":"down"`, `"reason_code":"not_resolved"`, `"updated_at"`} {
|
||||
if !strings.Contains(body, want) {
|
||||
t.Errorf("feedback JSON missing %s: %s", want, body)
|
||||
}
|
||||
}
|
||||
for _, forbidden := range []string{"out_track", "provider_message", "answer_message_id", "actor_id", "token", "callback", "params"} {
|
||||
if strings.Contains(strings.ToLower(body), forbidden) {
|
||||
t.Errorf("feedback JSON leaked %q: %s", forbidden, body)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestTurnSummaryFeedbackCountsJSON(t *testing.T) {
|
||||
raw, err := json.Marshal(TurnSummary{FeedbackUpCount: 2, FeedbackDownCount: 3, FeedbackUnresolvedCount: 1})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
body := string(raw)
|
||||
for _, want := range []string{`"feedback_up_count":2`, `"feedback_down_count":3`, `"feedback_unresolved_count":1`} {
|
||||
if !strings.Contains(body, want) {
|
||||
t.Errorf("summary JSON missing %s: %s", want, body)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,97 @@
|
||||
package audit
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
)
|
||||
|
||||
const (
|
||||
defaultLimit = 20
|
||||
maxLimit = 100
|
||||
maxChannelLen = 128
|
||||
maxTurnIDLen = 128
|
||||
maxSummaryLen = 200
|
||||
)
|
||||
|
||||
type Service struct {
|
||||
tenantKey string
|
||||
repo TurnRepository
|
||||
}
|
||||
|
||||
func NewService(tenantKey string, repo TurnRepository) (*Service, error) {
|
||||
tenantKey = strings.TrimSpace(tenantKey)
|
||||
if tenantKey == "" || repo == nil {
|
||||
return nil, fmt.Errorf("%w: tenant and repository are required", ErrInvalidArgument)
|
||||
}
|
||||
return &Service{tenantKey: tenantKey, repo: repo}, nil
|
||||
}
|
||||
|
||||
func (s *Service) ListTurns(ctx context.Context, req ListTurnsRequest) (ListTurnsResult, error) {
|
||||
if req.Limit == 0 {
|
||||
req.Limit = defaultLimit
|
||||
}
|
||||
if req.Limit < 1 || req.Limit > maxLimit {
|
||||
return ListTurnsResult{}, fmt.Errorf("%w: limit must be between 1 and %d", ErrInvalidArgument, maxLimit)
|
||||
}
|
||||
req.Channel = strings.TrimSpace(req.Channel)
|
||||
if len(req.Channel) > maxChannelLen {
|
||||
return ListTurnsResult{}, fmt.Errorf("%w: channel too long", ErrInvalidArgument)
|
||||
}
|
||||
if !validStatus(req.Status) {
|
||||
return ListTurnsResult{}, fmt.Errorf("%w: invalid status", ErrInvalidArgument)
|
||||
}
|
||||
|
||||
query := ListQuery{Limit: req.Limit + 1, Channel: req.Channel, Status: req.Status}
|
||||
if req.Cursor != "" {
|
||||
cursor, err := decodeCursor(req.Cursor)
|
||||
if err != nil {
|
||||
return ListTurnsResult{}, fmt.Errorf("%w: invalid cursor", ErrInvalidArgument)
|
||||
}
|
||||
query.BeforeCreatedAt = &cursor.CreatedAt
|
||||
query.BeforeID = cursor.ID
|
||||
}
|
||||
items, err := s.repo.ListTurns(ctx, s.tenantKey, query)
|
||||
if err != nil {
|
||||
return ListTurnsResult{}, err
|
||||
}
|
||||
for i := range items {
|
||||
items[i].QuestionSummary = truncateRunes(items[i].QuestionSummary, maxSummaryLen)
|
||||
items[i].AnswerSummary = truncateRunes(items[i].AnswerSummary, maxSummaryLen)
|
||||
}
|
||||
result := ListTurnsResult{Items: items}
|
||||
if result.Items == nil {
|
||||
result.Items = []TurnSummary{}
|
||||
}
|
||||
if len(result.Items) > req.Limit {
|
||||
result.Items = result.Items[:req.Limit]
|
||||
last := result.Items[len(result.Items)-1]
|
||||
result.NextCursor = encodeCursor(last.CreatedAt, last.ID)
|
||||
}
|
||||
return result, nil
|
||||
}
|
||||
|
||||
func (s *Service) GetTurn(ctx context.Context, turnID string) (TurnDetail, error) {
|
||||
turnID = strings.TrimSpace(turnID)
|
||||
if turnID == "" || len(turnID) > maxTurnIDLen {
|
||||
return TurnDetail{}, fmt.Errorf("%w: invalid turn id", ErrInvalidArgument)
|
||||
}
|
||||
return s.repo.GetTurn(ctx, s.tenantKey, turnID)
|
||||
}
|
||||
|
||||
func validStatus(status TurnStatus) bool {
|
||||
switch status {
|
||||
case "", StatusProcessing, StatusSucceeded, StatusFailed, StatusCancelled:
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
func truncateRunes(value string, limit int) string {
|
||||
runes := []rune(value)
|
||||
if len(runes) <= limit {
|
||||
return value
|
||||
}
|
||||
return string(runes[:limit])
|
||||
}
|
||||
@@ -0,0 +1,179 @@
|
||||
package audit
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
type stubRepo struct {
|
||||
listFn func(context.Context, string, ListQuery) ([]TurnSummary, error)
|
||||
getFn func(context.Context, string, string) (TurnDetail, error)
|
||||
}
|
||||
|
||||
func (r *stubRepo) ListTurns(ctx context.Context, tenant string, query ListQuery) ([]TurnSummary, error) {
|
||||
if r.listFn == nil {
|
||||
return nil, nil
|
||||
}
|
||||
return r.listFn(ctx, tenant, query)
|
||||
}
|
||||
|
||||
func (r *stubRepo) GetTurn(ctx context.Context, tenant, id string) (TurnDetail, error) {
|
||||
if r.getFn == nil {
|
||||
return TurnDetail{}, ErrNotFound
|
||||
}
|
||||
return r.getFn(ctx, tenant, id)
|
||||
}
|
||||
|
||||
func TestNewServiceRequiresTenantAndRepository(t *testing.T) {
|
||||
repo := &stubRepo{}
|
||||
if _, err := NewService("", repo); !errors.Is(err, ErrInvalidArgument) {
|
||||
t.Fatalf("empty tenant error=%v", err)
|
||||
}
|
||||
if _, err := NewService("tenant", nil); !errors.Is(err, ErrInvalidArgument) {
|
||||
t.Fatalf("nil repository error=%v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestServiceListUsesServerTenantDefaultLimitAndFetchesExtra(t *testing.T) {
|
||||
var gotTenant string
|
||||
var gotQuery ListQuery
|
||||
repo := &stubRepo{listFn: func(_ context.Context, tenant string, query ListQuery) ([]TurnSummary, error) {
|
||||
gotTenant, gotQuery = tenant, query
|
||||
return []TurnSummary{}, nil
|
||||
}}
|
||||
svc, err := NewService(" tenant-a ", repo)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
page, err := svc.ListTurns(context.Background(), ListTurnsRequest{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if gotTenant != "tenant-a" || gotQuery.Limit != 21 {
|
||||
t.Fatalf("tenant=%q query=%+v", gotTenant, gotQuery)
|
||||
}
|
||||
if page.Items == nil || page.NextCursor != "" {
|
||||
t.Fatalf("page=%+v", page)
|
||||
}
|
||||
}
|
||||
|
||||
func TestServiceListBuildsStableNextCursor(t *testing.T) {
|
||||
t3 := time.Date(2026, 7, 21, 3, 0, 0, 0, time.UTC)
|
||||
t2 := t3.Add(-time.Minute)
|
||||
t1 := t2.Add(-time.Minute)
|
||||
repo := &stubRepo{listFn: func(_ context.Context, _ string, query ListQuery) ([]TurnSummary, error) {
|
||||
if query.Limit != 3 {
|
||||
t.Fatalf("repository limit=%d want=3", query.Limit)
|
||||
}
|
||||
return []TurnSummary{{ID: "turn-3", CreatedAt: t3}, {ID: "turn-2", CreatedAt: t2}, {ID: "turn-1", CreatedAt: t1}}, nil
|
||||
}}
|
||||
svc, _ := NewService("tenant", repo)
|
||||
page, err := svc.ListTurns(context.Background(), ListTurnsRequest{Limit: 2})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(page.Items) != 2 || page.Items[1].ID != "turn-2" || page.NextCursor == "" {
|
||||
t.Fatalf("page=%+v", page)
|
||||
}
|
||||
decoded, err := decodeCursor(page.NextCursor)
|
||||
if err != nil || decoded.ID != "turn-2" || !decoded.CreatedAt.Equal(t2) {
|
||||
t.Fatalf("next cursor=%+v err=%v", decoded, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestServiceListDecodesCursorForRepository(t *testing.T) {
|
||||
created := time.Date(2026, 7, 21, 2, 0, 0, 123, time.UTC)
|
||||
cursor := encodeCursor(created, "turn-before")
|
||||
repo := &stubRepo{listFn: func(_ context.Context, _ string, query ListQuery) ([]TurnSummary, error) {
|
||||
if query.BeforeID != "turn-before" || query.BeforeCreatedAt == nil || !query.BeforeCreatedAt.Equal(created) {
|
||||
t.Fatalf("decoded query=%+v", query)
|
||||
}
|
||||
return nil, nil
|
||||
}}
|
||||
svc, _ := NewService("tenant", repo)
|
||||
if _, err := svc.ListTurns(context.Background(), ListTurnsRequest{Cursor: cursor}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestServiceListValidation(t *testing.T) {
|
||||
svc, _ := NewService("tenant", &stubRepo{})
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
req ListTurnsRequest
|
||||
}{
|
||||
{name: "negative limit", req: ListTurnsRequest{Limit: -1}},
|
||||
{name: "too large limit", req: ListTurnsRequest{Limit: 101}},
|
||||
{name: "bad status", req: ListTurnsRequest{Status: "bogus"}},
|
||||
{name: "long channel", req: ListTurnsRequest{Channel: strings.Repeat("x", 129)}},
|
||||
{name: "bad cursor", req: ListTurnsRequest{Cursor: "not-a-cursor"}},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
if _, err := svc.ListTurns(context.Background(), tc.req); !errors.Is(err, ErrInvalidArgument) {
|
||||
t.Fatalf("error=%v", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestServiceListAcceptsMaximumLimitAndFilters(t *testing.T) {
|
||||
repo := &stubRepo{listFn: func(_ context.Context, _ string, query ListQuery) ([]TurnSummary, error) {
|
||||
if query.Limit != 101 || query.Channel != "dingtalk" || query.Status != StatusSucceeded {
|
||||
t.Fatalf("query=%+v", query)
|
||||
}
|
||||
return nil, nil
|
||||
}}
|
||||
svc, _ := NewService("tenant", repo)
|
||||
if _, err := svc.ListTurns(context.Background(), ListTurnsRequest{Limit: 100, Channel: " dingtalk ", Status: StatusSucceeded}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestServiceListTruncatesUnicodeSummariesTo200Runes(t *testing.T) {
|
||||
long := strings.Repeat("初", 250)
|
||||
repo := &stubRepo{listFn: func(context.Context, string, ListQuery) ([]TurnSummary, error) {
|
||||
return []TurnSummary{{ID: "turn", CreatedAt: time.Now(), QuestionSummary: long, AnswerSummary: long}}, nil
|
||||
}}
|
||||
svc, _ := NewService("tenant", repo)
|
||||
page, err := svc.ListTurns(context.Background(), ListTurnsRequest{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len([]rune(page.Items[0].QuestionSummary)) != 200 || len([]rune(page.Items[0].AnswerSummary)) != 200 {
|
||||
t.Fatalf("summary lengths=%d/%d", len([]rune(page.Items[0].QuestionSummary)), len([]rune(page.Items[0].AnswerSummary)))
|
||||
}
|
||||
}
|
||||
|
||||
func TestServiceGetUsesServerTenant(t *testing.T) {
|
||||
var gotTenant, gotID string
|
||||
repo := &stubRepo{getFn: func(_ context.Context, tenant, id string) (TurnDetail, error) {
|
||||
gotTenant, gotID = tenant, id
|
||||
return TurnDetail{ID: id}, nil
|
||||
}}
|
||||
svc, _ := NewService("tenant-a", repo)
|
||||
if _, err := svc.GetTurn(context.Background(), " turn-1 "); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if gotTenant != "tenant-a" || gotID != "turn-1" {
|
||||
t.Fatalf("tenant=%q id=%q", gotTenant, gotID)
|
||||
}
|
||||
}
|
||||
|
||||
func TestServiceGetValidationAndNotFound(t *testing.T) {
|
||||
repo := &stubRepo{getFn: func(context.Context, string, string) (TurnDetail, error) {
|
||||
return TurnDetail{}, ErrNotFound
|
||||
}}
|
||||
svc, _ := NewService("tenant", repo)
|
||||
if _, err := svc.GetTurn(context.Background(), " "); !errors.Is(err, ErrInvalidArgument) {
|
||||
t.Fatalf("blank id error=%v", err)
|
||||
}
|
||||
if _, err := svc.GetTurn(context.Background(), strings.Repeat("x", 129)); !errors.Is(err, ErrInvalidArgument) {
|
||||
t.Fatalf("long id error=%v", err)
|
||||
}
|
||||
if _, err := svc.GetTurn(context.Background(), "missing"); !errors.Is(err, ErrNotFound) {
|
||||
t.Fatalf("missing error=%v", err)
|
||||
}
|
||||
}
|
||||
@@ -202,7 +202,11 @@ func (s *CardSender) buildCardBody(req CardRequest, outTrackID string) map[strin
|
||||
"callbackType": "STREAM",
|
||||
"cardData": map[string]any{
|
||||
"cardParamMap": map[string]any{
|
||||
"content": req.Content,
|
||||
"content": req.Content,
|
||||
"feedbackSelected": "",
|
||||
"feedbackReasonSelected": "",
|
||||
"showFeedbackReasons": "false",
|
||||
"feedbackMessage": "",
|
||||
},
|
||||
},
|
||||
"userIdType": 1,
|
||||
|
||||
@@ -218,6 +218,16 @@ func TestCardSendDirectPayload(t *testing.T) {
|
||||
if paramMap["content"] != "Hello from AI" {
|
||||
t.Errorf("content=%v", paramMap["content"])
|
||||
}
|
||||
for key, want := range map[string]any{
|
||||
"feedbackSelected": "",
|
||||
"feedbackReasonSelected": "",
|
||||
"showFeedbackReasons": "false",
|
||||
"feedbackMessage": "",
|
||||
} {
|
||||
if paramMap[key] != want {
|
||||
t.Errorf("%s=%v, want %v", key, paramMap[key], want)
|
||||
}
|
||||
}
|
||||
if body["userIdType"] != float64(1) {
|
||||
t.Errorf("userIdType=%v", body["userIdType"])
|
||||
}
|
||||
|
||||
@@ -0,0 +1,111 @@
|
||||
package dingtalk
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
|
||||
sdkcard "github.com/open-dingtalk/dingtalk-stream-sdk-go/card"
|
||||
|
||||
feedbackdomain "gitea.zheromiku.top/Weknora/dobotai-platform/internal/feedback"
|
||||
)
|
||||
|
||||
const (
|
||||
ActionFeedbackUp = "dobotai_feedback_up"
|
||||
ActionFeedbackDown = "dobotai_feedback_down"
|
||||
ActionReasonIncorrect = "dobotai_reason_incorrect"
|
||||
ActionReasonNotResolved = "dobotai_reason_not_resolved"
|
||||
ActionReasonIrrelevant = "dobotai_reason_irrelevant"
|
||||
ActionReasonUnclear = "dobotai_reason_unclear"
|
||||
ActionReasonOutdated = "dobotai_reason_outdated"
|
||||
ActionReasonOther = "dobotai_reason_other"
|
||||
)
|
||||
|
||||
type feedbackService interface {
|
||||
Apply(context.Context, feedbackdomain.Command) (feedbackdomain.Result, error)
|
||||
}
|
||||
|
||||
type CardFeedbackHandler struct {
|
||||
service feedbackService
|
||||
}
|
||||
|
||||
func NewCardFeedbackHandler(service feedbackService) (*CardFeedbackHandler, error) {
|
||||
if service == nil {
|
||||
return nil, errors.New("dingtalk card feedback: service is required")
|
||||
}
|
||||
return &CardFeedbackHandler{service: service}, nil
|
||||
}
|
||||
|
||||
func (h *CardFeedbackHandler) HandleSDK(ctx context.Context, request *sdkcard.CardRequest) (*sdkcard.CardResponse, error) {
|
||||
if request == nil {
|
||||
return nil, errors.New("dingtalk card feedback: request is nil")
|
||||
}
|
||||
value, reason, recognized := feedbackAction(request.CardActionData.CardPrivateData.ActionIdList)
|
||||
if !recognized {
|
||||
return &sdkcard.CardResponse{}, nil
|
||||
}
|
||||
_, err := h.service.Apply(ctx, feedbackdomain.Command{
|
||||
ExternalActorID: request.UserId,
|
||||
OutTrackID: request.OutTrackId,
|
||||
Value: value,
|
||||
ReasonCode: reason,
|
||||
Source: feedbackdomain.SourceDingTalkCard,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
showReasons := "false"
|
||||
message := "感谢反馈,我们会继续改进"
|
||||
if value == feedbackdomain.ValueUp {
|
||||
message = "感谢反馈"
|
||||
}
|
||||
if value == feedbackdomain.ValueDown && reason == feedbackdomain.ReasonEmpty {
|
||||
showReasons = "true"
|
||||
message = "请选择未解决原因"
|
||||
}
|
||||
return &sdkcard.CardResponse{
|
||||
CardUpdateOptions: &sdkcard.CardUpdateOptions{UpdatePrivateDataByKey: true},
|
||||
UserPrivateData: &sdkcard.CardDataDto{CardParamMap: map[string]string{
|
||||
"feedbackSelected": value,
|
||||
"feedbackReasonSelected": reason,
|
||||
"showFeedbackReasons": showReasons,
|
||||
"feedbackMessage": message,
|
||||
}},
|
||||
}, nil
|
||||
}
|
||||
|
||||
func feedbackAction(actions []string) (string, string, bool) {
|
||||
value, reason, matches := "", "", 0
|
||||
for _, action := range actions {
|
||||
candidateValue, candidateReason, ok := oneFeedbackAction(action)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
matches++
|
||||
value, reason = candidateValue, candidateReason
|
||||
}
|
||||
return value, reason, matches == 1
|
||||
}
|
||||
|
||||
func oneFeedbackAction(action string) (string, string, bool) {
|
||||
switch action {
|
||||
case ActionFeedbackUp:
|
||||
return feedbackdomain.ValueUp, feedbackdomain.ReasonEmpty, true
|
||||
case ActionFeedbackDown:
|
||||
return feedbackdomain.ValueDown, feedbackdomain.ReasonEmpty, true
|
||||
case ActionReasonIncorrect:
|
||||
return feedbackdomain.ValueDown, feedbackdomain.ReasonIncorrect, true
|
||||
case ActionReasonNotResolved:
|
||||
return feedbackdomain.ValueDown, feedbackdomain.ReasonNotResolved, true
|
||||
case ActionReasonIrrelevant:
|
||||
return feedbackdomain.ValueDown, feedbackdomain.ReasonIrrelevant, true
|
||||
case ActionReasonUnclear:
|
||||
return feedbackdomain.ValueDown, feedbackdomain.ReasonUnclear, true
|
||||
case ActionReasonOutdated:
|
||||
return feedbackdomain.ValueDown, feedbackdomain.ReasonOutdated, true
|
||||
case ActionReasonOther:
|
||||
return feedbackdomain.ValueDown, feedbackdomain.ReasonOther, true
|
||||
default:
|
||||
return "", "", false
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,185 @@
|
||||
package dingtalk
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
sdkcard "github.com/open-dingtalk/dingtalk-stream-sdk-go/card"
|
||||
"github.com/open-dingtalk/dingtalk-stream-sdk-go/payload"
|
||||
"github.com/open-dingtalk/dingtalk-stream-sdk-go/utils"
|
||||
|
||||
feedbackdomain "gitea.zheromiku.top/Weknora/dobotai-platform/internal/feedback"
|
||||
)
|
||||
|
||||
type fakeFeedbackService struct {
|
||||
calls int
|
||||
command feedbackdomain.Command
|
||||
result feedbackdomain.Result
|
||||
err error
|
||||
}
|
||||
|
||||
func (f *fakeFeedbackService) Apply(_ context.Context, command feedbackdomain.Command) (feedbackdomain.Result, error) {
|
||||
f.calls++
|
||||
f.command = command
|
||||
return f.result, f.err
|
||||
}
|
||||
|
||||
func cardRequest(action string) *sdkcard.CardRequest {
|
||||
return &sdkcard.CardRequest{
|
||||
OutTrackId: "track-1",
|
||||
UserId: "user-1",
|
||||
CorpId: "must-not-be-stored",
|
||||
Content: "must-not-be-stored",
|
||||
Extension: "must-not-be-stored",
|
||||
CardActionData: sdkcard.PrivateCardActionData{CardPrivateData: sdkcard.CardPrivateData{
|
||||
ActionIdList: []string{action},
|
||||
Params: map[string]any{
|
||||
"tenant_key": "attacker", "turn_id": "attacker", "user_id": "attacker", "token": "attacker",
|
||||
},
|
||||
}},
|
||||
}
|
||||
}
|
||||
|
||||
func TestCardFeedbackMapsWhitelistedActions(t *testing.T) {
|
||||
tests := []struct {
|
||||
action string
|
||||
value string
|
||||
reason string
|
||||
showReasons string
|
||||
message string
|
||||
}{
|
||||
{ActionFeedbackUp, feedbackdomain.ValueUp, "", "false", "感谢反馈"},
|
||||
{ActionFeedbackDown, feedbackdomain.ValueDown, "", "true", "请选择未解决原因"},
|
||||
{ActionReasonIncorrect, feedbackdomain.ValueDown, feedbackdomain.ReasonIncorrect, "false", "感谢反馈,我们会继续改进"},
|
||||
{ActionReasonNotResolved, feedbackdomain.ValueDown, feedbackdomain.ReasonNotResolved, "false", "感谢反馈,我们会继续改进"},
|
||||
{ActionReasonIrrelevant, feedbackdomain.ValueDown, feedbackdomain.ReasonIrrelevant, "false", "感谢反馈,我们会继续改进"},
|
||||
{ActionReasonUnclear, feedbackdomain.ValueDown, feedbackdomain.ReasonUnclear, "false", "感谢反馈,我们会继续改进"},
|
||||
{ActionReasonOutdated, feedbackdomain.ValueDown, feedbackdomain.ReasonOutdated, "false", "感谢反馈,我们会继续改进"},
|
||||
{ActionReasonOther, feedbackdomain.ValueDown, feedbackdomain.ReasonOther, "false", "感谢反馈,我们会继续改进"},
|
||||
}
|
||||
for _, tc := range tests {
|
||||
t.Run(tc.action, func(t *testing.T) {
|
||||
service := &fakeFeedbackService{result: feedbackdomain.Result{Created: true, Value: tc.value, ReasonCode: tc.reason}}
|
||||
handler, err := NewCardFeedbackHandler(service)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
response, err := handler.HandleSDK(context.Background(), cardRequest(tc.action))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if service.calls != 1 {
|
||||
t.Fatalf("calls=%d", service.calls)
|
||||
}
|
||||
if service.command.ExternalActorID != "user-1" || service.command.OutTrackID != "track-1" || service.command.Value != tc.value || service.command.ReasonCode != tc.reason || service.command.Source != feedbackdomain.SourceDingTalkCard {
|
||||
t.Fatalf("command=%+v", service.command)
|
||||
}
|
||||
if response == nil || response.CardUpdateOptions == nil || !response.CardUpdateOptions.UpdatePrivateDataByKey || response.CardUpdateOptions.UpdateCardDataByKey || response.CardData != nil || response.UserPrivateData == nil {
|
||||
t.Fatalf("response=%+v", response)
|
||||
}
|
||||
params := response.UserPrivateData.CardParamMap
|
||||
if params["feedbackSelected"] != tc.value || params["feedbackReasonSelected"] != tc.reason || params["showFeedbackReasons"] != tc.showReasons || params["feedbackMessage"] != tc.message {
|
||||
t.Fatalf("params=%+v", params)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestCardFeedbackUnknownOrAmbiguousActionIsSafeAck(t *testing.T) {
|
||||
for _, actions := range [][]string{{"unknown"}, {ActionFeedbackUp, ActionFeedbackDown}, nil} {
|
||||
service := &fakeFeedbackService{}
|
||||
handler, _ := NewCardFeedbackHandler(service)
|
||||
request := cardRequest("unknown")
|
||||
request.CardActionData.CardPrivateData.ActionIdList = actions
|
||||
response, err := handler.HandleSDK(context.Background(), request)
|
||||
if err != nil || response == nil || service.calls != 0 {
|
||||
t.Fatalf("actions=%v response=%+v error=%v calls=%d", actions, response, err, service.calls)
|
||||
}
|
||||
if response.CardUpdateOptions != nil || response.UserPrivateData != nil {
|
||||
t.Fatalf("safe ACK must not update card: %+v", response)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestCardFeedbackValidationAndServiceError(t *testing.T) {
|
||||
if _, err := NewCardFeedbackHandler(nil); err == nil {
|
||||
t.Fatal("nil service must fail")
|
||||
}
|
||||
serviceErr := errors.New("store failed")
|
||||
service := &fakeFeedbackService{err: serviceErr}
|
||||
handler, _ := NewCardFeedbackHandler(service)
|
||||
if _, err := handler.HandleSDK(context.Background(), nil); err == nil {
|
||||
t.Fatal("nil request must fail")
|
||||
}
|
||||
response, err := handler.HandleSDK(context.Background(), cardRequest(ActionFeedbackUp))
|
||||
if !errors.Is(err, serviceErr) || response != nil || service.calls != 1 {
|
||||
t.Fatalf("response=%+v error=%v calls=%d", response, err, service.calls)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStreamClientRegistersOptionalCardFeedbackRouter(t *testing.T) {
|
||||
config := StreamConfig{ClientID: "id", ClientSecret: "secret", TenantKey: "tenant"}
|
||||
chatHandler := mustTestHandler(t)
|
||||
feedbackHandler, _ := NewCardFeedbackHandler(&fakeFeedbackService{})
|
||||
client, err := NewStreamClient(config, chatHandler, feedbackHandler)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := client.inner.GetHandler(utils.SubscriptionTypeKCallback, payload.CardInstanceCallbackTopic); err != nil {
|
||||
t.Fatalf("card callback router not registered: %v", err)
|
||||
}
|
||||
|
||||
withoutFeedback, err := NewStreamClient(config, chatHandler)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := withoutFeedback.inner.GetHandler(utils.SubscriptionTypeKCallback, payload.CardInstanceCallbackTopic); err == nil {
|
||||
t.Fatal("card callback router must remain optional for old callers")
|
||||
}
|
||||
if _, err := NewStreamClient(config, chatHandler, nil); err == nil {
|
||||
t.Fatal("explicit nil feedback handler must fail")
|
||||
}
|
||||
if _, err := NewStreamClient(config, chatHandler, feedbackHandler, feedbackHandler); err == nil {
|
||||
t.Fatal("multiple feedback handlers must fail")
|
||||
}
|
||||
}
|
||||
|
||||
func TestStreamCardRouterHandlesRealSDKDataFrameEnvelope(t *testing.T) {
|
||||
service := &fakeFeedbackService{result: feedbackdomain.Result{Created: true, Value: feedbackdomain.ValueDown, ReasonCode: feedbackdomain.ReasonNotResolved}}
|
||||
feedbackHandler, _ := NewCardFeedbackHandler(service)
|
||||
client, err := NewStreamClient(
|
||||
StreamConfig{ClientID: "id", ClientSecret: "secret", TenantKey: "tenant"},
|
||||
mustTestHandler(t), feedbackHandler,
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
router, err := client.inner.GetHandler(utils.SubscriptionTypeKCallback, payload.CardInstanceCallbackTopic)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
actionData, _ := json.Marshal(map[string]any{"cardPrivateData": map[string]any{
|
||||
"actionIds": []string{ActionReasonNotResolved},
|
||||
"params": map[string]any{"tenant_key": "attacker", "token": "attacker"},
|
||||
}})
|
||||
outer, _ := json.Marshal(map[string]any{
|
||||
"outTrackId": "track-frame", "userId": "user-frame", "content": string(actionData),
|
||||
"corpId": "must-not-be-used", "extension": "must-not-be-used",
|
||||
})
|
||||
frameResponse, err := router(context.Background(), &payload.DataFrame{Data: string(outer)})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if frameResponse == nil || frameResponse.Code != payload.DataFrameResponseStatusCodeKOK {
|
||||
t.Fatalf("frame response=%+v", frameResponse)
|
||||
}
|
||||
if service.calls != 1 || service.command.ExternalActorID != "user-frame" || service.command.OutTrackID != "track-frame" || service.command.Value != feedbackdomain.ValueDown || service.command.ReasonCode != feedbackdomain.ReasonNotResolved {
|
||||
t.Fatalf("service calls=%d command=%+v", service.calls, service.command)
|
||||
}
|
||||
if strings.Contains(frameResponse.Data, "attacker") || !strings.Contains(frameResponse.Data, "feedbackReasonSelected") {
|
||||
t.Fatalf("unsafe or incomplete frame response: %s", frameResponse.Data)
|
||||
}
|
||||
}
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
"fmt"
|
||||
"strings"
|
||||
|
||||
sdkcard "github.com/open-dingtalk/dingtalk-stream-sdk-go/card"
|
||||
"github.com/open-dingtalk/dingtalk-stream-sdk-go/chatbot"
|
||||
dtsdk "github.com/open-dingtalk/dingtalk-stream-sdk-go/client"
|
||||
)
|
||||
@@ -232,7 +233,7 @@ type StreamClient struct {
|
||||
// credential and registers the chatbot callback router. The handler is
|
||||
// invoked synchronously from the SDK callback; its return value only
|
||||
// acknowledges the frame, replies always go through the session webhook.
|
||||
func NewStreamClient(config StreamConfig, handler *Handler) (*StreamClient, error) {
|
||||
func NewStreamClient(config StreamConfig, handler *Handler, feedbackHandlers ...*CardFeedbackHandler) (*StreamClient, error) {
|
||||
if err := config.Validate(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -242,6 +243,12 @@ func NewStreamClient(config StreamConfig, handler *Handler) (*StreamClient, erro
|
||||
if handler == nil {
|
||||
return nil, errors.New("dingtalk stream: handler is required")
|
||||
}
|
||||
if len(feedbackHandlers) > 1 {
|
||||
return nil, errors.New("dingtalk stream: at most one card feedback handler is allowed")
|
||||
}
|
||||
if len(feedbackHandlers) == 1 && feedbackHandlers[0] == nil {
|
||||
return nil, errors.New("dingtalk stream: card feedback handler is nil")
|
||||
}
|
||||
cred := dtsdk.NewAppCredentialConfig(strings.TrimSpace(config.ClientID), strings.TrimSpace(config.ClientSecret))
|
||||
// Keep the SDK default AutoReconnect=true so runtime disconnects
|
||||
// reconnect automatically; Close flips it off before closing.
|
||||
@@ -258,6 +265,11 @@ func NewStreamClient(config StreamConfig, handler *Handler) (*StreamClient, erro
|
||||
}
|
||||
return nil, nil
|
||||
})
|
||||
if len(feedbackHandlers) == 1 {
|
||||
inner.RegisterCardCallbackRouter(func(ctx context.Context, request *sdkcard.CardRequest) (*sdkcard.CardResponse, error) {
|
||||
return feedbackHandlers[0].HandleSDK(ctx, request)
|
||||
})
|
||||
}
|
||||
return &StreamClient{inner: inner}, nil
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,125 @@
|
||||
package feedback
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// Service handles answer feedback domain logic. It validates incoming
|
||||
// commands, normalizes fields, generates cryptographically random IDs,
|
||||
// and delegates persistence to a Repository.
|
||||
type Service struct {
|
||||
tenantKey string
|
||||
channel string
|
||||
repo Repository
|
||||
gen IDGenerator
|
||||
}
|
||||
|
||||
// NewService creates a feedback Service. It rejects empty tenant, empty
|
||||
// channel, and nil repo. An optional IDGenerator can be provided for
|
||||
// deterministic testing; if nil, a crypto/rand UUIDv4 generator is used.
|
||||
func NewService(tenantKey, channel string, repo Repository, gen ...IDGenerator) (*Service, error) {
|
||||
if strings.TrimSpace(tenantKey) == "" {
|
||||
return nil, errors.New("tenant_key must not be empty")
|
||||
}
|
||||
if strings.TrimSpace(channel) == "" {
|
||||
return nil, errors.New("channel must not be empty")
|
||||
}
|
||||
if repo == nil {
|
||||
return nil, errors.New("repository must not be nil")
|
||||
}
|
||||
if len(gen) > 1 {
|
||||
return nil, errors.New("at most one IDGenerator allowed")
|
||||
}
|
||||
var idGen IDGenerator
|
||||
if len(gen) == 1 {
|
||||
idGen = gen[0]
|
||||
} else {
|
||||
idGen = &defaultIDGenerator{}
|
||||
}
|
||||
if idGen == nil {
|
||||
return nil, errors.New("IDGenerator must not be nil")
|
||||
}
|
||||
return &Service{
|
||||
tenantKey: strings.TrimSpace(tenantKey),
|
||||
channel: strings.TrimSpace(channel),
|
||||
repo: repo,
|
||||
gen: idGen,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// defaultIDGenerator produces UUIDv4-style 128-bit random IDs.
|
||||
type defaultIDGenerator struct{}
|
||||
|
||||
func (g *defaultIDGenerator) Generate() (string, error) {
|
||||
b := make([]byte, 16)
|
||||
if _, err := rand.Read(b); err != nil {
|
||||
return "", err
|
||||
}
|
||||
b[6] = (b[6] & 0x0f) | 0x40 // version 4
|
||||
b[8] = (b[8] & 0x3f) | 0x80 // variant 8-9-a-b
|
||||
return fmt.Sprintf("%08x-%04x-%04x-%04x-%012x",
|
||||
b[0:4], b[4:6], b[6:8], b[8:10], b[10:16]), nil
|
||||
}
|
||||
|
||||
// Apply validates and processes a feedback Command. It trims inputs,
|
||||
// validates allowed values, normalizes reason codes, generates IDs,
|
||||
// and delegates to the Repository.
|
||||
func (s *Service) Apply(ctx context.Context, cmd Command) (Result, error) {
|
||||
cmd.ExternalActorID = strings.TrimSpace(cmd.ExternalActorID)
|
||||
cmd.OutTrackID = strings.TrimSpace(cmd.OutTrackID)
|
||||
cmd.Value = strings.TrimSpace(cmd.Value)
|
||||
cmd.ReasonCode = strings.TrimSpace(cmd.ReasonCode)
|
||||
cmd.Source = strings.TrimSpace(cmd.Source)
|
||||
|
||||
if cmd.ExternalActorID == "" || len(cmd.ExternalActorID) > 256 {
|
||||
return Result{}, fmt.Errorf("%w: external actor ID", ErrInvalidCommand)
|
||||
}
|
||||
if cmd.OutTrackID == "" || len(cmd.OutTrackID) > 256 {
|
||||
return Result{}, fmt.Errorf("%w: outTrackID", ErrInvalidCommand)
|
||||
}
|
||||
if cmd.Value != ValueUp && cmd.Value != ValueDown {
|
||||
return Result{}, fmt.Errorf("%w: value", ErrInvalidCommand)
|
||||
}
|
||||
if cmd.Source != SourceDingTalkCard {
|
||||
return Result{}, fmt.Errorf("%w: source", ErrInvalidCommand)
|
||||
}
|
||||
if cmd.Value == ValueUp {
|
||||
cmd.ReasonCode = ReasonEmpty
|
||||
} else if !validReason(cmd.ReasonCode) {
|
||||
return Result{}, fmt.Errorf("%w: reason code", ErrInvalidCommand)
|
||||
}
|
||||
|
||||
feedbackID, err := s.gen.Generate()
|
||||
if err != nil {
|
||||
return Result{}, fmt.Errorf("generate feedback ID: %w", err)
|
||||
}
|
||||
if strings.TrimSpace(feedbackID) == "" {
|
||||
return Result{}, errors.New("generate feedback ID: empty ID")
|
||||
}
|
||||
actorID, err := s.gen.Generate()
|
||||
if err != nil {
|
||||
return Result{}, fmt.Errorf("generate actor ID: %w", err)
|
||||
}
|
||||
if strings.TrimSpace(actorID) == "" {
|
||||
return Result{}, errors.New("generate actor ID: empty ID")
|
||||
}
|
||||
|
||||
return s.repo.RecordFeedback(ctx, Record{
|
||||
ID: feedbackID, TenantKey: s.tenantKey, Channel: s.channel, ActorID: actorID,
|
||||
ExternalActorID: cmd.ExternalActorID, OutTrackID: cmd.OutTrackID,
|
||||
Value: cmd.Value, ReasonCode: cmd.ReasonCode, Source: cmd.Source,
|
||||
})
|
||||
}
|
||||
|
||||
func validReason(reason string) bool {
|
||||
switch reason {
|
||||
case ReasonEmpty, ReasonIncorrect, ReasonNotResolved, ReasonIrrelevant, ReasonUnclear, ReasonOutdated, ReasonOther:
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,189 @@
|
||||
package feedback_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/feedback"
|
||||
)
|
||||
|
||||
type stubRepo struct {
|
||||
calls int
|
||||
lastRecord feedback.Record
|
||||
result feedback.Result
|
||||
err error
|
||||
}
|
||||
|
||||
func (s *stubRepo) RecordFeedback(_ context.Context, record feedback.Record) (feedback.Result, error) {
|
||||
s.calls++
|
||||
s.lastRecord = record
|
||||
return s.result, s.err
|
||||
}
|
||||
|
||||
type sequenceGenerator struct {
|
||||
ids []string
|
||||
errs map[int]error
|
||||
calls int
|
||||
}
|
||||
|
||||
func (g *sequenceGenerator) Generate() (string, error) {
|
||||
index := g.calls
|
||||
g.calls++
|
||||
if err := g.errs[index]; err != nil {
|
||||
return "", err
|
||||
}
|
||||
if index >= len(g.ids) {
|
||||
return "", errors.New("no id configured")
|
||||
}
|
||||
return g.ids[index], nil
|
||||
}
|
||||
|
||||
func TestNewServiceValidation(t *testing.T) {
|
||||
repo := &stubRepo{}
|
||||
gen := &sequenceGenerator{ids: []string{"a", "b"}, errs: map[int]error{}}
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
tenant string
|
||||
channel string
|
||||
repo feedback.Repository
|
||||
gens []feedback.IDGenerator
|
||||
}{
|
||||
{name: "empty tenant", channel: "dingtalk", repo: repo},
|
||||
{name: "empty channel", tenant: "tenant", repo: repo},
|
||||
{name: "nil repository", tenant: "tenant", channel: "dingtalk"},
|
||||
{name: "nil generator", tenant: "tenant", channel: "dingtalk", repo: repo, gens: []feedback.IDGenerator{nil}},
|
||||
{name: "too many generators", tenant: "tenant", channel: "dingtalk", repo: repo, gens: []feedback.IDGenerator{gen, gen}},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
if _, err := feedback.NewService(tc.tenant, tc.channel, tc.repo, tc.gens...); err == nil {
|
||||
t.Fatal("expected constructor error")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestApplyUpUsesServerScopeTrimsAndClearsReason(t *testing.T) {
|
||||
repo := &stubRepo{result: feedback.Result{Created: true, Value: feedback.ValueUp}}
|
||||
gen := &sequenceGenerator{ids: []string{"feedback-id", "actor-id"}, errs: map[int]error{}}
|
||||
service, err := feedback.NewService(" tenant-a ", " dingtalk ", repo, gen)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
result, err := service.Apply(context.Background(), feedback.Command{
|
||||
ExternalActorID: " user-a ", OutTrackID: " track-a ", Value: feedback.ValueUp,
|
||||
ReasonCode: feedback.ReasonIncorrect, Source: feedback.SourceDingTalkCard,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !result.Created || repo.calls != 1 || gen.calls != 2 {
|
||||
t.Fatalf("result=%+v repo_calls=%d gen_calls=%d", result, repo.calls, gen.calls)
|
||||
}
|
||||
got := repo.lastRecord
|
||||
if got.ID != "feedback-id" || got.ActorID != "actor-id" || got.TenantKey != "tenant-a" || got.Channel != "dingtalk" || got.ExternalActorID != "user-a" || got.OutTrackID != "track-a" || got.Value != feedback.ValueUp || got.ReasonCode != "" || got.Source != feedback.SourceDingTalkCard {
|
||||
t.Fatalf("record=%+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestApplyDownAcceptsEveryReasonAndEmpty(t *testing.T) {
|
||||
reasons := []string{
|
||||
feedback.ReasonEmpty, feedback.ReasonIncorrect, feedback.ReasonNotResolved,
|
||||
feedback.ReasonIrrelevant, feedback.ReasonUnclear, feedback.ReasonOutdated, feedback.ReasonOther,
|
||||
}
|
||||
for _, reason := range reasons {
|
||||
t.Run("reason_"+reason, func(t *testing.T) {
|
||||
repo := &stubRepo{result: feedback.Result{Changed: true, Value: feedback.ValueDown, ReasonCode: reason}}
|
||||
gen := &sequenceGenerator{ids: []string{"feedback-id", "actor-id"}, errs: map[int]error{}}
|
||||
service, _ := feedback.NewService("tenant", "dingtalk", repo, gen)
|
||||
result, err := service.Apply(context.Background(), feedback.Command{
|
||||
ExternalActorID: "user", OutTrackID: "track", Value: feedback.ValueDown,
|
||||
ReasonCode: " " + reason + " ", Source: feedback.SourceDingTalkCard,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if repo.lastRecord.ReasonCode != reason || result.ReasonCode != reason {
|
||||
t.Fatalf("record=%+v result=%+v", repo.lastRecord, result)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestApplyRejectsInvalidCommandsBeforeGeneratingIDs(t *testing.T) {
|
||||
tooLong := strings.Repeat("x", 257)
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
cmd feedback.Command
|
||||
}{
|
||||
{name: "empty actor", cmd: feedback.Command{OutTrackID: "track", Value: feedback.ValueUp, Source: feedback.SourceDingTalkCard}},
|
||||
{name: "empty track", cmd: feedback.Command{ExternalActorID: "user", Value: feedback.ValueUp, Source: feedback.SourceDingTalkCard}},
|
||||
{name: "long actor", cmd: feedback.Command{ExternalActorID: tooLong, OutTrackID: "track", Value: feedback.ValueUp, Source: feedback.SourceDingTalkCard}},
|
||||
{name: "long track", cmd: feedback.Command{ExternalActorID: "user", OutTrackID: tooLong, Value: feedback.ValueUp, Source: feedback.SourceDingTalkCard}},
|
||||
{name: "invalid value", cmd: feedback.Command{ExternalActorID: "user", OutTrackID: "track", Value: "maybe", Source: feedback.SourceDingTalkCard}},
|
||||
{name: "invalid source", cmd: feedback.Command{ExternalActorID: "user", OutTrackID: "track", Value: feedback.ValueUp, Source: "web"}},
|
||||
{name: "invalid down reason", cmd: feedback.Command{ExternalActorID: "user", OutTrackID: "track", Value: feedback.ValueDown, ReasonCode: "secret", Source: feedback.SourceDingTalkCard}},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
repo := &stubRepo{}
|
||||
gen := &sequenceGenerator{ids: []string{"feedback-id", "actor-id"}, errs: map[int]error{}}
|
||||
service, _ := feedback.NewService("tenant", "dingtalk", repo, gen)
|
||||
if _, err := service.Apply(context.Background(), tc.cmd); !errors.Is(err, feedback.ErrInvalidCommand) {
|
||||
t.Fatalf("error=%v", err)
|
||||
}
|
||||
if repo.calls != 0 || gen.calls != 0 {
|
||||
t.Fatalf("repo_calls=%d gen_calls=%d", repo.calls, gen.calls)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestApplyIDFailureNeverCallsRepository(t *testing.T) {
|
||||
idErr := errors.New("entropy unavailable")
|
||||
for _, failureIndex := range []int{0, 1} {
|
||||
t.Run(string(rune('0'+failureIndex)), func(t *testing.T) {
|
||||
repo := &stubRepo{}
|
||||
gen := &sequenceGenerator{ids: []string{"feedback-id", "actor-id"}, errs: map[int]error{failureIndex: idErr}}
|
||||
service, _ := feedback.NewService("tenant", "dingtalk", repo, gen)
|
||||
_, err := service.Apply(context.Background(), feedback.Command{ExternalActorID: "user", OutTrackID: "track", Value: feedback.ValueUp, Source: feedback.SourceDingTalkCard})
|
||||
if !errors.Is(err, idErr) || repo.calls != 0 {
|
||||
t.Fatalf("error=%v repo_calls=%d", err, repo.calls)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestApplyPassesThroughRepositoryResultAndError(t *testing.T) {
|
||||
repoErr := errors.New("repository failed")
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
result feedback.Result
|
||||
err error
|
||||
}{
|
||||
{name: "replayed", result: feedback.Result{Replayed: true, Value: feedback.ValueDown, ReasonCode: feedback.ReasonUnclear}},
|
||||
{name: "error", err: repoErr},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
repo := &stubRepo{result: tc.result, err: tc.err}
|
||||
gen := &sequenceGenerator{ids: []string{"feedback-id", "actor-id"}, errs: map[int]error{}}
|
||||
service, _ := feedback.NewService("tenant", "dingtalk", repo, gen)
|
||||
result, err := service.Apply(context.Background(), feedback.Command{ExternalActorID: "user", OutTrackID: "track", Value: feedback.ValueDown, ReasonCode: feedback.ReasonUnclear, Source: feedback.SourceDingTalkCard})
|
||||
if !errors.Is(err, tc.err) || result != tc.result {
|
||||
t.Fatalf("result=%+v error=%v", result, err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestRecordCannotBeSerializedIntoExternalJSON(t *testing.T) {
|
||||
record := feedback.Record{ID: "feedback-id", TenantKey: "tenant", Channel: "dingtalk", ActorID: "actor-id", ExternalActorID: "user", OutTrackID: "provider-track", Value: feedback.ValueDown, ReasonCode: feedback.ReasonOther, Source: feedback.SourceDingTalkCard}
|
||||
encoded, err := json.Marshal(record)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if string(encoded) != "{}" {
|
||||
t.Fatalf("internal record must not be JSON-exposable: %s", encoded)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,87 @@
|
||||
// Package feedback implements the domain logic for answer feedback
|
||||
// (B2 DingTalk card callback). It validates and normalizes incoming
|
||||
// feedback commands, generates cryptographically random IDs, and
|
||||
// delegates persistence to a Repository.
|
||||
//
|
||||
// The package is intentionally free of any webhook, token, raw-callback,
|
||||
// content, or extension columns — those belong to the channel layer.
|
||||
package feedback
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
)
|
||||
|
||||
var (
|
||||
ErrInvalidCommand = errors.New("invalid feedback command")
|
||||
ErrTargetNotFound = errors.New("feedback target not found")
|
||||
)
|
||||
|
||||
// Value constants for feedback.
|
||||
const (
|
||||
ValueUp = "up"
|
||||
ValueDown = "down"
|
||||
)
|
||||
|
||||
// ReasonCode constants for down feedback.
|
||||
const (
|
||||
ReasonEmpty = ""
|
||||
ReasonIncorrect = "incorrect"
|
||||
ReasonNotResolved = "not_resolved"
|
||||
ReasonIrrelevant = "irrelevant"
|
||||
ReasonUnclear = "unclear"
|
||||
ReasonOutdated = "outdated"
|
||||
ReasonOther = "other"
|
||||
)
|
||||
|
||||
// Source constants.
|
||||
const (
|
||||
SourceDingTalkCard = "dingtalk_card"
|
||||
)
|
||||
|
||||
// Command represents a feedback action from an external channel.
|
||||
// Only the caller-provided fields are exposed; tenant, channel, turn,
|
||||
// and message are fixed by the Service and never accepted from callers.
|
||||
type Command struct {
|
||||
ExternalActorID string // external actor identifier (required)
|
||||
OutTrackID string // outbound tracking/answer identifier (required)
|
||||
Value string // "up" or "down"
|
||||
ReasonCode string // optional reason for "down"
|
||||
Source string // must be "dingtalk_card"
|
||||
}
|
||||
|
||||
// Record is the full domain object passed to the Repository for persistence.
|
||||
// It contains the fixed tenant/channel, cryptographically random IDs, and
|
||||
// the normalized command fields.
|
||||
type Record struct {
|
||||
ID string `json:"-"`
|
||||
TenantKey string `json:"-"`
|
||||
Channel string `json:"-"`
|
||||
ActorID string `json:"-"`
|
||||
Value string `json:"-"`
|
||||
ReasonCode string `json:"-"`
|
||||
Source string `json:"-"`
|
||||
ExternalActorID string `json:"-"`
|
||||
OutTrackID string `json:"-"`
|
||||
}
|
||||
|
||||
// Result reports the outcome of a RecordFeedback call.
|
||||
// Exactly one of Created, Changed, or Replayed is true.
|
||||
type Result struct {
|
||||
Created bool
|
||||
Changed bool
|
||||
Replayed bool
|
||||
Value string
|
||||
ReasonCode string
|
||||
}
|
||||
|
||||
// Repository persists feedback records.
|
||||
type Repository interface {
|
||||
RecordFeedback(ctx context.Context, r Record) (Result, error)
|
||||
}
|
||||
|
||||
// IDGenerator produces unique IDs. The default implementation uses
|
||||
// crypto/rand to produce UUIDv4-style strings.
|
||||
type IDGenerator interface {
|
||||
Generate() (string, error)
|
||||
}
|
||||
@@ -0,0 +1,189 @@
|
||||
package httpapi
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/audit"
|
||||
)
|
||||
|
||||
type fakeAuditService struct {
|
||||
listRequest audit.ListTurnsRequest
|
||||
listResult audit.ListTurnsResult
|
||||
listErr error
|
||||
getID string
|
||||
getResult audit.TurnDetail
|
||||
getErr error
|
||||
}
|
||||
|
||||
func (f *fakeAuditService) ListTurns(_ context.Context, request audit.ListTurnsRequest) (audit.ListTurnsResult, error) {
|
||||
f.listRequest = request
|
||||
return f.listResult, f.listErr
|
||||
}
|
||||
|
||||
func (f *fakeAuditService) GetTurn(_ context.Context, id string) (audit.TurnDetail, error) {
|
||||
f.getID = id
|
||||
return f.getResult, f.getErr
|
||||
}
|
||||
|
||||
func TestAuditOptionsRequireCompleteDistinctConfiguration(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
opts Options
|
||||
}{
|
||||
{name: "service only", opts: Options{AuditService: &fakeAuditService{}}},
|
||||
{name: "token only", opts: Options{AdminToken: "admin-secret"}},
|
||||
{name: "same ingress and admin token", opts: Options{AuditService: &fakeAuditService{}, AdminToken: "ingress-secret"}},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
if _, err := NewWithOptions(fakeQA{}, "ingress-secret", tc.opts); err == nil {
|
||||
t.Fatal("expected configuration error")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestAuditRoutesAreAbsentWhenDisabled(t *testing.T) {
|
||||
server, err := New(fakeQA{}, "ingress-secret")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
result := httptest.NewRecorder()
|
||||
server.Handler().ServeHTTP(result, httptest.NewRequest(http.MethodGet, "/v1/admin/audit/turns", nil))
|
||||
if result.Code != http.StatusNotFound {
|
||||
t.Fatalf("status=%d body=%s", result.Code, result.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
func TestAuditListRequiresIndependentAdminBearer(t *testing.T) {
|
||||
service := &fakeAuditService{listResult: audit.ListTurnsResult{Items: []audit.TurnSummary{}}}
|
||||
server, err := NewWithOptions(fakeQA{}, "ingress-secret", Options{AuditService: service, AdminToken: "admin-secret"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, authorization := range []string{"", "Bearer wrong", "Bearer ingress-secret", "bearer admin-secret", "Bearer admin-secret"} {
|
||||
request := httptest.NewRequest(http.MethodGet, "/v1/admin/audit/turns", nil)
|
||||
request.Header.Set("Authorization", authorization)
|
||||
result := httptest.NewRecorder()
|
||||
server.Handler().ServeHTTP(result, request)
|
||||
if result.Code != http.StatusUnauthorized || !strings.Contains(result.Body.String(), "unauthorized") {
|
||||
t.Fatalf("auth=%q status=%d body=%s", authorization, result.Code, result.Body.String())
|
||||
}
|
||||
}
|
||||
request := httptest.NewRequest(http.MethodGet, "/v1/admin/audit/turns", nil)
|
||||
request.Header.Set("Authorization", "Bearer admin-secret")
|
||||
result := httptest.NewRecorder()
|
||||
server.Handler().ServeHTTP(result, request)
|
||||
if result.Code != http.StatusOK || result.Header().Get("Cache-Control") != "no-store" {
|
||||
t.Fatalf("status=%d cache=%q body=%s", result.Code, result.Header().Get("Cache-Control"), result.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
func TestAuditListParsesAllowedQueryOnly(t *testing.T) {
|
||||
service := &fakeAuditService{listResult: audit.ListTurnsResult{Items: []audit.TurnSummary{}}}
|
||||
server, _ := NewWithOptions(fakeQA{}, "ingress", Options{AuditService: service, AdminToken: "admin"})
|
||||
request := httptest.NewRequest(http.MethodGet, "/v1/admin/audit/turns?limit=100&cursor=opaque&channel=dingtalk&status=succeeded", nil)
|
||||
request.Header.Set("Authorization", "Bearer admin")
|
||||
result := httptest.NewRecorder()
|
||||
server.Handler().ServeHTTP(result, request)
|
||||
if result.Code != http.StatusOK {
|
||||
t.Fatalf("status=%d body=%s", result.Code, result.Body.String())
|
||||
}
|
||||
if service.listRequest.Limit != 100 || service.listRequest.Cursor != "opaque" || service.listRequest.Channel != "dingtalk" || service.listRequest.Status != audit.StatusSucceeded {
|
||||
t.Fatalf("request=%+v", service.listRequest)
|
||||
}
|
||||
|
||||
for _, query := range []string{"tenant_key=other", "unknown=x", "limit=0", "limit=101", "limit=x", "limit=1&limit=2"} {
|
||||
request := httptest.NewRequest(http.MethodGet, "/v1/admin/audit/turns?"+query, nil)
|
||||
request.Header.Set("Authorization", "Bearer admin")
|
||||
result := httptest.NewRecorder()
|
||||
server.Handler().ServeHTTP(result, request)
|
||||
if result.Code != http.StatusBadRequest {
|
||||
t.Fatalf("query=%q status=%d body=%s", query, result.Code, result.Body.String())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestAuditListMapsServiceErrorsWithoutLeaking(t *testing.T) {
|
||||
service := &fakeAuditService{listErr: errors.New("postgres password=private unavailable")}
|
||||
server, _ := NewWithOptions(fakeQA{}, "ingress", Options{AuditService: service, AdminToken: "admin"})
|
||||
request := httptest.NewRequest(http.MethodGet, "/v1/admin/audit/turns", nil)
|
||||
request.Header.Set("Authorization", "Bearer admin")
|
||||
result := httptest.NewRecorder()
|
||||
server.Handler().ServeHTTP(result, request)
|
||||
if result.Code != http.StatusInternalServerError || strings.Contains(result.Body.String(), "password") || strings.Contains(result.Body.String(), "private") {
|
||||
t.Fatalf("status=%d body=%s", result.Code, result.Body.String())
|
||||
}
|
||||
|
||||
service.listErr = audit.ErrInvalidArgument
|
||||
result = httptest.NewRecorder()
|
||||
server.Handler().ServeHTTP(result, request)
|
||||
if result.Code != http.StatusBadRequest {
|
||||
t.Fatalf("invalid query status=%d", result.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAuditDetailReturnsSafeFullQuestionAndAnswer(t *testing.T) {
|
||||
now := time.Date(2026, 7, 21, 0, 0, 0, 0, time.UTC)
|
||||
service := &fakeAuditService{getResult: audit.TurnDetail{
|
||||
ID: "turn-1", CreatedAt: now, Channel: "dingtalk", Status: audit.StatusSucceeded,
|
||||
Question: "full question", Answer: "full answer",
|
||||
References: []audit.SafeReference{{ID: "ref-1", KnowledgeTitle: "Manual", Score: 0.8}},
|
||||
AnswerMessage: &audit.MessageMeta{ID: "answer-1", Direction: "outbound", Role: "assistant", MessageType: "text", Status: "sent", CreatedAt: now, DeliveryMethod: "card"},
|
||||
}}
|
||||
server, _ := NewWithOptions(fakeQA{}, "ingress", Options{AuditService: service, AdminToken: "admin"})
|
||||
request := httptest.NewRequest(http.MethodGet, "/v1/admin/audit/turns/turn-1", nil)
|
||||
request.Header.Set("Authorization", "Bearer admin")
|
||||
result := httptest.NewRecorder()
|
||||
server.Handler().ServeHTTP(result, request)
|
||||
body := result.Body.String()
|
||||
if result.Code != http.StatusOK || service.getID != "turn-1" || !strings.Contains(body, "full question") || !strings.Contains(body, "full answer") {
|
||||
t.Fatalf("status=%d id=%q body=%s", result.Code, service.getID, body)
|
||||
}
|
||||
for _, forbidden := range []string{"weknora_session", "weknora_request", "provider_message", "sessionWebhook", "token", "raw_error", "error_message", "content"} {
|
||||
if strings.Contains(strings.ToLower(body), strings.ToLower(forbidden)) {
|
||||
t.Fatalf("detail leaked forbidden %q: %s", forbidden, body)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestAuditDetailMapsNotFoundInvalidAndInternal(t *testing.T) {
|
||||
service := &fakeAuditService{getErr: audit.ErrNotFound}
|
||||
server, _ := NewWithOptions(fakeQA{}, "ingress", Options{AuditService: service, AdminToken: "admin"})
|
||||
call := func(path string) *httptest.ResponseRecorder {
|
||||
request := httptest.NewRequest(http.MethodGet, path, nil)
|
||||
request.Header.Set("Authorization", "Bearer admin")
|
||||
result := httptest.NewRecorder()
|
||||
server.Handler().ServeHTTP(result, request)
|
||||
return result
|
||||
}
|
||||
if got := call("/v1/admin/audit/turns/missing"); got.Code != http.StatusNotFound {
|
||||
t.Fatalf("not found status=%d", got.Code)
|
||||
}
|
||||
service.getErr = audit.ErrInvalidArgument
|
||||
if got := call("/v1/admin/audit/turns/bad"); got.Code != http.StatusBadRequest {
|
||||
t.Fatalf("invalid status=%d", got.Code)
|
||||
}
|
||||
service.getErr = errors.New("raw database secret")
|
||||
if got := call("/v1/admin/audit/turns/turn"); got.Code != http.StatusInternalServerError || strings.Contains(got.Body.String(), "secret") {
|
||||
t.Fatalf("internal status=%d body=%s", got.Code, got.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
func TestValidateLoopbackListenAddress(t *testing.T) {
|
||||
for _, address := range []string{"127.0.0.1:8090", "localhost:8090", "[::1]:8090"} {
|
||||
if err := ValidateLoopbackListenAddress(address); err != nil {
|
||||
t.Errorf("address=%q error=%v", address, err)
|
||||
}
|
||||
}
|
||||
for _, address := range []string{"0.0.0.0:8090", "192.168.1.5:8090", ":8090", "127.0.0.1", "example.com:8090"} {
|
||||
if err := ValidateLoopbackListenAddress(address); err == nil {
|
||||
t.Errorf("address=%q should fail", address)
|
||||
}
|
||||
}
|
||||
}
|
||||
+155
-8
@@ -5,10 +5,14 @@ import (
|
||||
"crypto/subtle"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/audit"
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/core"
|
||||
)
|
||||
|
||||
@@ -16,39 +20,88 @@ type QAService interface {
|
||||
Ask(context.Context, core.QARequest) (core.QAResult, error)
|
||||
}
|
||||
|
||||
type AuditService interface {
|
||||
ListTurns(context.Context, audit.ListTurnsRequest) (audit.ListTurnsResult, error)
|
||||
GetTurn(context.Context, string) (audit.TurnDetail, error)
|
||||
}
|
||||
|
||||
type Options struct {
|
||||
AuditService AuditService
|
||||
AdminToken string
|
||||
}
|
||||
|
||||
type Server struct {
|
||||
service QAService
|
||||
token string
|
||||
mux *http.ServeMux
|
||||
service QAService
|
||||
ingressToken string
|
||||
auditService AuditService
|
||||
adminToken string
|
||||
mux *http.ServeMux
|
||||
}
|
||||
|
||||
func New(service QAService, ingressToken string) (*Server, error) {
|
||||
return NewWithOptions(service, ingressToken, Options{})
|
||||
}
|
||||
|
||||
func NewWithOptions(service QAService, ingressToken string, opts Options) (*Server, error) {
|
||||
if service == nil {
|
||||
return nil, errors.New("QA service is required")
|
||||
}
|
||||
if strings.TrimSpace(ingressToken) == "" {
|
||||
return nil, errors.New("DOBOTAI_INGRESS_TOKEN is required")
|
||||
}
|
||||
s := &Server{service: service, token: ingressToken, mux: http.NewServeMux()}
|
||||
adminToken := strings.TrimSpace(opts.AdminToken)
|
||||
hasAuditService := opts.AuditService != nil
|
||||
hasAdminToken := adminToken != ""
|
||||
if hasAuditService != hasAdminToken {
|
||||
return nil, errors.New("audit service and DOBOTAI_ADMIN_TOKEN must be configured together")
|
||||
}
|
||||
if hasAdminToken && constantTimeTokenEqual(adminToken, ingressToken) {
|
||||
return nil, errors.New("DOBOTAI_ADMIN_TOKEN must differ from DOBOTAI_INGRESS_TOKEN")
|
||||
}
|
||||
|
||||
s := &Server{
|
||||
service: service, ingressToken: ingressToken,
|
||||
auditService: opts.AuditService, adminToken: adminToken,
|
||||
mux: http.NewServeMux(),
|
||||
}
|
||||
s.mux.HandleFunc("GET /healthz", s.health)
|
||||
s.mux.HandleFunc("POST /v1/qa", s.authorize(s.ask))
|
||||
s.mux.HandleFunc("POST /v1/qa", s.authorizeToken(s.ingressToken, s.ask))
|
||||
if hasAuditService {
|
||||
s.mux.HandleFunc("GET /v1/admin/audit/turns", s.authorizeToken(s.adminToken, s.listAuditTurns))
|
||||
s.mux.HandleFunc("GET /v1/admin/audit/turns/{id}", s.authorizeToken(s.adminToken, s.getAuditTurn))
|
||||
}
|
||||
return s, nil
|
||||
}
|
||||
|
||||
func (s *Server) Handler() http.Handler { return s.mux }
|
||||
|
||||
func (s *Server) authorize(next http.HandlerFunc) http.HandlerFunc {
|
||||
func (s *Server) authorizeToken(expected string, next http.HandlerFunc) http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
provided := strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer ")
|
||||
if len(provided) != len(s.token) || subtle.ConstantTimeCompare([]byte(provided), []byte(s.token)) != 1 {
|
||||
if strings.HasPrefix(r.URL.Path, "/v1/admin/audit/") {
|
||||
w.Header().Set("Cache-Control", "no-store")
|
||||
}
|
||||
header := r.Header.Get("Authorization")
|
||||
if !strings.HasPrefix(header, "Bearer ") {
|
||||
writeError(w, http.StatusUnauthorized, "unauthorized")
|
||||
return
|
||||
}
|
||||
provided := strings.TrimPrefix(header, "Bearer ")
|
||||
if provided == "" || strings.ContainsAny(provided, " \t\r\n") || !constantTimeTokenEqual(provided, expected) {
|
||||
writeError(w, http.StatusUnauthorized, "unauthorized")
|
||||
return
|
||||
}
|
||||
next(w, r)
|
||||
}
|
||||
}
|
||||
|
||||
func constantTimeTokenEqual(provided, expected string) bool {
|
||||
return len(provided) == len(expected) && subtle.ConstantTimeCompare([]byte(provided), []byte(expected)) == 1
|
||||
}
|
||||
|
||||
func (s *Server) health(w http.ResponseWriter, _ *http.Request) {
|
||||
writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
|
||||
}
|
||||
|
||||
func (s *Server) ask(w http.ResponseWriter, r *http.Request) {
|
||||
r.Body = http.MaxBytesReader(w, r.Body, 1<<20)
|
||||
defer r.Body.Close()
|
||||
@@ -82,9 +135,103 @@ func (s *Server) ask(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
writeJSON(w, http.StatusOK, result)
|
||||
}
|
||||
|
||||
func (s *Server) listAuditTurns(w http.ResponseWriter, r *http.Request) {
|
||||
allowed := map[string]bool{"limit": true, "cursor": true, "channel": true, "status": true}
|
||||
for key, values := range r.URL.Query() {
|
||||
if !allowed[key] || len(values) != 1 {
|
||||
writeError(w, http.StatusBadRequest, "invalid audit query")
|
||||
return
|
||||
}
|
||||
}
|
||||
var request audit.ListTurnsRequest
|
||||
if raw, ok := singleQueryValue(r, "limit"); ok {
|
||||
limit, err := strconv.Atoi(raw)
|
||||
if err != nil || limit < 1 || limit > 100 {
|
||||
writeError(w, http.StatusBadRequest, "invalid audit query")
|
||||
return
|
||||
}
|
||||
request.Limit = limit
|
||||
}
|
||||
request.Cursor = r.URL.Query().Get("cursor")
|
||||
request.Channel = r.URL.Query().Get("channel")
|
||||
request.Status = audit.TurnStatus(r.URL.Query().Get("status"))
|
||||
if len(request.Cursor) > 512 || len(request.Channel) > 128 || !validAuditStatus(request.Status) {
|
||||
writeError(w, http.StatusBadRequest, "invalid audit query")
|
||||
return
|
||||
}
|
||||
result, err := s.auditService.ListTurns(r.Context(), request)
|
||||
if err != nil {
|
||||
writeAuditError(w, err)
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, result)
|
||||
}
|
||||
|
||||
func (s *Server) getAuditTurn(w http.ResponseWriter, r *http.Request) {
|
||||
if len(r.URL.Query()) != 0 {
|
||||
writeError(w, http.StatusBadRequest, "invalid audit query")
|
||||
return
|
||||
}
|
||||
result, err := s.auditService.GetTurn(r.Context(), r.PathValue("id"))
|
||||
if err != nil {
|
||||
writeAuditError(w, err)
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, result)
|
||||
}
|
||||
|
||||
func singleQueryValue(r *http.Request, key string) (string, bool) {
|
||||
values, ok := r.URL.Query()[key]
|
||||
if !ok || len(values) != 1 {
|
||||
return "", false
|
||||
}
|
||||
return values[0], true
|
||||
}
|
||||
|
||||
func validAuditStatus(status audit.TurnStatus) bool {
|
||||
switch status {
|
||||
case "", audit.StatusProcessing, audit.StatusSucceeded, audit.StatusFailed, audit.StatusCancelled:
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
func writeAuditError(w http.ResponseWriter, err error) {
|
||||
switch {
|
||||
case errors.Is(err, audit.ErrInvalidArgument):
|
||||
writeError(w, http.StatusBadRequest, "invalid audit query")
|
||||
case errors.Is(err, audit.ErrNotFound):
|
||||
writeError(w, http.StatusNotFound, "audit turn not found")
|
||||
default:
|
||||
writeError(w, http.StatusInternalServerError, "audit query failed")
|
||||
}
|
||||
}
|
||||
|
||||
func ValidateLoopbackListenAddress(address string) error {
|
||||
host, port, err := net.SplitHostPort(strings.TrimSpace(address))
|
||||
if err != nil || host == "" || port == "" {
|
||||
return errors.New("audit API requires a loopback host and explicit port")
|
||||
}
|
||||
portNumber, err := strconv.Atoi(port)
|
||||
if err != nil || portNumber < 1 || portNumber > 65535 {
|
||||
return errors.New("audit API requires a valid port")
|
||||
}
|
||||
if strings.EqualFold(host, "localhost") {
|
||||
return nil
|
||||
}
|
||||
ip := net.ParseIP(host)
|
||||
if ip == nil || !ip.IsLoopback() {
|
||||
return fmt.Errorf("audit API listen host %q is not loopback", host)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func writeError(w http.ResponseWriter, status int, message string) {
|
||||
writeJSON(w, status, map[string]string{"error": message})
|
||||
}
|
||||
|
||||
func writeJSON(w http.ResponseWriter, status int, value any) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(status)
|
||||
|
||||
@@ -0,0 +1,194 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/audit"
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/core"
|
||||
)
|
||||
|
||||
const listAuditTurnsSQL = `SELECT
|
||||
t.id,t.created_at,t.channel,t.status,
|
||||
COALESCE(c.conversation_type,''),COALESCE(c.title,''),
|
||||
COALESCE(a.external_actor_id,''),COALESCE(rm.actor_name_snapshot,''),
|
||||
LEFT(t.question,200),LEFT(t.answer,200),
|
||||
CASE WHEN jsonb_typeof(t.references_json)='array' THEN jsonb_array_length(t.references_json) ELSE 0 END,
|
||||
COALESCE(am.status,''),COALESCE(am.delivery_method,''),
|
||||
COALESCE(fc.up_count,0),COALESCE(fc.down_count,0),COALESCE(fc.unresolved_count,0)
|
||||
FROM turns t
|
||||
LEFT JOIN messages rm ON rm.id=t.request_message_id AND rm.tenant_key=t.tenant_key AND rm.channel=t.channel
|
||||
LEFT JOIN messages am ON am.id=t.answer_message_id AND am.tenant_key=t.tenant_key AND am.channel=t.channel
|
||||
LEFT JOIN channel_conversations c ON c.id=rm.conversation_id AND c.tenant_key=t.tenant_key AND c.channel=t.channel
|
||||
LEFT JOIN channel_actors a ON a.id=rm.actor_id AND a.tenant_key=t.tenant_key AND a.channel=t.channel
|
||||
LEFT JOIN LATERAL (
|
||||
SELECT
|
||||
(COUNT(*) FILTER (WHERE af.value='up'))::int AS up_count,
|
||||
(COUNT(*) FILTER (WHERE af.value='down'))::int AS down_count,
|
||||
(COUNT(*) FILTER (WHERE af.value='down' AND af.reason_code=''))::int AS unresolved_count
|
||||
FROM answer_feedback af
|
||||
WHERE af.tenant_key=t.tenant_key AND af.channel=t.channel AND af.turn_id=t.id
|
||||
) fc ON TRUE
|
||||
WHERE t.tenant_key=$1
|
||||
AND ($2='' OR t.channel=$2)
|
||||
AND ($3='' OR t.status=$3)
|
||||
AND ($4::timestamptz IS NULL OR (t.created_at,t.id)<($4,$5))
|
||||
ORDER BY t.created_at DESC,t.id DESC
|
||||
LIMIT $6`
|
||||
|
||||
const getAuditTurnSQL = `SELECT
|
||||
t.id,t.created_at,t.channel,t.status,
|
||||
COALESCE(c.conversation_type,''),COALESCE(c.title,''),
|
||||
COALESCE(a.external_actor_id,''),COALESCE(rm.actor_name_snapshot,''),
|
||||
t.question,t.answer,t.references_json,t.media_json,
|
||||
COALESCE(rm.id,''),COALESCE(rm.direction,''),COALESCE(rm.role,''),COALESCE(rm.message_type,''),COALESCE(rm.status,''),COALESCE(rm.actor_name_snapshot,''),rm.created_at,
|
||||
COALESCE(am.id,''),COALESCE(am.direction,''),COALESCE(am.role,''),COALESCE(am.message_type,''),COALESCE(am.status,''),am.created_at,am.delivered_at,COALESCE(am.delivery_method,''),COALESCE(am.failure_code,''),
|
||||
COALESCE(ff.feedback_json,'[]'::jsonb)
|
||||
FROM turns t
|
||||
LEFT JOIN messages rm ON rm.id=t.request_message_id AND rm.tenant_key=t.tenant_key AND rm.channel=t.channel
|
||||
LEFT JOIN messages am ON am.id=t.answer_message_id AND am.tenant_key=t.tenant_key AND am.channel=t.channel
|
||||
LEFT JOIN channel_conversations c ON c.id=rm.conversation_id AND c.tenant_key=t.tenant_key AND c.channel=t.channel
|
||||
LEFT JOIN channel_actors a ON a.id=rm.actor_id AND a.tenant_key=t.tenant_key AND a.channel=t.channel
|
||||
LEFT JOIN LATERAL (
|
||||
SELECT jsonb_agg(jsonb_build_object(
|
||||
'actor_external_id',fa.external_actor_id,
|
||||
'actor_display_name',fa.display_name,
|
||||
'value',af.value,
|
||||
'reason_code',af.reason_code,
|
||||
'updated_at',af.updated_at
|
||||
) ORDER BY af.updated_at DESC,af.id DESC) AS feedback_json
|
||||
FROM answer_feedback af
|
||||
JOIN channel_actors fa ON fa.id=af.actor_id AND fa.tenant_key=af.tenant_key AND fa.channel=af.channel
|
||||
WHERE af.tenant_key=t.tenant_key AND af.channel=t.channel AND af.turn_id=t.id
|
||||
) ff ON TRUE
|
||||
WHERE t.tenant_key=$1 AND t.id=$2`
|
||||
|
||||
func (s *Store) ListTurns(ctx context.Context, tenantKey string, query audit.ListQuery) ([]audit.TurnSummary, error) {
|
||||
var before any
|
||||
if query.BeforeCreatedAt != nil {
|
||||
before = *query.BeforeCreatedAt
|
||||
}
|
||||
rows, err := s.pool.Query(ctx, listAuditTurnsSQL, tenantKey, query.Channel, string(query.Status), before, query.BeforeID, query.Limit)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("list audit turns: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
items := make([]audit.TurnSummary, 0)
|
||||
for rows.Next() {
|
||||
var item audit.TurnSummary
|
||||
var status string
|
||||
if err := rows.Scan(
|
||||
&item.ID, &item.CreatedAt, &item.Channel, &status,
|
||||
&item.ConversationType, &item.ConversationTitle,
|
||||
&item.ActorExternalID, &item.ActorDisplayName,
|
||||
&item.QuestionSummary, &item.AnswerSummary, &item.ReferenceCount,
|
||||
&item.AnswerDeliveryStatus, &item.AnswerDeliveryMethod,
|
||||
&item.FeedbackUpCount, &item.FeedbackDownCount, &item.FeedbackUnresolvedCount,
|
||||
); err != nil {
|
||||
return nil, fmt.Errorf("scan audit turn summary: %w", err)
|
||||
}
|
||||
item.Status = audit.TurnStatus(status)
|
||||
items = append(items, item)
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, fmt.Errorf("iterate audit turns: %w", err)
|
||||
}
|
||||
return items, nil
|
||||
}
|
||||
|
||||
func (s *Store) GetTurn(ctx context.Context, tenantKey, turnID string) (audit.TurnDetail, error) {
|
||||
var detail audit.TurnDetail
|
||||
var status string
|
||||
var referencesJSON, mediaJSON, feedbackJSON []byte
|
||||
var request audit.MessageMeta
|
||||
var answer audit.MessageMeta
|
||||
var requestCreated, answerCreated, deliveredAt *time.Time
|
||||
err := s.pool.QueryRow(ctx, getAuditTurnSQL, tenantKey, turnID).Scan(
|
||||
&detail.ID, &detail.CreatedAt, &detail.Channel, &status,
|
||||
&detail.ConversationType, &detail.ConversationTitle,
|
||||
&detail.ActorExternalID, &detail.ActorDisplayName,
|
||||
&detail.Question, &detail.Answer, &referencesJSON, &mediaJSON,
|
||||
&request.ID, &request.Direction, &request.Role, &request.MessageType, &request.Status, &request.ActorName, &requestCreated,
|
||||
&answer.ID, &answer.Direction, &answer.Role, &answer.MessageType, &answer.Status, &answerCreated, &deliveredAt, &answer.DeliveryMethod, &answer.FailureCode,
|
||||
&feedbackJSON,
|
||||
)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return audit.TurnDetail{}, audit.ErrNotFound
|
||||
}
|
||||
if err != nil {
|
||||
return audit.TurnDetail{}, fmt.Errorf("get audit turn: %w", err)
|
||||
}
|
||||
detail.Status = audit.TurnStatus(status)
|
||||
|
||||
var references []core.Reference
|
||||
if len(referencesJSON) > 0 {
|
||||
if err := json.Unmarshal(referencesJSON, &references); err != nil {
|
||||
return audit.TurnDetail{}, fmt.Errorf("decode audit references: %w", err)
|
||||
}
|
||||
}
|
||||
detail.References = safeAuditReferences(references)
|
||||
detail.ReferenceCount = len(detail.References)
|
||||
|
||||
var media []core.MediaAudit
|
||||
if len(mediaJSON) > 0 {
|
||||
if err := json.Unmarshal(mediaJSON, &media); err != nil {
|
||||
return audit.TurnDetail{}, fmt.Errorf("decode audit media: %w", err)
|
||||
}
|
||||
}
|
||||
detail.Media = safeAuditMedia(media)
|
||||
if err := json.Unmarshal(feedbackJSON, &detail.Feedback); err != nil {
|
||||
return audit.TurnDetail{}, fmt.Errorf("decode audit feedback: %w", err)
|
||||
}
|
||||
|
||||
if request.ID != "" {
|
||||
if requestCreated != nil {
|
||||
request.CreatedAt = *requestCreated
|
||||
}
|
||||
detail.RequestMessage = &request
|
||||
}
|
||||
if answer.ID != "" {
|
||||
if answerCreated != nil {
|
||||
answer.CreatedAt = *answerCreated
|
||||
}
|
||||
answer.DeliveredAt = deliveredAt
|
||||
detail.AnswerMessage = &answer
|
||||
detail.AnswerDeliveryStatus = answer.Status
|
||||
detail.AnswerDeliveryMethod = answer.DeliveryMethod
|
||||
}
|
||||
return detail, nil
|
||||
}
|
||||
|
||||
func safeAuditReferences(references []core.Reference) []audit.SafeReference {
|
||||
if len(references) == 0 {
|
||||
return nil
|
||||
}
|
||||
safe := make([]audit.SafeReference, 0, len(references))
|
||||
for _, reference := range references {
|
||||
safe = append(safe, audit.SafeReference{
|
||||
ID: reference.ID, KnowledgeID: reference.KnowledgeID,
|
||||
KnowledgeTitle: reference.KnowledgeTitle, Score: reference.Score,
|
||||
})
|
||||
}
|
||||
return safe
|
||||
}
|
||||
|
||||
func safeAuditMedia(media []core.MediaAudit) []audit.SafeMedia {
|
||||
if len(media) == 0 {
|
||||
return nil
|
||||
}
|
||||
safe := make([]audit.SafeMedia, 0, len(media))
|
||||
for _, item := range media {
|
||||
safe = append(safe, audit.SafeMedia{
|
||||
Type: item.Type, MIMEType: item.MIMEType, SizeBytes: item.SizeBytes,
|
||||
Width: item.Width, Height: item.Height,
|
||||
})
|
||||
}
|
||||
return safe
|
||||
}
|
||||
|
||||
var _ audit.TurnRepository = (*Store)(nil)
|
||||
@@ -0,0 +1,98 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/core"
|
||||
)
|
||||
|
||||
func TestAuditListSQLTenantSafeAndKeyset(t *testing.T) {
|
||||
sql := strings.ToUpper(listAuditTurnsSQL)
|
||||
for _, want := range []string{
|
||||
"FROM TURNS T", "T.TENANT_KEY=$1", "RM.TENANT_KEY=T.TENANT_KEY", "RM.CHANNEL=T.CHANNEL",
|
||||
"AM.TENANT_KEY=T.TENANT_KEY", "AM.CHANNEL=T.CHANNEL",
|
||||
"C.TENANT_KEY=T.TENANT_KEY", "C.CHANNEL=T.CHANNEL",
|
||||
"A.TENANT_KEY=T.TENANT_KEY", "A.CHANNEL=T.CHANNEL",
|
||||
"(T.CREATED_AT,T.ID)<($4,$5)", "ORDER BY T.CREATED_AT DESC,T.ID DESC", "LIMIT $6",
|
||||
} {
|
||||
if !strings.Contains(sql, want) {
|
||||
t.Errorf("list audit SQL missing %q", want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestAuditListSQLHandlesHistoricalJSONNullReferences(t *testing.T) {
|
||||
sql := strings.ToUpper(listAuditTurnsSQL)
|
||||
if !strings.Contains(sql, "CASE WHEN JSONB_TYPEOF(T.REFERENCES_JSON)='ARRAY'") ||
|
||||
!strings.Contains(sql, "THEN JSONB_ARRAY_LENGTH(T.REFERENCES_JSON) ELSE 0 END") {
|
||||
t.Fatal("list audit SQL must not call jsonb_array_length on historical JSON null/scalars")
|
||||
}
|
||||
}
|
||||
|
||||
func TestAuditDetailSQLTenantSafeAndExcludesForbiddenData(t *testing.T) {
|
||||
sql := strings.ToUpper(getAuditTurnSQL)
|
||||
for _, want := range []string{
|
||||
"T.TENANT_KEY=$1", "T.ID=$2", "RM.TENANT_KEY=T.TENANT_KEY", "AM.TENANT_KEY=T.TENANT_KEY",
|
||||
"C.TENANT_KEY=T.TENANT_KEY", "A.TENANT_KEY=T.TENANT_KEY",
|
||||
} {
|
||||
if !strings.Contains(sql, want) {
|
||||
t.Errorf("detail audit SQL missing %q", want)
|
||||
}
|
||||
}
|
||||
for _, forbidden := range []string{"ERROR_MESSAGE", "WEKNORA_SESSION_ID", "WEKNORA_REQUEST_ID", "PROVIDER_MESSAGE_ID", "SAFE_METADATA_JSON"} {
|
||||
if strings.Contains(sql, forbidden) {
|
||||
t.Errorf("detail audit SQL must not select forbidden %q", forbidden)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestAuditFeedbackProjectionIsTenantChannelScoped(t *testing.T) {
|
||||
listSQL := strings.ToUpper(listAuditTurnsSQL)
|
||||
for _, want := range []string{
|
||||
"FROM ANSWER_FEEDBACK AF", "AF.TENANT_KEY=T.TENANT_KEY", "AF.CHANNEL=T.CHANNEL", "AF.TURN_ID=T.ID",
|
||||
"COUNT(*) FILTER (WHERE AF.VALUE='UP')", "COUNT(*) FILTER (WHERE AF.VALUE='DOWN')",
|
||||
"AF.VALUE='DOWN' AND AF.REASON_CODE=''",
|
||||
} {
|
||||
if !strings.Contains(listSQL, want) {
|
||||
t.Errorf("list feedback projection missing %q", want)
|
||||
}
|
||||
}
|
||||
detailSQL := strings.ToUpper(getAuditTurnSQL)
|
||||
for _, want := range []string{
|
||||
"JSONB_AGG", "FROM ANSWER_FEEDBACK AF", "JOIN CHANNEL_ACTORS FA",
|
||||
"AF.TENANT_KEY=T.TENANT_KEY", "AF.CHANNEL=T.CHANNEL", "AF.TURN_ID=T.ID",
|
||||
"FA.TENANT_KEY=AF.TENANT_KEY", "FA.CHANNEL=AF.CHANNEL",
|
||||
} {
|
||||
if !strings.Contains(detailSQL, want) {
|
||||
t.Errorf("detail feedback projection missing %q", want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestSafeAuditReferencesDropContentAndRaw(t *testing.T) {
|
||||
references := []core.Reference{{
|
||||
ID: "ref-1", KnowledgeID: "kb-1", KnowledgeTitle: "Manual", Score: 0.9,
|
||||
Content: "private chunk body", Raw: json.RawMessage(`{"private":"payload"}`),
|
||||
}}
|
||||
safe := safeAuditReferences(references)
|
||||
if len(safe) != 1 || safe[0].ID != "ref-1" || safe[0].KnowledgeID != "kb-1" || safe[0].KnowledgeTitle != "Manual" || safe[0].Score != 0.9 {
|
||||
t.Fatalf("safe references=%+v", safe)
|
||||
}
|
||||
raw, err := json.Marshal(safe)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
body := string(raw)
|
||||
if strings.Contains(body, "private chunk") || strings.Contains(body, "payload") || strings.Contains(strings.ToLower(body), "content") || strings.Contains(strings.ToLower(body), "raw") {
|
||||
t.Fatalf("safe references leaked forbidden data: %s", body)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSafeAuditMediaCopiesMetadataOnly(t *testing.T) {
|
||||
media := safeAuditMedia([]core.MediaAudit{{Type: "image", MIMEType: "image/png", SizeBytes: 42, Width: 10, Height: 20}})
|
||||
if len(media) != 1 || media[0].Type != "image" || media[0].MIMEType != "image/png" || media[0].SizeBytes != 42 || media[0].Width != 10 || media[0].Height != 20 {
|
||||
t.Fatalf("safe media=%+v", media)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,143 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/feedback"
|
||||
)
|
||||
|
||||
const findFeedbackAnswerSQL = `SELECT m.id,t.id
|
||||
FROM messages m
|
||||
JOIN turns t ON t.id=m.correlation_id AND t.tenant_key=m.tenant_key AND t.channel=m.channel AND t.answer_message_id=m.id
|
||||
WHERE m.tenant_key=$1 AND m.channel=$2 AND m.provider_message_id=$3
|
||||
AND m.direction='outbound' AND m.role='assistant' AND m.status='sent' AND m.delivery_method='card'
|
||||
AND t.status='succeeded'`
|
||||
|
||||
const upsertFeedbackActorSQL = `INSERT INTO channel_actors
|
||||
(id,tenant_key,channel,external_actor_id,actor_type,display_name)
|
||||
VALUES($1,$2,$3,$4,'user','')
|
||||
ON CONFLICT (tenant_key,channel,external_actor_id) DO UPDATE SET
|
||||
updated_at=channel_actors.updated_at
|
||||
RETURNING id`
|
||||
|
||||
const findFeedbackStateSQL = `SELECT id,value,reason_code
|
||||
FROM answer_feedback
|
||||
WHERE tenant_key=$1 AND channel=$2 AND answer_message_id=$3 AND actor_id=$4
|
||||
FOR UPDATE`
|
||||
|
||||
const insertFeedbackStateSQL = `INSERT INTO answer_feedback
|
||||
(id,tenant_key,channel,answer_message_id,turn_id,actor_id,value,reason_code,source)
|
||||
VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9)`
|
||||
|
||||
const updateFeedbackStateSQL = `UPDATE answer_feedback
|
||||
SET value=$1,reason_code=$2,source=$3,updated_at=NOW()
|
||||
WHERE id=$4 AND tenant_key=$5 AND channel=$6 AND answer_message_id=$7 AND actor_id=$8`
|
||||
|
||||
func (s *Store) RecordFeedback(ctx context.Context, record feedback.Record) (feedback.Result, error) {
|
||||
if err := validateFeedbackRecord(record); err != nil {
|
||||
return feedback.Result{}, err
|
||||
}
|
||||
tx, err := s.pool.BeginTx(ctx, pgx.TxOptions{})
|
||||
if err != nil {
|
||||
return feedback.Result{}, fmt.Errorf("begin feedback transaction: %w", err)
|
||||
}
|
||||
defer func() { _ = tx.Rollback(ctx) }()
|
||||
|
||||
var answerMessageID, turnID string
|
||||
err = tx.QueryRow(ctx, findFeedbackAnswerSQL, record.TenantKey, record.Channel, record.OutTrackID).Scan(&answerMessageID, &turnID)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return feedback.Result{}, feedback.ErrTargetNotFound
|
||||
}
|
||||
if err != nil {
|
||||
return feedback.Result{}, fmt.Errorf("find feedback answer: %w", err)
|
||||
}
|
||||
|
||||
lockKey := feedbackLockKey(record.TenantKey, record.Channel, answerMessageID, record.ExternalActorID)
|
||||
if _, err = tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1,0))`, lockKey); err != nil {
|
||||
return feedback.Result{}, fmt.Errorf("lock feedback state: %w", err)
|
||||
}
|
||||
|
||||
if err = tx.QueryRow(ctx, upsertFeedbackActorSQL,
|
||||
record.ActorID, record.TenantKey, record.Channel, record.ExternalActorID,
|
||||
).Scan(&record.ActorID); err != nil {
|
||||
return feedback.Result{}, fmt.Errorf("resolve feedback actor: %w", err)
|
||||
}
|
||||
|
||||
var existingID, existingValue, existingReason string
|
||||
err = tx.QueryRow(ctx, findFeedbackStateSQL,
|
||||
record.TenantKey, record.Channel, answerMessageID, record.ActorID,
|
||||
).Scan(&existingID, &existingValue, &existingReason)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
command, insertErr := tx.Exec(ctx, insertFeedbackStateSQL,
|
||||
record.ID, record.TenantKey, record.Channel, answerMessageID, turnID,
|
||||
record.ActorID, record.Value, record.ReasonCode, record.Source,
|
||||
)
|
||||
if insertErr != nil {
|
||||
return feedback.Result{}, fmt.Errorf("insert feedback state: %w", insertErr)
|
||||
}
|
||||
if command.RowsAffected() != 1 {
|
||||
return feedback.Result{}, errors.New("insert feedback state affected unexpected rows")
|
||||
}
|
||||
if err = tx.Commit(ctx); err != nil {
|
||||
return feedback.Result{}, fmt.Errorf("commit feedback create: %w", err)
|
||||
}
|
||||
return feedback.Result{Created: true, Value: record.Value, ReasonCode: record.ReasonCode}, nil
|
||||
}
|
||||
if err != nil {
|
||||
return feedback.Result{}, fmt.Errorf("find feedback state: %w", err)
|
||||
}
|
||||
|
||||
if sameFeedbackState(existingValue, existingReason, record.Value, record.ReasonCode) {
|
||||
if err = tx.Commit(ctx); err != nil {
|
||||
return feedback.Result{}, fmt.Errorf("commit feedback replay: %w", err)
|
||||
}
|
||||
return feedback.Result{Replayed: true, Value: existingValue, ReasonCode: existingReason}, nil
|
||||
}
|
||||
|
||||
command, err := tx.Exec(ctx, updateFeedbackStateSQL,
|
||||
record.Value, record.ReasonCode, record.Source, existingID,
|
||||
record.TenantKey, record.Channel, answerMessageID, record.ActorID,
|
||||
)
|
||||
if err != nil {
|
||||
return feedback.Result{}, fmt.Errorf("update feedback state: %w", err)
|
||||
}
|
||||
if command.RowsAffected() != 1 {
|
||||
return feedback.Result{}, errors.New("update feedback state affected unexpected rows")
|
||||
}
|
||||
if err = tx.Commit(ctx); err != nil {
|
||||
return feedback.Result{}, fmt.Errorf("commit feedback change: %w", err)
|
||||
}
|
||||
return feedback.Result{Changed: true, Value: record.Value, ReasonCode: record.ReasonCode}, nil
|
||||
}
|
||||
|
||||
func validateFeedbackRecord(record feedback.Record) error {
|
||||
if record.ID == "" || record.ActorID == "" || record.TenantKey == "" || record.Channel == "" ||
|
||||
record.ExternalActorID == "" || record.OutTrackID == "" || record.Source != feedback.SourceDingTalkCard {
|
||||
return feedback.ErrInvalidCommand
|
||||
}
|
||||
if record.Value != feedback.ValueUp && record.Value != feedback.ValueDown {
|
||||
return feedback.ErrInvalidCommand
|
||||
}
|
||||
if record.Value == feedback.ValueUp && record.ReasonCode != "" {
|
||||
return feedback.ErrInvalidCommand
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func feedbackLockKey(parts ...string) string {
|
||||
key := ""
|
||||
for _, part := range parts {
|
||||
key += fmt.Sprintf("%d:%s|", len(part), part)
|
||||
}
|
||||
return key
|
||||
}
|
||||
|
||||
func sameFeedbackState(existingValue, existingReason, value, reason string) bool {
|
||||
return existingValue == value && existingReason == reason
|
||||
}
|
||||
|
||||
var _ feedback.Repository = (*Store)(nil)
|
||||
@@ -0,0 +1,77 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/feedback"
|
||||
)
|
||||
|
||||
func TestFeedbackAnswerLookupRequiresExactDeliveredCard(t *testing.T) {
|
||||
sql := strings.ToUpper(findFeedbackAnswerSQL)
|
||||
for _, fragment := range []string{
|
||||
"FROM MESSAGES M", "JOIN TURNS T", "T.TENANT_KEY=M.TENANT_KEY", "T.CHANNEL=M.CHANNEL",
|
||||
"T.ANSWER_MESSAGE_ID=M.ID", "M.TENANT_KEY=$1", "M.CHANNEL=$2", "M.PROVIDER_MESSAGE_ID=$3",
|
||||
"M.DIRECTION='OUTBOUND'", "M.ROLE='ASSISTANT'", "M.STATUS='SENT'", "M.DELIVERY_METHOD='CARD'",
|
||||
"T.STATUS='SUCCEEDED'",
|
||||
} {
|
||||
if !strings.Contains(sql, fragment) {
|
||||
t.Errorf("answer lookup missing %q", fragment)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestFeedbackActorAndStateSQLAreTenantChannelSafe(t *testing.T) {
|
||||
actor := strings.ToUpper(upsertFeedbackActorSQL)
|
||||
for _, fragment := range []string{
|
||||
"INSERT INTO CHANNEL_ACTORS", "TENANT_KEY", "CHANNEL", "EXTERNAL_ACTOR_ID",
|
||||
"ON CONFLICT (TENANT_KEY,CHANNEL,EXTERNAL_ACTOR_ID)", "RETURNING ID",
|
||||
} {
|
||||
if !strings.Contains(actor, fragment) {
|
||||
t.Errorf("actor upsert missing %q", fragment)
|
||||
}
|
||||
}
|
||||
lookup := strings.ToUpper(findFeedbackStateSQL)
|
||||
for _, fragment := range []string{"TENANT_KEY=$1", "CHANNEL=$2", "ANSWER_MESSAGE_ID=$3", "ACTOR_ID=$4", "FOR UPDATE"} {
|
||||
if !strings.Contains(lookup, fragment) {
|
||||
t.Errorf("feedback lookup missing %q", fragment)
|
||||
}
|
||||
}
|
||||
update := strings.ToUpper(updateFeedbackStateSQL)
|
||||
for _, fragment := range []string{"VALUE=$1", "REASON_CODE=$2", "SOURCE=$3", "UPDATED_AT=NOW()", "ID=$4", "TENANT_KEY=$5", "CHANNEL=$6", "ANSWER_MESSAGE_ID=$7", "ACTOR_ID=$8"} {
|
||||
if !strings.Contains(update, fragment) {
|
||||
t.Errorf("feedback update missing %q", fragment)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestFeedbackStoreUsesShortSerializedTransaction(t *testing.T) {
|
||||
data, err := os.ReadFile("feedback.go")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
source := strings.ToUpper(string(data))
|
||||
for _, fragment := range []string{"BEGINTX", "PG_ADVISORY_XACT_LOCK", "ROLLBACK", "COMMIT"} {
|
||||
if !strings.Contains(source, fragment) {
|
||||
t.Errorf("feedback store missing transaction fragment %q", fragment)
|
||||
}
|
||||
}
|
||||
for _, forbidden := range []string{"SESSIONWEBHOOK", "ACCESS_TOKEN", "RAW_CALLBACK", "RAW_PAYLOAD", "CORP_ID", "EXTENSION"} {
|
||||
if strings.Contains(source, forbidden) {
|
||||
t.Errorf("feedback store contains forbidden %q", forbidden)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestSameFeedbackState(t *testing.T) {
|
||||
if !sameFeedbackState(feedback.ValueDown, feedback.ReasonUnclear, feedback.ValueDown, feedback.ReasonUnclear) {
|
||||
t.Fatal("identical feedback must be replayed")
|
||||
}
|
||||
if sameFeedbackState(feedback.ValueDown, feedback.ReasonUnclear, feedback.ValueDown, feedback.ReasonOther) {
|
||||
t.Fatal("reason change must be a state change")
|
||||
}
|
||||
if sameFeedbackState(feedback.ValueDown, "", feedback.ValueUp, "") {
|
||||
t.Fatal("value change must be a state change")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user